Coverage for tests/test_external_checks.py: 99%

1"""All external-check tests use mock transports, never live services.""" 

2 

3import hashlib 

4import json 

5from copy import deepcopy 

6from datetime import datetime, timezone 

7from pathlib import Path 

8 

9import httpx 

10import pytest 

11from conftest import course_payload 

12 

13from data_api.api.schemas.ping import DatasetHealth, PingResponse 

14from data_api.checks.common import CheckFailure, fetch_json, report 

15from data_api.checks.published import check_published 

16from data_api.checks.smoke import ENDPOINTS, check_endpoints, check_smoke 

17from data_api.core.config import PREFETCH_ENDPOINTS 

18from data_api.data.nthudata import Freshness 

19 

20 

21@pytest.fixture 

22def publication(): 

23 rss = json.loads( 

24 (Path(__file__).parent / "fixtures" / "library_rss_published_urls.json").read_text( 

25 encoding="utf-8" 

26 ) 

27 ) 

28 payloads = {f"/{path}": [] for path in PREFETCH_ENDPOINTS} 

29 payloads.update( 

30 { 

31 "/buses.json": {}, 

32 "/maps.json": {}, 

33 "/courses/latest.json": [course_payload()], 

34 "/courses/sources.json": {"semesters": ["11510"]}, 

35 "/courses/semesters/11510.json": [course_payload()], 

36 "/libraries/rss.json": rss, 

37 } 

38 ) 

39 documents = {path: json.dumps(payload).encode() for path, payload in payloads.items()} 

40 manifest = {"file_details": {"/": [], "libraries": []}} 

41 for path, content in documents.items(): 

42 section, _, name = path.lstrip("/").rpartition("/") 

43 manifest["file_details"].setdefault(section or "/", []).append( 

44 { 

45 "name": name, 

46 "version": "fixture", 

47 "last_commit": "not-the-version", 

48 "sha256": hashlib.sha256(content).hexdigest(), 

49 } 

50 ) 

51 return documents, manifest 

52 

53 

54def published_client(documents, manifest, overrides=None): 

55 def serve(request): 

56 if request.url.path in (overrides or {}): 

57 result = overrides[request.url.path] 

58 if isinstance(result, Exception): 

59 raise result 

60 return result 

61 if request.url.path == "/file_details.json": 

62 return httpx.Response(200, json=manifest) 

63 if request.url.path in documents: 

64 return httpx.Response( 

65 200, 

66 content=documents[request.url.path], 

67 headers={"content-type": "application/json"}, 

68 ) 

69 return httpx.Response(404) 

70 

71 return httpx.AsyncClient(transport=httpx.MockTransport(serve)) 

72 

73 

74async def test_complete_publication_accepts_url_regressions_and_serializes(publication): 

75 documents, manifest = publication 

76 original = deepcopy(documents) 

77 async with published_client(documents, manifest) as client: 

78 assert await check_published(client, "https://publisher.test/") == [] 

79 assert documents == original 

80 

81 

82async def test_publication_check_does_not_reset_an_active_runtime(publication, dataset_runtime): 

83 await dataset_runtime.get("courses/latest.json") 

84 snapshot = dataset_runtime.states["/courses/latest.json"].snapshot 

85 active_client = dataset_runtime.fetcher.client 

86 documents, manifest = publication 

87 async with published_client(documents, manifest) as client: 

88 assert await check_published(client, "https://publisher.test") == [] 

89 assert dataset_runtime.states["/courses/latest.json"].snapshot is snapshot 

90 assert dataset_runtime.fetcher.client is active_client 

91 assert not active_client.is_closed 

92 

93 

94async def test_legacy_manifest_and_empty_datasets_are_accepted(publication): 

95 documents, manifest = publication 

96 for entries in manifest["file_details"].values(): 

97 for entry in entries: 

98 entry.pop("version") 

99 entry.pop("sha256") 

100 documents["/courses/latest.json"] = b"[]" 

101 documents["/libraries/rss.json"] = json.dumps( 

102 {"news": [], "eresources": [], "exhibit": [], "branches": []} 

103 ).encode() 

104 async with published_client(documents, manifest) as client: 

105 assert await check_published(client, "https://publisher.test") == [] 

106 

107 

108@pytest.mark.parametrize( 

109 "response,stage,reason", 

110 [ 

111 (httpx.Response(404), "http", "status 404"), 

112 ( 

113 httpx.Response( 

114 200, text="<html>not a dataset</html>", headers={"content-type": "text/html"} 

115 ), 

116 "content-type", 

117 "expected application/json", 

118 ), 

119 ( 

120 httpx.Response(200, content=b"not json", headers={"content-type": "application/json"}), 

121 "checksum", 

122 "SHA-256 mismatch", 

123 ), 

124 (httpx.ReadTimeout("private URL"), "http", "timeout"), 

125 (httpx.ConnectError("private URL"), "http", "connection/request failure"), 

126 ], 

127) 

