subclassing

Author: sanderegg

tests/performance/common/base_user.py

@@ -0,0 +1,75 @@
+import logging
+
+from locust import FastHttpUser
+
+from .auth_settings import DeploymentAuth, OsparcAuth
+
+_logger = logging.getLogger(__name__)
+
+
+class OsparcUserBase(FastHttpUser):
+    """
+    Base class for Locust users that provides common functionality.
+    This class can be extended by specific user classes to implement
+    different behaviors or tasks.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.deploy_auth = DeploymentAuth()
+        _logger.debug("Using deployment auth: %s", self.deploy_auth)
+
+
+class OsparcWebUserBase(OsparcUserBase):
+    """
+    Base class for web users in Locust that provides common functionality.
+    This class can be extended by specific web user classes to implement
+    different behaviors or tasks.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if self.environment.parsed_options.requires_login:
+            self.osparc_auth = OsparcAuth()
+            _logger.debug("Using OsparcAuth for login: %s", self.osparc_auth)
+
+    def on_start(self) -> None:
+        """
+        Called when a web user starts. Can be overridden by subclasses
+        to implement custom startup behavior, such as logging in.
+        """
+        if self.environment.parsed_options.requires_login:
+            self._login()
+
+    def on_stop(self) -> None:
+        """
+        Called when a web user stops. Can be overridden by subclasses
+        to implement custom shutdown behavior, such as logging out.
+        """
+        if self.environment.parsed_options.requires_login:
+            self._logout()
+
+    def _login(self) -> None:
+        # Implement login logic here
+        logging.info(
+            "Loggin in user with email: %s",
+            {
+                "email": self.osparc_auth.OSPARC_USER_NAME,
+                "password": self.osparc_auth.OSPARC_PASSWORD.get_secret_value(),
+            },
+        )
+        response = self.client.post(
+            "/v0/auth/login",
+            json={
+                "email": self.osparc_auth.OSPARC_USER_NAME,
+                "password": self.osparc_auth.OSPARC_PASSWORD.get_secret_value(),
+            },
+            auth=self.deploy_auth.to_auth(),
+        )
+        response.raise_for_status()
+        logging.info("Logged in user with email: %s", self.osparc_auth)
+
+    def _logout(self) -> None:
+        # Implement logout logic here
+        self.client.post("/v0/auth/logout", auth=self.deploy_auth.to_auth())
+        logging.info("Logged out user with email: %s", self.osparc_auth)

tests/performance/locustfiles/deployment_max_rps_single_endpoint.py

@@ -11,8 +11,8 @@
 import logging
 
 import locust_plugins
-from common.auth_settings import DeploymentAuth, OsparcAuth
-from locust import FastHttpUser, events, task
+from common.base_user import OsparcWebUserBase
+from locust import events, task
 
 logging.basicConfig(level=logging.INFO)
 
@@ -45,51 +45,12 @@ def _(environment, **_kwargs) -> None:
     logging.info("Requires login: %s", environment.parsed_options.requires_login)
 
 
-class WebApiUser(FastHttpUser):
+class WebApiUser(OsparcWebUserBase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.deploy_auth = DeploymentAuth()
-        logging.debug("Using deployment auth: %s", self.deploy_auth)
-
-        if self.environment.parsed_options.requires_login:
-            self.osparc_auth = OsparcAuth()
-            logging.debug("Using OsparcAuth for login: %s", self.osparc_auth)
 
     @task
     def get_endpoint(self) -> None:
         self.client.get(
             self.environment.parsed_options.endpoint, auth=self.deploy_auth.to_auth()
         )
-
-    def _login(self) -> None:
-        # Implement login logic here
-        logging.info(
-            "Loggin in user with email: %s",
-            {
-                "email": self.osparc_auth.OSPARC_USER_NAME,
-                "password": self.osparc_auth.OSPARC_PASSWORD.get_secret_value(),
-            },
-        )
-        response = self.client.post(
-            "/v0/auth/login",
-            json={
-                "email": self.osparc_auth.OSPARC_USER_NAME,
-                "password": self.osparc_auth.OSPARC_PASSWORD.get_secret_value(),
-            },
-            auth=self.deploy_auth.to_auth(),
-        )
-        response.raise_for_status()
-        logging.info("Logged in user with email: %s", self.osparc_auth)
-
-    def _logout(self) -> None:
-        # Implement logout logic here
-        self.client.post("/v0/auth/logout", auth=self.deploy_auth.to_auth())
-        logging.info("Logged out user with email: %s", self.osparc_auth)
-
-    def on_start(self) -> None:
-        if self.environment.parsed_options.requires_login:
-            self._login()
-
-    def on_stop(self) -> None:
-        if self.environment.parsed_options.requires_login:
-            self._logout()