feat: AM-1 swiftly_auth module — Firebase REST anon-signUp + process cache

Replaces the static SWIFTLY_BEARER_TOKEN env-var lookup with a JIT
mint via the Firebase Identity Toolkit signUp endpoint, gated by the
firebaseApiKey published in luckysupermarkets.com/config.json.

- get_token(): returns cached JWT if exp > now+300s, else mints
- mint_anonymous_token(): fetches API key, posts signUp with
  Origin/Referer headers, validates iss + exp on the returned JWT
- SwiftlyAuthMintError surfaces verbatim to ScrapeLog.error_message
- Process-local cache only; threading.Lock around mutate

Tests: 4 unit tests covering fresh mint, cache hit, near-expiry
re-mint, and Firebase non-200. Full suite: 92/92 green.

Spec: docs/specs/2026-05-06-swiftly-token-auto-mint.md
Wiring into lucky_ca_scraper deferred to AM-2.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-06 15:19:30 -07:00
co-authored by Claude Opus 4.7
parent b4d92c07af
commit 5e0a49e4ae
2 changed files with 329 additions and 0 deletions
+152
View File
@@ -0,0 +1,152 @@
"""Offline tests for swiftly_auth — Firebase REST anon-signUp + cache.
All tests run with ``requests.get``/``requests.post`` patched. No live
network. The live integration check belongs in
``scripts/spike_swiftly_ingest.py --confirm-live``.
"""
from __future__ import annotations
import base64
import json
import sys
import time
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
import requests
BACKEND_DIR = Path(__file__).resolve().parent.parent
if str(BACKEND_DIR) not in sys.path:
sys.path.insert(0, str(BACKEND_DIR))
from app.services import swiftly_auth # noqa: E402
from app.services.swiftly_auth import ( # noqa: E402
SwiftlyAuthMintError,
get_token,
)
API_KEY = "AIzaSy-test-key"
EXPECTED_ISS = "https://securetoken.google.com/swiftly-lu-prod"
@pytest.fixture(autouse=True)
def _clear_cache():
swiftly_auth._reset_cache_for_tests()
yield
swiftly_auth._reset_cache_for_tests()
def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def _make_jwt(*, exp: float, iss: str = EXPECTED_ISS) -> str:
header = _b64url(json.dumps({"alg": "RS256", "typ": "JWT"}).encode())
payload = _b64url(
json.dumps({"iss": iss, "aud": "swiftly-lu-prod", "exp": int(exp)}).encode()
)
sig = _b64url(b"signature-bytes")
return f"{header}.{payload}.{sig}"
def _config_response() -> MagicMock:
resp = MagicMock(spec=requests.Response)
resp.status_code = 200
resp.json.return_value = {"firebaseApiKey": API_KEY}
return resp
def _signup_response(*, id_token: str, status_code: int = 200) -> MagicMock:
resp = MagicMock(spec=requests.Response)
resp.status_code = status_code
resp.text = "ok" if status_code == 200 else "boom"
if status_code == 200:
resp.json.return_value = {"idToken": id_token}
else:
resp.json.side_effect = ValueError("not json")
return resp
def test_get_token_mints_fresh_when_cache_empty():
"""First call fetches config.json, posts to Firebase, returns idToken."""
fresh_exp = time.time() + 3600
token = _make_jwt(exp=fresh_exp)
with patch.object(
swiftly_auth.requests, "get", return_value=_config_response()
) as mget, patch.object(
swiftly_auth.requests,
"post",
return_value=_signup_response(id_token=token),
) as mpost:
result = get_token()
assert result == token
mget.assert_called_once()
config_url = mget.call_args.args[0]
assert config_url == "https://luckysupermarkets.com/config.json"
mpost.assert_called_once()
post_url = mpost.call_args.args[0]
assert post_url == "https://identitytoolkit.googleapis.com/v1/accounts:signUp"
assert mpost.call_args.kwargs["params"] == {"key": API_KEY}
headers = mpost.call_args.kwargs["headers"]
assert headers["Origin"] == "https://luckysupermarkets.com"
assert headers["Referer"] == "https://luckysupermarkets.com/"
def test_get_token_returns_cached_without_reminting():
"""Second call within the refresh window does NOT re-hit Firebase."""
fresh_exp = time.time() + 3600 # well above the 300s refresh margin
token = _make_jwt(exp=fresh_exp)
with patch.object(
swiftly_auth.requests, "get", return_value=_config_response()
), patch.object(
swiftly_auth.requests,
"post",
return_value=_signup_response(id_token=token),
) as mpost:
first = get_token()
second = get_token()
assert first == second == token
assert mpost.call_count == 1, "expected exactly one signUp call across two get_token() calls"
def test_get_token_remints_when_cached_token_near_expiry():
"""A cached token within the refresh-margin window triggers a re-mint."""
near_expiry = time.time() + 120 # < 300s margin → must refresh
fresh_exp = time.time() + 3600
stale_token = _make_jwt(exp=near_expiry)
fresh_token = _make_jwt(exp=fresh_exp)
swiftly_auth._state["api_key"] = API_KEY # skip the config fetch
swiftly_auth._state["token"] = stale_token
swiftly_auth._state["exp"] = near_expiry
with patch.object(
swiftly_auth.requests,
"post",
return_value=_signup_response(id_token=fresh_token),
) as mpost:
result = get_token()
assert result == fresh_token
mpost.assert_called_once()
def test_mint_raises_on_firebase_non_200():
"""Firebase 403/400/etc. surfaces as SwiftlyAuthMintError with status detail."""
with patch.object(
swiftly_auth.requests, "get", return_value=_config_response()
), patch.object(
swiftly_auth.requests,
"post",
return_value=_signup_response(id_token="ignored", status_code=403),
):
with pytest.raises(SwiftlyAuthMintError) as excinfo:
get_token()
assert "403" in str(excinfo.value)