103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
from datetime import date
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.v1.endpoints import schemes as schemes_endpoint
|
|
|
|
|
|
def _build_client() -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(schemes_endpoint.router, prefix="/api/v1")
|
|
return TestClient(app)
|
|
|
|
|
|
def test_get_schemes_forwards_optional_scheme_type(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_get_all_schemes(network, scheme_type=None, query_date=None):
|
|
captured["network"] = network
|
|
captured["scheme_type"] = scheme_type
|
|
captured["query_date"] = query_date
|
|
return [
|
|
{
|
|
"scheme_id": 1,
|
|
"scheme_name": "burst_case",
|
|
"scheme_type": scheme_type,
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(schemes_endpoint, "get_all_schemes", fake_get_all_schemes)
|
|
|
|
response = _build_client().get(
|
|
"/api/v1/schemes",
|
|
params={"network": "demo", "scheme_type": "burst_analysis"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured == {
|
|
"network": "demo",
|
|
"scheme_type": "burst_analysis",
|
|
"query_date": None,
|
|
}
|
|
assert response.json()[0]["scheme_type"] == "burst_analysis"
|
|
|
|
|
|
def test_get_schemes_forwards_query_date(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_get_all_schemes(network, scheme_type=None, query_date=None):
|
|
captured["network"] = network
|
|
captured["scheme_type"] = scheme_type
|
|
captured["query_date"] = query_date
|
|
return []
|
|
|
|
monkeypatch.setattr(schemes_endpoint, "get_all_schemes", fake_get_all_schemes)
|
|
|
|
response = _build_client().get(
|
|
"/api/v1/schemes",
|
|
params={
|
|
"network": "demo",
|
|
"scheme_type": "dma_leak_identification",
|
|
"query_date": "2026-01-02T00:00:00+08:00",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured == {
|
|
"network": "demo",
|
|
"scheme_type": "dma_leak_identification",
|
|
"query_date": date(2026, 1, 2),
|
|
}
|
|
|
|
|
|
def test_get_scheme_detail_forwards_scheme_type(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_query_scheme_detail(name, scheme_name, scheme_type=None):
|
|
captured["name"] = name
|
|
captured["scheme_name"] = scheme_name
|
|
captured["scheme_type"] = scheme_type
|
|
return {
|
|
"scheme_name": scheme_name,
|
|
"scheme_type": scheme_type,
|
|
"rows": [{"Area": "1", "LeakageFlow_m3_per_s": 0.1}],
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
schemes_endpoint, "query_scheme_detail", fake_query_scheme_detail
|
|
)
|
|
|
|
response = _build_client().get(
|
|
"/api/v1/schemes/dma_001",
|
|
params={"network": "demo", "scheme_type": "dma_leak_identification"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured == {
|
|
"name": "demo",
|
|
"scheme_name": "dma_001",
|
|
"scheme_type": "dma_leak_identification",
|
|
}
|
|
assert response.json()["scheme_name"] == "dma_001"
|