128async def test_delivery_failures_identify_dataset_and_stage(publication, response, stage, reason): 

129 documents, manifest = publication 

130 async with published_client(documents, manifest, {"/directory.json": response}) as client: 

131 failures = await check_published(client, "https://publisher.test") 

132 assert failures == [CheckFailure("/directory.json", stage, reason)] 

133 

134 

135@pytest.mark.parametrize( 

136 "path,payload,expected_stage", 

137 [ 

138 ("/directory.json", [{"unexpected": "private payload"}], "candidate"), 

139 ("/courses/latest.json", [{"id": ""}], "candidate"), 

140 ("/buses.json", {"unexpected": []}, "candidate"), 

141 ( 

142 "/libraries/rss.json", 

143 {"news": [{"title": "Article", "description": "", "image": {"url": "https://"}}]}, 

144 "candidate", 

145 ), 

146 ], 

147) 

148async def test_schema_drift_fails_full_candidate(publication, path, payload, expected_stage): 

149 documents, manifest = publication 

150 documents[path] = json.dumps(payload).encode() 

151 for entries in manifest["file_details"].values(): 

152 for entry in entries: 

153 entry.pop("sha256") 

154 async with published_client(documents, manifest) as client: 

155 failures = await check_published(client, "https://publisher.test") 

156 assert len(failures) == 1 

157 assert failures[0].target == path 

158 assert failures[0].stage == expected_stage 

159 assert "private payload" not in str(failures[0]) 

160 

161 

162async def test_missing_required_manifest_entry_is_not_silently_loaded(publication): 

163 documents, manifest = publication 

164 manifest["file_details"]["libraries"] = [] 

165 async with published_client(documents, manifest) as client: 

166 failures = await check_published(client, "https://publisher.test") 

167 assert {failure.target for failure in failures} == { 

168 "/libraries/rss.json", 

169 "/libraries/calendars.json", 

170 } 

171 assert all(failure.stage == "manifest" for failure in failures) 

172 

173 

174async def test_published_semester_must_have_manifest_entry(publication): 

175 documents, manifest = publication 

176 manifest["file_details"]["courses/semesters"] = [] 

177 async with published_client(documents, manifest) as client: 

178 failures = await check_published(client, "https://publisher.test") 

179 assert failures == [ 

180 CheckFailure( 

181 "/courses/semesters/11510.json", 

182 "manifest", 

183 "published semester dataset is missing", 

184 ) 

185 ] 

186 

187 

188@pytest.mark.parametrize( 

189 "payload", 

190 [ 

191 [course_payload(id="11420OTHER", semester_code="11420")], 

192 [course_payload(), course_payload()], 

193 [{"id": "11510INVALID"}], 

194 ], 

195) 

196async def test_published_semester_model_and_identity_are_validated(publication, payload): 

197 documents, manifest = publication 

198 path = "/courses/semesters/11510.json" 

199 documents[path] = json.dumps(payload).encode() 

200 manifest["file_details"]["courses/semesters"][0].pop("sha256") 

201 async with published_client(documents, manifest) as client: 

202 failures = await check_published(client, "https://publisher.test") 

203 assert len(failures) == 1 

204 assert failures[0].target == path 

205 assert failures[0].stage == "candidate" 

206 

207 

208async def test_duplicate_manifest_is_rejected(publication): 

209 documents, manifest = publication 

210 manifest["file_details"]["/"].append(manifest["file_details"]["/"][0]) 

211 async with published_client(documents, manifest) as client: 

212 failures = await check_published(client, "https://publisher.test") 

213 assert failures == [CheckFailure("/file_details.json", "manifest", "manifest_parsing")] 

214 

215 

216@pytest.mark.parametrize( 

217 "content,stage", 

218 [ 

219 (b"invalid", "json"), 

220 (b'"\xff"', "json"), 

221 (b"null", "json"), 

222 ], 

223) 

224async def test_json_failures_without_checksum(content, stage): 

225 async with httpx.AsyncClient( 

226 transport=httpx.MockTransport( 

227 lambda request: httpx.Response( 

228 200, content=content, headers={"content-type": "application/json; charset=utf-8"} 

229 ) 

230 ) 

231 ) as client: 

232 with pytest.raises(CheckFailure) as error: 

233 await fetch_json(client, "https://publisher.test/test.json", "/test.json") 

234 assert error.value.stage == stage 

235 

236 

237def healthy_readiness(): 

