tests/test_api.py

from __future__ import annotations

import time

from fastapi.testclient import TestClient

from service.app import create_app
from service.runner import Runner
from service.settings import Settings


def make_settings(tmp_path, **overrides):
    values = {
        "data_dir": tmp_path,
        "api_key": "test-key",
        "callback_secret": "secret",
        "callback_allowed_hosts": ("example.com",),
        "public_base_url": "https://space.example",
        "runner_mode": "stub",
        "production_api_url": "",
        "production_api_key": "",
        "modelverse_api_url": "https://api.modelverse.cn/v1",
        "modelverse_api_key": "",
        "modelverse_image_model": "Qwen/Qwen-Image",
        "modelverse_video_t2v_model": "Wan-AI/Wan2.6-T2V",
        "modelverse_video_i2v_model": "Wan-AI/Wan2.6-I2V",
        "poll_interval_seconds": 0.01,
        "job_timeout_seconds": 30,
        "max_workers": 1,
        "command_json": "",
    }
    values.update(overrides)
    return Settings(**values)


def test_job_lifecycle(tmp_path):
    settings = make_settings(tmp_path)
    with TestClient(create_app(settings)) as client:
        headers = {"Authorization": "Bearer test-key"}
        created = client.post(
            "/v1/jobs",
            headers=headers,
            json={
                "content_item_id": "content-1",
                "title": "Test",
                "script": "A valid test script.",
                "duration_seconds": 2,
            },
        )
        assert created.status_code == 202
        job_id = created.json()["id"]
        deadline = time.monotonic() + 15
        body = created.json()
        while time.monotonic() < deadline:
            result = client.get(f"/v1/jobs/{job_id}", headers=headers)
            assert result.status_code == 200
            body = result.json()
            if body["status"] in {"completed", "failed"}:
                break
            time.sleep(0.1)
        assert body["status"] == "completed", body.get("error")
        assert body["output"]["video_url"].endswith("/output.mp4")
        artifact = client.get(f"/v1/jobs/{job_id}/artifacts/output.mp4", headers=headers)
        assert artifact.status_code == 200
        assert artifact.headers["content-type"] == "video/mp4"


def test_authentication(tmp_path):
    settings = make_settings(tmp_path, api_key="required")
    with TestClient(create_app(settings)) as client:
        assert client.get("/v1/capabilities").status_code == 401


def test_home_page_opens_without_authentication(tmp_path):
    settings = make_settings(tmp_path, api_key="required")
    with TestClient(create_app(settings)) as client:
        response = client.get("/")

        assert response.status_code == 200
        assert "OpenMontage API" in response.text
        assert 'href="/docs"' in response.text


def test_image_generation_contract(tmp_path, monkeypatch):
    def fake_generate_image(self, payload):
        return {
            "model": "Qwen/Qwen-Image",
            "image_url": "https://cdn.example/generated.png",
            "width": payload["width"],
            "height": payload["height"],
            "provider": "modelverse",
        }

    monkeypatch.setattr(Runner, "generate_image", fake_generate_image)
    with TestClient(create_app(make_settings(tmp_path))) as client:
        response = client.post(
            "/v1/images/generations",
            headers={"Authorization": "Bearer test-key"},
            json={"prompt": "A studio portrait", "width": 768, "height": 1024},
        )

    assert response.status_code == 200
    assert response.json()["model"] == "Qwen/Qwen-Image"
    assert response.json()["width"] == 768


def test_unsafe_video_prompt_is_rejected_before_job_creation(tmp_path):
    settings = make_settings(tmp_path, runner_mode="modelverse", modelverse_api_key="configured")
    with TestClient(create_app(settings)) as client:
        response = client.post(
            "/v1/jobs",
            headers={"Authorization": "Bearer test-key"},
            json={
                "content_item_id": "content-unsafe",
                "script": "一位女性脱掉上衣并走向镜头。",
                "duration_seconds": 5,
            },
        )

    assert response.status_code == 422
    assert "未提交给 ModelVerse" in response.json()["detail"]
    assert list((tmp_path / "jobs").iterdir()) == []


def test_safe_video_prompt_passes_preflight(tmp_path):
    runner = Runner(make_settings(tmp_path))
    runner.validate_video_request(
        {
            "prompt": "一位年轻女性站起身，微笑着走向窗边，整理外套，镜头缓慢推进，温暖自然光。"
        }
    )


def test_modelverse_failure_keeps_original_diagnostics_and_redacts_credentials(tmp_path):
    runner = Runner(make_settings(tmp_path))
    message = runner._modelverse_failure_message(
        {
            "output": {
                "task_id": "provider-task-1",
                "task_status": "Failure",
                "error_message": "content rejected by safety policy; Authorization=secret-value",
                "safety_status": "blocked",
            },
            "request_id": "request-1",
        }
    )

    assert '"output.task_status":"Failure"' in message
    assert '"output.error_message":"content rejected by safety policy; Authorization=[REDACTED]"' in message
    assert '"output.safety_status":"blocked"' in message
    assert '"request_id":"request-1"' in message
    assert "secret-value" not in message


def test_modelverse_failure_explains_when_provider_omits_error_message(tmp_path):
    runner = Runner(make_settings(tmp_path))
    message = runner._modelverse_failure_message(
        {"output": {"task_id": "provider-task-2", "task_status": "Failure"}}
    )

    assert "ModelVerse returned Failure but omitted output.error_message." in message


def test_first_frame_rejects_private_hosts_before_provider_call(tmp_path):
    runner = Runner(make_settings(tmp_path))

    try:
        runner._validate_first_frame_url("http://127.0.0.1/private.png")
    except RuntimeError as exception:
        assert "public network addresses" in str(exception)
    else:
        raise AssertionError("Private first-frame URL was accepted")


def test_video_model_mode_must_match_generation_mode(tmp_path):
    runner = Runner(make_settings(tmp_path))
    runner._validate_video_model("Wan-AI/Wan2.6-I2V", "i2v")

    try:
        runner._validate_video_model("Wan-AI/Wan2.6-T2V", "i2v")
    except RuntimeError as exception:
        assert "not I2V-compatible" in str(exception)
    else:
        raise AssertionError("T2V model was accepted for an I2V request")