Coverage for tests/test_bus_schedule.py: 99%
303 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Regression tests for unified bus schedules and stop filters."""
3from datetime import datetime, timezone
5import pytest
6from fastmcp import Client
7from httpx import ASGITransport, AsyncClient
9from data_api.api.api import app
10from data_api.api.routers import buses as bus_router
11from data_api.api.schemas.buses import BusCanonicalDetailedSchedule, BusCanonicalSchedule
12from data_api.data.nthudata import NTHUDataManager, Snapshot
13from data_api.domain.buses import services
14from data_api.domain.buses.enums import BusStopsName
15from data_api.mcp.server import mcp
16from data_api.mcp.tools import buses as bus_tools
17from data_api.runtime import default_runtime
19pytestmark = pytest.mark.usefixtures("dataset_runtime")
22def prepared_service(raw):
23 manager = NTHUDataManager()
24 service = services.BusesService(manager)
25 now = datetime.now(timezone.utc)
26 snapshot = Snapshot(raw, service.prepare(raw), "bus-fixture", now)
27 service.state.snapshot = snapshot
28 service.state.last_checked_at = now
29 service.state.next_check = float("inf")
30 service._snapshot = snapshot
31 service.last_commit_hash = snapshot.version
32 return service
35@pytest.fixture
36def populated_buses(monkeypatch):
37 service = prepared_service(
38 {
39 "towardTSMCBuildingInfo": {
40 "direction": "up",
41 "duration": "2026",
42 "route": "main-up",
43 "routeEN": "Main up",
44 },
45 "towardMainGateInfo": {
46 "direction": "down",
47 "duration": "2026",
48 "route": "main-down",
49 "routeEN": "Main down",
50 },
51 "towardNandaInfo": {
52 "direction": "up",
53 "duration": "2026",
54 "route": "nanda-up",
55 "routeEN": "Nanda up",
56 },
57 "towardMainCampusInfo": {
58 "direction": "down",
59 "duration": "2026",
60 "route": "nanda-down",
61 "routeEN": "Nanda down",
62 },
63 "weekdayBusScheduleTowardTSMCBuilding": [
64 {"time": "08:00", "description": "", "dep_stop": "校門", "line": "red"},
65 {"time": "08:10", "description": "", "dep_stop": "校門", "line": "green"},
66 {"time": "08:20", "description": "", "dep_stop": "校門", "line": "red"},
67 {"time": "08:30", "description": "", "dep_stop": "校門", "line": "green"},
68 ],
69 "weekdayBusScheduleTowardNanda": [
70 {"time": "08:15", "description": "路線二"},
71 ],
72 "weekendBusScheduleTowardTSMCBuilding": [
73 {"time": "09:00", "description": "", "dep_stop": "校門", "line": "red"}
74 ],
75 }
76 )
77 monkeypatch.setattr(services, "buses_service", service)
78 monkeypatch.setattr(default_runtime, "buses", service)
79 return service
82@pytest.mark.parametrize("route", ["main", "nanda", "all"])
83@pytest.mark.parametrize("day", ["weekday", "weekend"])
84@pytest.mark.parametrize("direction", ["up", "down", "all"])
85@pytest.mark.parametrize("stop", list(BusStopsName))
86@pytest.mark.parametrize("detailed", [False, True])
87def test_stop_filters_actual_route_membership(
88 populated_buses, route, day, direction, stop, detailed
89):
90 full = populated_buses.get_schedule(
91 route_type=route, day=day, direction=direction, detailed=True
92 )
93 expected = [
94 {
95 **(bus if detailed else bus["dep_info"]),
96 "arrive_time": next(
97 arrival["arrive_time"]
98 for arrival in bus["stops_time"]
99 if arrival["stop"] == stop.value
100 ),
101 }
102 for bus in full
103 if any(arrival["stop"] == stop.value for arrival in bus["stops_time"])
104 ]
105 result = populated_buses.get_schedule(
106 route_type=route, day=day, direction=direction, detailed=detailed, stop=stop
107 )
108 assert result == expected
109 upcoming = [bus for bus in expected if bus["arrive_time"] >= "08:05"]
110 assert (
111 populated_buses.query_schedule(
112 route_type=route,
113 day=day,
114 direction=direction,
115 detailed=detailed,
116 stop=stop,
117 after_time="08:05",
118 limit=1,
119 )
120 == upcoming[:1]
121 )
122 assert (
123 populated_buses.get_schedule(route_type=route, day=day, direction=direction, detailed=True)
124 == full
125 )
128async def get_schedule(client, path="/buses/schedule", **params):
129 return await client.get(
130 path, params={"route": "all", "day": "weekday", "direction": "all", **params}
131 )
134def canonical_schedules(schedules, details):
135 model = BusCanonicalDetailedSchedule if details else BusCanonicalSchedule
136 return [
137 model.model_validate(bus).model_dump(mode="json", exclude_none=True) for bus in schedules
138 ]
141@pytest.mark.parametrize("route", [None, "main", "nanda"])
142@pytest.mark.parametrize("direction", [None, "up", "down"])
143async def test_route_metadata_filters(populated_buses, route, direction):
144 params = {}
145 if route is not None:
146 params["route"] = route
147 if direction is not None:
148 params["direction"] = direction
149 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
150 response = await client.get("/buses/routes", params=params)
151 assert response.status_code == 200
152 expected_routes = [
153 f"{campus}-{bound}"
154 for campus in ["main", "nanda"]
155 for bound in ["up", "down"]
156 if (route is None or route == campus) and (direction is None or direction == bound)
157 ]
158 assert [info["route"] for info in response.json()] == expected_routes
159 assert response.headers["X-Data-Commit-Hash"] == "bus-fixture"
162@pytest.mark.parametrize("route", ["unknown", "", "all"])
163async def test_invalid_route_metadata_filter(populated_buses, route):
164 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
165 response = await client.get("/buses/routes", params={"route": route})
166 assert response.status_code == 422
167 assert any(error["loc"] == ["query", "route"] for error in response.json()["detail"])
170@pytest.mark.parametrize("path", ["/buses/schedule"])
171@pytest.mark.parametrize("limit", [1, 2])
172async def test_shared_limit_caps_bus_results(populated_buses, path, limit):
173 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
174 response = await get_schedule(client, path, limit=limit)
175 assert response.status_code == 200
176 assert len(response.json()) == limit
179@pytest.mark.parametrize("path", ["/buses/schedule"])
180@pytest.mark.parametrize("limit", [0, -1, "invalid"])
181async def test_invalid_shared_limit_is_rejected(populated_buses, path, limit):
182 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
183 response = await get_schedule(client, path, limit=limit)
184 assert response.status_code == 422
185 assert any(error["loc"] == ["query", "limit"] for error in response.json()["detail"])
188@pytest.mark.parametrize("details", [False, True])
189async def test_rest_filters_before_limit_and_preserves_shape(populated_buses, details):
190 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
191 unfiltered = await get_schedule(client, details=details, limit=100)
192 filtered = await get_schedule(
193 client, details=details, stop=BusStopsName.M3.value, time="08:05", limit=1
194 )
195 assert filtered.status_code == 200
196 expected = populated_buses.query_schedule(
197 route_type="all",
198 day="weekday",
199 direction="all",
200 detailed=details,
201 stop=BusStopsName.M3,
202 after_time="08:05",
203 limit=1,
204 )
205 assert len(expected) == 1
206 assert filtered.json() == canonical_schedules(expected, details)
207 entry = filtered.json()[0]
208 departure = entry["dep_info"] if details else entry
209 assert departure["time"] == "08:10"
210 assert "arrive_time" not in unfiltered.json()[0]
211 assert set(entry) == {*unfiltered.json()[0], "arrive_time"}
212 assert entry["arrive_time"] == "08:12"
213 if details:
214 assert "arrive_time" not in entry["dep_info"]
215 assert filtered.headers["X-Data-Commit-Hash"] == "bus-fixture"
216 if details:
217 assert entry == BusCanonicalDetailedSchedule.model_validate(expected[0]).model_dump(
218 mode="json", exclude_none=True
219 )
222@pytest.mark.parametrize("details", [False, True])
223@pytest.mark.parametrize(
224 "time,expected_times",
225 [
226 ("08:05", ["08:00", "08:10", "08:20", "08:30"]),
227 ("08:06", ["08:00", "08:10", "08:20", "08:30"]),
228 ("08:07", ["08:10", "08:20", "08:30"]),
229 ("08:16", ["08:10", "08:20", "08:30"]),
230 ("08:37", []),
231 ],
232)
233async def test_stop_arrival_filter_matches_mcp(populated_buses, details, time, expected_times):
234 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
235 response = await get_schedule(
236 client, stop=BusStopsName.M5.value, time=time, details=details, limit=100
237 )
238 assert response.status_code == 200
239 departures = [bus["dep_info"] if details else bus for bus in response.json()]
240 assert [bus["time"] for bus in departures] == expected_times
241 async with Client(mcp) as client:
242 result = await client.call_tool(
243 "get_bus_schedule",
244 {
245 "day": "weekday",
246 "stop": BusStopsName.M5.value,
247 "time": time,
248 "details": details,
249 "limit": 100,
250 },
251 )
252 assert result.data["buses"] == response.json()
255@pytest.mark.parametrize("details", [False, True])
256async def test_in_transit_bus_is_included_before_limit_for_selected_stop(populated_buses, details):
257 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
258 response = await get_schedule(
259 client, stop=BusStopsName.M5.value, time="08:05", details=details, limit=1
260 )
261 no_stop = await get_schedule(client, time="08:05", details=details, limit=1)
262 assert response.status_code == no_stop.status_code == 200
263 departure = response.json()[0]["dep_info"] if details else response.json()[0]
264 assert departure["time"] == "08:00"
265 departure_without_stop = no_stop.json()[0]["dep_info"] if details else no_stop.json()[0]
266 assert departure_without_stop["time"] == "08:10"
267 async with Client(mcp) as client:
268 result = await client.call_tool(
269 "get_bus_schedule",
270 {
271 "day": "weekday",
272 "stop": BusStopsName.M5.value,
273 "time": "08:05",
274 "details": details,
275 "limit": 1,
276 },
277 )
278 assert result.data["buses"] == response.json()
281@pytest.mark.parametrize(
282 "stop,time,expected_times",
283 [
284 (BusStopsName.M1, "08:01", ["08:10", "08:15", "08:20", "08:30"]),
285 (BusStopsName.M3, "08:02", ["08:00", "08:10", "08:20", "08:30"]),
286 (BusStopsName.M3, "08:05", ["08:10", "08:20", "08:30"]),
287 (BusStopsName.S1, "08:30", ["08:15"]),
288 ],
289)
290async def test_filter_uses_selected_stop_not_other_stops(
291 populated_buses, stop, time, expected_times
292):
293 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
294 response = await get_schedule(client, stop=stop.value, time=time, limit=100)
295 assert response.status_code == 200
296 assert [bus["time"] for bus in response.json()] == expected_times
299@pytest.mark.parametrize("details", [False, True])
300async def test_current_day_ignores_time(populated_buses, monkeypatch, details):
301 monkeypatch.setattr(bus_router, "get_current_time_state", lambda: ("weekend", "09:03"))
303 class FixedDatetime:
304 @staticmethod
305 def now():
306 return datetime(2026, 10, 4, 9, 3)
308 monkeypatch.setattr(bus_tools, "datetime", FixedDatetime)
309 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
310 response = await get_schedule(
311 client, day="current", stop=BusStopsName.M5.value, time="23:59", details=details
312 )
313 async with Client(mcp) as client:
314 result = await client.call_tool(
315 "get_bus_schedule",
316 {"day": "current", "stop": BusStopsName.M5.value, "time": "23:59", "details": details},
317 )
318 assert response.status_code == 200
319 assert len(response.json()) == 1
320 departure = response.json()[0]["dep_info"] if details else response.json()[0]
321 assert departure["time"] == "09:00"
322 assert result.data["buses"] == response.json()
325@pytest.mark.parametrize("details", [False, True])
326async def test_mcp_stop_time_limit_and_metadata(populated_buses, details):
327 async with Client(mcp) as client:
328 result = await client.call_tool(
329 "get_bus_schedule",
330 {
331 "day": "weekday",
332 "stop": BusStopsName.M3.value,
333 "time": "08:05",
334 "limit": 1,
335 "details": details,
336 },
337 )
338 data = result.data
339 assert len(data["buses"]) == 1
340 assert data["buses"][0]["arrive_time"] == "08:12"
341 if details:
342 assert "arrive_time" not in data["buses"][0]["dep_info"]
343 departure = data["buses"][0]["dep_info"] if details else data["buses"][0]
344 assert departure["time"] == "08:10"
345 assert departure["line"] == "main_green"
346 assert data["stop_name"] == BusStopsName.M3.value
347 assert len(data["stop_info"]) == 1
348 assert {"latitude", "longitude"} <= data["stop_info"][0].keys()
351async def test_no_matching_stop_returns_empty(populated_buses):
352 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
353 response = await get_schedule(client, route="main", stop=BusStopsName.S1.value)
354 assert response.status_code == 200
355 assert response.json() == []
358@pytest.mark.parametrize("details", [False, True])
359async def test_without_stop_preserves_existing_schedule(populated_buses, details):
360 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
361 response = await get_schedule(client, details=details, limit=100)
362 assert response.status_code == 200
363 assert len(response.json()) == 5
364 departures = [
365 entry["dep_info"]["time"] if details else entry["time"] for entry in response.json()
366 ]
367 assert departures == ["08:00", "08:10", "08:15", "08:20", "08:30"]
368 lines = [entry["dep_info"]["line"] if details else entry["line"] for entry in response.json()]
369 assert lines == ["main_red", "main_green", "nanda_route_2", "main_red", "main_green"]
372@pytest.mark.parametrize(
373 "params",
374 [{"stop": "unknown"}, {"stop": ""}, {"limit": 0}, {"route": "unknown"}, {"route": ""}],
375)
376async def test_invalid_rest_filters_are_rejected(populated_buses, params):
377 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
378 response = await get_schedule(client, **params)
379 assert response.status_code == 422
382@pytest.mark.parametrize("params", [{"stop": "unknown"}, {"stop": ""}, {"limit": 0}])
383async def test_invalid_mcp_filters_are_rejected(populated_buses, params):
384 async with Client(mcp) as client:
385 result = await client.call_tool("get_bus_schedule", params, raise_on_error=False)
386 assert result.is_error
389def test_openapi_bus_parameters():
390 paths = app.openapi()["paths"]
391 bus_paths = {path for path in paths if path.startswith("/buses")}
392 canonical_paths = {
393 path for path in bus_paths if not paths[path]["get"].get("deprecated", False)
394 }
395 assert canonical_paths == {"/buses/routes", "/buses/stops", "/buses/schedule"}
396 assert bus_paths == canonical_paths
397 stops = paths["/buses/stops"]["get"]
398 assert stops["operationId"] == "getBusStops"
399 assert not stops.get("parameters")
400 stop_schema = stops["responses"]["200"]["content"]["application/json"]["schema"]
401 assert stop_schema["type"] == "array"
402 assert stop_schema["items"] == {"$ref": "#/components/schemas/BusStopsInfo"}
403 new = paths["/buses/schedule"]["get"]
404 assert not new.get("deprecated", False)
405 assert new["operationId"] == "getBusSchedule"
406 assert "stop" in {parameter["name"] for parameter in new["parameters"]}
407 parameters = {parameter["name"]: parameter for parameter in new["parameters"]}
408 assert "預估到站時間篩選" in parameters["stop"]["description"]
409 assert set(parameters) == {"route", "direction", "day", "time", "stop", "limit", "details"}
410 assert not parameters["limit"]["required"]
411 assert parameters["limit"]["schema"]["default"] == 5
412 assert {"type": "integer", "minimum": 1} in parameters["limit"]["schema"]["anyOf"]
413 route_parameters = {
414 parameter["name"]: parameter for parameter in paths["/buses/routes"]["get"]["parameters"]
415 }
416 assert set(route_parameters) == {"route", "direction"}
417 assert not route_parameters["route"]["required"]
418 assert route_parameters["route"]["schema"]["enum"] == ["main", "nanda"]
419 for name, default in [("route", "all"), ("day", "current"), ("direction", "all")]:
420 assert not parameters[name]["required"]
421 assert parameters[name]["schema"]["default"] == default
424@pytest.mark.parametrize("day", ["weekday", "weekend"])
425@pytest.mark.parametrize(
426 "time", ["25:00", "24:00", "08:60", "noon", "", "8:00", "08:0", " 08:00", "08:00\n"]
427)
428async def test_invalid_mcp_time_is_rejected_before_query(populated_buses, monkeypatch, day, time):
429 def unexpected_query(**kwargs):
430 pytest.fail("Invalid time must be rejected before querying schedules")
432 monkeypatch.setattr(populated_buses, "query_schedule", unexpected_query)
433 async with Client(mcp) as client:
434 result = await client.call_tool(
435 "get_bus_schedule", {"day": day, "time": time}, raise_on_error=False
436 )
437 assert result.is_error
440@pytest.mark.parametrize("day", ["weekday", "weekend"])
441@pytest.mark.parametrize("time", ["00:00", "08:10", "23:59", None])
442async def test_valid_mcp_time_filters(populated_buses, day, time):
443 async with Client(mcp) as client:
444 result = await client.call_tool(
445 "get_bus_schedule", {"day": day, "time": time, "limit": 100, "details": False}
446 )
447 expected = populated_buses.query_schedule(
448 route_type="all", day=day, direction="all", after_time=time or "", limit=100
449 )
450 assert result.data["buses"] == canonical_schedules(expected, False)
453@pytest.mark.parametrize(
454 "params,expected_times",
455 [
456 ({}, ["08:10", "08:15", "08:20", "08:30"]),
457 ({"stop": BusStopsName.M5.value}, ["08:00", "08:10", "08:20", "08:30"]),
458 ({"route": "main", "direction": "up"}, ["08:10", "08:20", "08:30"]),
459 ({"route": "nanda"}, ["08:15"]),
460 ({"time": "23:59"}, ["08:10", "08:15", "08:20", "08:30"]),
461 ({"direction": "down"}, []),
462 ({"limit": 2}, ["08:10", "08:15"]),
463 ],
464)
465async def test_canonical_defaults_query_upcoming_departures(
466 populated_buses, monkeypatch, params, expected_times
467):
468 monkeypatch.setattr(bus_router, "get_current_time_state", lambda: ("weekday", "08:05"))
469 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
470 response = await client.get("/buses/schedule", params=params)
471 assert response.status_code == 200
472 assert [bus["time"] for bus in response.json()] == expected_times
473 assert response.headers["X-Data-Commit-Hash"] == "bus-fixture"
476async def test_canonical_default_limit(populated_buses, monkeypatch):
477 monkeypatch.setattr(bus_router, "get_current_time_state", lambda: ("weekday", "00:00"))
478 bus = populated_buses.raw_schedule_data[("all", "weekday", "all")][-1]
479 populated_buses.raw_schedule_data[("all", "weekday", "all")].append({**bus, "time": "08:40"})
480 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
481 response = await client.get("/buses/schedule")
482 assert response.status_code == 200
483 assert [bus["time"] for bus in response.json()] == ["08:00", "08:10", "08:15", "08:20", "08:30"]
486@pytest.mark.parametrize(
487 "line,expected",
488 [
489 ("red", "main_red"),
490 ("green", "main_green"),
491 ("route_1", "nanda_route_1"),
492 ("route_2", "nanda_route_2"),
493 ("main_red", "main_red"),
494 ("main_green", "main_green"),
495 ("nanda_route_1", "nanda_route_1"),
496 ("nanda_route_2", "nanda_route_2"),
497 ],
498)
499def test_canonical_line_identifiers(line, expected):
500 bus = {
501 "time": "11:30",
502 "description": "",
503 "dep_stop": "校門",
504 "line": line,
505 "bus_type": "large-sized_bus",
506 }
507 assert BusCanonicalSchedule.model_validate(bus).line == expected
508 detailed = BusCanonicalDetailedSchedule.model_validate({"dep_info": bus, "stops_time": []})
509 assert detailed.dep_info.line == expected
510 assert bus["line"] == line
513@pytest.mark.parametrize("details", [False, True])
514async def test_published_nanda_line_rest_and_mcp(populated_buses, monkeypatch, details):
515 published = {
516 "weekdayBusScheduleTowardNanda": [{"time": "11:30", "description": "", "line": "route2"}],
517 "weekdayBusScheduleTowardMainCampus": [
518 {"time": "11:30", "description": "", "line": "route1"}
519 ],
520 }
521 service = prepared_service(published)
522 monkeypatch.setattr(services, "buses_service", service)
523 monkeypatch.setattr(default_runtime, "buses", service)
524 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
525 response = await get_schedule(client, route="nanda", direction="up", details=details)
526 education = await get_schedule(
527 client, route="nanda", direction="up", stop=BusStopsName.M7.value, details=details
528 )
529 tsmc = await get_schedule(
530 client, route="nanda", direction="up", stop=BusStopsName.M5.value, details=details
531 )
532 down = await get_schedule(client, route="nanda", direction="down", details=details)
533 async with Client(mcp) as client:
534 result = await client.call_tool(
535 "get_bus_schedule",
536 {"route": "nanda", "day": "weekday", "direction": "up", "details": details},
537 )
538 assert response.status_code == education.status_code == 200
539 assert down.status_code == tsmc.status_code == 200
540 departure = response.json()[0]["dep_info"] if details else response.json()[0]
541 assert departure == {
542 "time": "11:30",
543 "description": "",
544 "dep_stop": "校門",
545 "line": "nanda_route_2",
546 "bus_type": "large-sized_bus",
547 }
548 assert education.json() == [{**response.json()[0], "arrive_time": "11:37"}]
549 assert tsmc.json() == []
550 down_departure = down.json()[0]["dep_info"] if details else down.json()[0]
551 assert down_departure["line"] == "nanda_route_1"
552 assert not result.is_error
553 assert result.data["buses"] == response.json()
554 if details:
555 assert [arrival["stop"] for arrival in response.json()[0]["stops_time"]] == [
556 BusStopsName.M1.value,
557 BusStopsName.M2.value,
558 BusStopsName.M6.value,
559 BusStopsName.M7.value,
560 BusStopsName.S1.value,
561 ]
562 assert service.get_schedule(
563 route_type="nanda", day="weekday", direction="up", stop=BusStopsName.M7
564 )
565 assert (
566 service.get_schedule(
567 route_type="nanda", day="weekday", direction="up", stop=BusStopsName.M5
568 )
569 == []
570 )
571 assert published["weekdayBusScheduleTowardNanda"][0]["line"] == "route2"
574@pytest.mark.parametrize("day", ["weekday", "weekend"])
575@pytest.mark.parametrize("direction", ["up", "down"])
576@pytest.mark.parametrize(
577 "line,description,expected",
578 [
579 ("route1", "路線二", "route_1"),
580 ("route2", "", "route_2"),
581 ("route_1", "", "route_1"),
582 ("route_2", "", "route_2"),
583 ("nanda_route_1", "", "route_1"),
584 ("nanda_route_2", "", "route_2"),
585 ("", "路線二", "route_2"),
586 ("", "", "route_1"),
587 ],
588)
589def test_nanda_source_line_controls_schedule_and_stops(day, direction, line, description, expected):
590 destination = "Nanda" if direction == "up" else "MainCampus"
591 published = {
592 f"{day}BusScheduleToward{destination}": [
593 {"time": "11:30", "description": description, "line": line}
594 ]
595 }
596 service = services.BusesService.prepare(published)
597 result = service.query_schedule(route_type="nanda", day=day, direction=direction, detailed=True)
598 assert result[0]["dep_info"]["line"] == expected
599 stops = [arrival["stop"] for arrival in result[0]["stops_time"]]
600 assert (BusStopsName.M7.value in stops) == (expected == "route_2")
601 assert (BusStopsName.M5.value in stops) == (expected == "route_1")
602 assert result[0]["stops_time"][0]["arrive_time"] == "11:30"
603 assert published[f"{day}BusScheduleToward{destination}"][0]["line"] == line
606def test_invalid_nanda_source_line_rejects_candidate():
607 with pytest.raises(ValueError, match="Invalid Nanda bus line"):
608 services.BusesService.prepare(
609 {
610 "weekdayBusScheduleTowardNanda": [
611 {"time": "11:30", "description": "", "line": "route3"}
612 ]
613 }
614 )