Coverage for tests/test_refactoring.py: 100%
204 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Regression coverage for contract-preserving internal refactors."""
3import asyncio
4import json
5from datetime import datetime
6from types import SimpleNamespace
7from unittest.mock import Mock
9import httpx
10import pytest
11from conftest import course_payload, make_course
13from data_api.api import api
14from data_api.core.config import PREFETCH_ENDPOINTS
15from data_api.core.exceptions import UpstreamException
16from data_api.data.nthudata import NTHUDataManager
17from data_api.domain.buses.query import resolve_bus_time
18from data_api.domain.buses.services import BusesService, PreparedBusData
19from data_api.domain.courses.models import Conditions
20from data_api.domain.energy.services import EnergyService
21from data_api.domain.libraries.services import LibrariesService
22from data_api.runtime import Runtime, default_runtime, get_runtime
25async def test_request_duration_uses_monotonic_clock(monkeypatch):
26 clock = SimpleNamespace(
27 time=Mock(side_effect=[100.0, 99.0]),
28 perf_counter=Mock(side_effect=[10.0, 10.25]),
29 )
30 monkeypatch.setattr(api, "time", clock)
31 async with httpx.AsyncClient(
32 transport=httpx.ASGITransport(app=api.app), base_url="http://test"
33 ) as client:
34 response = await client.get("/")
35 assert response.status_code == 200
36 assert response.headers["X-Process-Time"] == "0.25"
37 clock.time.assert_not_called()
40@pytest.mark.parametrize(
41 "day,time,instant,expected",
42 [
43 ("current", "23:59", "2026-10-16T23:59:00", ("weekday", "23:59")),
44 ("current", "23:59", "2026-10-17T00:00:00", ("weekend", "00:00")),
45 ("weekday", "09:03", "2026-10-17T00:00:00", ("weekday", "09:03")),
46 ("weekend", None, "2026-10-19T00:00:00", ("weekend", "")),
47 ],
48)
49def test_bus_time_resolution(day, time, instant, expected):
50 now = datetime.fromisoformat(instant)
51 state = resolve_bus_time(day, time, now)
52 expected_time = now.strftime("%H:%M")
53 assert (state.day, state.after_time) == expected
54 assert state.current_time == expected_time
57def test_course_conditions_prepare_once_and_do_not_retain_course():
58 conditions = Conditions(
59 list_build_target=[{"row_field": "id", "matcher": "^11510A", "regex_match": True}]
60 )
61 first = make_course(id="11510A")
62 second = make_course(id="11510B")
63 assert conditions.accept(first)
64 predicate = conditions._predicate
65 assert not conditions.accept(second)
66 assert conditions.accept(first)
67 assert conditions._predicate is predicate
68 assert not hasattr(conditions, "course")
71async def _mcp_request(client, method, params):
72 response = await client.post(
73 "/mcp",
74 headers={"Accept": "application/json, text/event-stream"},
75 json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
76 )
77 assert response.status_code == 200
78 messages = [
79 json.loads(line.removeprefix("data: "))
80 for line in response.text.splitlines()
81 if line.startswith("data: ")
82 ]
83 assert len(messages) == 1
84 assert "error" not in messages[0]
85 return messages[0]["result"]
88async def test_two_apps_isolate_snapshots_clients_and_mcp_http(mock_upstream):
89 calls = []
90 versions = {"first.test": "first", "second.test": "second"}
91 payloads = {f"/{path}": [] for path in PREFETCH_ENDPOINTS}
92 payloads.update(
93 {
94 "/buses.json": {},
95 "/maps.json": {},
96 "/courses/sources.json": {"semesters": ["11510"]},
97 "/libraries/rss.json": {
98 "news": [],
99 "eresources": [],
100 "exhibit": [],
101 "branches": [],
102 },
103 }
104 )
106 def handler(request):
107 calls.append((request.url.host, request.url.path))
108 marker = versions[request.url.host]
109 if request.url.path == "/file_details.json":
110 return httpx.Response(
111 200,
112 json={
113 "file_details": {
114 "/": [{"name": path, "version": marker} for path in PREFETCH_ENDPOINTS]
115 }
116 },
117 )
118 if request.url.path == "/courses/latest.json":
119 return httpx.Response(200, json=[course_payload(id=f"11510{marker}")])
120 return httpx.Response(200, json=payloads[request.url.path])
122 mock_upstream(handler)
123 first = api.create_app(Runtime.create(NTHUDataManager(base_url="https://first.test")))
124 second = api.create_app(Runtime.create(NTHUDataManager(base_url="https://second.test")))
125 assert first.state.runtime is not second.state.runtime
126 assert first.state.mcp is not second.state.mcp
127 assert first.state.runtime.courses.state is not second.state.runtime.courses.state
128 async with first.router.lifespan_context(first):
129 first_clients = (
130 first.state.datasets.fetcher.client,
131 first.state.runtime.energy.client,
132 first.state.runtime.libraries.client,
133 )
134 assert all(client is not None and not client.is_closed for client in first_clients)
135 async with second.router.lifespan_context(second):
136 second_clients = (
137 second.state.datasets.fetcher.client,
138 second.state.runtime.energy.client,
139 second.state.runtime.libraries.client,
140 )
141 assert all(a is not b for a, b in zip(first_clients, second_clients, strict=True))
142 async with (
143 httpx.AsyncClient(
144 transport=httpx.ASGITransport(app=first), base_url="http://first"
145 ) as first_rest,
146 httpx.AsyncClient(
147 transport=httpx.ASGITransport(app=second), base_url="http://second"
148 ) as second_rest,
149 ):
150 before = len(calls)
151 responses = await asyncio.gather(
152 first_rest.get("/courses"), second_rest.get("/courses")
153 )
154 for response, marker in zip(responses, ("first", "second"), strict=True):
155 assert response.json()[0]["id"] == f"11510{marker}"
156 assert response.headers["X-Data-Commit-Hash"] == marker
157 for rest, marker in ((first_rest, "first"), (second_rest, "second")):
158 initialized = await _mcp_request(
159 rest,
160 "initialize",
161 {
162 "protocolVersion": "2025-03-26",
163 "capabilities": {},
164 "clientInfo": {"name": "runtime-test", "version": "1"},
165 },
166 )
167 assert initialized["serverInfo"]["version"] == first.version
168 result = await _mcp_request(
169 rest, "tools/call", {"name": "search_courses", "arguments": {}}
170 )
171 assert not result.get("isError", False)
172 courses_payload = json.loads(result["content"][0]["text"])
173 expected_id = f"11510{marker}"
174 assert courses_payload["courses"][0]["id"] == expected_id
175 ping = await rest.get("/ping")
176 assert ping.status_code == 200
177 assert ping.json()["datasets"]["/courses/latest.json"]["version"] == marker
178 assert len(calls) == before
179 assert get_runtime() is default_runtime
180 assert all(client.is_closed for client in second_clients)
181 assert all(not client.is_closed for client in first_clients)
182 assert all(client.is_closed for client in first_clients)
183 assert first.state.runtime.energy.client is None
184 assert first.state.runtime.libraries.client is None
185 assert not first.state.runtime.courses.state.usable
186 assert not second.state.runtime.courses.state.usable
187 assert first.state.runtime.courses.state.snapshot is None
188 assert first.state.runtime.buses._snapshot is None
191async def test_runtime_closes_all_clients_on_startup_failure(monkeypatch):
192 runtime = Runtime.create()
193 application = api.create_app(runtime)
194 clients = []
196 async def fail(endpoints):
197 clients.extend(
198 [runtime.datasets.fetcher.client, runtime.energy.client, runtime.libraries.client]
199 )
200 raise RuntimeError("startup failed")
202 monkeypatch.setattr(runtime.datasets, "prefetch", fail)
203 app_lifespan = api.lifespan(application)
204 with pytest.raises(RuntimeError, match="startup failed"):
205 await app_lifespan.__aenter__()
206 assert len(clients) == 3
207 assert all(client.is_closed for client in clients)
208 assert runtime.datasets.fetcher.client is None
209 assert runtime.energy.client is None
210 assert runtime.libraries.client is None
213async def test_runtime_cannot_be_reentered_without_losing_live_clients():
214 runtime = Runtime.create()
215 async with runtime.lifespan():
216 energy = runtime.energy.client
217 library = runtime.libraries.client
218 nested_lifespan = runtime.lifespan()
219 with pytest.raises(RuntimeError, match="already running"):
220 await nested_lifespan.__aenter__()
221 assert runtime.energy.client is energy
222 assert runtime.libraries.client is library
223 assert not energy.is_closed
224 assert not library.is_closed
227async def test_energy_reuses_one_injected_client():
228 requests = []
230 def handler(request):
231 requests.append(request)
232 return httpx.Response(200, text='<img alt="kW: 1">')
234 service = EnergyService()
235 async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
236 service.client = client
237 assert len(await service.get_realtime_electricity_usage()) == 3
238 assert len(await service.get_realtime_electricity_usage()) == 3
239 assert not client.is_closed
240 assert len(requests) == 6
243def test_bus_preparation_returns_data_without_runtime_state():
244 data = BusesService.prepare({})
245 assert isinstance(data, PreparedBusData)
246 assert not isinstance(data, BusesService)
247 assert not hasattr(data, "manager")
248 assert not hasattr(data, "state")
251def test_app_factory_defaults_to_independent_runtimes():
252 first, second = api.create_app(), api.create_app()
253 assert first.state.runtime is not second.state.runtime
254 assert first.state.datasets is not second.state.datasets
255 assert first.state.datasets is not default_runtime.datasets
258async def test_isolated_app_reports_missing_snapshots_through_rest_and_mcp():
259 application = api.create_app()
260 async with application.router.lifespan_context(application):
261 async with httpx.AsyncClient(
262 transport=httpx.ASGITransport(app=application), base_url="http://test"
263 ) as client:
264 response = await client.get("/courses", headers={"Origin": "https://example.com"})
265 assert response.status_code == 503
266 assert response.json() == {"detail": "Service temporarily unavailable"}
267 assert response.headers["Access-Control-Allow-Origin"] == "*"
268 assert float(response.headers["X-Process-Time"]) >= 0
269 result = await _mcp_request(
270 client, "tools/call", {"name": "search_courses", "arguments": {}}
271 )
272 assert result["isError"]
273 assert "temporarily unavailable" in result["content"][0]["text"]
274 assert "https://" not in result["content"][0]["text"]
277async def test_energy_failure_cancels_siblings_before_releasing_client():
278 started = []
279 cancelled = []
280 ready = asyncio.Event()
282 async def handler(request):
283 system = request.url.path
284 started.append(system)
285 if len(started) == 3:
286 ready.set()
287 await ready.wait()
288 if system.endswith("1.aspx"):
289 raise httpx.ConnectError("Unavailable", request=request)
290 try:
291 await asyncio.Event().wait()
292 finally:
293 cancelled.append(system)
295 service = EnergyService()
296 async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
297 service.client = client
298 with pytest.raises(UpstreamException):
299 await service.get_realtime_electricity_usage()
300 assert len(cancelled) == 2
301 assert not client.is_closed
304async def test_library_live_calls_borrow_the_injected_client(
305 library_space_payload, lost_items_html
306):
307 methods = []
309 def handler(request):
310 methods.append(request.method)
311 if request.method == "GET":
312 return httpx.Response(200, json=library_space_payload)
313 return httpx.Response(200, text=lost_items_html(count=1))
315 service = LibrariesService()
316 async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
317 service.client = client
318 spaces = await service.get_space_availability()
319 assert spaces == library_space_payload["rows"]
320 assert len(await service.get_lost_and_found_items()) == 1
321 assert not client.is_closed
322 assert methods == ["GET", "POST"]