feat: add locust support

README.md

@@ -402,6 +402,12 @@ async def fetch_with_manual_retry():
                 return response.json()
 ```
 
+### Testing
+
+`pytest` is used as the testing framework.
+
+We use `locust` for integrated load testing. User behaviors are defined in `tests/locustfile.py`, you can use it to simulate concurrent users, and get QPS/response metrics for analysis.
+
 ## Development Setup
 
 - Python >= 3.12

justfile

@@ -27,6 +27,10 @@ run port=PORT:
 @test:
     uv run --frozen pytest -xvs tests
 
+# Run locust
+@benchmark:
+    uv run locust -f tests/locustfile.py
+
 # Generate database migration
 @db-migrate message:
     uv run alembic revision --autogenerate -m "{{message}}"

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
 [dependency-groups]
 dev = [
     "fakeredis>=2.32.1",
+    "locust>=2.32.5",
     "pytest>=8.3.3",
     "pytest-asyncio>=1.2.0",
     "pytest-cov>=6.1.1",
@@ -48,9 +49,7 @@ testpaths = ["tests"]
 python_files = ["test_*.py", "*_test.py"]
 python_classes = ["Test*"]
 python_functions = ["test_*"]
-markers = [
-    "integration: marks tests as integration tests",
-]
+markers = ["integration: marks tests as integration tests"]
 
 [[tool.uv.index]]
 name = "tsinghua"

tests/locustfile.py

@@ -0,0 +1,40 @@
+from locust import HttpUser, TaskSet, task, between
+import time
+
+
+class AuthAPITasks(TaskSet):
+    def on_start(self):
+        """Setup: register a user for this taskset"""
+        unique_id = f"{self.user.user_id}_{int(time.time())}"
+        self.email = f"loadtest_{unique_id}@example.com"
+        self.username = f"user_{unique_id}"
+        self.password = "TestPass123!"
+
+        self.client.post(
+            "/auth/register",
+            json={
+                "email": self.email,
+                "username": self.username,
+                "password": self.password,
+            },
+        )
+
+    @task(3)
+    def login(self):
+        self.client.post(
+            "/auth/jwt/login",
+            data={
+                "username": self.email,
+                "password": self.password,
+            },
+        )
+
+
+class WebsiteUser(HttpUser):
+    tasks = [AuthAPITasks]
+    wait_time = between(1, 3)
+    user_id = 0
+
+    def on_start(self):
+        WebsiteUser.user_id += 1
+        self.user_id = WebsiteUser.user_id