evOo-local/infra/media-worker/tests/test_schema.py

from __future__ import annotations

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from .conftest import AUTH


def _synth(media_root: Path, **extra: object) -> dict[str, object]:
    return {
        "path": f"{media_root}/testsrc.mp4",
        "track": [{"tMs": 0, "box": {"x": 0.1, "y": 0.1, "w": 0.2, "h": 0.2}}],
        "seed": 42,
        "outPath": f"{media_root}/synth/out.mkv",
        **extra,
    }


@pytest.mark.parametrize("field", ["targetFace", "targetImage", "faceImagePath", "referencePath", "image"])
def test_synthesize_identity_rejects_any_target_face_field(
    client: TestClient, media_root: Path, field: str
) -> None:
    response = client.post("/v1/synthesizeIdentity", headers=AUTH, json=_synth(media_root, **{field: "data"}))
    assert response.status_code == 400
    error = response.json()["error"]
    assert error["code"] == "bad_request"
    assert error["operation"] == "synthesizeIdentity"


def test_synthesize_identity_is_unsupported_without_licensed_model(
    client: TestClient, media_root: Path
) -> None:
    response = client.post("/v1/synthesizeIdentity", headers=AUTH, json=_synth(media_root))
    assert response.status_code == 501
    assert response.json()["error"] == {
        "code": "unsupported",
        "message": "no commercially licensed model",
        "operation": "synthesizeIdentity",
    }
    assert not (media_root / "synth" / "out.mkv").exists()


@pytest.mark.parametrize(
    ("operation", "body"),
    [
        ("detectScenes", {"path": "/x", "extra": 1}),
        ("detectFaces", {"path": "/x"}),
        ("detectFaces", {"path": "/x", "sampleFps": 0}),
        ("detectFaces", {"path": "/x", "sampleFps": "2"}),
        ("embedFrames", {"path": "/x", "timesMs": [1.5]}),
        ("embedFrames", {"path": "/x", "timesMs": [-1]}),
        ("segmentPrompted", {"path": "/x", "prompt": {"tMs": 0}, "propagate": True, "outDir": "/y"}),
        (
            "segmentPrompted",
            {
                "path": "/x",
                "prompt": {"tMs": 0, "box": {"x": 2, "y": 0, "w": 1, "h": 1}},
                "propagate": True,
                "outDir": "/y",
            },
        ),
        ("redact", {"path": "/x", "outPath": "/y", "redactions": []}),
        (
            "redact",
            {
                "path": "/x",
                "outPath": "/y",
                "redactions": [{"keyframes": [], "mode": "blur", "paddingPct": 0, "featherPct": 0}],
            },
        ),
        (
            "redact",
            {
                "path": "/x",
                "outPath": "/y",
                "redactions": [
                    {
                        "keyframes": [{"tMs": 0, "box": {"x": 0, "y": 0, "w": 1, "h": 1}}],
                        "mode": "reframe_out",
                        "paddingPct": 0,
                        "featherPct": 0,
                    }
                ],
            },
        ),
        (
            "redact",
            {
                "path": "/x",
                "outPath": "/y",
                "redactions": [
                    {
                        "keyframes": [{"tMs": 0, "box": {"x": 0, "y": 0, "w": 1, "h": 1}}],
                        "mode": "solid",
                        "paddingPct": 0,
                        "featherPct": 0,
                        "color": "red",
                    }
                ],
            },
        ),
        (
            "redact",
            {
                "path": "/x",
                "outPath": "/y",
                "redactions": [
                    {
                        "keyframes": [{"tMs": 0, "box": {"x": 0, "y": 0, "w": 1, "h": 1, "z": 1}}],
                        "mode": "blur",
                        "paddingPct": 0,
                        "featherPct": 0,
                    }
                ],
            },
        ),
    ],
)
def test_schema_violations_are_bad_requests(
    client: TestClient, operation: str, body: dict[str, object]
) -> None:
    response = client.post(f"/v1/{operation}", headers=AUTH, json=body)
    assert response.status_code == 400, response.text
    assert response.json()["error"]["code"] == "bad_request"
    assert response.json()["error"]["operation"] == operation


def test_malformed_json_and_unknown_operation(client: TestClient) -> None:
    response = client.post(
        "/v1/detectScenes", headers={**AUTH, "content-type": "application/json"}, content=b"{nope"
    )
    assert response.status_code == 400
    response = client.post("/v1/makeDeepfake", headers=AUTH, json={})
    assert response.status_code == 404
    assert response.json()["error"]["code"] == "bad_request"