Pouzor
7935b671d3
Exposes homelab topology to MCP-compatible AI clients (Claude Code, etc.) over LAN via HTTP/SSE on port 8001. - New mcp/ service: FastAPI + mcp SDK, SSE transport - Auth: X-API-Key for AI clients, X-MCP-Service-Key for backend (Docker-internal) - Resources: canvas, nodes, edges, scan/pending, scan/runs - Tools: create/update/delete nodes+edges, trigger scan, approve/hide devices - Backend deps.py: accepts JWT or MCP service key (no plain-text password) - 40 tests (auth, resources, tools) across asyncio + trio - docker-compose.yml: mcp service on port 8001 - README: MCP setup section with Claude Code/Desktop config examples
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
from app.tools import _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={})
|
|
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_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_unknown_tool():
|
|
with pytest.raises(ValueError, match="Unknown tool"):
|
|
await _dispatch("nonexistent", {})
|