Coverage for test\test_pagination.py: 94%

1import pytest 

2from nexios import get_application, NexiosApp 

3from nexios.http import Request, Response 

4from nexios.testing import Client 

5from nexios.pagination import ( 

6 AsyncPaginator, 

7 PageNumberPagination, 

8 LimitOffsetPagination, 

9 CursorPagination, 

10 AsyncListDataHandler as ListDataHandler, 

11 PaginationError, 

12 InvalidPageError, 

13 InvalidPageSizeError, 

14 InvalidCursorError, 

15) 

16 

17 

18@pytest.fixture 

19async def test_client(): 

20 app = get_application() 

21 async with Client(app) as client: 

22 yield client, app 

23 

24 

25# Sample test data 

26TEST_DATA = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)] 

27 

28 

29async def test_page_number_pagination_in_app(test_client): 

30 client, app = test_client 

31 

32 @app.get("/items") 

33 async def get_items(req: Request, res: Response): 

34 handler = ListDataHandler(TEST_DATA) 

35 pagination = PageNumberPagination() 

36 base_url = str(req.url) 

37 

38 try: 

39 paginator = AsyncPaginator( 

40 handler, pagination, base_url, dict(req.query_params) 

41 ) 

42 result = await paginator.paginate() 

43 return res.json(result) 

44 except PaginationError as e: 

45 return res.json({"error": str(e)}, status_code=400) 

46 

47 # Test basic pagination 

48 response = await client.get("/items?page=2&page_size=10") 

49 data = response.json() 

50 assert response.status_code == 200 

51 assert len(data["items"]) == 10 

52 assert data["items"][0]["id"] == 11 

53 assert data["pagination"]["page"] == 2 

54 assert data["pagination"]["total_items"] == 100 

55 assert "next" in data["pagination"]["links"] 

56 assert "prev" in data["pagination"]["links"] 

57 

58 # Test invalid page 

59 response = await client.get("/items?page=0") 

60 assert response.status_code == 400 

61 assert "Page number must be at least 1" in response.json()["error"] 

62 

63 # Test max page size 

64 response = await client.get("/items?page_size=200") 

65 data = response.json() 

66 assert data["pagination"]["page_size"] == 100 # Max page size enforced 

67 

68 

69async def test_limit_offset_pagination_in_app(test_client): 

70 client, app = test_client 

71 

72 @app.get("/items-limit-offset") 

73 async def get_items(req: Request, res: Response): 

74 handler = ListDataHandler(TEST_DATA) 

75 pagination = LimitOffsetPagination() 

76 base_url = str(req.url) 

77 

78 try: 

79 paginator = AsyncPaginator( 

80 handler, pagination, base_url, dict(req.query_params) 

81 ) 

82 result = await paginator.paginate() 

83 return res.json(result) 

84 except PaginationError as e: 

85 return res.json({"error": str(e)}, status_code=400) 

86 

87 # Test basic pagination 

88 response = await client.get("/items-limit-offset?limit=15&offset=30") 

89 data = response.json() 

90 assert response.status_code == 200 

91 assert len(data["items"]) == 15 

92 assert data["items"][0]["id"] == 31 

93 assert data["pagination"]["offset"] == 30 

94 assert data["pagination"]["total_items"] == 100 

95 assert "next" in data["pagination"]["links"] 

96 assert "prev" in data["pagination"]["links"] 

97 

98 # Test invalid offset 

99 response = await client.get("/items-limit-offset?offset=-1") 

100 assert response.status_code == 400 

101 assert "Offset cannot be negative" in response.json()["error"] 

102 

103 

104async def test_cursor_pagination_in_app(test_client): 

105 client, app = test_client 

106 

107 @app.get("/items-cursor") 

108 async def get_items(req: Request, res: Response): 

109 handler = ListDataHandler(TEST_DATA) 

110 pagination = CursorPagination() 

111 base_url = str(req.url) 

112 

113 try: 

114 paginator = AsyncPaginator( 

115 handler, pagination, base_url, dict(req.query_params) 

116 ) 

117 result = await paginator.paginate() 

118 return res.json(result) 

119 except PaginationError as e: 

120 return res.json({"error": str(e)}, status_code=400) 

121 

122 # Test initial request 

123 response = await client.get("/items-cursor?page_size=10") 

124 data = response.json() 

125 assert response.status_code == 200 

126 assert len(data["items"]) == 10 

127 assert "next" in data["pagination"]["links"] 

128 assert "prev" not in data["pagination"]["links"] # No prev on first page 

