smolkik-code
9b15f8b09c
- PostgreSQL schema: 14 tables, JSONB attributes, audit triggers, soft delete - FastAPI backend: CRUD, search/filter, relationship graph, bulk import, JWT RBAC - React frontend: CI table, detail card, force-graph, dashboard - Seed data: homelab scenario (Proxmox, Mikrotik, VMs, services) - Docker Compose + Kubernetes manifests - 20 backend tests (pytest + httpx)
200 lines
7.1 KiB
Python
200 lines
7.1 KiB
Python
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
from app.main import app
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def admin_token(client: AsyncClient):
|
|
resp = await client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_headers(admin_token: str):
|
|
return {"Authorization": f"Bearer {admin_token}"}
|
|
|
|
|
|
async def test_health(client: AsyncClient):
|
|
resp = await client.get("/api/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "healthy"
|
|
|
|
|
|
async def test_login_success(client: AsyncClient):
|
|
resp = await client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
|
|
assert resp.status_code == 200
|
|
assert "access_token" in resp.json()
|
|
|
|
|
|
async def test_login_invalid(client: AsyncClient):
|
|
resp = await client.post("/api/auth/login", json={"username": "admin", "password": "wrong"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_get_me(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/auth/me", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["username"] == "admin"
|
|
|
|
|
|
async def test_list_classes(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/classes", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) >= 4 # Hardware, NetworkDevice, Software, Storage
|
|
|
|
|
|
async def test_list_types(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/types", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) >= 6
|
|
|
|
|
|
async def test_list_locations(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/locations", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) >= 3
|
|
|
|
|
|
async def test_list_cis(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] >= 25
|
|
assert data["page"] == 1
|
|
|
|
|
|
async def test_list_cis_pagination(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci", headers=auth_headers, params={"page": 1, "page_size": 5})
|
|
data = resp.json()
|
|
assert len(data["items"]) == 5
|
|
assert data["page_size"] == 5
|
|
|
|
|
|
async def test_list_cis_search(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci", headers=auth_headers, params={"search": "proxmox"})
|
|
data = resp.json()
|
|
assert data["total"] >= 2
|
|
assert all("proxmox" in i["name"].lower() for i in data["items"])
|
|
|
|
|
|
async def test_list_cis_filter_status(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci", headers=auth_headers, params={"status": "active"})
|
|
data = resp.json()
|
|
assert all(i["status"] == "active" for i in data["items"])
|
|
|
|
|
|
async def test_get_ci_detail(client: AsyncClient, auth_headers: dict):
|
|
list_resp = await client.get("/api/ci", headers=auth_headers)
|
|
ci_id = list_resp.json()["items"][0]["id"]
|
|
resp = await client.get(f"/api/ci/{ci_id}", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
ci = resp.json()
|
|
assert "ip_addresses" in ci
|
|
assert "relationships_out" in ci
|
|
assert "relationships_in" in ci
|
|
|
|
|
|
async def test_create_ci(client: AsyncClient, auth_headers: dict):
|
|
types_resp = await client.get("/api/types", headers=auth_headers)
|
|
type_id = types_resp.json()[0]["id"]
|
|
resp = await client.post("/api/ci", headers=auth_headers, json={
|
|
"name": "test-ci-auto",
|
|
"ci_type_id": type_id,
|
|
"description": "Auto-created for testing",
|
|
"status": "active",
|
|
"tags": ["test", "automated"],
|
|
"attributes": {"cpu_cores": 2, "ram_gb": 4},
|
|
})
|
|
assert resp.status_code == 201
|
|
ci = resp.json()
|
|
assert ci["name"] == "test-ci-auto"
|
|
assert ci["version"] == 1
|
|
|
|
|
|
async def test_update_ci(client: AsyncClient, auth_headers: dict):
|
|
list_resp = await client.get("/api/ci", headers=auth_headers)
|
|
ci_id = list_resp.json()["items"][0]["id"]
|
|
resp = await client.patch(f"/api/ci/{ci_id}", headers=auth_headers, json={
|
|
"description": "Updated description"
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["description"] == "Updated description"
|
|
|
|
|
|
async def test_delete_ci_soft(client: AsyncClient, auth_headers: dict):
|
|
# Create then delete
|
|
types_resp = await client.get("/api/types", headers=auth_headers)
|
|
type_id = types_resp.json()[0]["id"]
|
|
create_resp = await client.post("/api/ci", headers=auth_headers, json={
|
|
"name": "to-be-deleted",
|
|
"ci_type_id": type_id,
|
|
"status": "active",
|
|
})
|
|
ci_id = create_resp.json()["id"]
|
|
del_resp = await client.delete(f"/api/ci/{ci_id}", headers=auth_headers)
|
|
assert del_resp.status_code == 204
|
|
# Should not appear in list
|
|
list_resp = await client.get("/api/ci", headers=auth_headers, params={"search": "to-be-deleted"})
|
|
assert list_resp.json()["total"] == 0
|
|
|
|
|
|
async def test_get_nonexistent_ci(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci/00000000-0000-0000-0000-000000000000", headers=auth_headers)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
async def test_dashboard_stats(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/dashboard/stats", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
stats = resp.json()
|
|
assert stats["total_cis"] > 0
|
|
assert stats["total_relationships"] > 0
|
|
|
|
|
|
async def test_graph(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/ci/graph/visualize", headers=auth_headers, params={"depth": 1})
|
|
assert resp.status_code == 200
|
|
graph = resp.json()
|
|
assert "nodes" in graph
|
|
assert "edges" in graph
|
|
assert len(graph["nodes"]) > 0
|
|
|
|
|
|
async def test_bulk_import(client: AsyncClient, auth_headers: dict):
|
|
types_resp = await client.get("/api/types", headers=auth_headers)
|
|
type_name = types_resp.json()[0]["name"]
|
|
resp = await client.post("/api/ci/bulk/import", headers=auth_headers, json={
|
|
"items": [
|
|
{"name": "bulk-1", "ci_type_name": type_name, "status": "active"},
|
|
{"name": "bulk-2", "ci_type_name": type_name, "status": "active"},
|
|
]
|
|
})
|
|
assert resp.status_code == 201
|
|
assert resp.json()["created"] == 2
|
|
|
|
|
|
async def test_viewer_cannot_create(client: AsyncClient):
|
|
resp = await client.post("/api/auth/login", json={"username": "viewer", "password": "view123"})
|
|
token = resp.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
types_resp = await client.get("/api/types", headers=headers)
|
|
type_id = types_resp.json()[0]["id"]
|
|
resp = await client.post("/api/ci", headers=headers, json={
|
|
"name": "should-fail",
|
|
"ci_type_id": type_id,
|
|
})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_unauthorized_access(client: AsyncClient):
|
|
resp = await client.get("/api/ci")
|
|
assert resp.status_code == 403 # No auth header → HTTPBearer returns 403
|