feat(image-render): nas_client webhook adapter (video-render 복제)

services/image-render/nas_client.py

@@ -0,0 +1,54 @@
+"""NAS webhook 어댑터 — Windows worker → NAS image-lab HTTP 위임.
+
+video-render nas_client 복제 (call-time os.getenv으로 테스트 격리).
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict, Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT = 10.0
+
+
+def _post(payload: Dict[str, Any]) -> None:
+    nas_base_url = os.getenv("NAS_BASE_URL", "http://192.168.45.54:18802")
+    internal_api_key = os.getenv("INTERNAL_API_KEY", "")
+    url = f"{nas_base_url}/api/internal/image/update"
+    try:
+        r = httpx.post(
+            url,
+            headers={"X-Internal-Key": internal_api_key},
+            json=payload,
+            timeout=_TIMEOUT,
+        )
+        if r.status_code != 200:
+            logger.error("webhook %s returned %d: %s",
+                         payload.get("task_id"), r.status_code, r.text[:200])
+    except Exception:
+        logger.exception("webhook %s 호출 실패", payload.get("task_id"))
+
+
+def webhook_update_task(
+    task_id: str,
+    status: str,
+    progress: int,
+    message: str = "",
+    image_url: Optional[str] = None,
+    error: Optional[str] = None,
+) -> None:
+    payload: Dict[str, Any] = {
+        "task_id": task_id,
+        "status": status,
+        "progress": progress,
+        "message": message,
+    }
+    if image_url is not None:
+        payload["image_url"] = image_url
+    if error is not None:
+        payload["error"] = error
+    _post(payload)

services/image-render/tests/test_nas_client.py

@@ -0,0 +1,20 @@
+import nas_client
+
+
+def test_webhook_includes_image_url(monkeypatch):
+    captured = {}
+
+    def fake_post(payload):
+        captured.update(payload)
+
+    monkeypatch.setattr(nas_client, "_post", fake_post)
+    nas_client.webhook_update_task("t1", "succeeded", 100, "done", image_url="/media/image/t1.png")
+    assert captured["task_id"] == "t1"
+    assert captured["image_url"] == "/media/image/t1.png"
+
+
+def test_webhook_omits_none_fields(monkeypatch):
+    captured = {}
+    monkeypatch.setattr(nas_client, "_post", lambda p: captured.update(p))
+    nas_client.webhook_update_task("t2", "processing", 10, "working")
+    assert "image_url" not in captured and "error" not in captured