marcus
bafa327412
redeploy_project: replace broken SYNO.Docker.Image/pull with a unified 4-step delete-before-start flow for all project states (RUNNING, STOPPED, BUILD_FAILED). Reads image tags from the project's compose.yaml via FileStation before stopping, deletes each cached image (non-fatal), then starts the project so DSM auto-pulls the latest version. Polls for RUNNING as before. update_image_tag: auto-update env vars whose value equals the numeric version prefix of the old tag when the new tag shares the same <digits>-<suffix> pattern (e.g. JENKINS_VERSION=2.558 → 2.560 when tag changes 2.558-jdk21 → 2.560-jdk21). Preview mode lists the pending auto-updates. Only triggers when the var exists and the pattern matches. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
406 lines
13 KiB
Python
406 lines
13 KiB
Python
"""Tests for modules/projects.py."""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
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.config import AppConfig, ConnectionConfig
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
|
|
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.config import AppConfig, ConnectionConfig
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
|
|
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.config import AppConfig, ConnectionConfig
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
|
|
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()
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Bug 2: status-aware redeploy
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def make_projects_tools(client):
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
from mcp_synology_container.modules.projects import register_projects
|
|
|
|
config = AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
|
|
return decorator
|
|
|
|
register_projects(MockMCP(), config, client)
|
|
return tools
|
|
|
|
|
|
def project_list(status: str) -> dict:
|
|
return {
|
|
"uuid-1": {
|
|
"id": "uuid-1",
|
|
"name": "myapp",
|
|
"status": status,
|
|
"path": "/volume1/docker/myapp",
|
|
"containerIds": ["abc123"],
|
|
"services": [],
|
|
}
|
|
}
|
|
|
|
|
|
def make_stateful_redeploy_mock(initial_status: str, stop_raises=None):
|
|
"""Create a stateful client mock for redeploy tests.
|
|
|
|
Returns (client, calls_list). After ``start`` is called, subsequent
|
|
``list`` calls return RUNNING so the polling loop terminates immediately.
|
|
asyncio.sleep is NOT patched here — patch it at call-site.
|
|
|
|
FileStation.List returns an empty file list so compose image detection is
|
|
skipped (image deletion is tested separately).
|
|
"""
|
|
client = AsyncMock()
|
|
calls = []
|
|
start_called = False
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
nonlocal start_called
|
|
calls.append((api, method))
|
|
if api == "SYNO.FileStation.List":
|
|
return {"files": []} # No compose file → skip image deletion
|
|
if method == "start":
|
|
start_called = True
|
|
if method == "stop" and stop_raises:
|
|
raise stop_raises
|
|
if method == "list":
|
|
return project_list("RUNNING") if start_called else project_list(initial_status)
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
client.post_request = AsyncMock()
|
|
return client, calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_running_project():
|
|
"""RUNNING project: stop then start; polls until RUNNING."""
|
|
client, calls = make_stateful_redeploy_mock("RUNNING")
|
|
tools = make_projects_tools(client)
|
|
|
|
with patch("mcp_synology_container.modules.projects.asyncio.sleep"):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "redeployed successfully" in result
|
|
methods = [m for _, m in calls]
|
|
assert "stop" in methods
|
|
assert "start" in methods
|
|
assert methods.index("stop") < methods.index("start")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_stopped_project_starts_directly():
|
|
"""STOPPED project: skip stop, just start; polls until RUNNING."""
|
|
client, calls = make_stateful_redeploy_mock("STOPPED")
|
|
tools = make_projects_tools(client)
|
|
|
|
with patch("mcp_synology_container.modules.projects.asyncio.sleep"):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "redeployed successfully" in result
|
|
methods = [m for _, m in calls]
|
|
assert "stop" not in methods
|
|
assert "start" in methods
|
|
assert "STOPPED" in result or "starting directly" in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_build_failed_project():
|
|
"""BUILD_FAILED project: stop → (delete images) → start; polls until RUNNING."""
|
|
client, calls = make_stateful_redeploy_mock("BUILD_FAILED")
|
|
tools = make_projects_tools(client)
|
|
|
|
with patch("mcp_synology_container.modules.projects.asyncio.sleep"):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "redeployed successfully" in result
|
|
methods = [m for _, m in calls]
|
|
assert "stop" in methods
|
|
assert "start" in methods
|
|
assert methods.index("stop") < methods.index("start")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_build_failed_stop_error_nonfatal():
|
|
"""BUILD_FAILED: stop failure is non-fatal and must not abort the redeploy."""
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
|
|
client, _ = make_stateful_redeploy_mock(
|
|
"BUILD_FAILED",
|
|
stop_raises=SynologyError("already stopped", code=2101),
|
|
)
|
|
tools = make_projects_tools(client)
|
|
|
|
with patch("mcp_synology_container.modules.projects.asyncio.sleep"):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "redeployed successfully" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_image_delete_failure_nonfatal():
|
|
"""Image deletion failure must be non-fatal: start must still be called."""
|
|
client = AsyncMock()
|
|
start_called = False
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
nonlocal start_called
|
|
if api == "SYNO.FileStation.List":
|
|
# Return a compose file so that image listing is attempted
|
|
return {"files": [{"name": "docker-compose.yml"}]}
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
# Return one image matching the compose service
|
|
return {
|
|
"images": [
|
|
{
|
|
"id": "sha256:abc123",
|
|
"repository": "nginx",
|
|
"tags": ["1.24"],
|
|
"size": 50000000,
|
|
}
|
|
]
|
|
}
|
|
if api == "SYNO.Docker.Container" and method == "list":
|
|
return {"containers": []}
|
|
if method == "start":
|
|
start_called = True
|
|
if method == "list":
|
|
return project_list("RUNNING") if start_called else project_list("RUNNING")
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
# Simulate FileStation download of compose.yaml
|
|
client.download_text = AsyncMock(return_value="services:\n web:\n image: nginx:1.24\n")
|
|
# post_request (image delete) raises an error — must be non-fatal
|
|
client.post_request = AsyncMock(side_effect=Exception("delete failed"))
|
|
|
|
tools = make_projects_tools(client)
|
|
|
|
with patch("mcp_synology_container.modules.projects.asyncio.sleep"):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "redeployed successfully" in result
|
|
assert start_called, "start must be called even when image deletion fails"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_poll_timeout():
|
|
"""If project never reaches RUNNING after start, a warning is emitted."""
|
|
client = AsyncMock()
|
|
start_called = False
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
nonlocal start_called
|
|
if api == "SYNO.FileStation.List":
|
|
return {"files": []}
|
|
if method == "start":
|
|
start_called = True
|
|
if method == "list":
|
|
# Before start: return RUNNING so initial status check picks a valid path.
|
|
# After start: return STARTING to simulate a stuck container — triggers timeout.
|
|
return project_list("STARTING") if start_called else project_list("RUNNING")
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
client.post_request = AsyncMock()
|
|
tools = make_projects_tools(client)
|
|
|
|
# Use tiny timeout so the test is instant (interval=1, timeout=1 → 1 poll)
|
|
with (
|
|
patch("mcp_synology_container.modules.projects.asyncio.sleep"),
|
|
patch("mcp_synology_container.modules.projects._POLL_TIMEOUT", 1),
|
|
patch("mcp_synology_container.modules.projects._POLL_INTERVAL", 1),
|
|
):
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "Warning" in result
|
|
assert "redeployed successfully" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redeploy_unknown_status_returns_error():
|
|
"""Unknown status must return a clear error with a workaround hint."""
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.FileStation.List":
|
|
return {"files": []}
|
|
if method == "list":
|
|
return project_list("UPDATING")
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
client.post_request = AsyncMock()
|
|
tools = make_projects_tools(client)
|
|
|
|
result = await tools["redeploy_project"]("myapp", confirmed=True)
|
|
|
|
assert "UPDATING" in result
|
|
assert "Workaround" in result or "stop_project" in result
|