homelable/mcp/tests/test_tools.py

import pytest
from unittest.mock import AsyncMock, patch
from app.tools import TOOLS, _dispatch


@pytest.fixture
def mock_backend():
    with patch("app.tools.backend") as m:
        m.post = AsyncMock(return_value={"id": "1"})
        m.patch = AsyncMock(return_value={"id": "1"})
        m.delete = AsyncMock(return_value={})
        m.get = AsyncMock(return_value=[])
        yield m


@pytest.mark.anyio
async def test_create_node(mock_backend):
    result = await _dispatch("create_node", {"type": "server", "label": "Proxmox"})
    mock_backend.post.assert_called_once_with("/api/v1/nodes", {"type": "server", "label": "Proxmox"})
    assert result == {"id": "1"}


@pytest.mark.anyio
async def test_update_node(mock_backend):
    await _dispatch("update_node", {"id": "42", "label": "New name"})
    mock_backend.patch.assert_called_once_with("/api/v1/nodes/42", {"label": "New name"})


@pytest.mark.anyio
async def test_update_node_parent_id(mock_backend):
    await _dispatch("update_node", {"id": "42", "parent_id": "proxmox-1"})
    mock_backend.patch.assert_called_once_with("/api/v1/nodes/42", {"parent_id": "proxmox-1"})


@pytest.mark.anyio
async def test_create_node_full_properties(mock_backend):
    args = {
        "type": "proxmox",
        "label": "pve1",
        "os": "Proxmox VE 8",
        "notes": "Main hypervisor",
        "services": [{"name": "ssh", "port": 22}],
        "cpu_count": 16,
        "cpu_model": "Ryzen 9 5950X",
        "ram_gb": 64,
        "disk_gb": 2000,
        "show_hardware": True,
        "properties": [{"name": "rack", "value": "A1"}],
    }
    await _dispatch("create_node", dict(args))
    # All extra fields forwarded to the backend unchanged.
    mock_backend.post.assert_called_once_with("/api/v1/nodes", args)


@pytest.mark.anyio
async def test_update_node_properties(mock_backend):
    await _dispatch("update_node", {
        "id": "42",
        "os": "Debian 12",
        "properties": [{"name": "role", "value": "db"}],
    })
    mock_backend.patch.assert_called_once_with("/api/v1/nodes/42", {
        "os": "Debian 12",
        "properties": [{"name": "role", "value": "db"}],
    })


@pytest.mark.anyio
async def test_delete_node(mock_backend):
    await _dispatch("delete_node", {"id": "42"})
    mock_backend.delete.assert_called_once_with("/api/v1/nodes/42")


@pytest.mark.anyio
async def test_create_edge(mock_backend):
    await _dispatch("create_edge", {"source": "1", "target": "2", "type": "ethernet"})
    mock_backend.post.assert_called_once_with("/api/v1/edges", {"source": "1", "target": "2", "type": "ethernet"})


@pytest.mark.anyio
async def test_delete_edge(mock_backend):
    await _dispatch("delete_edge", {"id": "99"})
    mock_backend.delete.assert_called_once_with("/api/v1/edges/99")


@pytest.mark.anyio
async def test_trigger_scan_no_ranges(mock_backend):
    await _dispatch("trigger_scan", {})
    mock_backend.post.assert_called_once_with("/api/v1/scan/trigger", {})


@pytest.mark.anyio
async def test_trigger_scan_with_ranges(mock_backend):
    await _dispatch("trigger_scan", {"ranges": ["192.168.1.0/24"]})
    mock_backend.post.assert_called_once_with("/api/v1/scan/trigger", {"ranges": ["192.168.1.0/24"]})


@pytest.mark.anyio
async def test_approve_device(mock_backend):
    await _dispatch("approve_device", {"id": "5", "type": "server", "label": "MyServer"})
    mock_backend.post.assert_called_once_with("/api/v1/scan/pending/5/approve", {"type": "server", "label": "MyServer"})


