ObsiGate/tests/test_admin.py

# 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)


# ═══════════════════════════════════════════════════════════════
# Admin dashboard PAGE (/admin.html)
# ═══════════════════════════════════════════════════════════════
class TestAdminPage:
    """Regression for ROADMAP #71 — Admin menu bounced back to the main page.

    Root cause: /admin.html had no explicit route, so the SPA catch-all
    ``/{full_path:path}`` served index.html. Guard the fix.
    """

    def test_page_served_as_admin(self, admin_client):
        token = _login(admin_client)
        resp = admin_client.get("/admin.html", headers=_bearer(token))
        assert resp.status_code == 200
        body = resp.text
        # Real admin page markers (NOT the main SPA index.html)
        assert "admin-main" in body or "ObsiGate — Admin" in body or "admin.js" in body
        # The regression: index.html markers must be absent
        assert "boot-splash" not in body
        # Asset paths must point to the actual static mount (/static), not /frontend
        assert "/static/js/admin.js" in body
        assert "/frontend/js/admin.js" not in body

    def test_page_requires_auth(self, admin_client):
        resp = admin_client.get("/admin.html")
        assert resp.status_code in (401, 403)

    def test_page_requires_admin_role(self, admin_client):
        token = _login(admin_client, username="normaluser", password="normal123")
        resp = admin_client.get("/admin.html", headers=_bearer(token))
        assert resp.status_code == 403