feat(admin): backend dashboard pour #71

- backend/admin.py : module FastAPI avec 4 endpoints admin-gated - GET /api/admin/stats (CPU/RAM/Disk/Uptime via psutil) - GET /api/admin/audit (filtres user/action/limit/offset) - GET /api/admin/backup-stats (count + size + age par vault) - GET /api/admin/stream (Server-Sent Events 5s) - backend/main.py : routeur monté + middleware gzip bypass pour SSE - backend/requirements.txt : ajout psutil>=5.9 - tests/test_admin.py : 13 tests (auth + filtres + format SSE), 100% verts Pas de frontend dans cette PR — page admin.html + admin.js restent à faire.
2026-09-07 08:54:25 -04:00
parent 99c9acccb0
commit b649c6b301
4 changed files with 581 additions and 5 deletions
@@ -0,0 +1,305 @@
+# tests/test_admin.py — Integration tests for the Admin Dashboard endpoints
+# (ROADMAP #71)
+#
+# These tests cover:
+# - GET /api/admin/stats       — CPU/RAM/Disk/Uptime snapshot
+# - GET /api/admin/audit       — recent audit log entries with filters
+# - GET /api/admin/backup-stats — backup counts/sizes per vault
+# - GET /api/admin/stream      — Server-Sent Events stream
+# All endpoints must require admin auth.
+
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def admin_client(tmp_path):
+    """TestClient with auth enabled, isolated temp data, admin user provisioned."""
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+
+    import json
+    from backend.auth.password import hash_password
+
+    pw_hash = hash_password("chab30")
+    users = {
+        "version": 1,
+        "users": {
+            "admin": {
+                "id": "admin-1",
+                "username": "admin",
+                "display_name": "admin",
+                "password_hash": pw_hash,
+                "role": "admin",
+                "vaults": ["*"],
+                "active": True,
+                "created_at": "2026-01-01T00:00:00",
+            },
+        "normaluser": {
+            "id": "user-1",
+            "username": "normaluser",
+            "display_name": "normal",
+            "password_hash": hash_password("normal123"),
+            "role": "user",
+            "vaults": ["TestVault"],
+            "active": True,
+            "created_at": "2026-01-01T00:00:00",
+        },
+        },
+    }
+    (data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
+
+    src_secret = Path("data/secret.key")
+    if src_secret.exists():
+        shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
+
+    orig_cwd = os.getcwd()
+    test_vault_path = os.path.abspath("test-vault")
+    os.chdir(str(tmp_path))
+
+    os.environ["VAULT_1_NAME"] = "TestVault"
+    os.environ["VAULT_1_PATH"] = test_vault_path
+    os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
+    os.environ["OBSIGATE_ADMIN_USER"] = "admin"
+    os.environ["OBSIGATE_ADMIN_PASSWORD"] = "chab30"
+    os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
+
+    import backend.main
+    backend.main._load_config = lambda: {"watcher_enabled": False}
+
+    from backend.main import app
+    from backend.indexer import build_index, index
+    import asyncio
+
+    for key in list(index.keys()):
+        del index[key]
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(build_index())
+
+    from backend.search import init_inverted_index
+    init_inverted_index()
+
+    from fastapi.testclient import TestClient
+    client = TestClient(app)
+    yield client
+
+    if hasattr(client, "close"):
+        client.close()
+    loop.run_until_complete(asyncio.sleep(0))
+
+    os.chdir(orig_cwd)
+    shutil.rmtree(str(tmp_path), ignore_errors=True)
+    for k in [
+        "VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
+        "OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED",
+    ]:
+        os.environ.pop(k, None)
+
+
+def _login(client, username="admin", password="chab30"):
+    resp = client.post("/api/auth/login", json={"username": username, "password": password})
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _bearer(token):
+    return {"Authorization": f"Bearer {token}"}
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/stats
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminStats:
+    EXPECTED_KEYS = {
+        "cpu_pct", "mem_used_mb", "mem_total_mb",
+        "disk_used_gb", "disk_total_gb",
+        "uptime_seconds", "active_sessions",
+    }
+
+    def test_stats_ok_as_admin(self, admin_client):
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        missing = self.EXPECTED_KEYS - set(data.keys())
+        assert not missing, f"Missing keys: {missing}"
+        # Numeric sanity (allow 0 for any metric — depending on platform)
+        for key in ("cpu_pct", "mem_used_mb", "mem_total_mb",
+                    "disk_used_gb", "disk_total_gb", "uptime_seconds",
+                    "active_sessions"):
+            assert isinstance(data[key], (int, float)), f"{key} not numeric"
+            assert data[key] >= 0, f"{key} is negative"
+
+    def test_stats_requires_auth(self, admin_client):
+        resp = admin_client.get("/api/admin/stats")
+        assert resp.status_code in (401, 403)
+
+    def test_stats_requires_admin_role(self, admin_client):
+        token = _login(admin_client, username="normaluser", password="normal123")
+        resp = admin_client.get("/api/admin/stats", headers=_bearer(token))
+        assert resp.status_code == 403
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/audit
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminAudit:
+    def _seed_audit(self, tmp_path_factory=None):
+        """Append a few entries to the audit log so filtering has data."""
+        from backend.audit import _write_entry
+        from datetime import datetime, timezone
+        entries = [
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_save", "username": "alice", "vault": "TestVault",
+             "size": 100, "ip": "127.0.0.1"},
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_delete", "username": "bob", "vault": "TestVault",
+             "ip": "127.0.0.1"},
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_save", "username": "bob", "vault": "TestVault",
+             "size": 50, "ip": "127.0.0.1"},
+        ]
+        for e in entries:
+            _write_entry(e)
+
+    def test_audit_ok(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "entries" in data
+        assert "total" in data
+        # We just seeded at least 3 entries
+        assert data["total"] >= 3
+
+    def test_audit_filter_by_user(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?user=bob", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert all("bob" in str(e.get("username", "")).lower() for e in data["entries"])
+
+    def test_audit_filter_by_action(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?action=file_delete", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert all(e.get("action") == "file_delete" for e in data["entries"])
+
+    def test_audit_limit_param(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?limit=2", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data["entries"]) <= 2
+
+    def test_audit_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/audit")
+        assert resp.status_code in (401, 403)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/backup-stats
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminBackupStats:
+    def _create_backup_files(self, tmp_path_factory=None):
+        """Create a couple of fake .bak files in the default backup dir."""
+        from backend.indexer import vault_config
+        import time as _time
+        for vault_name, cfg in list(vault_config.items()):
+            vault_root = Path(cfg["path"])
+            backup_root = vault_root / ".obsigate-backup" / vault_name / "subdir"
+            backup_root.mkdir(parents=True, exist_ok=True)
+            ts1 = int(_time.time()) - 86400
+            ts2 = int(_time.time())
+            (backup_root / f"note.md.{ts1}.bak").write_text("old version")
+            (backup_root / f"note.md.{ts2}.bak").write_text("newer version with more content")
+
+    def test_backup_stats_ok(self, admin_client):
+        self._create_backup_files()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/backup-stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        for k in ("total_backups", "total_size_mb", "oldest_age_days",
+                  "newest_age_days", "by_vault"):
+            assert k in data, f"missing key {k}"
+        assert data["total_backups"] >= 2
+        assert data["total_size_mb"] >= 0
+        # We created one set per vault — at least one vault entry
+        assert isinstance(data["by_vault"], dict)
+
+    def test_backup_stats_empty(self, admin_client):
+        """Even with no backups, endpoint returns 200 with zero counts."""
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/backup-stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total_backups"] >= 0
+        # If no backups exist, both age fields are 0
+        if data["total_backups"] == 0:
+            assert data["newest_age_days"] == 0.0
+            assert data["oldest_age_days"] == 0.0
+
+    def test_backup_stats_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/backup-stats")
+        assert resp.status_code in (401, 403)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/stream
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminStream:
+    def test_stream_content_type(self, admin_client):
+        """Verify SSE endpoint returns text/event-stream with valid first frame.
+
+        The /api/admin/stream endpoint is an infinite generator (yields every 5s).
+        We can't easily consume a streaming response from a sync TestClient
+        (the context manager blocks on entry for infinite responses). Instead,
+        we verify the endpoint contract from two angles:
+          1. Without auth → 401/403 (proves the endpoint is mounted and gated)
+          2. Direct invocation of the underlying generator yields a valid SSE
+             frame on the first iteration (proves the format contract).
+        """
+        # 1) Endpoint is gated behind admin auth.
+        resp = admin_client.get("/api/admin/stream")
+        assert resp.status_code in (401, 403), (
+            f"unauthenticated should be rejected, got {resp.status_code}"
+        )
+
+        # 2) Direct generator check — read the first frame, then close.
+        import asyncio
+        from backend.admin import _stats_event_generator
+
+        async def _first_frame():
+            gen = _stats_event_generator()
+            return await gen.__anext__()
+
+        first = asyncio.run(_first_frame())
+        assert isinstance(first, str), f"expected str, got {type(first).__name__}"
+        assert first.startswith("event: stats"), f"unexpected frame: {first[:100]!r}"
+        assert "data: " in first
+        # The data line should be parseable JSON.
+        import json
+        data_line = [ln for ln in first.splitlines() if ln.startswith("data: ")][0]
+        payload = json.loads(data_line[len("data: "):])
+        assert isinstance(payload, dict)
+        assert "cpu_pct" in payload or "error" in payload
+
+    def test_stream_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/stream")
+        assert resp.status_code in (401, 403)