jiang
5966d039de
Reorganize algorithm packages by business responsibility, move orchestration into services, and keep database access behind pooled repositories. Harden analysis API validation, remove unsafe legacy simulation endpoints, and add regression and architecture boundary coverage. BREAKING CHANGE: legacy algorithm module paths and obsolete simulation endpoints are removed.
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.v1.endpoints import burst_detection, burst_location
|
|
|
|
|
|
def _client(module) -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(module.router, prefix="/api/v1")
|
|
app.dependency_overrides[module.get_current_keycloak_username] = lambda: "tester"
|
|
return TestClient(app)
|
|
|
|
|
|
def test_burst_detection_runs_service_outside_event_loop(monkeypatch):
|
|
captured = {}
|
|
|
|
async def fake_threadpool(func, **kwargs):
|
|
captured["func"] = func
|
|
captured["kwargs"] = kwargs
|
|
return {"status": "completed"}
|
|
|
|
monkeypatch.setattr(burst_detection, "run_in_threadpool", fake_threadpool)
|
|
response = _client(burst_detection).post(
|
|
"/api/v1/burst-detections",
|
|
json={"network": "demo", "observed_pressure_data": {"S1": [1.0]}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured["func"] is burst_detection.run_burst_detection
|
|
assert captured["kwargs"]["username"] == "tester"
|
|
|
|
|
|
def test_burst_location_runs_service_outside_event_loop(monkeypatch):
|
|
captured = {}
|
|
|
|
async def fake_threadpool(func, **kwargs):
|
|
captured["func"] = func
|
|
captured["kwargs"] = kwargs
|
|
return {"status": "completed"}
|
|
|
|
monkeypatch.setattr(burst_location, "run_in_threadpool", fake_threadpool)
|
|
response = _client(burst_location).post(
|
|
"/api/v1/burst-locations",
|
|
json={
|
|
"network": "demo",
|
|
"burst_leakage": 0.1,
|
|
"burst_pressure": {"S1": 1.0},
|
|
"normal_pressure": {"S1": 1.1},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured["func"] is burst_location.run_burst_location_by_network
|
|
assert captured["kwargs"]["username"] == "tester"
|