60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""REST API 压测脚本（Issue #94）- Locust
|
||
|
||
覆盖完整用户路径：登录 -> 设备查询 -> 数据上报 -> 报表查询
|
||
并发梯度：locust -u 10/50/100/500/1000 --spawn-rate 10
|
||
|
||
运行：
|
||
pip install locust
|
||
locust -f tests/performance/locustfile.py --host http://127.0.0.1:8000 \
|
||
-u 100 --spawn-rate 10 --run-time 5m --headless \
|
||
--html reports/locust_$(date +%F_%H%M).html
|
||
"""
|
||
|
||
from locust import HttpUser, between, task
|
||
|
||
|
||
class WaterManagementUser(HttpUser):
|
||
wait_time = between(0.5, 2.0)
|
||
token = None
|
||
|
||
def on_start(self):
|
||
"""登录获取 token（若接口不存在则以匿名继续，保证压测可运行）"""
|
||
resp = self.client.post("/api/auth/login",
|
||
json={"username": "perf", "password": "perf123"},
|
||
catch_response=True)
|
||
if resp.status_code == 200:
|
||
try:
|
||
self.token = resp.json().get("data", {}).get("access_token")
|
||
except Exception:
|
||
self.token = None
|
||
else:
|
||
resp.success() # 登录接口未实现时不计入失败
|
||
|
||
@property
|
||
def headers(self):
|
||
return {"Authorization": f"Bearer {self.token}"} if self.token else {}
|
||
|
||
@task(5)
|
||
def list_devices(self):
|
||
self.client.get("/api/devices?page=1&page_size=20", headers=self.headers,
|
||
name="/api/devices")
|
||
|
||
@task(3)
|
||
def device_detail(self):
|
||
self.client.get("/api/devices/1", headers=self.headers, name="/api/devices/{id}")
|
||
|
||
@task(2)
|
||
def report_data(self):
|
||
self.client.post("/api/data/report",
|
||
json={"device_id": 1, "metric": "flow", "value": 12.5},
|
||
headers=self.headers, name="/api/data/report")
|
||
|
||
@task(2)
|
||
def billing_report(self):
|
||
self.client.get("/api/billing/report?month=2026-07", headers=self.headers,
|
||
name="/api/billing/report")
|
||
|
||
@task(1)
|
||
def health(self):
|
||
self.client.get("/health", name="/health")
|