generated from john/python-template
130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
"""Tests for Step 5 operator access control behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.testclient import TestClient
|
|
|
|
from transcription.api.errors import register_error_handlers
|
|
from transcription.errors import build_error_envelope
|
|
from transcription.security import AccessDeniedError
|
|
from transcription.security import enforce_request_access
|
|
|
|
|
|
def _basic_header(username: str, password: str) -> str:
|
|
token = base64.b64encode(f"{username}:{password}".encode()).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"}
|