test(packs-lab): DSM client mock 테스트 (login/share/logout 순서)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

packs-lab/tests/test_dsm_client.py

@@ -0,0 +1,111 @@
+"""DSM 7.x API client 테스트 — httpx mock으로 외부 호출 차단."""
+import asyncio
+from unittest.mock import patch, MagicMock
+
+import pytest
+import httpx
+
+from app.dsm_client import create_share_link, DSMError
+
+
+@pytest.fixture(autouse=True)
+def _dsm_env(monkeypatch):
+    monkeypatch.setenv("DSM_HOST", "https://test-nas:5001")
+    monkeypatch.setenv("DSM_USER", "test-user")
+    monkeypatch.setenv("DSM_PASS", "test-pass")
+    from app import dsm_client
+    monkeypatch.setattr(dsm_client, "DSM_HOST", "https://test-nas:5001")
+    monkeypatch.setattr(dsm_client, "DSM_USER", "test-user")
+    monkeypatch.setattr(dsm_client, "DSM_PASS", "test-pass")
+
+
+def _make_response(json_data, status_code=200):
+    """httpx.Response mock."""
+    mock = MagicMock(spec=httpx.Response)
+    mock.json.return_value = json_data
+    mock.status_code = status_code
+    mock.raise_for_status = MagicMock()
+    return mock
+
+
+def test_create_share_link_login_logout():
+    """login → Sharing.create → logout 순서가 보장되어야 한다."""
+    call_order = []
+
+    async def fake_get(self, url, *, params=None, **kw):
+        api = (params or {}).get("api", "")
+        method = (params or {}).get("method", "")
+        call_order.append(f"{api}.{method}")
+        if api == "SYNO.API.Auth" and method == "login":
+            return _make_response({"success": True, "data": {"sid": "fake-sid"}})
+        if api == "SYNO.API.Auth" and method == "logout":
+            return _make_response({"success": True})
+        if api == "SYNO.FileStation.Sharing" and method == "create":
+            return _make_response({
+                "success": True,
+                "data": {"links": [{"url": "https://test-nas:5001/sharing/abc"}]},
+            })
+        return _make_response({"success": False, "error": "unexpected"})
+
+    with patch.object(httpx.AsyncClient, "get", new=fake_get):
+        url, expires_at = asyncio.run(create_share_link("/volume1/test/file.zip", expires_in_sec=3600))
+
+    assert url == "https://test-nas:5001/sharing/abc"
+    assert call_order == [
+        "SYNO.API.Auth.login",
+        "SYNO.FileStation.Sharing.create",
+        "SYNO.API.Auth.logout",
+    ]
+
+
+def test_create_share_link_returns_url_and_expiry():
+    """응답 파싱 — links[0].url 사용."""
+    async def fake_get(self, url, *, params=None, **kw):
+        method = (params or {}).get("method", "")
+        if method == "login":
+            return _make_response({"success": True, "data": {"sid": "sid"}})
+        if method == "create":
+            return _make_response({
+                "success": True,
+                "data": {"links": [{"url": "https://nas/sharing/xyz"}]},
+            })
+        return _make_response({"success": True})
+
+    with patch.object(httpx.AsyncClient, "get", new=fake_get):
+        url, expires_at = asyncio.run(create_share_link("/volume1/test/file.zip", expires_in_sec=7200))
+
+    assert url == "https://nas/sharing/xyz"
+    assert expires_at is not None
+
+
+def test_dsm_login_failure_raises():
+    """login API success=False → DSMError."""
+    async def fake_get(self, url, *, params=None, **kw):
+        return _make_response({"success": False, "error": {"code": 400}})
+
+    with patch.object(httpx.AsyncClient, "get", new=fake_get):
+        with pytest.raises(DSMError, match="login 실패"):
+            asyncio.run(create_share_link("/volume1/test/file.zip"))
+
+
+def test_dsm_share_failure_logs_out():
+    """Sharing.create 실패해도 logout 호출 (try/finally)."""
+    call_order = []
+
+    async def fake_get(self, url, *, params=None, **kw):
+        method = (params or {}).get("method", "")
+        call_order.append(method)
+        if method == "login":
+            return _make_response({"success": True, "data": {"sid": "sid"}})
+        if method == "create":
+            return _make_response({"success": False, "error": {"code": 401}})
+        if method == "logout":
+            return _make_response({"success": True})
+        return _make_response({"success": False})
+
+    with patch.object(httpx.AsyncClient, "get", new=fake_get):
+        with pytest.raises(DSMError, match="Sharing.create 실패"):
+            asyncio.run(create_share_link("/volume1/test/file.zip"))
+
+    assert "login" in call_order
+    assert "logout" in call_order, "logout이 호출되지 않음 (finally 누락 의심)"