feat(video-render): scaffold + nas_client webhook adapter (SP-7)
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>
This commit is contained in:
70
services/video-render/tests/test_nas_client.py
Normal file
70
services/video-render/tests/test_nas_client.py
Normal file
@@ -0,0 +1,70 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user