diff --git a/packs-lab/tests/test_dsm_client.py b/packs-lab/tests/test_dsm_client.py new file mode 100644 index 0000000..e488402 --- /dev/null +++ b/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 누락 의심)"