gahusb
f32792e4a9
Dockerfile (python:3.12-slim), requirements (openai + google-cloud-storage + httpx + redis). .env.example: OPENAI/GOOGLE/PIAPI/SEEDANCE keys + VIDEO_MEDIA_ROOT. nas_client.webhook_update_task: call-time os.getenv (테스트 격리), respx mock 5 tests. Plan-B-Video Phase 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""nas_client — webhook adapter for video-render."""
|
|
import pytest
|
|
import respx
|
|
import httpx
|
|
|
|
from nas_client import webhook_update_task
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(monkeypatch):
|
|
monkeypatch.setenv("NAS_BASE_URL", "http://nas-test:18801")
|
|
monkeypatch.setenv("INTERNAL_API_KEY", "test-key")
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_sends_x_internal_key():
|
|
route = respx.post("http://nas-test:18801/api/internal/video/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
webhook_update_task("task-1", "processing", 30, message="downloading")
|
|
assert route.called
|
|
req = route.calls[0].request
|
|
assert req.headers["X-Internal-Key"] == "test-key"
|
|
import json
|
|
body = json.loads(req.content)
|
|
assert body["task_id"] == "task-1"
|
|
assert body["status"] == "processing"
|
|
assert body["progress"] == 30
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_with_video_url():
|
|
route = respx.post("http://nas-test:18801/api/internal/video/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
webhook_update_task("task-2", "succeeded", 100, message="완료",
|
|
video_url="/media/video/task-2.mp4")
|
|
import json
|
|
payload = json.loads(route.calls[0].request.content)
|
|
assert payload["video_url"] == "/media/video/task-2.mp4"
|
|
assert payload["status"] == "succeeded"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_with_error():
|
|
route = respx.post("http://nas-test:18801/api/internal/video/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
webhook_update_task("task-3", "failed", 0, error="Sora API rate limit")
|
|
import json
|
|
payload = json.loads(route.calls[0].request.content)
|
|
assert payload["error"] == "Sora API rate limit"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_swallows_network_error(caplog):
|
|
respx.post("http://nas-test:18801/api/internal/video/update").mock(
|
|
side_effect=httpx.ConnectError("no host")
|
|
)
|
|
webhook_update_task("task-5", "processing", 10)
|
|
assert "task-5" in caplog.text
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_swallows_non_200(caplog):
|
|
respx.post("http://nas-test:18801/api/internal/video/update").mock(
|
|
return_value=httpx.Response(500, text="server error")
|
|
)
|
|
webhook_update_task("task-6", "processing", 50)
|
|
assert "task-6" in caplog.text
|