Coverage for tests/test_buses.py: 100%
83 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Tests for buses endpoints."""
3import pytest
4from fastmcp import Client
5from httpx import ASGITransport, AsyncClient
7from data_api.api import schemas
8from data_api.api.api import app
9from data_api.domain.buses.services import PreparedBusData, buses_service
10from data_api.mcp.server import mcp
12pytestmark = pytest.mark.usefixtures("dataset_runtime")
15def test_missing_route_metadata_is_not_registered():
16 service = PreparedBusData()
17 service._res_json = {"towardMainCampusInfo": {"direction": "down"}}
19 service._populate_info_data()
21 assert service.get_route_info("nanda", "up") == []
22 assert service.get_route_info("nanda", "down") == [{"direction": "down"}]
25@pytest.mark.parametrize(
26 "path,method",
27 [
28 ("/buses/routes", "get_route_info"),
29 ("/buses/stops", "gen_bus_stops_info"),
30 (
31 "/buses/schedule?route=main&day=weekday&direction=up",
32 "query_schedule",
33 ),
34 (
35 "/buses/schedule?route=main&day=weekday&direction=up&stop=台積館",
36 "query_schedule",
37 ),
38 ],
39)
40async def test_internal_bus_errors_are_safe(monkeypatch, caplog, path, method):
41 def fail(*args, **kwargs):
42 raise RuntimeError("Private bus failure at https://upstream.example")
44 monkeypatch.setattr(buses_service, method, fail)
45 async with AsyncClient(
46 transport=ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test"
47 ) as client:
48 response = await client.get(path, headers={"Origin": "https://example.com"})
49 assert response.status_code == 500
50 assert response.json() == {"detail": "Internal server error"}
51 assert response.headers["Access-Control-Allow-Origin"] == "*"
52 assert float(response.headers["X-Process-Time"]) >= 0
53 assert "Private bus failure" in caplog.text
56async def test_unexpected_bus_tool_errors_are_masked(monkeypatch, caplog):
57 async def fail():
58 raise RuntimeError("Private bus failure at https://upstream.example")
60 monkeypatch.setattr(buses_service, "update_data", fail)
61 async with Client(mcp) as client:
62 result = await client.call_tool("get_bus_schedule", {}, raise_on_error=False)
63 assert result.is_error
64 assert "Private" not in result.content[0].text
65 assert "http" not in result.content[0].text
66 assert "Private bus failure" in caplog.text
69class TestBusesRoutes:
70 """Tests for bus routes endpoints."""
72 @pytest.fixture
73 async def client(self):
74 """Create async test client."""
75 async with AsyncClient(
76 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True
77 ) as client:
78 yield client
80 @pytest.mark.parametrize("campus", ["main", "nanda"])
81 @pytest.mark.parametrize("direction", ["up", "down"])
82 async def test_get_bus_routes(self, client: AsyncClient, campus: str, direction: str):
83 """Test getting bus routes by campus and direction."""
84 response = await client.get(f"/buses/routes/?route={campus}&direction={direction}")
85 assert response.status_code == 200
88class TestBusesStopMetadata:
89 """Tests for the bus stop collection endpoint."""
91 @pytest.fixture
92 async def client(self):
93 """Create async test client."""
94 async with AsyncClient(
95 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True
96 ) as client:
97 yield client
99 async def test_get_bus_stops_info(self, client: AsyncClient):
100 """Test getting bus stops information."""
101 response = await client.get("/buses/stops")
102 assert response.status_code == 200
103 expected = [
104 schemas.buses.BusStopsInfo.model_validate(stop).model_dump()
105 for stop in buses_service.gen_bus_stops_info()
106 ]
107 assert expected
108 assert response.json() == expected
109 assert response.headers["X-Data-Commit-Hash"] == buses_service.last_commit_hash
112class TestBusesSchedules:
113 """Tests for bus schedules endpoints."""
115 @pytest.fixture
116 async def client(self):
117 """Create async test client."""
118 async with AsyncClient(
119 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True
120 ) as client:
121 yield client
123 @pytest.mark.parametrize("path", ["/buses/schedule"])
124 @pytest.mark.parametrize(
125 "bus_type",
126 [_.value for _ in schemas.buses.BusRouteType],
127 )
128 @pytest.mark.parametrize(
129 "day",
130 [_.value for _ in schemas.buses.BusDayWithCurrent],
131 )
132 @pytest.mark.parametrize(
133 "direction",
134 [_.value for _ in schemas.buses.BusDirection],
135 )
136 async def test_get_bus_schedules(
137 self, client: AsyncClient, bus_type: str, day: str, direction: str, path: str
138 ):
139 """Test getting bus schedules by type, day and direction."""
140 response = await client.get(f"{path}?route={bus_type}&day={day}&direction={direction}")
141 assert response.status_code == 200
144class TestBusesStops:
145 """Tests for bus stops endpoints."""
147 @pytest.fixture
148 async def client(self):
149 """Create async test client."""
150 async with AsyncClient(
151 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True
152 ) as client:
153 yield client
155 @pytest.mark.parametrize(
156 "stop_name",
157 [_.value for _ in schemas.buses.BusStopsName],
158 )
159 @pytest.mark.parametrize(
160 "bus_type",
161 [_.value for _ in schemas.buses.BusRouteType],
162 )
163 @pytest.mark.parametrize(
164 "day",
165 [_.value for _ in schemas.buses.BusDayWithCurrent],
166 )
167 @pytest.mark.parametrize(
168 "direction",
169 [_.value for _ in schemas.buses.BusDirection],
170 )
171 @pytest.mark.parametrize("details", [True, False])
172 async def test_get_bus_stop_arrivals(
173 self,
174 client: AsyncClient,
175 stop_name: str,
176 bus_type: str,
177 day: str,
178 direction: str,
179 details: bool,
180 ):
181 """Test getting bus arrivals at a specific stop."""
182 response = await client.get(
183 "/buses/schedule",
184 params={
185 "stop": stop_name,
186 "route": bus_type,
187 "day": day,
188 "direction": direction,
189 "details": details,
190 },
191 )
192 assert response.status_code == 200