5e2094f2ad
FastAPI service in infra/media-worker implementing every MediaInferenceProvider operation behind HTTP API v1, plus capabilities. Bearer auth, MEDIA_ROOT realpath enforcement, strict request schemas, cancellation on disconnect, GPU/CPU gates with 503 overflow, input size/duration limits, lazy SHA256-pinned model downloads, fail-closed redaction with atomic output, offline pytest suite with fake backends, CPU/CUDA/ROCm Dockerfile and a license review of every model. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.config import ConfigError, Settings
|
|
|
|
from .conftest import AUTH, TOKEN
|
|
|
|
|
|
def test_healthz_needs_no_token(client: TestClient) -> None:
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json()["ok"] is True
|
|
assert isinstance(response.json()["version"], str)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"headers",
|
|
[
|
|
{},
|
|
{"Authorization": "Bearer wrong-token-" + "y" * 32},
|
|
{"Authorization": TOKEN},
|
|
{"Authorization": "Basic " + TOKEN},
|
|
],
|
|
)
|
|
def test_routes_reject_missing_or_wrong_token(client: TestClient, headers: dict[str, str]) -> None:
|
|
for method, url in [("GET", "/v1/capabilities"), ("POST", "/v1/detectScenes")]:
|
|
response = client.request(method, url, headers=headers, json={"path": "/x"})
|
|
assert response.status_code == 401
|
|
assert response.json()["error"]["code"] == "unauthorized"
|
|
|
|
|
|
def test_valid_token_is_accepted(client: TestClient) -> None:
|
|
assert client.get("/v1/capabilities", headers=AUTH).status_code == 200
|
|
|
|
|
|
def test_refuses_to_start_without_a_long_token(tmp_path: Path) -> None:
|
|
env = {"MEDIA_ROOT": str(tmp_path)}
|
|
with pytest.raises(ConfigError):
|
|
Settings.from_env(env)
|
|
with pytest.raises(ConfigError):
|
|
Settings.from_env({**env, "MEDIA_WORKER_TOKEN": "short"})
|
|
assert Settings.from_env({**env, "MEDIA_WORKER_TOKEN": "z" * 32}).token == "z" * 32
|
|
assert Settings.from_env({**env, "MEDIA_WORKER_INSECURE_DEV": "1"}).insecure_dev
|
|
|
|
|
|
def test_settings_validate_root_and_defaults(tmp_path: Path) -> None:
|
|
with pytest.raises(ConfigError):
|
|
Settings.from_env({"MEDIA_WORKER_TOKEN": "z" * 32, "MEDIA_ROOT": "relative/path"})
|
|
settings = Settings.from_env({"MEDIA_WORKER_TOKEN": "z" * 32, "MEDIA_ROOT": str(tmp_path)})
|
|
assert settings.host == "127.0.0.1"
|
|
assert settings.port == 8790
|
|
assert settings.gpu_concurrency == 1
|
|
assert settings.max_input_bytes == 20 * 1024**3
|
|
assert settings.max_seconds == 4 * 3600
|