from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from fastapi.testclient import TestClient from sqlmodel import SQLModel, create_engine @pytest.fixture def api_client(tmp_path, monkeypatch): db_path = tmp_path / "appops_test.db" monkeypatch.setenv("APPOPS_DB_PATH", str(db_path)) monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token") from backend.app.db import session from backend.app.db.init_db import init_db from backend.app.main import app engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False}) session.engine = engine SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) init_db() with TestClient(app) as client: yield client def test_statistics_summary_uses_live_audit_job_and_unraid_data(api_client, monkeypatch): from backend.app.api import routes_observability from backend.app.connectors.unraid import UnraidConnector monkeypatch.setattr( UnraidConnector, "get_system_metrics", lambda self: {"status": "ok", "cpuUsage": 37, "ramUsage": 64}, raising=True, ) routes_observability._STATISTICS_HISTORY.clear() now = datetime.now(UTC).replace(tzinfo=None) api_client.post( "/api/actions/execute", json={ "intent": "volume.delete", "target": "postgres-data", "source": "manual", "risk": "critical", "plan": [], }, ) api_client.post( "/api/actions/execute", json={ "intent": "container.restart", "target": "portainer", "source": "voice", "risk": "high", "plan": [], }, ) response = api_client.get("/api/statistics/summary") assert response.status_code == 200 payload = response.json() assert payload["cpuHistory"][-1]["value"] == 37 assert payload["ramHistory"][-1]["value"] == 64 assert payload["blockedActions"] >= 1 assert payload["voiceActionCount"] >= 1 from sqlmodel import Session, select from backend.app.db import session as database from backend.app.db.models import AuditEvent with Session(database.engine) as db: events = db.exec(select(AuditEvent)).all() voice_events = [event for event in events if (event.redacted_metadata or {}).get("requestSource") == "voice"] assert voice_events assert all(event.identity == "authentik:feature-test-admin" for event in voice_events) assert all(point["time"] != "00:00" for point in payload["cpuHistory"]) assert payload["generatedAt"] >= (now - timedelta(minutes=1)).isoformat() def test_statistics_summary_keeps_partial_unraid_metrics_explicit(api_client, monkeypatch): from backend.app.api import routes_observability from backend.app.connectors.unraid import UnraidConnector monkeypatch.setattr( UnraidConnector, "get_system_metrics", lambda self: { "status": "partial", "cpuUsage": None, "ramUsage": 92, "reason": "cpu_unavailable", }, raising=True, ) routes_observability._STATISTICS_HISTORY.clear() response = api_client.get("/api/statistics/summary") assert response.status_code == 200 payload = response.json() assert payload["sources"]["unraid"] == "partial" assert payload["cpuHistory"][-1]["value"] is None assert payload["ramHistory"][-1]["value"] == 92 assert payload["sourceWarnings"][0]["source"] == "unraid" assert payload["sourceWarnings"][0]["surface"] == "metrics" def test_statistics_preserves_precision_and_measured_gpu_load(api_client, monkeypatch): from types import SimpleNamespace from backend.app.api import routes_observability as routes from backend.app.connectors.unraid import UnraidConnector monkeypatch.setattr(UnraidConnector, 'get_system_metrics', lambda self: {'status': 'ok', 'cpuUsage': 12.34, 'ramUsage': 45.67}) gpu = {'status': 'partial', 'source': 'gpu_hot', 'observedAt': None, 'reason': 'gpu_metrics_timestamp_unavailable', 'devices': [ {'id': '0', 'name': 'GPU A', 'utilizationPercent': 68.91, 'memoryUsedMiB': 1800, 'memoryTotalMiB': 16000, 'temperatureC': 51}, {'id': '1', 'name': 'GPU B', 'utilizationPercent': 12.35, 'memoryUsedMiB': None, 'memoryTotalMiB': None, 'temperatureC': None}, ]} monkeypatch.setattr(routes, 'GpuMetricsConnector', lambda **kwargs: SimpleNamespace(get_metrics=lambda: gpu), raising=False) routes._STATISTICS_HISTORY.clear() payload = api_client.get('/api/statistics/summary').json() assert payload['cpuHistory'][-1]['value'] == 12.34 assert payload['ramHistory'][-1]['value'] == 45.67 assert payload['gpuHistory'][-1]['value'] == 68.91 assert payload['gpu']['devices'] == gpu['devices'] assert payload['gpu']['receivedAt'] == payload['generatedAt'] assert payload['gpu']['observedAt'] is None assert payload['sources']['gpu'] == 'partial' def test_statistics_records_unavailable_metrics_as_a_gap(api_client, monkeypatch): from types import SimpleNamespace from backend.app.api import routes_observability as routes from backend.app.connectors.unraid import UnraidConnector monkeypatch.setattr(UnraidConnector, 'get_system_metrics', lambda self: {'status': 'error', 'reason': 'unavailable', 'cpuUsage': None, 'ramUsage': None}) monkeypatch.setattr(routes, 'GpuMetricsConnector', lambda **kwargs: SimpleNamespace(get_metrics=lambda: {'status': 'unavailable', 'devices': []}), raising=False) routes._STATISTICS_HISTORY[:] = [{'time': '2026-01-01T00:00:00Z', 'cpu': 34, 'ram': 56}] payload = api_client.get('/api/statistics/summary').json() assert payload['cpuHistory'][-1]['value'] is None assert payload['ramHistory'][-1]['value'] is None assert payload['gpuHistory'][-1]['value'] is None def test_statistics_history_grows_with_interleaved_viewers(): from backend.app.api import routes_observability as routes routes._STATISTICS_HISTORY.clear() start = datetime(2026, 10, 1, 12, 0, tzinfo=UTC) for index in range(25): timestamp = (start + timedelta(seconds=index * 2.5)).isoformat() samples = routes._record_statistics_sample(timestamp, 12.34, 45.67, 0.0) assert len(samples) == 13 assert samples[-1]['time'] == (start + timedelta(minutes=1)).isoformat() assert samples[-1]['gpu'] == 0.0 samples = routes._record_statistics_sample((start + timedelta(minutes=31)).isoformat(), None, None, None) assert len(samples) == 2 assert samples[-1]['cpu'] is None def test_storage_summary_exposes_cache_pool_inventory(api_client, monkeypatch): from backend.app.connectors.portainer import PortainerConnector from backend.app.connectors.unraid import UnraidConnector monkeypatch.setattr( UnraidConnector, "get_cache_status", lambda self: { "status": "ok", "cacheUsage": 88, "pools": [ { "name": "cache", "status": "DISK_OK", "healthy": True, "type": "CACHE", "usedGiB": 88, "sizeGiB": 100, "freeGiB": 12, "usagePercent": 88, } ], }, raising=True, ) monkeypatch.setattr( UnraidConnector, "get_share_summary", lambda self: {"status": "ok", "shares": [{"name": "appdata", "usedGiB": 88, "sizeGiB": 100, "critical": True}]}, raising=True, ) monkeypatch.setattr( UnraidConnector, "get_disk_warnings", lambda self: { "status": "ok", "disks": [{"name": "disk1", "status": "DISK_OK", "healthy": True, "type": "DATA", "usedGiB": 50, "sizeGiB": 100, "freeGiB": 50}], "diskWarnings": 0, }, raising=True, ) monkeypatch.setattr(PortainerConnector, "list_containers", lambda self: {"status": "ok", "containers": []}, raising=True) response = api_client.get("/api/storage/summary") assert response.status_code == 200 payload = response.json() assert payload["cacheUsage"] == 88 assert payload["storagePools"] == [ { "name": "cache", "status": "DISK_OK", "healthy": True, "type": "CACHE", "usedGiB": 88, "sizeGiB": 100, "freeGiB": 12, "usagePercent": 88, } ] assert payload["disks"][0]["name"] == "disk1"