login

Author: sanderegg

tests/performance2/common/deploy_auth.py

@@ -9,3 +9,12 @@ class MonitoringBasicAuth(BaseSettings):
 
     def to_auth(self) -> tuple[str, str]:
         return (self.SC_USER_NAME, self.SC_PASSWORD)
+
+
+class OsparcAuth(BaseSettings):
+    model_config = SettingsConfigDict(extra="ignore")
+    OSPARC_USER_NAME: str = Field(default=..., examples=["<your username>"])
+    OSPARC_PASSWORD: str = Field(default=..., examples=["<your password>"])
+
+    def to_auth(self) -> tuple[str, str]:
+        return (self.OS_USER_NAME, self.OS_PASSWORD)

tests/performance2/locustfiles/deployment_max_rps_single_endpoint.py

@@ -11,7 +11,7 @@
 import logging
 
 import locust_plugins
-from common.deploy_auth import MonitoringBasicAuth
+from common.deploy_auth import MonitoringBasicAuth, OsparcAuth
 from locust import events, task
 from locust.contrib.fasthttp import FastHttpUser
 
@@ -26,7 +26,7 @@
 _endpoint_holder = {"endpoint": "/"}
 
 
-def add_endpoint_argument(parser):
+def add_endpoint_argument(parser) -> None:
     parser.add_argument(
         "--endpoint",
         type=str,
@@ -42,7 +42,7 @@ def _(parser):
 
 
 @events.init.add_listener
-def _(environment, **_kwargs):
+def _(environment, **_kwargs) -> None:
     _endpoint_holder["endpoint"] = environment.parsed_options.endpoint
     logging.info("Testing endpoint: %s", _endpoint_holder["endpoint"])
 
@@ -52,7 +52,25 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.auth = MonitoringBasicAuth().to_auth()
         self.endpoint = _endpoint_holder["endpoint"]
+        self.osparc_auth = OsparcAuth()
+        self.requires_login = False
 
     @task
-    def get_endpoint(self):
+    def get_endpoint(self) -> None:
         self.client.get(self.endpoint, auth=self.auth)
+
+    def _login(self) -> None:
+        # Implement login logic here
+        logging.info("Logging in user with email: %s", self.osparc_auth)
+
+    def _logout(self) -> None:
+        # Implement logout logic here
+        logging.info("Logging out user with email: %s", self.osparc_auth)
+
+    def on_start(self) -> None:
+        if self.requires_login:
+            self._login()
+
+    def on_stop(self) -> None:
+        if self.requires_login:
+            self._logout()