marcus
59f7fc1d6c
create_network:
- Switch from GET request() to post_request()
- name, subnet, gateway, iprange → json.dumps(value) per DSM convention
- disable_masquerade=json.dumps(False) added as required fixed param
delete_network:
- Switch from GET request() to post_request() with method "remove"
(was "delete" — wrong method name)
- Send full network object from list response as
networks=json.dumps([{...}]) including all DSM fields plus _key=id
(ipv6_gateway, ipv6_subnet, ipv6_iprange, disable_masquerade with
safe defaults for fields absent from the list response)
Tests updated: mock post_request, assert correct method/params structure.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
347 lines
11 KiB
Python
347 lines
11 KiB
Python
"""Tests for modules/networks.py."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
|
|
def make_mock_mcp():
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
|
|
return decorator
|
|
|
|
return MockMCP(), tools
|
|
|
|
|
|
def make_config():
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
|
|
return AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
|
|
|
|
SAMPLE_NETWORKS = {
|
|
"network": [
|
|
{
|
|
"id": "b741915823aa",
|
|
"name": "vault_default",
|
|
"driver": "bridge",
|
|
"subnet": "172.22.0.0/16",
|
|
"gateway": "172.22.0.1",
|
|
"iprange": "",
|
|
"enable_ipv6": False,
|
|
"containers": ["vault"],
|
|
},
|
|
{
|
|
"id": "2fc7ebae3901",
|
|
"name": "host",
|
|
"driver": "host",
|
|
"subnet": "",
|
|
"gateway": "",
|
|
"iprange": "",
|
|
"enable_ipv6": False,
|
|
"containers": [],
|
|
},
|
|
{
|
|
"id": "aabbcc112233",
|
|
"name": "my_bridge",
|
|
"driver": "bridge",
|
|
"subnet": "10.10.0.0/24",
|
|
"gateway": "10.10.0.1",
|
|
"iprange": "",
|
|
"enable_ipv6": False,
|
|
"containers": [],
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# list_networks
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_networks_shows_all():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["list_networks"]()
|
|
assert "vault_default" in result
|
|
assert "host" in result
|
|
assert "my_bridge" in result
|
|
assert "3 total" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_networks_shows_subnet_and_gateway():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["list_networks"]()
|
|
assert "172.22.0.0/16" in result
|
|
assert "172.22.0.1" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_networks_shows_attached_containers():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["list_networks"]()
|
|
assert "vault" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_networks_empty():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {"network": []}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["list_networks"]()
|
|
assert "No networks found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_networks_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("API error", code=102)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["list_networks"]()
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# create_network
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_network_preview():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["create_network"](name="mynet", driver="bridge")
|
|
assert "Preview" in result
|
|
assert "mynet" in result
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_network_confirmed():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={"id": "deadbeef1234567890"})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["create_network"](
|
|
name="mynet", driver="bridge", subnet="192.168.100.0/24", confirmed=True
|
|
)
|
|
assert "created" in result
|
|
assert "mynet" in result
|
|
|
|
params = client.post_request.call_args.kwargs.get("params", {})
|
|
assert json.loads(params["name"]) == "mynet"
|
|
assert params["driver"] == "bridge"
|
|
assert json.loads(params["subnet"]) == "192.168.100.0/24"
|
|
assert json.loads(params["enable_ipv6"]) is False
|
|
assert json.loads(params["disable_masquerade"]) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_network_with_ipv6():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={"id": "abc123"})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
await tools["create_network"](name="ipv6net", enable_ipv6=True, confirmed=True)
|
|
|
|
params = client.post_request.call_args.kwargs.get("params", {})
|
|
assert json.loads(params["enable_ipv6"]) is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_network_optional_params_not_sent():
|
|
"""subnet/gateway/ip_range must not appear in params when not provided."""
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
await tools["create_network"](name="bare", confirmed=True)
|
|
|
|
params = client.post_request.call_args.kwargs.get("params", {})
|
|
assert "subnet" not in params
|
|
assert "gateway" not in params
|
|
assert "iprange" not in params
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_network_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(side_effect=SynologyError("create failed", code=100))
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["create_network"](name="mynet", confirmed=True)
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# delete_network
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_network_preview():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_network"](name="my_bridge")
|
|
assert "Preview" in result
|
|
assert "my_bridge" in result
|
|
assert "confirmed=True" in result
|
|
# Only the list GET was called; post_request must not be called
|
|
client.request.assert_called_once()
|
|
assert client.request.call_args.args[1] == "list"
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_network_confirmed():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_network"](name="my_bridge", confirmed=True)
|
|
assert "deleted" in result
|
|
assert "my_bridge" in result
|
|
|
|
# post_request must use method "remove" with networks JSON array
|
|
post_call = client.post_request.call_args
|
|
assert post_call.args[1] == "remove"
|
|
params = post_call.kwargs.get("params", {})
|
|
networks_list = json.loads(params["networks"])
|
|
assert len(networks_list) == 1
|
|
obj = networks_list[0]
|
|
assert obj["name"] == "my_bridge"
|
|
assert obj["id"] == "aabbcc112233"
|
|
assert obj["_key"] == "aabbcc112233"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_network_not_found():
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_network"](name="nonexistent", confirmed=True)
|
|
assert "not found" in result
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_network_blocked_by_containers():
|
|
"""Deletion must be refused when containers are attached."""
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
# vault_default has ["vault"] attached
|
|
result = await tools["delete_network"](name="vault_default", confirmed=True)
|
|
assert "Cannot delete" in result
|
|
assert "vault" in result
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_network_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.networks import register_networks
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_NETWORKS
|
|
client.post_request = AsyncMock(side_effect=SynologyError("remove failed", code=100))
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_networks(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_network"](name="my_bridge", confirmed=True)
|
|
assert "Error" in result
|