Coverage for tests/test_courses_query_grammar.py: 100%

1"""Regression coverage for the advanced course query grammar and tree budgets.""" 

2 

3from unittest.mock import Mock 

4 

5import pytest 

6from conftest import make_course, set_courses 

7from httpx import ASGITransport, AsyncClient 

8from pydantic import ValidationError 

9 

10from data_api.api.api import app 

11from data_api.api.schemas.courses import ( 

12 COURSE_QUERY_MAX_DEPTH, 

13 COURSE_QUERY_MAX_NODES, 

14 CourseCondition, 

15 CourseQueryCondition, 

16) 

17from data_api.domain.courses.models import Conditions 

18from data_api.domain.courses.services import courses_service 

19 

20ALICE = {"row_field": "lecturers.label", "matcher": "Alice"} 

21BOB = {"row_field": "lecturers.label", "matcher": "Bob"} 

22THREE_CREDITS = {"row_field": "credits", "matcher": 3} 

23FOUR_CREDITS = {"row_field": "credits", "matcher": 4} 

24 

25INVALID_QUERIES = [ 

26 pytest.param([ALICE, BOB], id="adjacent-two"), 

27 pytest.param([ALICE, BOB, ALICE], id="adjacent-three"), 

28 pytest.param([ALICE, "and", BOB, ALICE], id="extra-trailing-condition"), 

29 pytest.param([ALICE, "and", "or", BOB], id="consecutive-operators"), 

30 pytest.param([ALICE, "and", "or", "and", BOB], id="operator-in-operand-position"), 

31 pytest.param(["and", ALICE], id="leading-operator"), 

32 pytest.param([ALICE, "or"], id="trailing-operator"), 

33 pytest.param(["and"], id="only-operator"), 

34 pytest.param([ALICE, "xor", BOB], id="unknown-operator"), 

35 pytest.param([ALICE, "AND", BOB], id="uppercase-operator"), 

36 pytest.param([ALICE, "and", [ALICE, BOB]], id="nested-adjacent"), 

37 pytest.param([[ALICE, BOB, ALICE]], id="wrapped-adjacent"), 

38 pytest.param([ALICE, "or", [BOB, "or"]], id="nested-trailing-operator"), 

39 pytest.param([ALICE, "and", True], id="boolean-operand"), 

40 pytest.param([42], id="numeric-operand"), 

41 pytest.param([None], id="null-operand"), 

42] 

43 

44VALID_QUERIES = [ 

45 pytest.param(ALICE, ["A"], id="single-object"), 

46 pytest.param([ALICE], ["A"], id="single-array"), 

47 pytest.param([], ["A", "B"], id="empty-array"), 

48 pytest.param([ALICE, "and", THREE_CREDITS], ["A"], id="and"), 

49 pytest.param([ALICE, "or", BOB], ["A", "B"], id="or"), 

50 pytest.param([[ALICE, "or", BOB], "and", THREE_CREDITS], ["A"], id="nested"), 

51 pytest.param([ALICE, "or", BOB, "and", FOUR_CREDITS], ["B"], id="left-to-right"), 

52 pytest.param([ALICE, "or", [BOB, "and", FOUR_CREDITS]], ["A", "B"], id="grouped"), 

53 pytest.param([[], "and", ALICE], ["A"], id="empty-and"), 

54 pytest.param([[], "or", ALICE], ["A", "B"], id="empty-or"), 

55 pytest.param([[]], ["A", "B"], id="nested-empty"), 

56 pytest.param( 

57 [{"row_field": "lecturers.label", "matcher": "^A", "regex_match": True}, "or", BOB], 

58 ["A", "B"], 

59 id="regex", 

60 ), 

61] 

62 

63 

64def nested_query(depth): 

65 body = ALICE 

66 for _ in range(depth): 

67 body = [body] 

68 return body 

69 

70 

71def wide_query(conditions, operand=ALICE): 

72 body = [operand] 

73 for _ in range(conditions - 1): 

74 body.extend(["or", operand]) 

75 return body 

76 

77 

78LIMIT_QUERIES = [ 

79 pytest.param(nested_query(32), True, id="depth-at-limit"), 

80 pytest.param(nested_query(33), False, id="depth-over-limit"), 

81 pytest.param(nested_query(256), False, id="depth-beyond-recursive-parser-limit"), 

82 pytest.param(wide_query(512), True, id="1024-nodes"), 

83 pytest.param([wide_query(512)], False, id="1025-nodes"), 

84 pytest.param(wide_query(513), False, id="1026-nodes"), 

85 pytest.param(wide_query(512, []), True, id="1024-array-and-operator-nodes"), 

86 pytest.param([wide_query(512, [])], False, id="1025-array-and-operator-nodes"), 

87 pytest.param([wide_query(255), "and", wide_query(256)], True, id="1024-distributed-nodes"), 

88 pytest.param([wide_query(256), "and", wide_query(256)], False, id="1026-distributed-nodes"), 

89] 

90 

91 

92@pytest.fixture 

93def courses(): 

