✅ test(server): CWA-proxy auth tests + Basic-auth integration

conftest fakes CWA via httpx.MockTransport. test_auth covers cache
hit/miss/TTL/503/non-Basic. Integration tests use Basic auth helpers
keyed by the cwa_users fixture.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: vitofico

server/opds_sync/core/auth.py

@@ -3,7 +3,8 @@
 import logging
 import time
 from collections import OrderedDict
-from typing import Annotated, Callable
+from collections.abc import Callable
+from typing import Annotated
 
 import httpx
 from fastapi import Depends, HTTPException, Request, status
@@ -45,7 +46,9 @@ def __init__(
     async def validate(self, auth_header: str) -> str:
         """Returns the user_id (lowercased CWA username) or raises HTTPException(401/503)."""
         if not auth_header.lower().startswith("basic "):
-            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="basic auth required")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail="basic auth required"
+            )
 
         b64 = auth_header[6:].strip()
         key = hashlib.sha256(b64.encode("ascii")).digest()
@@ -56,7 +59,9 @@ async def validate(self, auth_header: str) -> str:
             self._cache.move_to_end(key)
             if cached.is_valid:
                 return cached.user_id  # type: ignore[return-value]
-            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid credentials")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid credentials"
+            )
 
         try:
             resp = await self._client.get(

server/tests/conftest.py

@@ -1,12 +1,10 @@
-import time
+import base64
 from collections.abc import AsyncIterator, Iterator
 
-import jwt
+import httpx
 import pytest
 from alembic import command
 from alembic.config import Config as AlembicConfig
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import rsa
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
     AsyncSession,
@@ -33,7 +31,7 @@ def _key(item):
 @pytest.fixture(scope="session")
 def postgres_url() -> Iterator[str]:
     with PostgresContainer("postgres:16-alpine") as pg:
-        sync_url = pg.get_connection_url()  # postgresql+psycopg2://...
+        sync_url = pg.get_connection_url()
         async_url = sync_url.replace("postgresql+psycopg2://", "postgresql+asyncpg://")
         yield async_url
 
@@ -60,72 +58,64 @@ async def session(engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
         await s.rollback()
 
 
-@pytest.fixture(scope="session")
-def signing_key():
-    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
-    priv = key.private_bytes(
-        encoding=serialization.Encoding.PEM,
-        format=serialization.PrivateFormat.PKCS8,
-        encryption_algorithm=serialization.NoEncryption(),
-    )
-    pub = key.public_key()
-    return priv, pub
+# ---- CWA mock --------------------------------------------------------------
 
 
 @pytest.fixture
-def issuer() -> str:
-    return "https://test-iss.example/"
+def cwa_users() -> dict[str, str]:
+    """Mutable per-test dict of valid CWA username → password."""
+    return {"alice": "alicepass", "bob": "bobpass"}
 
 
 @pytest.fixture
-def audience() -> str:
-    return "test-aud"
+def cwa_transport(cwa_users: dict[str, str]) -> httpx.MockTransport:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if not request.url.path.endswith("/opds"):
+            return httpx.Response(404)
+        auth = request.headers.get("authorization", "")
+        if not auth.lower().startswith("basic "):
+            return httpx.Response(401)
+        try:
+            decoded = base64.b64decode(auth[6:].strip()).decode("utf-8")
+            user, pw = decoded.split(":", 1)
+        except Exception:
+            return httpx.Response(401)
+        if cwa_users.get(user) == pw:
+            return httpx.Response(200, text="<feed/>")
+        return httpx.Response(401)
+
+    return httpx.MockTransport(handler)
 
 
 @pytest.fixture
-def make_token(signing_key, issuer, audience):
-    priv, _ = signing_key
-
-    def _make(sub: str, ttl: int = 60, **extra) -> str:
-        now = int(time.time())
-        claims = {
-            "sub": sub,
-            "iss": issuer,
-            "aud": audience,
-            "iat": now,
-            "nbf": now,
-            "exp": now + ttl,
-            **extra,
-        }
-        return jwt.encode(claims, priv, algorithm="RS256", headers={"kid": "test-key"})
+def basic_header():
+    def _make(user: str, pw: str) -> str:
+        token = base64.b64encode(f"{user}:{pw}".encode()).decode("ascii")
+        return f"Basic {token}"
 
     return _make
 
 
 @pytest.fixture
-def app_under_test(postgres_url, alembic_upgrade, monkeypatch, signing_key, issuer, audience):
-    """A FastAPI app wired to the test Postgres + a static JWKS fetcher."""
+def app_under_test(postgres_url, alembic_upgrade, monkeypatch, cwa_transport):
+    """A FastAPI app wired to the test Postgres + a mock CWA transport."""
     monkeypatch.setenv("OPDS_SYNC_DATABASE_URL", postgres_url)
-    monkeypatch.setenv("OPDS_SYNC_AUTHENTIK_ISSUER", issuer)
-    monkeypatch.setenv("OPDS_SYNC_AUTHENTIK_AUDIENCE", audience)
+    monkeypatch.setenv("OPDS_SYNC_CWA_BASE_URL", "http://test-cwa")
 
-    # Bust the lru_cache so the env vars take effect.
     from opds_sync.config import get_settings
 
     get_settings.cache_clear()
 
-    from opds_sync.core.auth import JwksFetcher, JwtValidator
+    from opds_sync.core.auth import CalibreAuthValidator
     from opds_sync.main import create_app
 
-    _, pub = signing_key
-
-    class StaticJwks(JwksFetcher):
-        async def get_signing_key(self, kid):  # type: ignore[override]
-            class _K:
-                key = pub  # noqa: E701
-
-            return _K()
-
     app = create_app()
-    app.state.jwt_validator = JwtValidator(jwks=StaticJwks(), issuer=issuer, audience=audience)
+    test_client = httpx.AsyncClient(
+        transport=cwa_transport, base_url="http://test-cwa", timeout=3.0
+    )
+    app.state.httpx_client = test_client
+    app.state.auth_validator = CalibreAuthValidator(
+        client=test_client,
+        cwa_base_url="http://test-cwa",
+    )
     return app

server/tests/integration/test_health.py

@@ -1,28 +1,16 @@
 from httpx import ASGITransport, AsyncClient
 
 
-async def test_healthz_returns_200():
-    from opds_sync.main import create_app
-
-    app = create_app()
-    transport = ASGITransport(app=app)
+async def test_healthz_returns_200(app_under_test):
+    transport = ASGITransport(app=app_under_test)
     async with AsyncClient(transport=transport, base_url="http://test") as client:
         r = await client.get("/sync/v1/healthz")
     assert r.status_code == 200
     assert r.json() == {"status": "ok"}
 
 
-async def test_readyz_returns_200_when_db_reachable(postgres_url, alembic_upgrade, monkeypatch):
-    monkeypatch.setenv("OPDS_SYNC_DATABASE_URL", postgres_url)
-    # Bust the lru_cache on get_settings so the new env var is honoured
-    from opds_sync.config import get_settings
-
-    get_settings.cache_clear()
-
-    from opds_sync.main import create_app
-
-    app = create_app()
-    transport = ASGITransport(app=app)
+async def test_readyz_returns_200_when_db_reachable(app_under_test):
+    transport = ASGITransport(app=app_under_test)
     async with AsyncClient(transport=transport, base_url="http://test") as client:
         r = await client.get("/sync/v1/readyz")
     assert r.status_code == 200

server/tests/integration/test_progress.py

@@ -1,13 +1,16 @@
+import base64
+
 from httpx import ASGITransport, AsyncClient
 
 
-def _bearer(token: str) -> dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
+def _basic(user: str, pw: str) -> dict[str, str]:
+    token = base64.b64encode(f"{user}:{pw}".encode()).decode("ascii")
+    return {"Authorization": f"Basic {token}"}
 
 
-async def test_post_progress_creates_document_and_progress(app_under_test, make_token):
+async def test_post_progress_creates_document_and_progress(app_under_test):
     transport = ASGITransport(app=app_under_test)
-    headers = _bearer(make_token("alice"))
+    headers = _basic("alice", "alicepass")
     body = {
         "items": [
             {
@@ -27,9 +30,9 @@ async def test_post_progress_creates_document_and_progress(app_under_test, make_
     assert data["results"][0]["server_client_updated_at"] == "2026-05-05T12:00:00+00:00"
 
 
-async def test_post_progress_lww_keeps_newer(app_under_test, make_token):
+async def test_post_progress_lww_keeps_newer(app_under_test):
     transport = ASGITransport(app=app_under_test)
-    headers = _bearer(make_token("alice"))
+    headers = _basic("alice", "alicepass")
     base = {
         "document": {"metadata_id": "abc", "content_hash": "hash1"},
         "locator": "loc",
@@ -67,12 +70,12 @@ async def test_post_progress_lww_keeps_newer(app_under_test, make_token):
         assert items[0]["percent"] == 0.5
 
 
-async def test_get_progress_filters_by_user(app_under_test, make_token):
+async def test_get_progress_filters_by_user(app_under_test):
     transport = ASGITransport(app=app_under_test)
     async with AsyncClient(transport=transport, base_url="http://test") as c:
         await c.post(
             "/sync/v1/progress",
-            headers=_bearer(make_token("alice")),
+            headers=_basic("alice", "alicepass"),
             json={
                 "items": [
                     {
@@ -85,7 +88,7 @@ async def test_get_progress_filters_by_user(app_under_test, make_token):
             },
         )
         r = await c.get(
-            "/sync/v1/progress?since=2026-01-01T00:00:00+00:00", headers=_bearer(make_token("bob"))
+            "/sync/v1/progress?since=2026-01-01T00:00:00+00:00", headers=_basic("bob", "bobpass")
         )
     assert r.status_code == 200
     assert r.json()["items"] == []