Coverage for tests/test_openapi_contract.py: 100%

1"""Pin the public structure and match representative real responses to declarations.""" 

2 

3import json 

4from copy import deepcopy 

5from pathlib import Path 

6 

7import httpx 

8import pytest 

9 

10from data_api.api.api import app 

11from data_api.api.schemas.calendars import CalendarEvent 

12from data_api.api.schemas.errors import ErrorResponse, ValidationErrorResponse 

13from data_api.api.schemas.responses import COUNT_HEADERS, DATA_VERSION_HEADERS 

14from data_api.checks.openapi import contract_differences, contract_projection 

15from data_api.domain.calendars.services import calendars_service 

16 

17BASELINE = Path(__file__).parent / "fixtures" / "openapi_contract.json" 

18 

19 

20def test_structured_openapi_baseline(): 

21 expected = json.loads(BASELINE.read_text(encoding="utf-8")) 

22 changes = contract_differences(expected, contract_projection(app.openapi())) 

23 assert not changes, "\n".join(changes) 

24 

25 

26@pytest.mark.parametrize("change", ["route", "parameter", "operationId", "default", "response"]) 

27def test_compatibility_check_identifies_structural_changes(change): 

28 expected = contract_projection(app.openapi()) 

29 actual = deepcopy(expected) 

30 operation = actual["operations"]["GET /courses"] 

31 match change: 

32 case "route": 

33 del actual["operations"]["GET /courses"] 

34 location = "GET /courses" 

35 case "parameter": 

36 del operation["parameters"]["query:lecturers.label"] 

37 location = "query:lecturers.label" 

38 case "operationId": 

39 operation["operationId"] = "unexpectedRename" 

40 location = "operationId" 

41 case "default": 

42 operation["parameters"]["query:type"]["schema"]["default"] = "xclass" 

43 location = "query:type.schema.default" 

44 case "response": 

45 del operation["responses"]["503"] 

46 location = "responses.503" 

47 assert any(location in difference for difference in contract_differences(expected, actual)) 

48 

49 

50def test_documentation_changes_are_ignored_but_title_fields_are_not(): 

51 schema = deepcopy(app.openapi()) 

52 expected = contract_projection(schema) 

53 schema["paths"]["/courses"]["get"]["description"] = "New documentation" 

54 schema["components"]["schemas"]["CourseData"]["properties"]["id"]["description"] = "New label" 

55 assert contract_projection(schema) == expected 

56 assert "title" in expected["schemas"]["LibraryRssItem"]["properties"] 

57 del schema["components"]["schemas"]["LibraryRssItem"]["properties"]["title"] 

58 assert contract_differences(expected, contract_projection(schema)) 

59 

60 

61def test_every_snapshot_operation_documents_availability_and_headers(): 

62 schema = app.openapi() 

63 live_prefixes = ("/energy/", "/libraries/spaces", "/libraries/lost") 

64 for path, methods in schema["paths"].items(): 

65 for operation in methods.values(): 

66 responses = operation["responses"] 

67 if path.startswith(live_prefixes): 

68 assert "503" not in responses 

69 assert "X-Data-Commit-Hash" not in responses["200"].get("headers", {}) 

70 continue 

71 assert responses["503"]["content"]["application/json"]["schema"] == { 

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

73 }, path 

74 headers = responses["200"]["headers"] 

75 counted = path.startswith("/courses") or path == "/calendars/{calendar_id}/events" 

76 assert headers == (COUNT_HEADERS if counted else DATA_VERSION_HEADERS), path 

77 

78 

79def test_manual_errors_and_course_validation_have_body_schemas(): 

80 schema = app.openapi() 

81 for path, methods in schema["paths"].items(): 

82 for operation in methods.values(): 

83 for status in ["400", "404", "500", "502", "503", "504"]: 

84 if status in operation["responses"]: 

85 assert operation["responses"][status]["content"]["application/json"][ 

86 "schema" 

87 ] == {"$ref": "#/components/schemas/ErrorResponse"}, (path, status) 

88 detail = schema["components"]["schemas"]["ValidationErrorResponse"]["properties"]["detail"] 

89 assert {choice["type"] for choice in detail["anyOf"]} == {"string", "array"} 

