Adobe HDS Load Testing Tool using Locust Framework. See inline for details

hds_loadtest.py

# 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