52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.v1.endpoints import burst_detection as burst_detection_endpoint
|
|
|
|
|
|
def _build_client() -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(burst_detection_endpoint.router, prefix="/api/v1")
|
|
app.dependency_overrides[
|
|
burst_detection_endpoint.get_current_keycloak_username
|
|
] = lambda: "tester"
|
|
return TestClient(app)
|
|
|
|
|
|
def test_detect_burst_forwards_target_mode_fields(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_run_burst_detection(**kwargs):
|
|
captured.update(kwargs)
|
|
return {
|
|
"network": kwargs["network"],
|
|
"sensor_nodes": [],
|
|
"observed_source": "historical_monitoring",
|
|
"sample_count": 0,
|
|
"points_per_day": 96,
|
|
"day_count": 0,
|
|
"rows": [],
|
|
"summary": {"burst_detected": False},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
burst_detection_endpoint, "run_burst_detection", fake_run_burst_detection
|
|
)
|
|
|
|
response = _build_client().post(
|
|
"/api/v1/burst-detections",
|
|
json={
|
|
"network": "demo",
|
|
"scheme_name": "burst-history",
|
|
"target_time": "2026-06-20T13:30:00+08:00",
|
|
"sampling_interval_minutes": 15,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured["username"] == "tester"
|
|
assert captured["network"] == "demo"
|
|
assert captured["scheme_name"] == "burst-history"
|
|
assert captured["target_time"].isoformat() == "2026-06-20T13:30:00+08:00"
|
|
assert captured["sampling_interval_minutes"] == 15
|