gahusb
11e73f6960
- _post_update payload·헤더 검증 - _process_one 정상 흐름 (processing + succeeded) - _process_one 예외 시 failed webhook Plan-B-Insta Phase 2 mature. Phase 3 cutover 준비 완료. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""worker.py — Redis BLPOP + webhook 단위 테스트."""
|
|
import json
|
|
import pytest
|
|
import httpx
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import worker
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_slate():
|
|
return {
|
|
"id": 42,
|
|
"cover_copy": json.dumps({"headline": "테스트 H", "body": "테스트 B", "accent_color": "#FF0000"}),
|
|
"body_copies": json.dumps([{"headline": "본문1", "body": "..."} for _ in range(8)]),
|
|
"cta_copy": json.dumps({"headline": "CTA", "body": "...", "cta": "Click"}),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_update_sends_correct_payload(monkeypatch):
|
|
monkeypatch.setenv("INTERNAL_API_KEY", "test-secret")
|
|
monkeypatch.setenv("NAS_BASE_URL", "http://nas.test")
|
|
# worker 모듈 환경변수 재로딩
|
|
worker.NAS_BASE_URL = "http://nas.test"
|
|
worker.INTERNAL_API_KEY = "test-secret"
|
|
|
|
captured = {}
|
|
async def fake_post(self, url, headers=None, json=None, **kw):
|
|
captured["url"] = url
|
|
captured["headers"] = headers
|
|
captured["json"] = json
|
|
class R:
|
|
status_code = 200
|
|
text = "ok"
|
|
return R()
|
|
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
await worker._post_update(client, "t-1", "processing", 30)
|
|
|
|
assert captured["url"] == "http://nas.test/api/internal/insta/update"
|
|
assert captured["headers"]["X-Internal-Key"] == "test-secret"
|
|
assert captured["json"]["status"] == "processing"
|
|
assert captured["json"]["progress"] == 30
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_one_success_calls_webhook_twice(monkeypatch, fake_slate):
|
|
"""processing(50) → succeeded(100) 두 번 호출 + render 한 번."""
|
|
calls: list = []
|
|
|
|
async def fake_post(self, url, headers=None, json=None, **kw):
|
|
calls.append({"status": json["status"], "progress": json["progress"]})
|
|
class R:
|
|
status_code = 200
|
|
text = "ok"
|
|
return R()
|
|
|
|
async def fake_get(self, url, **kw):
|
|
class R:
|
|
status_code = 200
|
|
def json(self_inner): return fake_slate
|
|
def raise_for_status(self_inner): pass
|
|
return R()
|
|
|
|
async def fake_render(slate, slate_id, template="default/card.html.j2"):
|
|
return [f"/tmp/{slate_id}/{i:02d}.png" for i in range(1, 11)]
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
|
|
monkeypatch.setattr(httpx.AsyncClient, "get", fake_get)
|
|
monkeypatch.setattr(worker, "render_slate", fake_render)
|
|
worker.INTERNAL_API_KEY = "test"
|
|
worker.NAS_BASE_URL = "http://nas.test"
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
await worker._process_one(client, {
|
|
"task_id": "t-2",
|
|
"params": {"slate_id": 42, "theme": "default"},
|
|
})
|
|
|
|
statuses = [c["status"] for c in calls]
|
|
assert "processing" in statuses
|
|
assert "succeeded" in statuses
|
|
assert calls[-1]["progress"] == 100
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_one_render_failure_reports_failed(monkeypatch, fake_slate):
|
|
"""render 예외 시 failed webhook 호출."""
|
|
calls: list = []
|
|
|
|
async def fake_post(self, url, headers=None, json=None, **kw):
|
|
calls.append(json)
|
|
class R: status_code = 200; text = "ok"
|
|
return R()
|
|
|
|
async def fake_get(self, url, **kw):
|
|
class R:
|
|
status_code = 200
|
|
def json(self_inner): return fake_slate
|
|
def raise_for_status(self_inner): pass
|
|
return R()
|
|
|
|
async def fake_render(*a, **k):
|
|
raise RuntimeError("Chromium crashed")
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
|
|
monkeypatch.setattr(httpx.AsyncClient, "get", fake_get)
|
|
monkeypatch.setattr(worker, "render_slate", fake_render)
|
|
worker.INTERNAL_API_KEY = "test"
|
|
worker.NAS_BASE_URL = "http://nas.test"
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
await worker._process_one(client, {
|
|
"task_id": "t-3",
|
|
"params": {"slate_id": 99},
|
|
})
|
|
|
|
last = calls[-1]
|
|
assert last["status"] == "failed"
|
|
assert "Chromium" in last["error"]
|