From 0449754f614a4dd4c3d767df58576b8933993cc4 Mon Sep 17 00:00:00 2001 From: sascha Date: Sun, 16 Aug 2026 20:36:18 +0200 Subject: [PATCH] Add operational safety suite --- tests/test_app.py | 111 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 110 insertions(+), 1 deletion(-) diff --git a/tests/test_app.py b/tests/test_app.py index a9a0ae1..645220a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -71,6 +71,28 @@ def test_info_advertises_capabilities_endpoint(): response = client.get("/info", headers={"Authorization": "Bearer test-token"}) assert response.status_code == 200 assert response.json()["endpoints"]["capabilities"] == "/capabilities" + assert response.json()["endpoints"]["doctor"] == "/doctor/{target}" + assert response.json()["endpoints"]["drift"] == "/drift" + assert response.json()["endpoints"]["maintenance_preflight"] == "/maintenance/preflight" + + +def test_audit_persists_and_redacts_secrets(tmp_path, monkeypatch): + db = tmp_path / "audit.sqlite3" + monkeypatch.setattr(app, "AUDIT_DB_PATH", str(db)) + app._init_audit_db() + app._audit("/danger", "POST", 200, "host=x token=abc password=hunter2 api_key=secret") + app._audit_log.clear() + + with TestClient(app.app) as client: + response = client.get("/audit", headers={"Authorization": "Bearer test-token"}) + + assert response.status_code == 200 + entry = response.json()[0] + assert entry["endpoint"] == "/danger" + assert "abc" not in entry["detail"] + assert "hunter2" not in entry["detail"] + assert "secret" not in entry["detail"] + assert entry["detail"].count("[REDACTED]") == 3 def test_wireguard_status_returns_redacted_live_state(monkeypatch): @@ -633,6 +655,7 @@ def test_backup_collection_runs_hosts_concurrently(monkeypatch): {"name": f"vm-{index}", "user": "sascha", "ip": f"10.1.1.{index}"} for index in range(1, 5) ]) + monkeypatch.setattr(app, "_config", {}) def fake_ssh(*_args, **_kwargs): nonlocal active, max_active @@ -647,7 +670,24 @@ def test_backup_collection_runs_hosts_concurrently(monkeypatch): assert max_active > 1 assert result["summary"] == { - "total": 4, "healthy": 4, "warning": 0, "critical": 0, "unknown": 0 + "total": 4, "healthy": 4, "warning": 0, "critical": 0, "unknown": 0, "exempt": 0 + } + + +def test_backup_policy_exempts_host_without_borg_call(monkeypatch): + monkeypatch.setattr(app, "_get_inventory_hosts", lambda: [ + {"name": "guck-vps", "user": "debian", "ip": "141.94.237.199"} + ]) + monkeypatch.setattr(app, "_config", {"backup": {"exempt_hosts": ["guck-vps"]}}) + monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not call borg"))) + + result = asyncio.run(app._collect_backup_status()) + + assert result["summary"] == { + "total": 1, "healthy": 0, "warning": 0, "critical": 0, "unknown": 0, "exempt": 1 + } + assert result["hosts"]["guck-vps"] == { + "state": "exempt", "ok": True, "reason": "backup policy exemption" } @@ -661,6 +701,75 @@ def test_overview_openapi_has_stable_enums_and_schema(): assert finding_schema["properties"]["severity"]["enum"] == ["healthy", "warning", "critical"] +def test_doctor_correlates_host_layers(monkeypatch): + async def snapshot(): + return { + "services": {}, + "hosts": {"guck-vps": {"reachable": True, "containers": ["caddy: Up 3 days"]}}, + "backups": {"summary": {}, "hosts": {"guck-vps": {"state": "exempt", "ok": True}}}, + "disks": {"guck-vps": {"pct": "8%"}}, + } + + monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot) + with TestClient(app.app) as client: + response = client.get("/doctor/guck-vps", headers={"Authorization": "Bearer test-token"}) + + assert response.status_code == 200 + result = response.json() + assert result["state"] == "healthy" + assert result["layers"]["host"]["reachable"] is True + assert result["layers"]["backup"]["state"] == "exempt" + assert result["layers"]["disk"]["pct"] == "8%" + assert result["findings"] == [] + + +def test_drift_reports_inventory_coverage_gaps(monkeypatch): + async def snapshot(): + return { + "services": {}, + "hosts": {"vm-a": {"reachable": True}, "vm-b": {"reachable": True}, "node1": {"reachable": True}}, + "backups": {"summary": {}, "hosts": {"vm-a": {"state": "healthy"}, "orphan": {"state": "healthy"}}}, + "disks": {"vm-a": {"pct": "10%"}, "node1": {"pct": "20%"}}, + } + + monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot) + with TestClient(app.app) as client: + response = client.get("/drift", headers={"Authorization": "Bearer test-token"}) + + assert response.status_code == 200 + result = response.json() + assert result["state"] == "warning" + assert {item["code"] for item in result["findings"]} == { + "inventory_missing_backup", "inventory_missing_disk", "backup_without_inventory" + } + + +def test_maintenance_preflight_blocks_active_target_backup(monkeypatch): + async def snapshot(): + return { + "services": {}, + "hosts": {"emby-chris": {"reachable": True, "containers": ["emby: Up 2 days"]}}, + "backups": {"summary": {}, "hosts": {"emby-chris": {"state": "healthy"}}}, + "disks": {"emby-chris": {"pct": "30%"}}, + } + + async def active_backups(): + return {"emby-chris": "active"} + + monkeypatch.setattr(app, "_collect_operational_snapshot", snapshot) + monkeypatch.setattr(app, "_collect_active_backups", active_backups) + with TestClient(app.app) as client: + response = client.get( + "/maintenance/preflight?action=docker&target=emby-chris", + headers={"Authorization": "Bearer test-token"}, + ) + + assert response.status_code == 200 + result = response.json() + assert result["safe"] is False + assert result["blockers"] == [{"code": "backup_active", "target": "emby-chris"}] + + def test_overview_is_compact_deterministic_and_light_model_friendly(monkeypatch): async def service_data(): return {