Coverage for src/data_api/checks/smoke.py: 88%

1"""Opt-in deployed API smoke checks: python -m data_api.checks.smoke.""" 

2 

3import argparse 

4import asyncio 

5import math 

6from dataclasses import dataclass 

7from typing import Any 

8 

9import httpx 

10from pydantic import TypeAdapter, ValidationError 

11 

12from data_api.api.schemas.announcements import AnnouncementDetail, AnnouncementSource 

13from data_api.api.schemas.buses import BusCanonicalDetailedSchedule, BusInfo, BusStopsInfo 

14from data_api.api.schemas.calendars import Calendar 

15from data_api.api.schemas.courses import CourseData 

16from data_api.api.schemas.departments import Department 

17from data_api.api.schemas.dining import DiningBuilding 

18from data_api.api.schemas.libraries import LibraryRssItem, LibraryRssType 

19from data_api.api.schemas.locations import LocationDetail 

20from data_api.api.schemas.newsletters import NewsletterInfo, NewsletterSource 

21from data_api.api.schemas.ping import PingResponse 

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

23from data_api.core.config import PREFETCH_ENDPOINTS 

24from data_api.models.courses import SemesterCode 

25 

26BUSES_DATASET = "/buses.json" 

27PING_PATH = "/ping" 

28 

29 

30@dataclass(frozen=True) 

31class EndpointCheck: 

32 path: str 

33 adapter: TypeAdapter[Any] 

34 dataset: str 

35 counted: bool = False 

36 

37 

38ENDPOINTS = [ 

39 EndpointCheck("/courses", TypeAdapter(list[CourseData]), "/courses/latest.json", counted=True), 

40 EndpointCheck( 

41 "/courses/semesters", TypeAdapter(list[SemesterCode]), "/courses/sources.json", counted=True 

42 ), 

43 EndpointCheck("/directory", TypeAdapter(list[Department]), "/directory.json"), 

44 EndpointCheck("/announcements", TypeAdapter(list[AnnouncementDetail]), "/announcements.json"), 

45 EndpointCheck( 

46 "/announcements/sources", 

47 TypeAdapter(list[AnnouncementSource]), 

48 "/announcements_list.json", 

49 ), 

50 EndpointCheck("/newsletters", TypeAdapter(list[NewsletterInfo]), "/newsletters.json"), 

51 EndpointCheck("/newsletters/sources", TypeAdapter(list[NewsletterSource]), "/newsletters.json"), 

52 EndpointCheck("/dining", TypeAdapter(list[DiningBuilding]), "/dining.json"), 

53 EndpointCheck("/locations", TypeAdapter(list[LocationDetail]), "/maps.json"), 

54 EndpointCheck("/buses/routes", TypeAdapter(list[BusInfo]), BUSES_DATASET), 

55 EndpointCheck("/buses/stops", TypeAdapter(list[BusStopsInfo]), BUSES_DATASET), 

56 EndpointCheck( 

57 "/buses/schedule?day=weekday&details=true&limit=5", 

58 TypeAdapter(list[BusCanonicalDetailedSchedule]), 

59 BUSES_DATASET, 

60 ), 

61 EndpointCheck("/calendars", TypeAdapter(list[Calendar]), "/calendars.json"), 

62 *[ 

63 EndpointCheck( 

64 f"/libraries/rss/{feed.value}", TypeAdapter(list[LibraryRssItem]), "/libraries/rss.json" 

65 ) 

66 for feed in LibraryRssType 

67 ], 

68] 

69 

70 

71def _validate_endpoint_headers( 

72 check: EndpointCheck, 

73 payload: Any, 

74 response: httpx.Response, 

75 expected_versions: dict[str, str | None] | None, 

76) -> None: 

77 try: 

78 duration = float(response.headers["X-Process-Time"]) 

79 if not math.isfinite(duration) or duration < 0: 

80 raise ValueError 

81 if check.counted and int(response.headers["X-Total-Count"]) != len(payload): 

82 raise ValueError 

83 version = response.headers.get("X-Data-Commit-Hash") 

84 if version == "" or ( 

85 expected_versions is not None and version != expected_versions[check.path] 

86 ): 

87 raise ValueError 

88 except (KeyError, ValueError) as exc: 

89 raise CheckFailure( 

90 check.path, "headers", "missing or invalid timing/count/version" 

91 ) from exc 

92 

93 

94async def check_endpoints( 

95 client: httpx.AsyncClient, 

96 api_url: str, 

97 expected_versions: dict[str, str | None] | None = None, 

98 *, 

99 checks: list[EndpointCheck] | None = None, 

100) -> list[CheckFailure]: 

101 failures = [] 

102 for check in ENDPOINTS if checks is None else checks: 

103 try: 

104 payload, response = await fetch_json(client, f"{api_url}{check.path}", check.path) 

105 try: 

106 check.adapter.validate_json(response.content, strict=True) 

107 except ValidationError as exc: 

108 raise CheckFailure(check.path, "response-schema", validation_reason(exc)) from exc 

109 _validate_endpoint_headers(check, payload, response, expected_versions) 

110 except CheckFailure as exc: 

111 failures.append(exc) 

112 return failures 

113 

114 

115async def check_smoke(client: httpx.AsyncClient, api_url: str) -> list[CheckFailure]: 

116 api_url = api_url.rstrip("/") 

117 failures = [] 

118 try: 

119 _, response = await fetch_json(client, f"{api_url}{PING_PATH}", PING_PATH) 

120 try: 

121 health = PingResponse.model_validate_json(response.content, strict=True) 

122 except ValidationError as exc: 

123 raise CheckFailure(PING_PATH, "response-schema", validation_reason(exc)) from exc 

124 if "no-store" not in response.headers.get("Cache-Control", "").lower(): 

125 raise CheckFailure(PING_PATH, "headers", "expected Cache-Control: no-store") 

126 required = {f"/{endpoint}" for endpoint in PREFETCH_ENDPOINTS} 

127 missing = required - health.datasets.keys() 

128 if missing: 

129 raise CheckFailure( 

130 PING_PATH, "readiness", f"missing datasets: {', '.join(sorted(missing))}" 

131 ) 

132 if ( 

133 not health.ready 

134 or health.status == "unavailable" 

135 or any(not dataset.usable for dataset in health.datasets.values()) 

136 ): 

137 raise CheckFailure(PING_PATH, "readiness", "one or more snapshots are unavailable") 

138 except CheckFailure as exc: 

139 failures.append(exc) 

140 failures.extend(await check_endpoints(client, api_url)) 

141 return failures 

142 

143 

144async def run(api_url: str, request_timeout: float) -> int: 

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

146 return report(await check_smoke(client, api_url), "deployment-smoke") 

147 

148 

149def main() -> int: 

150 parser = argparse.ArgumentParser(description=__doc__) 

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

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

153 args = parser.parse_args() 

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

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

156 return asyncio.run(run(args.api_url, args.timeout)) 

157 

158 

159if __name__ == "__main__": 

160 raise SystemExit(main())