238 return PingResponse( 

239 status="ok", 

240 ready=True, 

241 checked_at=datetime.now(timezone.utc), 

242 datasets={ 

243 f"/{path}": DatasetHealth(usable=True, freshness=Freshness.CURRENT) 

244 for path in PREFETCH_ENDPOINTS 

245 }, 

246 ).model_dump(mode="json") 

247 

248 

249def smoke_client(overrides=None): 

250 def serve(request): 

251 if request.url.path in (overrides or {}): 

252 return overrides[request.url.path] 

253 if request.url.path == "/ping": 

254 return httpx.Response( 

255 200, json=healthy_readiness(), headers={"Cache-Control": "no-store"} 

256 ) 

257 return httpx.Response( 

258 200, json=[], headers={"X-Process-Time": "0.01", "X-Total-Count": "0"} 

259 ) 

260 

261 return httpx.AsyncClient(transport=httpx.MockTransport(serve)) 

262 

263 

264async def test_smoke_accepts_empty_feeds_and_unknown_versions(): 

265 async with smoke_client() as client: 

266 assert await check_smoke(client, "https://api.test/") == [] 

267 rss_paths = [check.path for check in ENDPOINTS if check.path.startswith("/libraries/rss/")] 

268 assert set(rss_paths) == { 

269 "/libraries/rss/news", 

270 "/libraries/rss/eresources", 

271 "/libraries/rss/exhibit", 

272 "/libraries/rss/branches", 

273 } 

274 

275 

276@pytest.mark.parametrize("version", [None, "wrong-version", "expected-version"]) 

277async def test_published_serialization_checks_expected_version_header(version): 

278 headers = {"X-Process-Time": "0.01", "X-Total-Count": "0"} 

279 if version is not None: 

280 headers["X-Data-Commit-Hash"] = version 

281 expected = {check.path: None for check in ENDPOINTS} 

282 expected["/courses"] = "expected-version" 

283 async with smoke_client({"/courses": httpx.Response(200, json=[], headers=headers)}) as client: 

284 failures = await check_endpoints(client, "https://api.test", expected) 

285 if version == "expected-version": 

286 assert failures == [] 

287 else: 

288 assert len(failures) == 1 

289 assert failures[0].target == "/courses" 

290 assert failures[0].stage == "headers" 

291 

292 

293@pytest.mark.parametrize( 

294 "path,response,stage", 

295 [ 

296 ("/libraries/rss/news", httpx.Response(503), "http"), 

297 ("/libraries/rss/branches", httpx.Response(200, json={}), "response-schema"), 

298 ("/directory", httpx.Response(200, json=[{"unexpected": "private"}]), "response-schema"), 

299 ("/courses", httpx.Response(200, json=[], headers={"X-Process-Time": "0.01"}), "headers"), 

300 ( 

301 "/courses", 

302 httpx.Response(200, json=[], headers={"X-Process-Time": "nan", "X-Total-Count": "0"}), 

303 "headers", 

304 ), 

305 ( 

306 "/courses", 

307 httpx.Response(200, json=[], headers={"X-Process-Time": "0.01", "X-Total-Count": "53"}), 

308 "headers", 

309 ), 

310 ("/ping", httpx.Response(503), "http"), 

311 ("/ping", httpx.Response(200, json={}), "response-schema"), 

312 ("/ping", httpx.Response(200, json=healthy_readiness()), "headers"), 

313 ], 

314) 

315async def test_smoke_reports_endpoint_schema_headers_and_availability(path, response, stage): 

316 async with smoke_client({path: response}) as client: 

317 failures = await check_smoke(client, "https://api.test") 

318 assert len(failures) == 1 

319 assert failures[0].target == path 

320 assert failures[0].stage == stage 

321 

322 

323@pytest.mark.parametrize("missing,unusable", [(True, False), (False, True)]) 

324async def test_smoke_rejects_false_readiness(missing, unusable): 

325 payload = healthy_readiness() 

326 if missing: 

327 del payload["datasets"]["/courses/latest.json"] 

328 if unusable: 

329 payload["datasets"]["/courses/latest.json"]["usable"] = False 

330 async with smoke_client( 

331 {"/ping": httpx.Response(200, json=payload, headers={"Cache-Control": "no-store"})} 

332 ) as client: 

333 failures = await check_smoke(client, "https://api.test") 

334 assert len(failures) == 1 

335 assert failures[0].stage == "readiness" 

336 

337 

338def test_external_report_exit_status_and_safe_diagnostic(capsys): 

339 assert report([], "published-data") == 0 

340 assert report([CheckFailure("/rss.json", "candidate", "validation")], "published-data") == 1 

341 output = capsys.readouterr().out 

342 assert "PASS published-data" in output 

343 assert "/rss.json: stage=candidate reason=validation" in output