Implement all endpoints stress tests

.github/workflows/api.yml

@@ -85,6 +85,12 @@ jobs:
       QUALICHARGE_OIDC_IS_ENABLED: False
       QUALICHARGE_ALLOWED_HOSTS: '["http://localhost:8000"]'
       QUALICHARGE_API_STATIQUE_BULK_CREATE_MAX_SIZE: 1000
+      QUALICHARGE_DEBUG: 0
+      QUALICHARGE_PROFILING: 0
+      QUALICHARGE_UVICORN_WORKERS: 1
+      QUALICHARGE_DB_CONNECTION_MAX_OVERFLOW: 200
+      QUALICHARGE_DB_CONNECTION_POOL_SIZE: 50
+      QUALICHARGE_STATIQUE_DATA_PATH: /home/runner/work/qualicharge/qualicharge/data/irve-statique.json.gz
       # This is a fake setting required to run the app
       QUALICHARGE_OIDC_PROVIDER_BASE_URL: http://localhost:8000/fake
       QUALICHARGE_OAUTH2_TOKEN_ENCODING_KEY: thisissupersecret

.gitignore

@@ -49,6 +49,7 @@ scripts/*.csv
 
 # -- Tools
 .coverage
+src/api/bench_*.csv
 
 # Prefect
 src/prefect/.htpasswd

Makefile

@@ -31,7 +31,16 @@ data/afirev-charging.csv: data
 
 # -- Docker/compose
 bench: ## run API benchmark
-bench: run
+	# FIXME
+	#
+	# $(COMPOSE) stop
+	# $(COMPOSE) up -d --wait --force-recreate postgresql
+	# $(MAKE) migrate-api
+	# $(MAKE) create-api-superuser
+	# $(COMPOSE) up -d --wait api
+	# zcat data/irve-statique.json.gz | head -n 500 | \
+	# 	bin/qcc static bulk --chunk-size 1000
+
 	$(COMPOSE_RUN_API_PIPENV) \
 		locust \
 		  -f /mnt/bench/locustfile.py \
@@ -299,7 +308,6 @@ jupytext--to-ipynb: ## convert remote md files into ipynb
 
 reset-db: ## Reset the PostgreSQL database
 	$(COMPOSE) stop
-	# $(COMPOSE) down postgresql metabase
 	$(MAKE) migrate-api
 	$(MAKE) create-api-superuser
 	$(MAKE) create-api-test-db

env.d/api

@@ -9,10 +9,13 @@ QUALICHARGE_DB_PORT=5432
 QUALICHARGE_DB_USER=qualicharge
 QUALICHARGE_DEBUG=0
 QUALICHARGE_PROFILING=0
-QUALICHARGE_UVICORN_WORKERS=4
+QUALICHARGE_UVICORN_WORKERS=1
 QUALICHARGE_EXECUTION_ENVIRONMENT=development
 QUALICHARGE_OIDC_PROVIDER_BASE_URL=http://keycloak:8080/realms/qualicharge
 QUALICHARGE_OIDC_IS_ENABLED=False
 QUALICHARGE_OAUTH2_TOKEN_ENCODING_KEY=thisissupersecret
 QUALICHARGE_OAUTH2_TOKEN_ISSUER=http://localhost:8010
 QUALICHARGE_TEST_DB_NAME=test-qualicharge-api
+QUALICHARGE_STATIQUE_DATA_PATH=/data/irve-statique.json.gz
+QUALICHARGE_DB_CONNECTION_MAX_OVERFLOW=200
+QUALICHARGE_DB_CONNECTION_POOL_SIZE=50

src/bench/locustfile.py

@@ -1,10 +1,24 @@
 """QualiCharge API load tests."""
 
+import os
+from datetime import datetime, timedelta, timezone
+from functools import cache
+from pathlib import Path
+
+import pandas as pd
 import requests
 from locust import FastHttpUser, task
 
 API_ADMIN_USER = "admin"
 API_ADMIN_PASSWORD = "admin"
+STATIC_DB_OFFSET = 500
+
+
+@cache
+def load_statique_db(db_path: Path):
+    """Load statique database."""
+    db = pd.read_json(db_path, dtype_backend="pyarrow", lines=True)
+    return db.to_dict(orient="records")
 
 
 class BaseAPIUser(FastHttpUser):
@@ -13,6 +27,24 @@ class BaseAPIUser(FastHttpUser):
     abstract: bool = True
     username: str
     password: str
+    counter: dict = {
+        "static": {"create": STATIC_DB_OFFSET, "bulk": 10000},
+        "status": 0,
+        "session": 0,
+    }
+
+    def on_start(self):
+        """Log user in on start."""
+        credentials = {
+            "username": self.username,
+            "password": self.password,
+        }
+        response = requests.post(f"{self.host}/auth/token", data=credentials)
+        token = response.json()["access_token"]
+        self.client.auth_header = f"Bearer {token}"
+        self.statique_db = load_statique_db(
+            Path(os.environ["QUALICHARGE_STATIQUE_DATA_PATH"])
+        )
 
     @task
     def whoami(self):
@@ -29,15 +61,155 @@ def statique_list(self):
             if response.js["size"] == 0:
                 response.failure("Database contains no statique entries")
 
-    def on_start(self):
-        """Log user in on start."""
-        credentials = {
-            "username": self.username,
-            "password": self.password,
+    @task
+    def statique_get(self):
+        """Assess the /statique/{id_pdc_itinerance} GET endpoint."""
+        id_pdc_itinerance = "FRALLEGO002006P3"
+        with self.rest("GET", f"/statique/{id_pdc_itinerance}") as response:
+            if "id_pdc_itinerance" not in response.js:
+                response.failure(
+                    f"Database does not contain target statique entry {id_pdc_itinerance}"
+                )
+
+    @task
+    def statique_update(self):
+        """Assess the /statique/{id_pdc_itinerance} PUT endpoint."""
+        data = self.statique_db[22]
+        data["date_maj"] = "2024-11-21"
+        id_pdc_itinerance = data["id_pdc_itinerance"]
+
+        with self.rest("PUT", f"/statique/{id_pdc_itinerance}", json=data) as response:
+            if "id_pdc_itinerance" not in response.js:
+                response.failure(
+                    f"Database does not contain target statique entry {id_pdc_itinerance}"
+                )
+
+    @task
+    def statique_create(self):
+        """Assess the /statique/ POST endpoint."""
+        index = self.counter["static"]["create"]
+        data = self.statique_db[index]
+
+        with self.rest("POST", "/statique/", json=data) as response:
+            if response.status_code == 200 and response.js["size"] == 0:
+                response.failure("No Statique entry was created")
+            self.counter["static"]["create"] += 1
+
+    @task
+    def statique_bulk(self):
+        """Assess the /statique/bulk POST endpoint."""
+        start = self.counter["static"]["bulk"]
+        limit = 50
+        end = start + limit
+        data = self.statique_db[start:end]
+
+        with self.rest("POST", "/statique/bulk", json=data) as response:
+            if response.js["size"] == 0:
+                response.failure("No Statique entry was created")
+            self.counter["static"]["bulk"] = end
+
+    @task
+    def status_list(self):
+        """Assess the /dynamique/status/ GET endpoint."""
+        with self.rest("GET", "/dynamique/status/") as response:
+            if response.status_code == 404:
+                response.success()
+
+    @task
+    def status_get(self):
+        """Assess the /dynamique/status/{id_pdc_itinerance} GET endpoint."""
+        id_pdc_itinerance = "FRALLEGO002006P3"
+        with self.rest("GET", f"/dynamique/status/{id_pdc_itinerance}") as response:
+            if response.status_code == 404:
+                response.success()
+
+    @task
+    def status_history(self):
+        """Assess the /dynamique/status/{id_pdc_itinerance}/history GET endpoint."""
+        id_pdc_itinerance = "FRALLEGO002006P3"
+        with self.rest(
+            "GET", f"/dynamique/status/{id_pdc_itinerance}/history"
+        ) as response:
+            if response.status_code == 404:
+                response.success()
+
+    @task
+    def status_create(self):
+        """Assess the /dynamique/status/ POST endpoint."""
+        now = datetime.now(timezone.utc)
+        data = {
+            "etat_pdc": "en_service",
+            "occupation_pdc": "libre",
+            "horodatage": now.isoformat(),
+            "etat_prise_type_2": "fonctionnel",
+            "etat_prise_type_combo_ccs": "fonctionnel",
+            "etat_prise_type_chademo": "fonctionnel",
+            "etat_prise_type_ef": "fonctionnel",
+            "id_pdc_itinerance": "FRALLEGO002006P3",
         }
-        response = requests.post(f"{self.host}/auth/token", data=credentials)
-        token = response.json()["access_token"]
-        self.client.auth_header = f"Bearer {token}"
+        with self.rest("POST", "/dynamique/status/", json=data) as response:
+            pass
+
+    @task
+    def status_bulk(self):
+        """Assess the /dynamique/status/bulk POST endpoint."""
+        now = datetime.now(timezone.utc)
+        base = {
+            "etat_pdc": "en_service",
+            "occupation_pdc": "libre",
+            "horodatage": now.isoformat(),
+            "etat_prise_type_2": "fonctionnel",
+            "etat_prise_type_combo_ccs": "fonctionnel",
+            "etat_prise_type_chademo": "fonctionnel",
+            "etat_prise_type_ef": "fonctionnel",
+            "id_pdc_itinerance": "FRALLEGO002006P3",
+        }
+        delta = timedelta(seconds=3)
+        size = 10
+        data = [
+            dict(base, horodatage=(now - delta * n).isoformat()) for n in range(size)
+        ]
+
+        with self.rest("POST", "/dynamique/status/bulk", json=data) as response:
+            pass
+
+    @task
+    def session_create(self):
+        """Assess the /dynamique/session/ POST endpoint."""
+        now = datetime.now(timezone.utc)
+        data = {
+            "start": (now - timedelta(hours=1)).isoformat(),
+            "end": (now - timedelta(minutes=1)).isoformat(),
+            "energy": 666.66,
+            "id_pdc_itinerance": "FRALLEGO002006P3",
+        }
+        with self.rest("POST", "/dynamique/session/", json=data) as response:
+            pass
+
+    @task
+    def session_bulk(self):
+        """Assess the /dynamique/session/bulk POST endpoint."""
+        now = datetime.now(timezone.utc)
+        base = {
+            "start": (now - timedelta(hours=1)).isoformat(),
+            "end": (now - timedelta(minutes=1)).isoformat(),
+            "energy": 666.66,
+            "id_pdc_itinerance": "FRALLEGO002006P3",
+        }
+        delta = timedelta(seconds=3)
+        size = 10
+        data = [
+            dict(
+                base,
+                start=(now - timedelta(hours=n, minutes=30)).isoformat(),
+                end=(now - timedelta(hours=n)).isoformat(),
+                energy=666.66 * (n + 1),
+            )
+            for n in range(size)
+        ]
+
+        with self.rest("POST", "/dynamique/session/bulk", json=data) as response:
+            pass
 
 
 class APIAdminUser(BaseAPIUser):