gahusb
9eef2c5015
NAS DB 직접 접근 불가 → webhook_update_task/webhook_add_track으로 변환. X-Internal-Key 헤더 자동 첨부. 실패 시 raise 안 함 (logger.error). env var는 call time에 읽어 monkeypatch 테스트 호환성 확보. Plan-B-Music Phase 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
"""nas_client — webhook adapter tests."""
|
|
import os
|
|
import pytest
|
|
import respx
|
|
import httpx
|
|
|
|
from nas_client import webhook_update_task, webhook_add_track
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(monkeypatch):
|
|
monkeypatch.setenv("NAS_BASE_URL", "http://nas-test:18600")
|
|
monkeypatch.setenv("INTERNAL_API_KEY", "test-key")
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_sends_x_internal_key():
|
|
route = respx.post("http://nas-test:18600/api/internal/music/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
|
|
assert body["message"] == "downloading"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_with_audio_url():
|
|
route = respx.post("http://nas-test:18600/api/internal/music/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
webhook_update_task("task-2", "succeeded", 100, message="완료",
|
|
audio_url="/media/music/task-2.mp3")
|
|
import json
|
|
payload = json.loads(route.calls[0].request.content)
|
|
assert payload["audio_url"] == "/media/music/task-2.mp3"
|
|
assert payload["status"] == "succeeded"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_update_task_with_error():
|
|
route = respx.post("http://nas-test:18600/api/internal/music/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
webhook_update_task("task-3", "failed", 0, error="API rate limit")
|
|
import json
|
|
payload = json.loads(route.calls[0].request.content)
|
|
assert payload["error"] == "API rate limit"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_add_track_uses_track_field():
|
|
"""add_track은 update와 동시에 (succeeded 시)."""
|
|
route = respx.post("http://nas-test:18600/api/internal/music/update").mock(
|
|
return_value=httpx.Response(200, json={"ok": True})
|
|
)
|
|
track = {"title": "x", "audio_url": "/media/music/t.mp3", "provider": "suno"}
|
|
webhook_add_track("task-4", "succeeded", 100, message="ok",
|
|
audio_url="/media/music/t.mp3", track=track)
|
|
import json
|
|
payload = json.loads(route.calls[0].request.content)
|
|
assert payload["track"]["title"] == "x"
|
|
assert payload["status"] == "succeeded"
|
|
|
|
|
|
@respx.mock
|
|
def test_webhook_swallows_network_error(caplog):
|
|
"""webhook 실패해도 raise 안 함 (logger.error)."""
|
|
respx.post("http://nas-test:18600/api/internal/music/update").mock(
|
|
side_effect=httpx.ConnectError("no host")
|
|
)
|
|
# raise 안 하면 통과
|
|
webhook_update_task("task-5", "processing", 10)
|