Coverage for tests/test_courses.py: 100%

1"""Course REST query regression tests using normalized offline snapshots.""" 

2 

3import pytest 

4from conftest import make_course, set_courses 

5from httpx import ASGITransport, AsyncClient 

6 

7from data_api.api.api import app 

8from data_api.models.courses import BOOLEAN_FIELDS, INTEGER_FIELDS, CourseFieldName 

9 

10INVALID_REGEX_PATTERNS = ["[", "(", "\\", "*", "(?invalid)", "a{99999999999999999999}", "(" * 1000] 

11 

12 

13@pytest.fixture 

14def courses(): 

15 return [ 

16 make_course( 

17 id=f"11510{identity}", 

18 chinese_title=title, 

19 english_title=english, 

20 lecturers=[{"label": teacher}], 

21 credits=credits, 

22 note=note, 

23 language_code="en" if identity == "B" else (None if identity == "C" else "zh"), 

24 ) 

25 for identity, teacher, title, english, credits, note in [ 

26 ("A", "林福仁", "服務學習", "C++ [AI]", 0.5, "X-Class"), 

27 ("B", "林福仁", "程式設計", "Programming", 3, ""), 

28 ("C", "黃老師", "服務設計", "Design", 1.5, ""), 

29 ("D", "孫老師", "跨域學習", "Learning", 2, "X-Class"), 

30 ] 

31 ] 

32 

33 

34@pytest.fixture 

35async def client(monkeypatch, courses): 

36 set_courses(monkeypatch, courses) 

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

38 yield client 

39 

40 

41def test_course_search_query_model_keeps_flat_optional_parameters(): 

42 operation = app.openapi()["paths"]["/courses"]["get"] 

43 assert "requestBody" not in operation 

44 parameters = {parameter["name"]: parameter for parameter in operation["parameters"]} 

45 assert set(parameters) == {field.value for field in CourseFieldName} | {"type"} 

46 assert all(p["in"] == "query" and not p["required"] for p in parameters.values()) 

47 

48 

49@pytest.mark.parametrize("field", [field.value for field in CourseFieldName]) 

50async def test_every_get_field_is_accepted(client, field): 

51 value = ( 

52 "false" 

53 if field in BOOLEAN_FIELDS 

54 else ( 

55 "0" 

56 if field in INTEGER_FIELDS | {"credits"} 

57 else ("zh" if field == "language_code" else "中") 

58 ) 

59 ) 

60 assert (await client.get("/courses", params={field: value})).status_code == 200 

61 

62 

63@pytest.mark.parametrize( 

64 "params,ids", 

65 [ 

66 ({}, ["A", "B", "C", "D"]), 

67 ({"lecturers.label": ""}, ["A", "B", "C", "D"]), 

68 ({"lecturers.label": "林福仁"}, ["A", "B"]), 

69 ({"chinese_title": "服務"}, ["A", "C"]), 

70 ({"lecturers.label": "林福仁", "chinese_title": "服務"}, ["A"]), 

71 ({"type": "microcredits"}, ["A", "C"]), 

72 ({"type": "xclass"}, ["A", "D"]), 

73 ({"type": "xclass", "lecturers.label": "林福仁"}, ["A"]), 

74 ({"type": "microcredits", "lecturers.label": "林福仁"}, ["A"]), 

75 ({"type": "xclass", "lecturers.label": "黃"}, []), 

76 ({"english_title": r"^C\+\+", "id": "^11510A"}, ["A"]), 

77 ({"language_code": "zh|en"}, ["A", "B", "D"]), 

78 ({"language_code": "^en$"}, ["B"]), 

79 ({"language_code": "中"}, []), 

80 ({"credits": "3", "is_cancelled": "false", "reserved_for_freshmen": "0"}, ["B"]), 

81 ({"is_cancelled": "true"}, []), 

82 ], 

83) 

84async def test_get_collection_filters(client, params, ids): 

85 response = await client.get("/courses", params=params) 

86 assert response.status_code == 200, response.text 

87 assert [course["id"] for course in response.json()] == [f"11510{i}" for i in ids] 

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

