Files
python-github-backup/tests/test_pagination.py
Rodos 46140b0ff1 Fix retry logic for HTTP 5xx errors and network failures
Refactors error handling to retry all 5xx errors (not just 502), network errors (URLError, socket.error, IncompleteRead), and JSON parse errors with exponential backoff and jitter. Respects retry-after and rate limit headers per GitHub API requirements. Consolidates retry logic into make_request_with_retry() wrapper and adds clear logging for retry attempts and failures. Removes dead code from 2016 (errors list, _request_http_error, _request_url_error) that was intentionally disabled in commit 1e5a9048 to fix #29.

Fixes #140, #110, #138
2025-12-16 21:55:47 +11:00

148 lines
4.7 KiB
Python

"""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."""
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 = github_backup.retrieve_data(
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 = github_backup.retrieve_data(
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 = github_backup.retrieve_data(
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