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_create_node_with_design_id(mock_backend):
    # design_id is forwarded to the backend, which attaches the node to that canvas.
    args = {"type": "server", "label": "Proxmox", "design_id": "design-2"}
    await _dispatch("create_node", dict(args))
    mock_backend.post.assert_called_once_with("/api/v1/nodes", args)


@pytest.mark.anyio
async def test_create_edge_with_design_id(mock_backend):
    args = {"source": "1", "target": "2", "type": "ethernet", "design_id": "design-2"}
    await _dispatch("create_edge", dict(args))
    mock_backend.post.assert_called_once_with("/api/v1/edges", args)


@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"}]


@pytest.mark.anyio
async def test_get_canvas_with_design_id(mock_backend):
    mock_backend.get = AsyncMock(return_value={"nodes": [], "edges": []})
    await _dispatch("get_canvas", {"design_id": "design-2"})
    mock_backend.get.assert_called_once_with("/api/v1/canvas?design_id=design-2")


@pytest.mark.anyio
async def test_get_canvas_without_design_id_uses_default(mock_backend):
    # Backward compatible: no design_id -> unqualified canvas endpoint (first design).
    mock_backend.get = AsyncMock(return_value={"nodes": [], "edges": []})
    await _dispatch("get_canvas", {})
    mock_backend.get.assert_called_once_with("/api/v1/canvas")


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


def test_design_id_exposed_on_canvas_targeting_tools():
    # create_node/create_edge/get_canvas can target a specific canvas.
    assert "design_id" in _tool_schema("create_node")
    assert "design_id" in _tool_schema("create_edge")
    assert "design_id" in _tool_schema("get_canvas")


def test_update_node_schema_has_no_design_id():
    # The backend NodeUpdate schema can't move a node between designs, so the
    # update_node tool must not advertise design_id.
    assert "design_id" not in _tool_schema("update_node")


@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_list_designs(mock_backend):
    mock_backend.get = AsyncMock(return_value=[{"id": "d1", "name": "Network Topology", "node_count": 12}])
    result = await _dispatch("list_designs", {})
    mock_backend.get.assert_called_once_with("/api/v1/designs")
    assert result == [{"id": "d1", "name": "Network Topology", "node_count": 12}]


@pytest.mark.anyio
async def test_create_design(mock_backend):
    mock_backend.post = AsyncMock(return_value={"id": "d2", "name": "Scan Devices"})
    result = await _dispatch("create_design", {"name": "Scan Devices"})
    mock_backend.post.assert_called_once_with("/api/v1/designs", {"name": "Scan Devices"})
    assert result == {"id": "d2", "name": "Scan Devices"}


def test_create_design_schema_requires_name():
    tool = next(t for t in TOOLS if t.name == "create_design")
    assert tool.inputSchema["required"] == ["name"]


@pytest.mark.anyio
async def test_delete_design(mock_backend):
    await _dispatch("delete_design", {"design_id": "d-old"})
    mock_backend.delete.assert_called_once_with("/api/v1/designs/d-old")


def test_delete_design_schema_requires_design_id():
    tool = next(t for t in TOOLS if t.name == "delete_design")
    assert tool.inputSchema["required"] == ["design_id"]
    assert "design_id" in tool.inputSchema["properties"]


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


@pytest.mark.anyio
async def test_list_pending_devices_filters_non_pending(mock_backend):
    """The tool promises devices *not yet approved or hidden*; the backend
    endpoint returns the whole inventory including approved rows (they carry
    the canvas-presence badge). The tool must filter to status == "pending"
    and keep legacy rows that lack the field."""
    mock_backend.get = AsyncMock(return_value=[
        {"id": "p1", "ip": "192.168.1.50", "status": "pending"},
        {"id": "a1", "ip": "192.168.1.60", "status": "approved"},
        {"id": "h1", "ip": "192.168.1.70", "status": "hidden"},
        {"id": "legacy", "ip": "192.168.1.80"},
    ])
    result = await _dispatch("list_pending_devices", {})
    assert [d["id"] for d in result] == ["p1", "legacy"]


_INVENTORY = [
    {"id": "p1", "ip": "192.168.1.50", "status": "pending"},
    {"id": "a1", "ip": "192.168.1.60", "status": "approved"},
    {"id": "legacy", "ip": "192.168.1.80"},
]


@pytest.mark.anyio
async def test_list_inventory_default_returns_all(mock_backend):
    """No status filter returns the whole non-hidden inventory verbatim."""
    mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
    result = await _dispatch("list_inventory", {})
    mock_backend.get.assert_called_once_with("/api/v1/scan/pending")
    assert [d["id"] for d in result] == ["p1", "a1", "legacy"]


@pytest.mark.anyio
async def test_list_inventory_filters_approved(mock_backend):
    mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
    result = await _dispatch("list_inventory", {"status": "approved"})
    assert [d["id"] for d in result] == ["a1"]


@pytest.mark.anyio
async def test_list_inventory_pending_keeps_legacy(mock_backend):
    """Legacy rows without a status field count as pending."""
    mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
    result = await _dispatch("list_inventory", {"status": "pending"})
    assert [d["id"] for d in result] == ["p1", "legacy"]


@pytest.mark.anyio
async def test_list_hidden_devices(mock_backend):
    await _dispatch("list_hidden_devices", {})
    mock_backend.get.assert_called_once_with("/api/v1/scan/hidden")


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