Coverage for src/data_api/checks/published.py: 85%

1"""Opt-in publisher checks: python -m data_api.checks.published.""" 

2 

3import argparse 

4import asyncio 

5import json 

6import math 

7from typing import Any 

8 

9import httpx 

10from pydantic import TypeAdapter, ValidationError 

11 

12from data_api.api.api import create_app 

13from data_api.api.schemas.buses import BusCanonicalDetailedSchedule, BusCanonicalSchedule 

14from data_api.api.schemas.courses import CourseData 

15from data_api.checks.common import CheckFailure, fetch_json, report, validation_reason 

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

17from data_api.core.config import PREFETCH_ENDPOINTS 

18from data_api.data.nthudata import ( 

19 FetchFailure, 

20 JsonData, 

21 ManifestEntry, 

22 NTHUDataManager, 

23 parse_manifest, 

24) 

25from data_api.domain.buses.services import PreparedBusData 

26from data_api.domain.courses.services import prepare_catalog, prepare_courses 

27from data_api.runtime import Runtime 

28 

29MANIFEST_PATH = "/file_details.json" 

30 

31 

32def _check_serialization(path: str, candidate: Any) -> None: 

33 try: 

34 if path == "/courses/latest.json" or path.startswith("/courses/semesters/"): 

35 adapter = TypeAdapter(list[CourseData]) 

36 parsed = adapter.validate_python(candidate, from_attributes=True) 

37 adapter.dump_json(parsed, warnings="error") 

38 elif isinstance(candidate, PreparedBusData): 

39 for model, store in [ 

40 (BusCanonicalSchedule, candidate.raw_schedule_data), 

41 (BusCanonicalDetailedSchedule, candidate.detailed_schedule_data), 

42 ]: 

43 adapter = TypeAdapter(list[model]) 

44 for rows in store.values(): 

45 adapter.dump_json(adapter.validate_python(rows), warnings="error") 

46 else: 

47 TypeAdapter(JsonData).dump_json(candidate, warnings="error") 

48 except (ValidationError, ValueError) as exc: 

49 reason = validation_reason(exc) if isinstance(exc, ValidationError) else type(exc).__name__ 

50 raise CheckFailure(path, "serialization", reason) from exc 

51 

52 

53async def _check_download( 

54 client: httpx.AsyncClient, data_url: str, path: str, checksum: str | None, runtime: Runtime 

55) -> bytes: 

56 payload, response = await fetch_json(client, f"{data_url}{path}", path, checksum) 

57 try: 

58 if path.startswith("/courses/semesters/"): 

59 semester = path.rsplit("/", 1)[1].removesuffix(".json") 

60 candidate = prepare_courses(payload, semester) 

61 else: 

62 candidate = runtime.datasets.state_for(path).prepare(payload) 

63 except ValidationError as exc: 

64 raise CheckFailure(path, "candidate", validation_reason(exc)) from exc 

65 except FetchFailure as exc: 

66 raise CheckFailure(path, "candidate", exc.category) from exc 

67 except ValueError as exc: 

68 raise CheckFailure(path, "transformation", type(exc).__name__) from exc 

69 _check_serialization(path, candidate) 

70 return response.content 

71 

72 

73async def _check_local_responses( 

74 documents: dict[str, bytes], entries: dict[str, ManifestEntry], runtime: Runtime 

75) -> list[CheckFailure]: 

76 def serve_downloaded(request: httpx.Request) -> httpx.Response: 

77 content = documents.get(request.url.path) 

78 if content is None: 

79 return httpx.Response(404) 

80 return httpx.Response(200, content=content, headers={"content-type": "application/json"}) 

81 

82 # Serialization uses exactly the checked bytes, never a second publisher download or live API. 

83 failures = [] 

84 nthudata = runtime.datasets 

85 async with nthudata.lifespan( 

86 httpx.AsyncClient(transport=httpx.MockTransport(serve_downloaded)) 

87 ): 

88 results = await nthudata.prefetch(PREFETCH_ENDPOINTS) 

89 for endpoint, usable in results.items(): 

