test: improve coverage across frontend and backend

Frontend (22 → 37 tests):
- canvasStore: add tests for onNodesChange, onEdgesChange, onConnect,
  addNode with parentId, updateEdge, deleteEdge, setProxmoxContainerMode
  ON/OFF, loadCanvas parent-before-child ordering
- edgeColors: 6 tests for EDGE_DEFAULT_COLORS (all types, hex format, values)

Backend (40 → 80 tests):
- test_canvas (new): load empty canvas, default viewport, save creates/updates/
  deletes nodes+edges, viewport upsert, custom_colors, edge custom_color+path_style,
  auth guard
- test_fingerprint (new): match_port (known, unknown, wrong protocol, banner match,
  banner no-match, no banner), fingerprint_ports (matched, unknown, mixed, empty,
  default protocol), suggest_node_type (proxmox, server, generic, priority)
- test_nodes: update/delete 404, custom_colors CRUD, container_mode CRUD, auth guard
- test_edges: update edge, update/delete 404, custom_color, path_style, auth guard
This commit is contained in:
Pouzor
2026-03-07 15:01:00 +01:00
parent 07d8c4e58b
commit ba91d0f545
6 changed files with 458 additions and 0 deletions
+125
View File
@@ -0,0 +1,125 @@
import uuid
from unittest.mock import patch
import pytest
from httpx import AsyncClient
TOKEN_HASH = "$2b$12$o/LWyvmBc978CNpSsHxcveXN0WqjAGW/gBR0.U.HURWbaYD3GCDqS"
@pytest.fixture
async def headers(client: AsyncClient):
with patch("app.api.routes.auth._load_credentials", return_value=("admin", TOKEN_HASH)):
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
return {"Authorization": f"Bearer {res.json()['access_token']}"}
def node_payload(**kwargs):
return {"id": str(uuid.uuid4()), "type": "server", "label": "N", "status": "unknown", "pos_x": 0, "pos_y": 0, **kwargs}
def edge_payload(src, tgt, **kwargs):
return {"id": str(uuid.uuid4()), "source": src, "target": tgt, "type": "ethernet", **kwargs}
# ── load_canvas ───────────────────────────────────────────────────────────────
async def test_load_canvas_empty(client: AsyncClient, headers: dict):
res = await client.get("/api/v1/canvas", headers=headers)
assert res.status_code == 200
data = res.json()
assert data["nodes"] == []
assert data["edges"] == []
assert data["viewport"] == {"x": 0, "y": 0, "zoom": 1}
async def test_load_canvas_requires_auth(client: AsyncClient):
res = await client.get("/api/v1/canvas")
assert res.status_code == 403
# ── save_canvas ───────────────────────────────────────────────────────────────
async def test_save_canvas_creates_nodes_and_edges(client: AsyncClient, headers: dict):
n1 = node_payload(label="Router", type="router")
n2 = node_payload(label="Switch", type="switch")
e1 = edge_payload(n1["id"], n2["id"])
res = await client.post("/api/v1/canvas/save", json={"nodes": [n1, n2], "edges": [e1], "viewport": {"x": 1, "y": 2, "zoom": 1.5}}, headers=headers)
assert res.status_code == 200
assert res.json() == {"saved": True}
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert len(canvas["nodes"]) == 2
assert len(canvas["edges"]) == 1
assert canvas["viewport"] == {"x": 1, "y": 2, "zoom": 1.5}
async def test_save_canvas_updates_existing_node(client: AsyncClient, headers: dict):
n1 = node_payload(label="Old Label")
await client.post("/api/v1/canvas/save", json={"nodes": [n1], "edges": [], "viewport": {}}, headers=headers)
n1_updated = {**n1, "label": "New Label", "ip": "10.0.0.1"}
await client.post("/api/v1/canvas/save", json={"nodes": [n1_updated], "edges": [], "viewport": {}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert len(canvas["nodes"]) == 1
assert canvas["nodes"][0]["label"] == "New Label"
assert canvas["nodes"][0]["ip"] == "10.0.0.1"
async def test_save_canvas_deletes_removed_nodes(client: AsyncClient, headers: dict):
n1 = node_payload(label="Keep")
n2 = node_payload(label="Remove")
await client.post("/api/v1/canvas/save", json={"nodes": [n1, n2], "edges": [], "viewport": {}}, headers=headers)
await client.post("/api/v1/canvas/save", json={"nodes": [n1], "edges": [], "viewport": {}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert len(canvas["nodes"]) == 1
assert canvas["nodes"][0]["label"] == "Keep"
async def test_save_canvas_deletes_removed_edges(client: AsyncClient, headers: dict):
n1 = node_payload()
n2 = node_payload()
e1 = edge_payload(n1["id"], n2["id"])
await client.post("/api/v1/canvas/save", json={"nodes": [n1, n2], "edges": [e1], "viewport": {}}, headers=headers)
await client.post("/api/v1/canvas/save", json={"nodes": [n1, n2], "edges": [], "viewport": {}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert canvas["edges"] == []
async def test_save_canvas_persists_viewport_on_update(client: AsyncClient, headers: dict):
await client.post("/api/v1/canvas/save", json={"nodes": [], "edges": [], "viewport": {"x": 10, "y": 20, "zoom": 2}}, headers=headers)
await client.post("/api/v1/canvas/save", json={"nodes": [], "edges": [], "viewport": {"x": 5, "y": 5, "zoom": 0.5}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert canvas["viewport"] == {"x": 5, "y": 5, "zoom": 0.5}
async def test_save_canvas_persists_custom_colors(client: AsyncClient, headers: dict):
n1 = node_payload(custom_colors={"border": "#ff0000", "icon": "#00ff00"})
await client.post("/api/v1/canvas/save", json={"nodes": [n1], "edges": [], "viewport": {}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
assert canvas["nodes"][0]["custom_colors"] == {"border": "#ff0000", "icon": "#00ff00"}
async def test_save_canvas_persists_edge_custom_color_and_path_style(client: AsyncClient, headers: dict):
n1 = node_payload()
n2 = node_payload()
e1 = edge_payload(n1["id"], n2["id"], custom_color="#a855f7", path_style="smooth")
await client.post("/api/v1/canvas/save", json={"nodes": [n1, n2], "edges": [e1], "viewport": {}}, headers=headers)
canvas = (await client.get("/api/v1/canvas", headers=headers)).json()
edge = canvas["edges"][0]
assert edge["custom_color"] == "#a855f7"
assert edge["path_style"] == "smooth"
async def test_save_canvas_requires_auth(client: AsyncClient):
res = await client.post("/api/v1/canvas/save", json={"nodes": [], "edges": [], "viewport": {}})
assert res.status_code == 403
+48
View File
@@ -52,3 +52,51 @@ async def test_delete_edge(client: AsyncClient, headers: dict, two_nodes):
res = await client.delete(f"/api/v1/edges/{edge_id}", headers=headers)
assert res.status_code == 204
assert len((await client.get("/api/v1/edges", headers=headers)).json()) == 0
async def test_delete_edge_not_found(client: AsyncClient, headers: dict):
res = await client.delete("/api/v1/edges/nonexistent", headers=headers)
assert res.status_code == 404
async def test_update_edge(client: AsyncClient, headers: dict, two_nodes):
src, tgt = two_nodes
edge_id = (await client.post("/api/v1/edges", json={"source": src, "target": tgt, "type": "ethernet"}, headers=headers)).json()["id"]
res = await client.patch(f"/api/v1/edges/{edge_id}", json={"type": "wifi", "label": "uplink"}, headers=headers)
assert res.status_code == 200
assert res.json()["type"] == "wifi"
assert res.json()["label"] == "uplink"
async def test_update_edge_not_found(client: AsyncClient, headers: dict):
res = await client.patch("/api/v1/edges/nonexistent", json={"type": "wifi"}, headers=headers)
assert res.status_code == 404
async def test_create_edge_with_custom_color(client: AsyncClient, headers: dict, two_nodes):
src, tgt = two_nodes
res = await client.post("/api/v1/edges", json={"source": src, "target": tgt, "type": "ethernet", "custom_color": "#a855f7"}, headers=headers)
assert res.status_code == 201
assert res.json()["custom_color"] == "#a855f7"
async def test_create_edge_with_path_style(client: AsyncClient, headers: dict, two_nodes):
src, tgt = two_nodes
res = await client.post("/api/v1/edges", json={"source": src, "target": tgt, "type": "ethernet", "path_style": "smooth"}, headers=headers)
assert res.status_code == 201
assert res.json()["path_style"] == "smooth"
async def test_update_edge_custom_color_and_path_style(client: AsyncClient, headers: dict, two_nodes):
src, tgt = two_nodes
edge_id = (await client.post("/api/v1/edges", json={"source": src, "target": tgt, "type": "ethernet"}, headers=headers)).json()["id"]
res = await client.patch(f"/api/v1/edges/{edge_id}", json={"custom_color": "#39d353", "path_style": "smooth"}, headers=headers)
assert res.status_code == 200
assert res.json()["custom_color"] == "#39d353"
assert res.json()["path_style"] == "smooth"
async def test_create_edge_requires_auth(client: AsyncClient, two_nodes):
src, tgt = two_nodes
res = await client.post("/api/v1/edges", json={"source": src, "target": tgt, "type": "ethernet"})
assert res.status_code == 403
+118
View File
@@ -0,0 +1,118 @@
from unittest.mock import patch
import pytest
from app.services.fingerprint import fingerprint_ports, match_port, suggest_node_type
MOCK_SIGNATURES = [
{"port": 80, "protocol": "tcp", "banner_regex": None, "service_name": "HTTP", "icon": "🌐", "category": "web", "suggested_node_type": "server"},
{"port": 443, "protocol": "tcp", "banner_regex": None, "service_name": "HTTPS", "icon": "🔒", "category": "web", "suggested_node_type": "server"},
{"port": 22, "protocol": "tcp", "banner_regex": None, "service_name": "SSH", "icon": "🖥", "category": "remote", "suggested_node_type": None},
{"port": 8006, "protocol": "tcp", "banner_regex": "proxmox", "service_name": "Proxmox API", "icon": "🔧", "category": "web", "suggested_node_type": "proxmox"},
{"port": 8006, "protocol": "tcp", "banner_regex": None, "service_name": "Web UI", "icon": "🌐", "category": "web", "suggested_node_type": "server"},
]
@pytest.fixture(autouse=True)
def mock_signatures():
with patch("app.services.fingerprint._load", return_value=MOCK_SIGNATURES):
yield
# ── match_port ────────────────────────────────────────────────────────────────
def test_match_port_known_port():
result = match_port(80, "tcp")
assert result is not None
assert result["service_name"] == "HTTP"
def test_match_port_unknown_port():
result = match_port(9999, "tcp")
assert result is None
def test_match_port_wrong_protocol():
result = match_port(80, "udp")
assert result is None
def test_match_port_with_banner_match():
result = match_port(8006, "tcp", banner="proxmox virtual environment")
assert result is not None
assert result["service_name"] == "Proxmox API"
def test_match_port_with_banner_no_match_falls_through_to_next():
# banner doesn't match proxmox regex → skips first sig, matches second (no banner_regex)
result = match_port(8006, "tcp", banner="something else")
assert result is not None
assert result["service_name"] == "Web UI"
def test_match_port_no_banner_matches_first_sig():
# no banner → banner check is skipped entirely, first matching sig wins
result = match_port(8006, "tcp", banner=None)
assert result is not None
assert result["service_name"] == "Proxmox API"
# ── fingerprint_ports ─────────────────────────────────────────────────────────
def test_fingerprint_ports_known_ports():
results = fingerprint_ports([{"port": 80, "protocol": "tcp"}, {"port": 443, "protocol": "tcp"}])
assert len(results) == 2
assert results[0]["service_name"] == "HTTP"
assert results[1]["service_name"] == "HTTPS"
assert results[0]["port"] == 80
assert results[0]["category"] == "web"
def test_fingerprint_ports_unknown_port():
results = fingerprint_ports([{"port": 9999, "protocol": "tcp"}])
assert len(results) == 1
assert results[0]["service_name"] == "unknown_service"
assert results[0]["icon"] is None
assert results[0]["category"] is None
def test_fingerprint_ports_mixed():
results = fingerprint_ports([{"port": 22, "protocol": "tcp"}, {"port": 9999, "protocol": "tcp"}])
assert results[0]["service_name"] == "SSH"
assert results[1]["service_name"] == "unknown_service"
def test_fingerprint_ports_empty():
assert fingerprint_ports([]) == []
def test_fingerprint_ports_defaults_protocol_to_tcp():
results = fingerprint_ports([{"port": 80}])
assert results[0]["service_name"] == "HTTP"
# ── suggest_node_type ─────────────────────────────────────────────────────────
def test_suggest_node_type_returns_proxmox_from_port():
result = suggest_node_type([{"port": 8006, "protocol": "tcp"}])
assert result == "proxmox"
def test_suggest_node_type_returns_server_from_http():
result = suggest_node_type([{"port": 80, "protocol": "tcp"}])
assert result == "server"
def test_suggest_node_type_no_match_returns_generic():
result = suggest_node_type([{"port": 9999, "protocol": "tcp"}])
assert result == "generic"
def test_suggest_node_type_empty_returns_generic():
assert suggest_node_type([]) == "generic"
def test_suggest_node_type_priority_proxmox_over_server():
# both proxmox and server ports open → proxmox wins (higher priority)
result = suggest_node_type([{"port": 80, "protocol": "tcp"}, {"port": 8006, "protocol": "tcp"}])
assert result == "proxmox"
+45
View File
@@ -65,3 +65,48 @@ async def test_list_nodes_returns_all(client: AsyncClient, headers: dict):
await client.post("/api/v1/nodes", json={"type": "generic", "label": f"Node {i}", "status": "unknown"}, headers=headers)
res = await client.get("/api/v1/nodes", headers=headers)
assert len(res.json()) == 3
async def test_update_node_not_found(client: AsyncClient, headers: dict):
res = await client.patch("/api/v1/nodes/nonexistent", json={"label": "X"}, headers=headers)
assert res.status_code == 404
async def test_delete_node_not_found(client: AsyncClient, headers: dict):
res = await client.delete("/api/v1/nodes/nonexistent", headers=headers)
assert res.status_code == 404
async def test_create_node_with_custom_colors(client: AsyncClient, headers: dict):
payload = {"type": "server", "label": "Styled", "status": "unknown", "custom_colors": {"border": "#ff0000", "background": "#001122", "icon": "#ffffff"}}
res = await client.post("/api/v1/nodes", json=payload, headers=headers)
assert res.status_code == 201
assert res.json()["custom_colors"] == {"border": "#ff0000", "background": "#001122", "icon": "#ffffff"}
async def test_update_node_custom_colors(client: AsyncClient, headers: dict):
create = await client.post("/api/v1/nodes", json={"type": "server", "label": "N", "status": "unknown"}, headers=headers)
node_id = create.json()["id"]
res = await client.patch(f"/api/v1/nodes/{node_id}", json={"custom_colors": {"border": "#a855f7"}}, headers=headers)
assert res.status_code == 200
assert res.json()["custom_colors"] == {"border": "#a855f7"}
async def test_create_proxmox_node_with_container_mode(client: AsyncClient, headers: dict):
payload = {"type": "proxmox", "label": "PVE", "status": "unknown", "container_mode": True}
res = await client.post("/api/v1/nodes", json=payload, headers=headers)
assert res.status_code == 201
assert res.json()["container_mode"] is True
async def test_update_node_container_mode(client: AsyncClient, headers: dict):
create = await client.post("/api/v1/nodes", json={"type": "proxmox", "label": "PVE", "status": "unknown"}, headers=headers)
node_id = create.json()["id"]
res = await client.patch(f"/api/v1/nodes/{node_id}", json={"container_mode": True}, headers=headers)
assert res.status_code == 200
assert res.json()["container_mode"] is True
async def test_create_node_requires_auth(client: AsyncClient):
res = await client.post("/api/v1/nodes", json={"type": "server", "label": "N", "status": "unknown"})
assert res.status_code == 403
@@ -93,4 +93,91 @@ describe('canvasStore', () => {
useCanvasStore.getState().setSelectedNode(null)
expect(useCanvasStore.getState().selectedNodeId).toBeNull()
})
it('onNodesChange marks unsaved', () => {
useCanvasStore.getState().addNode(makeNode('n1'))
useCanvasStore.getState().markSaved()
useCanvasStore.getState().onNodesChange([{ type: 'select', id: 'n1', selected: true }])
expect(useCanvasStore.getState().hasUnsavedChanges).toBe(true)
})
it('onEdgesChange marks unsaved', () => {
useCanvasStore.setState((s) => ({ edges: [...s.edges, makeEdge('e1', 'n1', 'n2')] }))
useCanvasStore.getState().markSaved()
useCanvasStore.getState().onEdgesChange([{ type: 'select', id: 'e1', selected: true }])
expect(useCanvasStore.getState().hasUnsavedChanges).toBe(true)
})
it('onConnect adds an edge between two nodes', () => {
useCanvasStore.getState().onConnect({ source: 'n1', target: 'n2', sourceHandle: null, targetHandle: null })
const { edges, hasUnsavedChanges } = useCanvasStore.getState()
expect(edges).toHaveLength(1)
expect(edges[0].source).toBe('n1')
expect(edges[0].target).toBe('n2')
expect(hasUnsavedChanges).toBe(true)
})
it('addNode with parent_id sets parentId and extent', () => {
useCanvasStore.getState().addNode(makeNode('parent'))
useCanvasStore.getState().addNode(makeNode('child', { parent_id: 'parent' }))
const child = useCanvasStore.getState().nodes.find((n) => n.id === 'child')
expect(child?.parentId).toBe('parent')
expect(child?.extent).toBe('parent')
})
it('updateEdge updates edge data and marks unsaved', () => {
useCanvasStore.setState((s) => ({ edges: [...s.edges, makeEdge('e1', 'n1', 'n2')] }))
useCanvasStore.getState().markSaved()
useCanvasStore.getState().updateEdge('e1', { type: 'wifi', label: 'uplink' })
const edge = useCanvasStore.getState().edges.find((e) => e.id === 'e1')
expect(edge?.data?.type).toBe('wifi')
expect(edge?.data?.label).toBe('uplink')
expect(useCanvasStore.getState().hasUnsavedChanges).toBe(true)
})
it('deleteEdge removes the edge and marks unsaved', () => {
useCanvasStore.setState((s) => ({ edges: [...s.edges, makeEdge('e1', 'n1', 'n2'), makeEdge('e2', 'n2', 'n3')] }))
useCanvasStore.getState().markSaved()
useCanvasStore.getState().deleteEdge('e1')
const { edges, hasUnsavedChanges } = useCanvasStore.getState()
expect(edges.find((e) => e.id === 'e1')).toBeUndefined()
expect(edges.find((e) => e.id === 'e2')).toBeDefined()
expect(hasUnsavedChanges).toBe(true)
})
it('setProxmoxContainerMode ON nests children inside proxmox', () => {
const proxmox: Node<NodeData> = { id: 'px', type: 'proxmox', position: { x: 0, y: 0 }, data: { label: 'px', type: 'proxmox', status: 'unknown', services: [], container_mode: false } }
const child = makeNode('vm1', { parent_id: 'px', type: 'vm' })
useCanvasStore.setState({ nodes: [proxmox, child] })
useCanvasStore.getState().setProxmoxContainerMode('px', true)
const { nodes } = useCanvasStore.getState()
const updatedProxy = nodes.find((n) => n.id === 'px')
const updatedChild = nodes.find((n) => n.id === 'vm1')
expect(updatedProxy?.data.container_mode).toBe(true)
expect(updatedProxy?.width).toBe(300)
expect(updatedChild?.parentId).toBe('px')
expect(updatedChild?.extent).toBe('parent')
})
it('setProxmoxContainerMode OFF detaches children', () => {
const proxmox: Node<NodeData> = { id: 'px', type: 'proxmox', position: { x: 0, y: 0 }, data: { label: 'px', type: 'proxmox', status: 'unknown', services: [], container_mode: true }, parentId: undefined }
const child: Node<NodeData> = { id: 'vm1', type: 'vm', position: { x: 0, y: 0 }, data: { label: 'vm1', type: 'vm', status: 'unknown', services: [], parent_id: 'px' }, parentId: 'px', extent: 'parent' }
useCanvasStore.setState({ nodes: [proxmox, child] })
useCanvasStore.getState().setProxmoxContainerMode('px', false)
const { nodes } = useCanvasStore.getState()
const updatedChild = nodes.find((n) => n.id === 'vm1')
expect(nodes.find((n) => n.id === 'px')?.data.container_mode).toBe(false)
expect(updatedChild?.parentId).toBeUndefined()
expect(updatedChild?.extent).toBeUndefined()
})
it('loadCanvas sorts parents before children', () => {
const parent = makeNode('p1')
const child: Node<NodeData> = { ...makeNode('c1', { parent_id: 'p1' }), parentId: 'p1', extent: 'parent' }
useCanvasStore.getState().loadCanvas([child, parent], [])
const { nodes } = useCanvasStore.getState()
const parentIdx = nodes.findIndex((n) => n.id === 'p1')
const childIdx = nodes.findIndex((n) => n.id === 'c1')
expect(parentIdx).toBeLessThan(childIdx)
})
})
@@ -0,0 +1,35 @@
import { describe, it, expect } from 'vitest'
import { EDGE_DEFAULT_COLORS } from '../edgeColors'
import type { EdgeType } from '@/types'
const EDGE_TYPES: EdgeType[] = ['ethernet', 'wifi', 'iot', 'vlan', 'virtual']
describe('EDGE_DEFAULT_COLORS', () => {
it('has an entry for every EdgeType', () => {
for (const type of EDGE_TYPES) {
expect(EDGE_DEFAULT_COLORS[type]).toBeDefined()
}
})
it('all colors are valid hex strings', () => {
for (const color of Object.values(EDGE_DEFAULT_COLORS)) {
expect(color).toMatch(/^#[0-9a-fA-F]{6}$/)
}
})
it('ethernet default is the dark gray neutral color', () => {
expect(EDGE_DEFAULT_COLORS.ethernet).toBe('#30363d')
})
it('wifi default is cyan', () => {
expect(EDGE_DEFAULT_COLORS.wifi).toBe('#00d4ff')
})
it('iot default is amber', () => {
expect(EDGE_DEFAULT_COLORS.iot).toBe('#e3b341')
})
it('virtual default is muted gray', () => {
expect(EDGE_DEFAULT_COLORS.virtual).toBe('#8b949e')
})
})