gabriel / musehub public
test_musehub_pagination.py python
385 lines 12.7 KB
Raw
c0f0b481 release: merge dev → main (#5) Gabriel Cardona <cgcardona@gmail.com> 5d ago
1 """Tests for RFC 8288 Link header pagination on MuseHub list endpoints.
2
3 Covers acceptance criteria:
4 - PaginationParams dependency parses page/per_page and cursor/limit query params
5 - build_link_header emits correct RFC 8288 rel links for first/last/prev/next
6 - build_cursor_link_header emits a rel="next" link with cursor and limit
7 - paginate_list slices correctly and returns accurate total
8 - GET /repos/{repo_id}/issues returns Link header and total field
9 - GET /repos/{repo_id}/pull-requests returns Link header and total field
10 - GET /repos/{repo_id}/commits returns Link header when per_page > 0
11 - GET /musehub/repos returns rel="next" Link header when next_cursor is present
12
13 All tests use fixtures from conftest.py. No live external APIs are called.
14 """
15 from __future__ import annotations
16
17 import pytest
18 from httpx import AsyncClient
19 from sqlalchemy.ext.asyncio import AsyncSession
20 from starlette.requests import Request as StarletteRequest
21
22 from musehub.api.routes.musehub.pagination import (
23 PaginationParams,
24 build_cursor_link_header,
25 build_link_header,
26 paginate_list,
27 )
28
29
30 # ---------------------------------------------------------------------------
31 # Unit tests — pagination helpers
32 # ---------------------------------------------------------------------------
33
34
35 def _make_request(url: str) -> StarletteRequest:
36 """Build a minimal Starlette Request for testing URL construction."""
37 scope = {
38 "type": "http",
39 "method": "GET",
40 "path": url.split("?")[0],
41 "query_string": url.split("?")[1].encode() if "?" in url else b"",
42 "headers": [],
43 }
44 return StarletteRequest(scope)
45
46
47 def test_paginate_list_first_page() -> None:
48 """paginate_list returns the first page slice and correct total."""
49 items = list(range(55))
50 page, total = paginate_list(items, page=1, per_page=20)
51 assert total == 55
52 assert page == list(range(20))
53
54
55 def test_paginate_list_middle_page() -> None:
56 """paginate_list returns the correct middle page slice."""
57 items = list(range(55))
58 page, total = paginate_list(items, page=2, per_page=20)
59 assert total == 55
60 assert page == list(range(20, 40))
61
62
63 def test_paginate_list_last_partial_page() -> None:
64 """paginate_list returns a partial slice on the final page."""
65 items = list(range(55))
66 page, total = paginate_list(items, page=3, per_page=20)
67 assert total == 55
68 assert page == list(range(40, 55))
69
70
71 def test_paginate_list_beyond_last_page_returns_empty() -> None:
72 """paginate_list returns an empty slice when page exceeds total."""
73 items = list(range(10))
74 page, total = paginate_list(items, page=5, per_page=10)
75 assert total == 10
76 assert page == []
77
78
79 def test_paginate_list_empty_input() -> None:
80 """paginate_list handles empty input gracefully."""
81 page_items: list[int]
82 page_items, total = paginate_list([], page=1, per_page=20)
83 assert total == 0
84 assert page_items == []
85
86
87 def test_build_link_header_single_page() -> None:
88 """build_link_header emits only first and last when there is exactly one page."""
89 req = _make_request("http://test/api/v1/repos/r1/issues?page=1&per_page=20")
90 header = build_link_header(req, total=5, page=1, per_page=20)
91 assert 'rel="first"' in header
92 assert 'rel="last"' in header
93 assert 'rel="next"' not in header
94 assert 'rel="prev"' not in header
95
96
97 def test_build_link_header_first_of_many() -> None:
98 """build_link_header emits first, last, and next (but not prev) on page 1 of N."""
99 req = _make_request("http://test/api/v1/repos/r1/issues?page=1&per_page=10")
100 header = build_link_header(req, total=55, page=1, per_page=10)
101 assert 'rel="first"' in header
102 assert 'rel="last"' in header
103 assert 'rel="next"' in header
104 assert 'rel="prev"' not in header
105 assert "page=2" in header
106 assert "page=6" in header # last page for 55 items at 10/page
107
108
109 def test_build_link_header_middle_page() -> None:
110 """build_link_header emits all four rels on an interior page."""
111 req = _make_request("http://test/api/v1/repos/r1/issues?page=3&per_page=10")
112 header = build_link_header(req, total=55, page=3, per_page=10)
113 assert 'rel="first"' in header
114 assert 'rel="last"' in header
115 assert 'rel="next"' in header
116 assert 'rel="prev"' in header
117 assert "page=4" in header
118 assert "page=2" in header
119
120
121 def test_build_link_header_last_page() -> None:
122 """build_link_header emits prev (but not next) on the last page."""
123 req = _make_request("http://test/api/v1/repos/r1/issues?page=6&per_page=10")
124 header = build_link_header(req, total=55, page=6, per_page=10)
125 assert 'rel="first"' in header
126 assert 'rel="last"' in header
127 assert 'rel="prev"' in header
128 assert 'rel="next"' not in header
129
130
131 def test_build_link_header_preserves_existing_query_params() -> None:
132 """build_link_header keeps non-pagination query params on generated URLs."""
133 req = _make_request("http://test/api/v1/repos/r1/issues?state=open&page=1&per_page=10")
134 header = build_link_header(req, total=30, page=1, per_page=10)
135 assert "state=open" in header
136
137
138 def test_build_cursor_link_header_emits_next_only() -> None:
139 """build_cursor_link_header emits exactly one rel="next" with cursor and limit encoded."""
140 req = _make_request("http://test/api/v1/repos?limit=20")
141 header = build_cursor_link_header(req, next_cursor="abc123", limit=20)
142 assert 'rel="next"' in header
143 assert "cursor=abc123" in header
144 assert "limit=20" in header
145 assert 'rel="prev"' not in header
146 assert 'rel="first"' not in header
147
148
149 # ---------------------------------------------------------------------------
150 # Integration tests — issues list endpoint
151 # ---------------------------------------------------------------------------
152
153
154 async def _create_repo(client: AsyncClient, auth_headers: dict[str, str], name: str) -> str:
155 r = await client.post(
156 "/api/v1/repos",
157 json={"name": name, "owner": "testuser"},
158 headers=auth_headers,
159 )
160 assert r.status_code == 201
161 repo_id: str = r.json()["repoId"]
162 return repo_id
163
164
165 async def _create_issue(
166 client: AsyncClient,
167 auth_headers: dict[str, str],
168 repo_id: str,
169 title: str,
170 ) -> None:
171 r = await client.post(
172 f"/api/v1/repos/{repo_id}/issues",
173 json={"title": title, "body": ""},
174 headers=auth_headers,
175 )
176 assert r.status_code == 201
177
178
179 @pytest.mark.anyio
180 async def test_list_issues_link_header_present(
181 client: AsyncClient,
182 auth_headers: dict[str, str],
183 ) -> None:
184 """GET /issues includes a Link header when pagination is active."""
185 repo_id = await _create_repo(client, auth_headers, "pagination-issues-link")
186 for i in range(5):
187 await _create_issue(client, auth_headers, repo_id, f"Issue {i}")
188
189 r = await client.get(
190 f"/api/v1/repos/{repo_id}/issues?page=1&per_page=2",
191 headers=auth_headers,
192 )
193 assert r.status_code == 200
194 assert "Link" in r.headers
195 link = r.headers["Link"]
196 assert 'rel="first"' in link
197 assert 'rel="last"' in link
198 assert 'rel="next"' in link
199
200
201 @pytest.mark.anyio
202 async def test_list_issues_total_field_returned(
203 client: AsyncClient,
204 auth_headers: dict[str, str],
205 ) -> None:
206 """GET /issues response body includes ``total`` with the count across all pages."""
207 repo_id = await _create_repo(client, auth_headers, "pagination-issues-total")
208 for i in range(7):
209 await _create_issue(client, auth_headers, repo_id, f"Track issue {i}")
210
211 r = await client.get(
212 f"/api/v1/repos/{repo_id}/issues?page=1&per_page=3",
213 headers=auth_headers,
214 )
215 assert r.status_code == 200
216 body = r.json()
217 assert body["total"] == 7
218 assert len(body["issues"]) == 3
219
220
221 @pytest.mark.anyio
222 async def test_list_issues_last_page_no_next(
223 client: AsyncClient,
224 auth_headers: dict[str, str],
225 ) -> None:
226 """GET /issues Link header on the last page has no rel=\"next\"."""
227 repo_id = await _create_repo(client, auth_headers, "pagination-issues-last")
228 for i in range(4):
229 await _create_issue(client, auth_headers, repo_id, f"Issue {i}")
230
231 r = await client.get(
232 f"/api/v1/repos/{repo_id}/issues?page=2&per_page=3",
233 headers=auth_headers,
234 )
235 assert r.status_code == 200
236 link = r.headers["Link"]
237 assert 'rel="next"' not in link
238 assert 'rel="prev"' in link
239
240
241 @pytest.mark.anyio
242 async def test_list_issues_default_page_returns_all_when_small(
243 client: AsyncClient,
244 auth_headers: dict[str, str],
245 ) -> None:
246 """GET /issues with no pagination params returns results on page 1 (default)."""
247 repo_id = await _create_repo(client, auth_headers, "pagination-issues-default")
248 for i in range(3):
249 await _create_issue(client, auth_headers, repo_id, f"Default page {i}")
250
251 r = await client.get(
252 f"/api/v1/repos/{repo_id}/issues",
253 headers=auth_headers,
254 )
255 assert r.status_code == 200
256 body = r.json()
257 assert len(body["issues"]) == 3
258 assert body["total"] == 3
259
260
261 # ---------------------------------------------------------------------------
262 # Integration tests — PRs list endpoint
263 # ---------------------------------------------------------------------------
264
265
266
267 @pytest.mark.anyio
268 async def test_list_prs_link_header_present(
269 client: AsyncClient,
270 auth_headers: dict[str, str],
271 db_session: AsyncSession,
272 ) -> None:
273 """GET /pull-requests includes a Link header when pagination is active."""
274 from musehub.db.musehub_models import MusehubPullRequest
275
276 repo_id = await _create_repo(client, auth_headers, "pagination-prs-link")
277
278 # Insert PRs directly to avoid branch validation complexity in tests
279 for i in range(3):
280 db_session.add(MusehubPullRequest(
281 repo_id=repo_id,
282 title=f"PR {i}",
283 from_branch=f"feat/{i}",
284 to_branch="main",
285 author="testuser",
286 ))
287 await db_session.commit()
288
289 r = await client.get(
290 f"/api/v1/repos/{repo_id}/pull-requests?page=1&per_page=2",
291 headers=auth_headers,
292 )
293 assert r.status_code == 200
294 assert "Link" in r.headers
295 link = r.headers["Link"]
296 assert 'rel="first"' in link
297 assert 'rel="next"' in link
298
299
300 @pytest.mark.anyio
301 async def test_list_prs_total_field_returned(
302 client: AsyncClient,
303 auth_headers: dict[str, str],
304 db_session: AsyncSession,
305 ) -> None:
306 """GET /pull-requests response body includes ``total`` field."""
307 from musehub.db.musehub_models import MusehubPullRequest
308
309 repo_id = await _create_repo(client, auth_headers, "pagination-prs-total")
310
311 for i in range(4):
312 db_session.add(MusehubPullRequest(
313 repo_id=repo_id,
314 title=f"PR {i}",
315 from_branch=f"feat/{i}",
316 to_branch="main",
317 author="testuser",
318 ))
319 await db_session.commit()
320
321 r = await client.get(
322 f"/api/v1/repos/{repo_id}/pull-requests",
323 headers=auth_headers,
324 )
325 assert r.status_code == 200
326 body = r.json()
327 assert "total" in body
328 assert body["total"] == 4
329
330
331 # ---------------------------------------------------------------------------
332 # Integration tests — commits list endpoint
333 # ---------------------------------------------------------------------------
334
335
336 @pytest.mark.anyio
337 async def test_list_commits_link_header_with_per_page(
338 client: AsyncClient,
339 auth_headers: dict[str, str],
340 db_session: AsyncSession,
341 ) -> None:
342 """GET /commits with per_page > 0 includes an RFC 8288 Link header."""
343 from datetime import datetime, timezone, timedelta
344 from musehub.db.musehub_models import MusehubCommit
345
346 repo_id = await _create_repo(client, auth_headers, "pagination-commits-link")
347 now = datetime.now(tz=timezone.utc)
348
349 for i in range(5):
350 db_session.add(MusehubCommit(
351 commit_id=f"sha-{i:04d}",
352 repo_id=repo_id,
353 branch="main",
354 parent_ids=[],
355 message=f"Commit {i}",
356 author="testuser",
357 timestamp=now + timedelta(seconds=i),
358 ))
359 await db_session.commit()
360
361 r = await client.get(
362 f"/api/v1/repos/{repo_id}/commits?page=1&per_page=2",
363 headers=auth_headers,
364 )
365 assert r.status_code == 200
366 assert "Link" in r.headers
367 link = r.headers["Link"]
368 assert 'rel="first"' in link
369 assert 'rel="next"' in link
370
371
372 @pytest.mark.anyio
373 async def test_list_commits_no_link_header_without_per_page(
374 client: AsyncClient,
375 auth_headers: dict[str, str],
376 ) -> None:
377 """GET /commits without per_page does NOT add a Link header (legacy mode)."""
378 repo_id = await _create_repo(client, auth_headers, "pagination-commits-no-link")
379
380 r = await client.get(
381 f"/api/v1/repos/{repo_id}/commits",
382 headers=auth_headers,
383 )
384 assert r.status_code == 200
385 assert "Link" not in r.headers