90 if not usable: 

91 state = nthudata.state_for(endpoint) 

92 failures.append( 

93 CheckFailure( 

94 f"/{endpoint}", 

95 "installation", 

96 state.last_error.category if state.last_error else "unavailable", 

97 ) 

98 ) 

99 if failures: 

100 return failures 

101 await runtime.courses.get_snapshot() 

102 await runtime.buses.update_data() 

103 expected_versions = { 

104 check.path: entries[check.dataset].active_version for check in ENDPOINTS 

105 } 

106 expected_versions["/calendars"], _ = await runtime.calendars.get_all_calendars() 

107 checks = list(ENDPOINTS) 

108 for semester in prepare_catalog(json.loads(documents["/courses/sources.json"])): 

109 path = f"/courses/{semester}" 

110 dataset = f"/courses/semesters/{semester}.json" 

111 checks.append(EndpointCheck(path, TypeAdapter(list[CourseData]), dataset, counted=True)) 

112 expected_versions[path] = entries[dataset].active_version 

113 async with httpx.AsyncClient( 

114 transport=httpx.ASGITransport(app=create_app(runtime)), 

115 base_url="http://contract.test", 

116 ) as local: 

117 failures = await check_endpoints( 

118 local, "http://contract.test", expected_versions, checks=checks 

119 ) 

120 return [ 

121 CheckFailure(failure.target, f"serialization/{failure.stage}", failure.reason) 

122 for failure in failures 

123 ] 

124 

125 

126async def check_published(client: httpx.AsyncClient, data_url: str) -> list[CheckFailure]: 

127 data_url = data_url.rstrip("/") 

128 try: 

129 manifest, manifest_response = await fetch_json( 

130 client, f"{data_url}{MANIFEST_PATH}", MANIFEST_PATH 

131 ) 

132 try: 

133 entries = parse_manifest(manifest) 

134 except FetchFailure as exc: 

135 raise CheckFailure(MANIFEST_PATH, "manifest", exc.category) from exc 

136 except CheckFailure as exc: 

137 return [exc] 

138 

139 failures = [] 

140 runtime = Runtime.create(NTHUDataManager(base_url=data_url)) 

141 documents = {MANIFEST_PATH: manifest_response.content} 

142 for endpoint in PREFETCH_ENDPOINTS: 

143 path = f"/{endpoint}" 

144 try: 

145 if path not in entries: 

146 raise CheckFailure(path, "manifest", "required dataset is missing") 

147 documents[path] = await _check_download( 

148 client, data_url, path, entries[path].sha256, runtime 

149 ) 

150 except CheckFailure as exc: 

151 failures.append(exc) 

152 if failures: 

153 return failures 

154 semesters = prepare_catalog(json.loads(documents["/courses/sources.json"])) 

155 for semester in semesters: 

156 path = f"/courses/semesters/{semester}.json" 

157 try: 

158 if path not in entries: 

159 raise CheckFailure(path, "manifest", "published semester dataset is missing") 

160 documents[path] = await _check_download( 

161 client, data_url, path, entries[path].sha256, runtime 

162 ) 

163 except CheckFailure as exc: 

164 failures.append(exc) 

165 if failures: 

166 return failures 

167 return await _check_local_responses(documents, entries, runtime) 

168 

169 

170async def run(data_url: str, request_timeout: float) -> int: 

171 async with httpx.AsyncClient(timeout=request_timeout, follow_redirects=True) as client: 

172 return report(await check_published(client, data_url), "published-data") 

173 

174 

175def main() -> int: 

176 parser = argparse.ArgumentParser(description=__doc__) 

177 parser.add_argument("--data-url", default="https://data.nthusa.tw") 

178 parser.add_argument("--timeout", type=float, default=15) 

179 args = parser.parse_args() 

180 if not math.isfinite(args.timeout) or args.timeout <= 0: 

181 parser.error("--timeout must be finite and positive") 

182 return asyncio.run(run(args.data_url, args.timeout)) 

183 

184 

185if __name__ == "__main__": 

186 raise SystemExit(main())