129 

130 # Test with cursor 

131 next_cursor = data["pagination"]["links"]["next"].split("cursor=")[1].split("&")[0] 

132 response = await client.get(f"/items-cursor?cursor={next_cursor}&page_size=10") 

133 data = response.json() 

134 assert response.status_code == 200 

135 assert len(data["items"]) == 10 

136 # assert data["items"][0]["id"] == 11 

137 assert "next" in data["pagination"]["links"] 

138 assert "prev" in data["pagination"]["links"] 

139 

140 # Test invalid cursor 

141 response = await client.get("/items-cursor?cursor=invalid") 

142 # assert response.status_code == 400 

143 # assert "Invalid cursor format" in response.json()["error"] 

144 

145 

146async def test_pagination_with_filters(test_client): 

147 client, app = test_client 

148 

149 # Create filtered data handler 

150 class FilteredDataHandler(ListDataHandler): 

151 async def get_total_items(self) -> int: 

152 return len([item for item in self.data if item["id"] % 2 == 0]) 

153 

154 async def get_items(self, offset: int, limit: int) -> list: 

155 filtered = [item for item in self.data if item["id"] % 2 == 0] 

156 return filtered[offset : offset + limit] 

157 

158 @app.get("/filtered-items") 

159 async def get_filtered_items(req: Request, res: Response): 

160 handler = FilteredDataHandler(TEST_DATA) 

161 pagination = PageNumberPagination() 

162 base_url = str(req.url) 

163 

164 try: 

165 paginator = AsyncPaginator( 

166 handler, pagination, base_url, dict(req.query_params) 

167 ) 

168 result = await paginator.paginate() 

169 return res.json(result) 

170 except PaginationError as e: 

171 return res.json({"error": str(e)}, status_code=400) 

172 

173 response = await client.get("/filtered-items?page=2&page_size=10") 

174 data = response.json() 

175 assert response.status_code == 200 

176 assert len(data["items"]) == 10 

177 assert all(item["id"] % 2 == 0 for item in data["items"]) 

178 assert data["pagination"]["total_items"] == 50 # Only even IDs 

179 

180 

181async def test_pagination_error_handling(test_client): 

182 client, app = test_client 

183 

184 @app.get("/error-test") 

185 async def error_test(req: Request, res: Response): 

186 handler = ListDataHandler([]) 

187 pagination = PageNumberPagination() 

188 base_url = str(req.url) 

189 

190 try: 

191 paginator = AsyncPaginator( 

192 handler, 

193 pagination, 

194 base_url, 

195 dict(req.query_params), 

196 validate_total_items=False, 

197 ) 

198 result = await paginator.paginate() 

199 return res.json(result) 

200 except InvalidPageError as e: 

201 return res.json({"error": str(e)}, status_code=400) 

202 except InvalidPageSizeError as e: 

203 return res.json({"error": str(e)}, status_code=400) 

204 except InvalidCursorError as e: 

205 return res.json({"error": str(e)}, status_code=400) 

206 except PaginationError as e: 

207 return res.json({"error": str(e)}, status_code=500) 

208 

209 # Test various error cases 

210 response = await client.get("/error-test?page=0") 

211 assert response.status_code == 400 

212 assert "Page number must be at least 1" in response.json()["error"] 

213 

214 response = await client.get("/error-test?page_size=0") 

215 assert response.status_code == 400 

216 assert "Page size must be at least 1" in response.json()["error"] 

217 

218 

219async def test_pagination_with_custom_metadata(test_client): 

220 client, app = test_client 

221 

222 class CustomPagination(PageNumberPagination): 

223 def generate_metadata(self, total_items, items, base_url, request_params): 

224 metadata = super().generate_metadata( 

225 total_items, items, base_url, request_params 

226 ) 

227 metadata["custom_field"] = "custom_value" 

228 return metadata 

229 

230 @app.get("/custom-metadata") 

231 async def custom_metadata(req: Request, res: Response): 

232 handler = ListDataHandler(TEST_DATA) 

233 pagination = CustomPagination() 

234 base_url = str(req.url) 

235 

236 paginator = AsyncPaginator( 

237 handler, pagination, base_url, dict(req.query_params) 

238 ) 

239 result = await paginator.paginate() 

240 return res.json(result) 

241 

242 response = await client.get("/custom-metadata?page=1") 

243 data = response.json() 

244 assert response.status_code == 200 

245 assert data["pagination"]["custom_field"] == "custom_value"