@pytest.mark.anyio
async def test_hide_device(mock_backend):
    await _dispatch("hide_device", {"id": "5"})
    mock_backend.post.assert_called_once_with("/api/v1/scan/pending/5/hide", {})


@pytest.mark.anyio
async def test_get_canvas(mock_backend):
    mock_backend.get = AsyncMock(return_value={
        "nodes": [
            {
                "id": "n1",
                "type": "router",
                "position": {"x": 100, "y": 200},
                "width": 160,
                "height": 80,
                "data": {"label": "Freebox", "ip": "192.168.1.1", "status": "online"},
            }
        ],
        "edges": [
            {"id": "e1", "source": "n1", "target": "n2", "type": "ethernet", "animated": True, "style": {"stroke": "#fff"}},
        ],
        "viewport": {"x": 0, "y": 0, "zoom": 1},
    })
    result = await _dispatch("get_canvas", {})
    mock_backend.get.assert_called_once_with("/api/v1/canvas")
    # Layout/style fields stripped, only semantic data kept
    assert result["nodes"] == [{"id": "n1", "node_type": "router", "label": "Freebox", "ip": "192.168.1.1", "status": "online"}]
    assert result["edges"] == [{"id": "e1", "source": "n1", "target": "n2", "type": "ethernet"}]
    assert "viewport" not in result


@pytest.mark.anyio
async def test_get_canvas_keeps_documentation_fields(mock_backend):
    mock_backend.get = AsyncMock(return_value={
        "nodes": [
            {
                "id": "n1",
                "type": "proxmox",
                "position": {"x": 0, "y": 0},
                "data": {
                    "label": "pve1",
                    "os": "Proxmox VE 8",
                    "notes": "Main hypervisor",
                    "cpu_count": 16,
                    "ram_gb": 64,
                    "properties": [{"name": "rack", "value": "A1"}],
                },
            }
        ],
        "edges": [],
    })
    result = await _dispatch("get_canvas", {})
    node = result["nodes"][0]
    assert node["os"] == "Proxmox VE 8"
    assert node["notes"] == "Main hypervisor"
    assert node["cpu_count"] == 16
    assert node["ram_gb"] == 64
    assert node["properties"] == [{"name": "rack", "value": "A1"}]


def _tool_schema(name: str) -> dict:
    tool = next(t for t in TOOLS if t.name == name)
    return tool.inputSchema["properties"]


def test_create_node_schema_exposes_full_node_fields():
    props = _tool_schema("create_node")
    for field in ("os", "notes", "services", "cpu_count", "ram_gb", "disk_gb", "properties", "mac"):
        assert field in props, f"create_node schema missing {field}"
    # type stays an enum of the canonical node types
    assert "enum" in props["type"]


def test_update_node_schema_exposes_full_node_fields():
    props = _tool_schema("update_node")
    for field in ("os", "notes", "services", "cpu_count", "ram_gb", "disk_gb", "properties", "mac"):
        assert field in props, f"update_node schema missing {field}"
    assert "id" in props


@pytest.mark.anyio
async def test_list_nodes(mock_backend):
    mock_backend.get = AsyncMock(return_value=[{"id": "1", "label": "Freebox"}])
    result = await _dispatch("list_nodes", {})
    mock_backend.get.assert_called_once_with("/api/v1/nodes")
    assert result == [{"id": "1", "label": "Freebox"}]


@pytest.mark.anyio
async def test_list_pending_devices(mock_backend):
    mock_backend.get = AsyncMock(return_value=[{"id": "p1", "ip": "192.168.1.50"}])
    result = await _dispatch("list_pending_devices", {})
    mock_backend.get.assert_called_once_with("/api/v1/scan/pending")
    assert result == [{"id": "p1", "ip": "192.168.1.50"}]


@pytest.mark.anyio
async def test_unknown_tool():
    with pytest.raises(ValueError, match="Unknown tool"):
        await _dispatch("nonexistent", {})