| """Secret scrubbing, friendly errors, and retry-on-429 behavior.""" |
|
|
| import pytest |
| import requests |
|
|
| from app import netutil |
|
|
|
|
| |
|
|
| def test_scrubs_key_param(): |
| url = "https://api.tomtom.com/route?key=2RHRTgSecret123&routeType=fastest" |
| scrubbed = netutil.scrub_secrets(url) |
| assert "2RHRTgSecret123" not in scrubbed |
| assert "key=***" in scrubbed |
| assert "routeType=fastest" in scrubbed |
|
|
|
|
| def test_scrubs_various_key_names(): |
| assert "secret" not in netutil.scrub_secrets("x?apiKey=secret") |
| assert "secret" not in netutil.scrub_secrets("x?api_key=secret") |
| assert "secret" not in netutil.scrub_secrets("x?apikey=secret&y=1") |
|
|
|
|
| def test_scrub_handles_exceptions(): |
| error = RuntimeError("429 for url: https://x.com/?key=abc123") |
| assert "abc123" not in netutil.scrub_secrets(error) |
|
|
|
|
| |
|
|
| def test_friendly_429(): |
| message = netutil.friendly_error("429 Client Error: Too Many Requests for url: x?key=abc") |
| assert "rate limit" in message.lower() |
| assert "abc" not in message |
|
|
|
|
| def test_friendly_403(): |
| assert "key" in netutil.friendly_error("403 Forbidden").lower() |
|
|
|
|
| def test_friendly_timeout(): |
| assert "timeout" in netutil.friendly_error("HTTPSConnectionPool: Read timed out").lower() |
|
|
|
|
| def test_friendly_unknown_is_scrubbed(): |
| message = netutil.friendly_error("weird failure at url?key=secret99") |
| assert "secret99" not in message |
|
|
|
|
| |
|
|
| class SeqResponse: |
| def __init__(self, status_code, payload=None): |
| self.status_code = status_code |
| self._payload = payload or {} |
| self.text = str(payload) |
|
|
| def json(self): |
| return self._payload |
|
|
| def raise_for_status(self): |
| if self.status_code >= 400: |
| raise requests.HTTPError(f"HTTP {self.status_code}", response=self) |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def no_sleep(monkeypatch): |
| monkeypatch.setattr(netutil.time, "sleep", lambda seconds: None) |
|
|
|
|
| def test_retries_on_429_then_succeeds(monkeypatch): |
| responses = [SeqResponse(429), SeqResponse(200, {"ok": True})] |
| calls = {"count": 0} |
|
|
| def fake_get(url, params=None, timeout=None): |
| response = responses[calls["count"]] |
| calls["count"] += 1 |
| return response |
|
|
| monkeypatch.setattr(netutil.requests, "get", fake_get) |
| result = netutil.get_with_retry("https://x.com", retries=3) |
| assert result.json() == {"ok": True} |
| assert calls["count"] == 2 |
|
|
|
|
| def test_raises_after_exhausting_retries(monkeypatch): |
| monkeypatch.setattr(netutil.requests, "get", lambda *a, **k: SeqResponse(429)) |
| with pytest.raises(requests.HTTPError, match="429"): |
| netutil.get_with_retry("https://x.com", retries=3) |
|
|
|
|
| def test_client_error_does_not_retry(monkeypatch): |
| calls = {"count": 0} |
|
|
| def fake_get(url, params=None, timeout=None): |
| calls["count"] += 1 |
| return SeqResponse(403) |
|
|
| monkeypatch.setattr(netutil.requests, "get", fake_get) |
| with pytest.raises(requests.HTTPError, match="403"): |
| netutil.get_with_retry("https://x.com", retries=3) |
| assert calls["count"] == 1 |
|
|
|
|
| def test_retries_on_server_error(monkeypatch): |
| responses = [SeqResponse(503), SeqResponse(200, {"ok": 1})] |
| calls = {"count": 0} |
|
|
| def fake_get(url, params=None, timeout=None): |
| response = responses[calls["count"]] |
| calls["count"] += 1 |
| return response |
|
|
| monkeypatch.setattr(netutil.requests, "get", fake_get) |
| assert netutil.get_with_retry("https://x.com").json() == {"ok": 1} |
|
|
|
|
| def test_retries_on_connection_error(monkeypatch): |
| calls = {"count": 0} |
|
|
| def fake_get(url, params=None, timeout=None): |
| calls["count"] += 1 |
| if calls["count"] == 1: |
| raise requests.ConnectionError("reset") |
| return SeqResponse(200, {"ok": 1}) |
|
|
| monkeypatch.setattr(netutil.requests, "get", fake_get) |
| assert netutil.get_with_retry("https://x.com").json() == {"ok": 1} |
|
|