94 return [ 

95 make_course( 

96 {"id": f"11510{course_id}", "lecturers": [{"label": teacher}], "credits": credit} 

97 ) 

98 for course_id, teacher, credit in [("A", "Alice", 3), ("B", "Bob", 4)] 

99 ] 

100 

101 

102@pytest.mark.parametrize("body", INVALID_QUERIES) 

103def test_schema_rejects_invalid_grammar(body): 

104 with pytest.raises(ValidationError): 

105 CourseQueryCondition.model_validate(body) 

106 

107 

108@pytest.mark.parametrize("body,expected_ids", VALID_QUERIES) 

109def test_schema_preserves_legal_query_semantics(body, expected_ids, courses): 

110 schema = CourseCondition if isinstance(body, dict) else CourseQueryCondition 

111 parsed = schema.model_validate(body) 

112 conditions = Conditions(list_build_target=[parsed.model_dump(mode="json")]) 

113 assert [course.id[5:] for course in courses if conditions.accept(course)] == expected_ids 

114 

115 

116@pytest.mark.parametrize("body,accepted", LIMIT_QUERIES) 

117def test_schema_enforces_exact_tree_limits(body, accepted): 

118 assert COURSE_QUERY_MAX_DEPTH == 32 

119 assert COURSE_QUERY_MAX_NODES == 1024 

120 if accepted: 

121 assert ( 

122 CourseQueryCondition.model_validate(body).model_dump(mode="json", exclude_unset=True) 

123 == body 

124 ) 

125 else: 

126 with pytest.raises(ValidationError, match="cannot exceed"): 

127 CourseQueryCondition.model_validate(body) 

128 

129 

130def test_limits_include_prevalidated_nested_models(): 

131 wide = CourseQueryCondition.model_validate(wide_query(512)) 

132 with pytest.raises(ValidationError, match="1024 nodes"): 

133 CourseQueryCondition.model_validate([wide]) 

134 deep = CourseQueryCondition.model_validate(nested_query(32)) 

135 with pytest.raises(ValidationError, match="32 array levels"): 

136 CourseQueryCondition.model_validate([deep]) 

137 

138 

139def test_schema_accepts_prevalidated_conditions_and_nested_models(): 

140 condition = CourseCondition.model_validate(ALICE) 

141 nested = CourseQueryCondition.model_validate([condition]) 

142 parsed = CourseQueryCondition.model_validate([nested, "or", condition]) 

143 assert parsed.model_dump(mode="json") == [ 

144 [ALICE | {"regex_match": False}], 

145 "or", 

146 ALICE | {"regex_match": False}, 

147 ] 

148 

149 

150@pytest.mark.parametrize("path", ["/courses/query"]) 

151@pytest.mark.parametrize("empty_data", [False, True]) 

152class TestCourseQueryGrammarEndpoints: 

153 @pytest.fixture 

154 async def client(self, monkeypatch, courses, empty_data): 

155 set_courses(monkeypatch, [] if empty_data else courses, "grammar-test-version") 

156 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: 

157 yield client 

158 

159 @pytest.mark.parametrize("body", INVALID_QUERIES) 

160 async def test_invalid_grammar_returns_422(self, client, monkeypatch, path, empty_data, body): 

161 query = Mock(side_effect=AssertionError("Invalid queries must not be evaluated")) 

162 monkeypatch.setattr(courses_service, "query", query) 

163 response = await client.post(path, json=body) 

164 assert response.status_code == 422 

165 assert isinstance(response.json()["detail"], list) 

166 assert "Traceback" not in response.text 

167 query.assert_not_called() 

168 

169 @pytest.mark.parametrize("body,expected_ids", VALID_QUERIES) 

170 async def test_legal_queries_keep_results(self, client, path, empty_data, body, expected_ids): 

171 response = await client.post(path, json=body) 

172 assert response.status_code == 200 

173 expected_ids = [] if empty_data else expected_ids 

174 assert [course["id"][5:] for course in response.json()] == expected_ids 

175 assert response.headers["X-Total-Count"] == str(len(expected_ids)) 

176 assert response.headers["X-Data-Commit-Hash"] == "grammar-test-version" 

177 

178 @pytest.mark.parametrize("body,accepted", LIMIT_QUERIES) 

179 async def test_tree_limits_return_422(self, client, path, empty_data, body, accepted): 

180 response = await client.post(path, json=body) 

181 assert response.status_code == (200 if accepted else 422) 

182 if accepted: 

183 expected_ids = ( 

184 [] if empty_data else (["A", "B"] if body == wide_query(512, []) else ["A"]) 

185 ) 

186 assert [course["id"][5:] for course in response.json()] == expected_ids 

187 assert response.headers["X-Total-Count"] == str(len(expected_ids)) 

188 else: 

189 assert any("cannot exceed" in error["msg"] for error in response.json()["detail"]) 

190 

191 

192@pytest.mark.parametrize("path", ["/courses/query"]) 

193def test_openapi_documents_grammar_validation(path): 

194 operation = app.openapi()["paths"][path]["post"] 

195 assert "grammar" in operation["responses"]["422"]["description"] 

196 schema = app.openapi()["components"]["schemas"]["CourseQueryCondition"] 

197 assert "32 array levels and 1024 nodes" in schema["description"]