"""Secret scrubbing, friendly errors, and retry-on-429 behavior.""" import pytest import requests from app import netutil # ---- scrub_secrets ---------------------------------------------------------- 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) # ---- friendly_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 # ---- get_with_retry ---------------------------------------------------------- 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 # no retry on a hard client error 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}