ver1 - Step 5 implementation complete.

tests/api/test_access_control.py

@@ -0,0 +1,128 @@
+"""Tests for Step 5 operator access control behavior."""
+
+from __future__ import annotations
+
+import base64
+from types import SimpleNamespace
+
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+from fastapi.testclient import TestClient
+import pytest
+
+from transcription.api.errors import register_error_handlers
+from transcription.errors import build_error_envelope
+from transcription.security import AccessDeniedError, enforce_request_access
+
+
+def _basic_header(username: str, password: str) -> str:
+    token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
+    return f"Basic {token}"
+
+
+def _build_app(*, settings) -> FastAPI:
+    app = FastAPI()
+    register_error_handlers(app)
+
+    @app.middleware("http")
+    async def operator_access_middleware(request, call_next):
+        try:
+            enforce_request_access(request=request, settings=settings)
+        except AccessDeniedError as exc:
+            envelope = build_error_envelope(exc)
+            headers = {"WWW-Authenticate": "Basic"} if exc.should_challenge else None
+            return JSONResponse(status_code=401, content=envelope.__dict__, headers=headers)
+        return await call_next(request)
+
+    @app.get("/healthz")
+    def healthz():
+        return {"status": "ok"}
+
+    @app.get("/api/jobs")
+    def get_jobs():
+        return [{"id": "demo"}]
+
+    @app.get("/ui")
+    def ui_root():
+        return {"ok": True}
+
+    return app
+
+
+@pytest.mark.integration
+class TestAccessControl:
+    """Verify protected routes enforce operator auth when enabled."""
+
+    def test_protected_api_requires_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/api/jobs")
+
+        assert response.status_code == 401
+        assert response.headers.get("WWW-Authenticate") == "Basic"
+        payload = response.json()
+        assert payload["category"] == "user_input_error"
+        assert payload["suggestion"]
+
+    def test_protected_api_rejects_invalid_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get(
+            "/api/jobs",
+            headers={"Authorization": _basic_header("operator", "wrong")},
+        )
+
+        assert response.status_code == 401
+        payload = response.json()
+        assert payload["message"] == "Invalid operator credentials"
+
+    def test_protected_api_allows_valid_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get(
+            "/api/jobs",
+            headers={"Authorization": _basic_header("operator", "secret")},
+        )
+
+        assert response.status_code == 200
+        assert response.json() == [{"id": "demo"}]
+
+    def test_protected_ui_path_requires_credentials(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/ui")
+
+        assert response.status_code == 401
+
+    def test_healthz_is_not_protected(self):
+        settings = SimpleNamespace(
+            operator_access_enabled=True,
+            operator_username="operator",
+            operator_password="secret",
+        )
+        client = TestClient(_build_app(settings=settings), raise_server_exceptions=False)
+
+        response = client.get("/healthz")
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "ok"}