Coverage for tests/test_api_endpoints.py: 100%

1"""Tests for API endpoints.""" 

2 

3import pytest 

4from httpx import ASGITransport, AsyncClient 

5 

6from data_api.api.api import app 

7 

8pytestmark = pytest.mark.usefixtures("dataset_runtime") 

9 

10 

11ROOT_ENDPOINTS = [ 

12 ("/announcements", "getAnnouncements", {"title": "Notice", "fuzzy": "false"}), 

13 ("/calendars", "getAllCalendars", {}), 

14 ("/dining", "getDiningData", {"restaurant_name": "Test", "fuzzy": "false"}), 

15 ("/locations", "getLocations", {"name": "台積", "fuzzy": "false"}), 

16 ("/newsletters", "getAllNewsletters", {"name": "Test", "fuzzy": "false"}), 

17] 

18 

19 

20@pytest.mark.parametrize("path,operation_id,params", ROOT_ENDPOINTS) 

21async def test_root_endpoints_and_slash_redirects(path, operation_id, params): 

22 async with AsyncClient( 

23 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=False 

24 ) as client: 

25 response = await client.get(path, params=params) 

26 legacy = await client.get(f"{path}/", params=params) 

27 

28 assert response.status_code == 200 

29 assert legacy.status_code == 307 

30 assert "location" not in response.headers 

31 assert legacy.headers["location"].split("?")[0] == f"http://test{path}" 

32 assert response.headers["X-Data-Commit-Hash"] 

33 

34 

35@pytest.mark.parametrize("path,operation_id,params", ROOT_ENDPOINTS) 

36def test_root_endpoints_openapi_migration(path, operation_id, params): 

37 paths = app.openapi()["paths"] 

38 operation = paths[path]["get"] 

39 assert f"{path}/" not in paths 

40 assert operation["operationId"] == operation_id 

41 assert not operation.get("deprecated", False) 

42 

43 

44def test_openapi_operation_ids_are_unique(): 

45 operations = [ 

46 operation["operationId"] 

47 for path in app.openapi()["paths"].values() 

48 for method, operation in path.items() 

49 if method in {"get", "post", "put", "patch", "delete", "head", "options", "trace"} 

50 ] 

51 assert len(operations) == len(set(operations)) 

52 

53 

54class TestAnnouncementsEndpoints: 

55 """Tests for announcements endpoints.""" 

56 

57 @pytest.fixture 

58 async def client(self): 

59 """Create async test client.""" 

60 async with AsyncClient( 

61 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

62 ) as client: 

63 yield client 

64 

65 async def test_announcements_with_title_filter(self, client: AsyncClient): 

66 """Test filtering announcements by title.""" 

67 params = {"title": "公告"} 

68 response = await client.get("/announcements", params=params) 

69 assert response.status_code == 200 

70 

71 async def test_announcements_with_fuzzy_search(self, client: AsyncClient): 

72 """Test fuzzy search for announcements.""" 

73 params = {"department": "學生", "fuzzy": True} 

74 response = await client.get("/announcements", params=params) 

75 assert response.status_code == 200 

76 

77 async def test_announcements_sources(self, client: AsyncClient): 

78 """Test getting announcement sources.""" 

79 response = await client.get("/announcements/sources") 

80 assert response.status_code == 200 

81 

82 async def test_announcements_sources_with_department(self, client: AsyncClient): 

83 """Test getting announcement sources with department filter.""" 

84 params = {"department": "清華公佈欄"} 

85 response = await client.get("/announcements/sources", params=params) 

86 assert response.status_code == 200 

87 

88 

89class TestDiningEndpoints: 

90 """Tests for dining endpoints.""" 

91 

92 @pytest.fixture 

93 async def client(self): 

94 """Create async test client.""" 

95 async with AsyncClient( 

96 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

97 ) as client: 

98 yield client 

99 

100 async def test_dining_with_restaurant_name(self, client: AsyncClient): 

101 """Test filtering dining by restaurant name.""" 

102 params = {"restaurant_name": "麥當勞"} 

103 response = await client.get("/dining", params=params) 

104 assert response.status_code == 200 

105 

106 async def test_dining_with_building_name(self, client: AsyncClient): 

107 """Test filtering dining by building name.""" 

108 params = {"building_name": "小吃部"} 

109 response = await client.get("/dining", params=params) 

110 assert response.status_code == 200 

111 

112 async def test_dining_fuzzy_search(self, client: AsyncClient): 

113 """Test dining fuzzy search.""" 

114 params = {"restaurant_name": "便利", "fuzzy": True} 

115 response = await client.get("/dining", params=params) 

116 assert response.status_code == 200 

117 

118 

119class TestLocationsEndpoints: 

120 """Tests for locations endpoints.""" 

121 

122 @pytest.fixture 

123 async def client(self): 

