File size: 4,141 Bytes
8b96826 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """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}
|