marcus/mcp-synology-container
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix create_network + delete_network: POST with DevTools-confirmed format

Browse Source 
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>
This commit is contained in:
Marcus van Elst
2026-04-13 18:56:21 +02:00
parent 4cee16922f
commit 59f7fc1d6c
2 changed files with 67 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mcp_synology_container/modules/networks.py
+32 -6
View File
@@ -89,20 +89,22 @@ def register_networks(mcp: FastMCP, config: AppConfig, client: DsmClient) -> Non
+ f"\n\nCall create_network(name={name!r}, confirmed=True) to confirm."
)
# DevTools-confirmed POST format: all string/bool params as json.dumps.
params: dict[str, Any] = {
"name": name,
"name": json.dumps(name),
"driver": driver,
"enable_ipv6": json.dumps(enable_ipv6),
"disable_masquerade": json.dumps(False),
}
if subnet is not None:
params["subnet"] = subnet
params["subnet"] = json.dumps(subnet)
if gateway is not None:
params["gateway"] = gateway
params["gateway"] = json.dumps(gateway)
if ip_range is not None:
params["iprange"] = ip_range
params["iprange"] = json.dumps(ip_range)
try:
result = await client.request("SYNO.Docker.Network", "create", params=params)
result = await client.post_request("SYNO.Docker.Network", "create", params=params)
except Exception as e:
return f"Error creating network '{name}': {e}"
@@ -147,8 +149,32 @@ def register_networks(mcp: FastMCP, config: AppConfig, client: DsmClient) -> Non
f"Call delete_network(name={name!r}, confirmed=True) to confirm."
)
# DevTools-confirmed POST format: method="remove", full network object
# as a JSON array in the "networks" parameter. The object must include
# all fields from the list response plus "_key" set to the network ID.
net_id = target.get("id", "")
network_obj: dict[str, Any] = {
"containers": target.get("containers") or [],
"driver": target.get("driver", ""),
"enable_ipv6": target.get("enable_ipv6", False),
"ipv6_gateway": target.get("ipv6_gateway", ""),
"ipv6_subnet": target.get("ipv6_subnet", ""),
"ipv6_iprange": target.get("ipv6_iprange", ""),
"gateway": target.get("gateway", ""),
"id": net_id,
"iprange": target.get("iprange", ""),
"name": name,
"subnet": target.get("subnet", ""),
"disable_masquerade": target.get("disable_masquerade", False),
"_key": net_id,
}
try:
await client.request("SYNO.Docker.Network", "delete", params={"name": name})
await client.post_request(
"SYNO.Docker.Network",
"remove",
params={"networks": json.dumps([network_obj])},
)
except Exception as e:
return f"Error deleting network '{name}': {e}"
tests/test_modules/test_networks.py
+35 -33
View File
@@ -155,6 +155,7 @@ 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)
@@ -164,6 +165,7 @@ async def test_create_network_preview():
assert "mynet" in result
assert "confirmed=True" in result
client.request.assert_not_called()
client.post_request.assert_not_called()
@pytest.mark.asyncio
@@ -171,7 +173,7 @@ async def test_create_network_confirmed():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {"id": "deadbeef1234567890"}
client.post_request = AsyncMock(return_value={"id": "deadbeef1234567890"})
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
@@ -182,12 +184,12 @@ async def test_create_network_confirmed():
assert "created" in result
assert "mynet" in result
call = client.request.call_args
params = call.kwargs.get("params") or {}
assert params["name"] == "mynet"
params = client.post_request.call_args.kwargs.get("params", {})
assert json.loads(params["name"]) == "mynet"
assert params["driver"] == "bridge"
assert params["subnet"] == "192.168.100.0/24"
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
@@ -195,15 +197,14 @@ async def test_create_network_with_ipv6():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {"id": "abc123"}
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)
call = client.request.call_args
params = call.kwargs.get("params") or {}
params = client.post_request.call_args.kwargs.get("params", {})
assert json.loads(params["enable_ipv6"]) is True
@@ -213,15 +214,14 @@ async def test_create_network_optional_params_not_sent():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.return_value = {}
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)
call = client.request.call_args
params = call.kwargs.get("params") or {}
params = client.post_request.call_args.kwargs.get("params", {})
assert "subnet" not in params
assert "gateway" not in params
assert "iprange" not in params
@@ -233,7 +233,7 @@ async def test_create_network_api_error():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
client.request.side_effect = SynologyError("create failed", code=100)
client.post_request = AsyncMock(side_effect=SynologyError("create failed", code=100))
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
@@ -253,6 +253,7 @@ async def test_delete_network_preview():
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)
@@ -261,9 +262,10 @@ async def test_delete_network_preview():
assert "Preview" in result
assert "my_bridge" in result
assert "confirmed=True" in result
# Only the list call should have been made, not delete
assert client.request.call_count == 1
# 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
@@ -271,15 +273,8 @@ async def test_delete_network_confirmed():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
async def mock_request(api, method, **kwargs):
if method == "list":
return SAMPLE_NETWORKS
if method == "delete":
return {}
return {}
client.request.side_effect = mock_request
client.request.return_value = SAMPLE_NETWORKS
client.post_request = AsyncMock(return_value={})
mcp, tools = make_mock_mcp()
register_networks(mcp, make_config(), client)
@@ -288,6 +283,17 @@ async def test_delete_network_confirmed():
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():
@@ -295,12 +301,14 @@ async def test_delete_network_not_found():
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
@@ -310,6 +318,7 @@ async def test_delete_network_blocked_by_containers():
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)
@@ -318,9 +327,7 @@ async def test_delete_network_blocked_by_containers():
result = await tools["delete_network"](name="vault_default", confirmed=True)
assert "Cannot delete" in result
assert "vault" in result
# delete API must not be called
assert client.request.call_count == 1
assert client.request.call_args.args[1] == "list"
client.post_request.assert_not_called()
@pytest.mark.asyncio
@@ -329,13 +336,8 @@ async def test_delete_network_api_error():
from mcp_synology_container.modules.networks import register_networks
client = AsyncMock()
async def mock_request(api, method, **kwargs):
if method == "list":
return SAMPLE_NETWORKS
raise SynologyError("delete failed", code=100)
client.request.side_effect = mock_request
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)