112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""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 누락 의심)"
|