Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Adobe HDS Load Testing Tool using Locust Framework. See inline for details
# HDS Load Testing Tool using Locust Load Testing Framework and HDS_Fragmenter library
# Simulates user access by fetching the bootstrap file to parse out fragment names.
# Bootstrap files are fetched 1/10 times.
#
# Usage:
# 1. Fetch dependencies given below
# 2. Edit "url_f4m_path" below to the stream-level manifest path/URI
# 3. Run: `locust -f hdslt.py -H http://my_origin_server`
# pip install locustio
from locust import HttpLocust, TaskSet, task
# wget https://raw.githubusercontent.com/use-sparingly/HDS_Fragmenter/master/hds_server_reader.py
# wget https://raw.githubusercontent.com/use-sparingly/HDS_Fragmenter/master/f4v.py
from hds_server_reader import HdsServerReader
from collections import deque
import random
class HDSClient(TaskSet):
def on_start(self):
self.frags_names = deque(maxlen=10)
@task(1)
def bootstrap(self):
""" Get bootstrap and update playlist window """
# CHANGE ME!
# Set path to stream-level f4m file. Do not use master set-level f4m.
url_f4m_path = "/hds/38_3.f4m"
url_path = url_f4m_path.rstrip(".f4m")
url_bootstrap = url_path + ".bootstrap"
resp = self.client.get(url_bootstrap)
try:
resp.raise_for_status()
hsr = HdsServerReader(resp.content)
frag_name_suffix = hsr.get_latest_frag_suffix()
frag_tuple = (url_path, frag_name_suffix)
if frag_tuple not in self.frags_names:
self.frags_names.append(frag_tuple)
except:
pass
@task(10)
def fragment(self):
if len(self.frags_names) <= 0:
return True
frag_tuple = random.choice(self.frags_names)
frag_url_base_path = frag_tuple[0]
frag_url = frag_url_base_path + frag_tuple[1]
self.client.get(frag_url, name=frag_url_base_path)
class WebsiteUser(HttpLocust):
task_set = HDSClient
min_wait=50
max_wait=1000
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.