89 assert response.headers["X-Data-Commit-Hash"] == "course-test-version" 

90 assert not response.history 

91 

92 

93@pytest.mark.parametrize( 

94 "params", 

95 [ 

96 {"type": "unknown"}, 

97 {"type": ""}, 

98 {"type": "XCLASS"}, 

99 {"credits": "bad"}, 

100 {"capacity": "1.5"}, 

101 {"is_cancelled": "bad"}, 

102 {"teacher": "old"}, 

103 {"credit": "3"}, 

104 ], 

105) 

106async def test_invalid_get_filters(client, params): 

107 assert (await client.get("/courses", params=params)).status_code == 422 

108 

109 

110@pytest.mark.parametrize("empty", [False, True]) 

111@pytest.mark.parametrize("pattern", INVALID_REGEX_PATTERNS) 

112@pytest.mark.parametrize("field", ["chinese_title", "lecturers.label", "id", "language_code"]) 

113async def test_get_invalid_regex(client, monkeypatch, empty, pattern, field): 

114 if empty: 

115 set_courses(monkeypatch, []) 

116 response = await client.get("/courses", params={field: pattern}) 

117 assert response.status_code == 422 

118 assert response.json() == {"detail": f"Invalid regular expression for course field '{field}'."} 

119 

120 

121@pytest.mark.parametrize("matcher", ["C++ [AI]", "C++", "["]) 

122async def test_post_exact_text(client, matcher): 

123 response = await client.post( 

124 "/courses/query", json={"row_field": "english_title", "matcher": matcher} 

125 ) 

126 assert response.status_code == 200 

127 assert [course["id"] for course in response.json()] == ( 

128 ["11510A"] if matcher == "C++ [AI]" else [] 

129 ) 

130 

131 

132@pytest.mark.parametrize("empty", [False, True]) 

133@pytest.mark.parametrize("nested", [False, True]) 

134@pytest.mark.parametrize("pattern", INVALID_REGEX_PATTERNS) 

135async def test_post_invalid_regex(client, monkeypatch, empty, nested, pattern): 

136 if empty: 

137 set_courses(monkeypatch, []) 

138 body = {"row_field": "lecturers.label", "matcher": pattern, "regex_match": True} 

139 if nested: 

140 body = [{"row_field": "id", "matcher": "absent"}, "and", [body]] 

141 response = await client.post("/courses/query", json=body) 

142 assert response.status_code == 422 

143 assert any("Invalid regular expression" in e["msg"] for e in response.json()["detail"]) 

144 

145 

146async def test_post_nested_regex_and_typed_matchers(client): 

147 response = await client.post( 

148 "/courses/query", 

149 json=[ 

150 {"row_field": "is_cancelled", "matcher": False}, 

151 "and", 

152 [ 

153 {"row_field": "credits", "matcher": 0.5}, 

154 "or", 

155 {"row_field": "lecturers.label", "matcher": "黃", "regex_match": True}, 

156 ], 

157 ], 

158 ) 

159 assert response.status_code == 200 

160 assert [c["id"] for c in response.json()] == ["11510A", "11510C"] 

161 assert response.headers["X-Total-Count"] == "2" 

162 assert response.headers["X-Data-Commit-Hash"] == "course-test-version" 

163 

164 

165async def test_large_integer_credit_matcher_does_not_overflow(client): 

166 response = await client.post( 

167 "/courses/query", json={"row_field": "credits", "matcher": 10**400} 

168 ) 

169 assert response.status_code == 200 

170 assert response.json() == [] 

171 

172 

173async def test_response_is_full_normalized_shape_and_nullable(client, monkeypatch): 

174 course = make_course(credits=None, capacity_unlimited=None, language_code=None) 

175 set_courses(monkeypatch, [course], version=None) 

176 response = await client.get("/courses") 

177 assert response.json() == [course.model_dump(mode="json")] 

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

179 assert ( 

180 await client.post("/courses/query", json={"row_field": "credits", "matcher": None}) 

181 ).json() == response.json() 

182 assert (await client.get("/courses?type=microcredits")).json() == []