Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
153 changes: 153 additions & 0 deletions tests/test_pagination.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
"""Tests for Link header pagination handling."""

import json
from unittest.mock import Mock, patch

import pytest

from github_backup import github_backup


class MockHTTPResponse:
"""Mock HTTP response for paginated API calls."""

def __init__(self, data, link_header=None):
self._content = json.dumps(data).encode("utf-8")
self._link_header = link_header
self._read = False
self.reason = "OK"

def getcode(self):
return 200

def read(self):
if self._read:
return b""
self._read = True
return self._content

def get_header(self, name, default=None):
"""Mock method for headers.get()."""
return self.headers.get(name, default)

@property
def headers(self):
headers = {"x-ratelimit-remaining": "5000"}
if self._link_header:
headers["Link"] = self._link_header
return headers


@pytest.fixture
def mock_args():
"""Mock args for retrieve_data_gen."""
args = Mock()
args.as_app = False
args.token_fine = None
args.token_classic = "fake_token"
args.username = None
args.password = None
args.osx_keychain_item_name = None
args.osx_keychain_item_account = None
args.throttle_limit = None
args.throttle_pause = 0
return args


def test_cursor_based_pagination(mock_args):
"""Link header with 'after' cursor parameter works correctly."""

# Simulate issues endpoint behavior: returns cursor in Link header
responses = [
# Issues endpoint returns 'after' cursor parameter (not 'page')
MockHTTPResponse(
data=[{"issue": i} for i in range(1, 101)], # Page 1 contents
link_header='<https://api.github.com/repos/owner/repo/issues?per_page=100&after=ABC123&page=2>; rel="next"',
),
MockHTTPResponse(
data=[{"issue": i} for i in range(101, 151)], # Page 2 contents
link_header=None, # No Link header - signals end of pagination
),
]
requests_made = []

def mock_urlopen(request, *args, **kwargs):
url = request.get_full_url()
requests_made.append(url)
return responses[len(requests_made) - 1]

with patch("github_backup.github_backup.urlopen", side_effect=mock_urlopen):
results = list(
github_backup.retrieve_data_gen(
mock_args, "https://api.github.com/repos/owner/repo/issues"
)
)

# Verify all items retrieved and cursor was used in second request
assert len(results) == 150
assert len(requests_made) == 2
assert "after=ABC123" in requests_made[1]


def test_page_based_pagination(mock_args):
"""Link header with 'page' parameter works correctly."""

# Simulate pulls/repos endpoint behavior: returns page numbers in Link header
responses = [
# Pulls endpoint uses traditional 'page' parameter (not cursor)
MockHTTPResponse(
data=[{"pull": i} for i in range(1, 101)], # Page 1 contents
link_header='<https://api.github.com/repos/owner/repo/pulls?per_page=100&page=2>; rel="next"',
),
MockHTTPResponse(
data=[{"pull": i} for i in range(101, 181)], # Page 2 contents
link_header=None, # No Link header - signals end of pagination
),
]
requests_made = []

def mock_urlopen(request, *args, **kwargs):
url = request.get_full_url()
requests_made.append(url)
return responses[len(requests_made) - 1]

with patch("github_backup.github_backup.urlopen", side_effect=mock_urlopen):
results = list(
github_backup.retrieve_data_gen(
mock_args, "https://api.github.com/repos/owner/repo/pulls"
)
)

# Verify all items retrieved and page parameter was used (not cursor)
assert len(results) == 180
assert len(requests_made) == 2
assert "page=2" in requests_made[1]
assert "after" not in requests_made[1]


def test_no_link_header_stops_pagination(mock_args):
"""Pagination stops when Link header is absent."""

# Simulate endpoint with results that fit in a single page
responses = [
MockHTTPResponse(
data=[{"label": i} for i in range(1, 51)], # Page contents
link_header=None, # No Link header - signals end of pagination
)
]
requests_made = []

def mock_urlopen(request, *args, **kwargs):
requests_made.append(request.get_full_url())
return responses[len(requests_made) - 1]

with patch("github_backup.github_backup.urlopen", side_effect=mock_urlopen):
results = list(
github_backup.retrieve_data_gen(
mock_args, "https://api.github.com/repos/owner/repo/labels"
)
)

# Verify pagination stopped after first request
assert len(results) == 50
assert len(requests_made) == 1