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")