dash_test_locustfile.py

from locust import HttpUser, task, between, SequentialTaskSet, events
from locust.exception import StopUser
import time
import logging
from bs4 import BeautifulSoup
import csv
import itertools
import json
import random

# https://www.blazemeter.com/blog/how-to-run-locust-with-different-users
global USER_CREDENTIALS
USER_CREDENTIALS = None

with open("random_users.csv", "r") as f:
    reader = csv.reader(f)
    next(reader)
    USER_CREDENTIALS = list(reader)
    random.shuffle(USER_CREDENTIALS)


counter = itertools.count()
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")


class DashboardTasks(SequentialTaskSet):
    lti_request_data = {}
    authenticity_token = None

    def on_start(self):
        self.logger = logging.getLogger(f"locust-{next(counter)}")
        self.logger.debug("Hatching locust")
        self.start_time = time.time()

        # NOTE here you set user specific values that will be used during authentication
        # NOTE this step will be specific to your authentication system
        if len(USER_CREDENTIALS) > 0:
            (
                self.user_number,
                self.gid,
                self.gidNumber,
                self.firstname,
                self.lastname,
            ) = USER_CREDENTIALS.pop(0)

            self.email = f"{self.firstname}{self.lastname}@gmail.com"

            self.lti_request_data = {
                "lti_message_type": "basic-lti-launch-request",
                "lti_version": "LTI-1p0",
                "oauth_consumer_key": "key",
                "resource_link_id": "resource",
                "lis_person_contact_email_primary": self.email,
                "lis_person_name_given": self.firstname,
                "lis_person_name_family": self.lastname,
                "custom_canvas_course_id": self.gid[2:],
                "custom_canvas_user_id": self.user_number,
            }

    # NOTE this is where you make the necessary requests to authenticate your user
    # NOTE this step will be specific to your authentication system
    @task
    def authenticate_user(self):

        # lti verification
        self.logger.debug("Posting to lti parse")
        response = self.client.post("/lti/launch", data=self.lti_request_data)

        # waiting page
        self.logger.debug("Visiting the waiting page")
        self.client.get(f"/verify_account_request/verify_account.html")

        # poll for success status of account before continuing, performing the same task that javascript would have on the waiting page
        while True:
            status_response = self.client.get(f"/verify_account_request/status")
            self.logger.info(status_response.json())

            if status_response.json()["status"] == "true":
                break

            time.sleep(10)

    @task
    def dashboard_page(self):
        self.logger.debug("Visiting the dashboard page")
        response = self.client.get(f"/pun/sys/dashboard")

        self.end_time = time.time()
        self.logger.info(
            f"Dashboard_Tasks_Elapsed_Time: {self.end_time - self.start_time}"
        )

        while True:
            # simulate the user by refreshing the dashboard every 50 to 200 seconds
            time.sleep(random.randint(50, 200))
            self.logger.debug("Visiting the dashboard page")
            response = self.client.get(f"/pun/sys/dashboard")

    @task
    def on_finish(self):
        raise StopUser()


class AcademicClusterUser(HttpUser):
    tasks = [DashboardTasks]
    wait_time = between(0, 0)
    host = "YOUR HOSTNAME"