164 lines
4.5 KiB
Python
164 lines
4.5 KiB
Python
"""Tests for modules/projects.py."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from mcp_synology_container.modules.projects import _find_project, _format_project_detail
|
|
|
|
|
|
SAMPLE_PROJECTS = {
|
|
"uuid-1": {
|
|
"id": "uuid-1",
|
|
"name": "myapp",
|
|
"status": "RUNNING",
|
|
"path": "/volume1/docker/myapp",
|
|
"share_path": "/docker/myapp",
|
|
"created_at": "2025-01-01T00:00:00Z",
|
|
"updated_at": "2025-01-02T00:00:00Z",
|
|
"containerIds": ["abc123def456"],
|
|
"services": [{"display_name": "myapp (project)"}],
|
|
},
|
|
"uuid-2": {
|
|
"id": "uuid-2",
|
|
"name": "database",
|
|
"status": "STOPPED",
|
|
"path": "/volume1/docker/database",
|
|
"share_path": "/docker/database",
|
|
"created_at": "2025-01-01T00:00:00Z",
|
|
"updated_at": "2025-01-01T00:00:00Z",
|
|
"containerIds": [],
|
|
"services": [],
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_project_found():
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_PROJECTS
|
|
|
|
result = await _find_project(client, "myapp")
|
|
assert result is not None
|
|
assert result["name"] == "myapp"
|
|
assert result["status"] == "RUNNING"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_project_not_found():
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_PROJECTS
|
|
|
|
result = await _find_project(client, "nonexistent")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_project_api_error():
|
|
client = AsyncMock()
|
|
client.request.side_effect = Exception("API error")
|
|
|
|
result = await _find_project(client, "myapp")
|
|
assert result is None
|
|
|
|
|
|
def test_format_project_detail():
|
|
project = SAMPLE_PROJECTS["uuid-1"]
|
|
output = _format_project_detail(project)
|
|
|
|
assert "myapp" in output
|
|
assert "RUNNING" in output
|
|
assert "/volume1/docker/myapp" in output
|
|
assert "uuid-1" in output
|
|
|
|
|
|
def test_format_project_detail_no_containers():
|
|
project = SAMPLE_PROJECTS["uuid-2"]
|
|
output = _format_project_detail(project)
|
|
|
|
assert "database" in output
|
|
assert "STOPPED" in output
|
|
assert "Containers: 0" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_projects_tool():
|
|
"""Test list_projects tool via function registration."""
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
|
|
config = AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_PROJECTS
|
|
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
return decorator
|
|
|
|
register_projects(MockMCP(), config, client)
|
|
assert "list_projects" in tools
|
|
|
|
result = await tools["list_projects"]()
|
|
assert "myapp" in result
|
|
assert "database" in result
|
|
assert "RUNNING" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_project_requires_confirmation():
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
|
|
config = AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
client = AsyncMock()
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
return decorator
|
|
|
|
register_projects(MockMCP(), config, client)
|
|
|
|
result = await tools["stop_project"]("myapp", confirmed=False)
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_project_requires_confirmation():
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
|
|
config = AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
client = AsyncMock()
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
return decorator
|
|
|
|
register_projects(MockMCP(), config, client)
|
|
|
|
result = await tools["redeploy_project"]("myapp", confirmed=False)
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|