124 """Create async test client.""" 

125 async with AsyncClient( 

126 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

127 ) as client: 

128 yield client 

129 

130 async def test_locations_all(self, client: AsyncClient): 

131 """Test getting all locations.""" 

132 response = await client.get("/locations") 

133 assert response.status_code == 200 

134 

135 

136class TestNewslettersEndpoints: 

137 """Tests for newsletters endpoints.""" 

138 

139 @pytest.fixture 

140 async def client(self): 

141 """Create async test client.""" 

142 async with AsyncClient( 

143 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

144 ) as client: 

145 yield client 

146 

147 async def test_newsletters_all(self, client: AsyncClient): 

148 """Test getting all newsletters.""" 

149 response = await client.get("/newsletters") 

150 assert response.status_code == 200 

151 

152 

153class TestCoursesEndpoints: 

154 """Tests for courses endpoints.""" 

155 

156 @pytest.fixture 

157 async def client(self): 

158 """Create async test client.""" 

159 async with AsyncClient( 

160 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

161 ) as client: 

162 yield client 

163 

164 async def test_courses_all(self, client: AsyncClient): 

165 """Test getting all courses.""" 

166 response = await client.get("/courses") 

167 assert response.status_code == 200 

168 

169 async def test_courses_search_with_params(self, client: AsyncClient): 

170 """Test searching courses with parameters.""" 

171 params = {"chinese_title": "微積分"} 

172 response = await client.get("/courses", params=params) 

173 assert response.status_code == 200 

174 

175 

176class TestBusesEndpoints: 

177 """Tests for buses endpoints.""" 

178 

179 @pytest.fixture 

180 async def client(self): 

181 """Create async test client.""" 

182 async with AsyncClient( 

183 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

184 ) as client: 

185 yield client 

186 

187 async def test_buses_routes_basic(self, client: AsyncClient): 

188 """Test basic bus routes.""" 

189 response = await client.get("/buses/routes/") 

190 assert response.status_code == 200 

191 

192 async def test_buses_routes_with_params(self, client: AsyncClient): 

193 """Test bus routes with parameters.""" 

194 response = await client.get("/buses/routes/?route=main&direction=up") 

195 assert response.status_code == 200 

196 

197 async def test_buses_routes_nanda(self, client: AsyncClient): 

198 """Test Nanda bus routes.""" 

199 response = await client.get("/buses/routes/?route=nanda&direction=up") 

200 assert response.status_code == 200 

201 

202 

203class TestDirectoryEndpoints: 

204 """Tests for directory endpoints.""" 

205 

206 @pytest.fixture 

207 async def client(self): 

208 """Create async test client.""" 

209 async with AsyncClient( 

210 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

211 ) as client: 

212 yield client 

213 

214 async def test_directory_search_multiple_keywords(self, client: AsyncClient): 

215 """Test directory search with different keywords.""" 

216 queries = ["教務處", "學務處", "研發處"] 

217 for query in queries: 

218 params = {"query": query} 

219 response = await client.get("/directory", params=params) 

220 assert response.status_code == 200 

221 

222 

223@pytest.mark.parametrize( 

224 "path,statuses", 

225 [ 

226 ("/buses/routes", {"500"}), 

227 ("/buses/stops", {"500"}), 

228 ("/energy/electricity", {"500", "502", "504"}), 

229 ("/courses", {"422"}), 

230 ("/libraries/spaces", {"500", "502", "504"}), 

231 ("/libraries/lost-and-found", {"500", "502", "504"}), 

232 ("/libraries/rss/{rss_type}", {"404"}), 

233 ("/calendars/{calendar_id}", {"404"}), 

234 ("/calendars/{calendar_id}/events", {"400", "404"}), 

235 ("/calendars/{calendar_id}/events/{event_id}", {"404"}), 

236 ], 

237) 

238def test_openapi_documents_route_errors(path, statuses): 

239 responses = app.openapi()["paths"][path]["get"]["responses"] 

240 assert statuses <= responses.keys() 

241 

242 

243@pytest.mark.parametrize( 

244 "path", 

245 [ 

246 "/energy/electricity", 

247 "/libraries/spaces", 

248 "/libraries/lost-and-found", 

249 ], 

250) 

251def test_openapi_live_errors_have_detail_schema(path): 

252 schema = app.openapi() 

253 responses = schema["paths"][path]["get"]["responses"] 

254 assert "403" not in responses 

255 assert "404" not in responses 

256 for status in ["500", "502", "504"]: 

257 assert responses[status]["content"]["application/json"]["schema"] == { 

258 "$ref": "#/components/schemas/ErrorResponse" 

259 } 

260 error_schema = schema["components"]["schemas"]["ErrorResponse"] 

261 assert error_schema["required"] == ["detail"] 

262 assert error_schema["properties"]["detail"]["type"] == "string"