90 assert schema["components"]["schemas"]["LibraryRssItem"]["properties"]["link"]["anyOf"] == [ 

91 {"type": "string"}, 

92 {"type": "null"}, 

93 ] 

94 for path, method in [ 

95 ("/courses", "get"), 

96 ("/courses/query", "post"), 

97 ]: 

98 assert schema["paths"][path][method]["responses"]["422"]["content"]["application/json"][ 

99 "schema" 

100 ] == {"$ref": "#/components/schemas/ValidationErrorResponse"} 

101 

102 

103def assert_error_matches_schema(response, schema_path, status, method="get"): 

104 assert response.status_code == status 

105 declared = app.openapi()["paths"][schema_path][method]["responses"][str(status)] 

106 ref = declared["content"]["application/json"]["schema"]["$ref"] 

107 model = ValidationErrorResponse if ref.endswith("ValidationErrorResponse") else ErrorResponse 

108 model.model_validate(response.json()) 

109 assert float(response.headers["X-Process-Time"]) >= 0 

110 

111 

112@pytest.mark.parametrize( 

113 "url,schema_path,status,detail_type", 

114 [ 

115 ("/calendars/missing", "/calendars/{calendar_id}", 404, str), 

116 ( 

117 "/calendars/missing/events?start=2026-10-02&end=2026-10-01", 

118 "/calendars/{calendar_id}/events", 

119 400, 

120 str, 

121 ), 

122 ("/courses?type=invalid", "/courses", 422, list), 

123 ("/courses?lecturers.label=%5B", "/courses", 422, str), 

124 ], 

125) 

126async def test_actual_manual_and_validation_errors( 

127 dataset_runtime, url, schema_path, status, detail_type 

128): 

129 async with httpx.AsyncClient( 

130 transport=httpx.ASGITransport(app=app), base_url="http://test" 

131 ) as client: 

132 response = await client.get(url) 

133 assert_error_matches_schema(response, schema_path, status) 

134 assert isinstance(response.json()["detail"], detail_type) 

135 

136 

137async def test_snapshot_unavailable_matches_openapi(): 

138 async with httpx.AsyncClient( 

139 transport=httpx.ASGITransport(app=app), base_url="http://test" 

140 ) as client: 

141 response = await client.get("/libraries/rss/news") 

142 assert_error_matches_schema(response, "/libraries/rss/{rss_type}", 503) 

143 assert "X-Data-Commit-Hash" not in response.headers 

144 

145 

146@pytest.mark.parametrize("timeout,status", [(False, 502), (True, 504)]) 

147async def test_live_upstream_error_matches_openapi(mock_upstream, timeout, status): 

148 def upstream(request): 

149 if timeout: 

150 raise httpx.ReadTimeout("private upstream timeout") 

151 return httpx.Response(503) 

152 

153 mock_upstream(upstream) 

154 async with httpx.AsyncClient( 

155 transport=httpx.ASGITransport(app=app), base_url="http://test" 

156 ) as client: 

157 response = await client.get("/libraries/spaces") 

158 assert_error_matches_schema(response, "/libraries/spaces", status) 

159 

160 

161async def test_calendar_pagination_body_and_headers_match_openapi(monkeypatch): 

162 events = [ 

163 { 

164 "id": f"event-{index}", 

165 "title": "Event", 

166 "start": "2026-10-01", 

167 "end": "2026-10-02", 

168 "all_day": True, 

169 } 

170 for index in range(3) 

171 ] 

172 

173 async def search(*args, **kwargs): 

174 return "calendar-version", events 

175 

176 monkeypatch.setattr(calendars_service, "search_calendar_events", search) 

177 async with httpx.AsyncClient( 

178 transport=httpx.ASGITransport(app=app), base_url="http://test" 

179 ) as client: 

180 response = await client.get("/calendars/academic/events?limit=1&offset=1") 

181 assert response.status_code == 200 

182 assert len(response.json()) == 1 

183 CalendarEvent.model_validate(response.json()[0]) 

184 assert response.headers["X-Total-Count"] == "3" 

185 assert response.headers["X-Data-Commit-Hash"] == "calendar-version" 

186 declared = app.openapi()["paths"]["/calendars/{calendar_id}/events"]["get"]["responses"]["200"] 

187 assert declared["headers"] == COUNT_HEADERS