62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
"""Regression tests for the shared SpaceDevs request budget."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from agenda import thespacedevs
|
|
|
|
|
|
def test_budget_survives_separate_calls(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
request = Mock(return_value=Mock(status_code=200))
|
|
monkeypatch.setattr("agenda.thespacedevs.requests.get", request)
|
|
monkeypatch.setattr("agenda.thespacedevs.time.time", lambda: 10000.0)
|
|
for _ in range(12):
|
|
thespacedevs.api_get(str(tmp_path), "https://example.com")
|
|
with pytest.raises(thespacedevs.RateLimitDeferred):
|
|
thespacedevs.api_get(str(tmp_path), "https://example.com")
|
|
assert request.call_count == 12
|
|
monkeypatch.setattr("agenda.thespacedevs.time.time", lambda: 13601.0)
|
|
thespacedevs.api_get(str(tmp_path), "https://example.com")
|
|
assert request.call_count == 13
|
|
|
|
|
|
@pytest.mark.parametrize("headers", [{}, {"Retry-After": "1832"}])
|
|
def test_throttle_persists_cooldown(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, headers: dict[str, str]
|
|
) -> None:
|
|
request = Mock(
|
|
return_value=Mock(
|
|
status_code=429,
|
|
headers=headers,
|
|
text='{"detail": "Request was throttled. Expected available in 1832 seconds."}',
|
|
)
|
|
)
|
|
monkeypatch.setattr("agenda.thespacedevs.requests.get", request)
|
|
monkeypatch.setattr("agenda.thespacedevs.time.time", lambda: 10000.0)
|
|
with pytest.raises(thespacedevs.RateLimitDeferred):
|
|
thespacedevs.api_get(str(tmp_path), "https://example.com")
|
|
state = json.loads((tmp_path / "api_budget.json").read_text())
|
|
assert state["retry_at"] == 11833.0
|
|
monkeypatch.setattr("agenda.thespacedevs.time.time", lambda: 11000.0)
|
|
with pytest.raises(thespacedevs.RateLimitDeferred):
|
|
thespacedevs.api_get(str(tmp_path), "https://example.com")
|
|
assert request.call_count == 1
|
|
|
|
|
|
def test_incomplete_pagination_keeps_cached_flights(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
cached = [{"slug": "still-in-orbit"}]
|
|
thespacedevs.write_active_crewed_flights_cache(str(tmp_path), cached)
|
|
response = Mock(status_code=200, ok=True)
|
|
response.json.return_value = {"results": [], "next": "https://example.com/page2"}
|
|
request = Mock(side_effect=[response, thespacedevs.RateLimitDeferred()])
|
|
monkeypatch.setattr(thespacedevs, "api_get", request)
|
|
assert thespacedevs.get_active_crewed_flights(str(tmp_path), refresh=True) == cached
|
|
loaded = thespacedevs.load_active_crewed_flights_cache(str(tmp_path))
|
|
assert loaded is not None and loaded[1] == cached
|