import json import os from pathlib import Path import subprocess import sys import pytest from fastapi.testclient import TestClient from sqlmodel import Session, SQLModel, create_engine ROOT = Path(__file__).resolve().parents[2] def test_settings_load_voice_token_from_file(tmp_path): voice_token_file = tmp_path / "voice-token" voice_token_file.write_text("persistent-voice-token-value", encoding="utf-8") environment = dict(os.environ) environment.pop("APP_OPS_VOICE_TOKEN", None) environment["APP_OPS_VOICE_TOKEN_FILE"] = str(voice_token_file) result = subprocess.run( [ sys.executable, "-c", "from backend.app.core.config import Settings; " "loaded = Settings(); " "raise SystemExit(0 if loaded.voice_token == 'persistent-voice-token-value' " "and loaded.voice_token_file else 1)", ], cwd=ROOT, env=environment, check=False, ) assert result.returncode == 0 @pytest.fixture def api_client(tmp_path, monkeypatch): db_path = tmp_path / "runtime_test.db" data_dir = tmp_path / "data" data_dir.mkdir() monkeypatch.setenv("APPOPS_DB_PATH", str(db_path)) monkeypatch.setenv("APPOPS_DATA_DIR", str(data_dir)) monkeypatch.setenv("APPOPS_ENV", "production") monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token") from backend.app.core import security from backend.app.core.config import settings from backend.app.db import session from backend.app.db.init_db import init_db from backend.app.main import app previous = { "db_path": settings.db_path, "data_dir": settings.data_dir, "environment": settings.environment, "voice_token": settings.voice_token, "security_voice_token": security.settings.voice_token, "apis": dict(settings.apis), } settings.db_path = str(db_path) settings.data_dir = str(data_dir) settings.environment = "production" settings.voice_token = "test-voice-token" security.settings.voice_token = "test-voice-token" for key in ["PORTAINER_URL", "PORTAINER_API_KEY", "PORTAINER_ENDPOINT_ID", "UNRAID_URL", "UNRAID_API_KEY"]: settings.apis[key] = None 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 settings.db_path = previous["db_path"] settings.data_dir = previous["data_dir"] settings.environment = previous["environment"] settings.voice_token = previous["voice_token"] settings.apis = previous["apis"] security.settings.voice_token = previous["security_voice_token"] def test_system_summary_prefers_setup_config_over_stale_connector_status(api_client): from sqlmodel import Session from backend.app.db import session as db_session from backend.app.db.models import Connector with Session(db_session.engine) as session: connector = session.get(Connector, "portainer") connector.status = "connected" session.add(connector) session.commit() payload = api_client.get("/api/system/summary").json() assert payload["portainerStatus"] == "missing_config" assert payload["gatewayHealth"] == "critical" def test_system_summary_reports_latest_audit_action_without_secret_leak(api_client): from sqlmodel import Session from backend.app.db import session as db_session from backend.app.db.models import AuditEvent secret = "SUMMARY_SECRET_TOKEN" with Session(db_session.engine) as session: session.add( AuditEvent( id="audit-summary-action", source="manual", identity="operator", action_id="container.restart", target=f"widefrog api_key={secret}", target_type="container", risk="medium", policy_decision="allowed", result="success", ) ) session.commit() payload = api_client.get("/api/system/summary").json() assert payload["lastAction"].startswith("container.restart -> widefrog") assert "api_key=[REDACTED]" in payload["lastAction"] assert secret not in payload["lastAction"] assert secret not in str(payload) def test_runtime_info_is_secret_safe(api_client): secret = "PORTAINER_RUNTIME_SECRET" api_client.patch( "/api/settings/connectors/portainer", json={ "url": "http://portainer.local", "endpointId": "1", "apiKey": secret, "verifyTls": False, }, ) payload = api_client.get("/api/runtime/info").json() assert payload["version"] == "1.8.0" assert payload["environment"] == "production" assert payload["dataDirConfigured"] is True assert payload["dbStatus"] == "ok" assert payload["connectorConfigSummary"]["portainer"]["hasSecret"] is True assert secret not in str(payload) def test_private_request_metrics_are_bounded_and_request_ids_are_returned(api_client): response = api_client.get( "/api/runtime/info", headers={"X-Request-ID": "review-request-1234"} ) assert response.status_code == 200 assert response.headers["X-Request-ID"] == "review-request-1234" assert response.headers["Server-Timing"].startswith("app;dur=") metrics = api_client.get("/api/admin/observability/requests").json() assert metrics["sampleLimit"] == 2_000 assert metrics["sampleCount"] >= 1 assert any(item["path"] == "/api/runtime/info" for item in metrics["slowEndpoints"]) assert all("query" not in item for item in metrics["slowEndpoints"]) openmetrics = api_client.get("/api/admin/observability/metrics") assert openmetrics.status_code == 200 assert "appops_http_request_duration_p95_milliseconds" in openmetrics.text assert "review-request-1234" not in openmetrics.text maintenance = api_client.get("/api/admin/maintenance/status").json() assert maintenance["status"] == "running" assert "notificationEvaluation" in maintenance["intervalsSeconds"] assert "observabilityPersistence" in maintenance["intervalsSeconds"] def test_sqlite_checkpoint_is_backup_first_and_reports_safe_sizes(api_client): status = api_client.get("/api/admin/database/status").json() result = api_client.post("/api/admin/database/checkpoint").json() assert status["status"] == "ready" assert status["databaseBytes"] > 0 assert status["journalMode"] in {"delete", "wal"} assert result["success"] is True assert result["status"] == "checkpointed" assert result["backup"]["name"].startswith("appops-db-") assert "auditId" in result def test_latest_backup_restore_drill_is_read_only(api_client): from backend.app.db.session import engine from backend.app.services.runtime_service import verify_latest_backup_drill created = api_client.post("/api/admin/backups/sqlite").json() with Session(engine) as session: result = verify_latest_backup_drill(session) assert created["success"] is True assert result["success"] is True assert result["integrityCheck"] == "ok" assert result["restoreCheck"] == "ok" assert result["backup"]["name"] == created["backup"]["name"] def test_app_connector_read_routes_load_stored_runtime_config(api_client, monkeypatch): from backend.app.connectors.plex import PlexConnector from backend.app.core.config import settings secret = "PLEX_RUNTIME_SECRET" api_client.patch( "/api/settings/connectors/plex", json={"url": "http://plex.local:32400", "apiKey": secret, "verifyTls": False}, ) settings.apis["PLEX_URL"] = None settings.apis["PLEX_TOKEN"] = None monkeypatch.setattr( PlexConnector, "_request", lambda self, path, params=None: { "MediaContainer": { "friendlyName": "Lab Plex", "version": "1.2.3", "machineIdentifier": "machine-1", "size": "0", } }, raising=True, ) response = api_client.get("/api/apps/plex/health") payload = response.json() assert response.status_code == 200 assert payload["status"] == "connected" assert payload["friendlyName"] == "Lab Plex" assert secret not in response.text def test_deep_health_is_available_without_admin_token(api_client): payload = api_client.get("/api/health/deep").json() assert payload["status"] == "ok" assert payload["checks"]["dbReachable"] is True assert payload["checks"]["dataDirWritable"] is True assert "setup" in payload def test_legacy_unlock_and_anonymous_admin_switch_are_removed(api_client): assert api_client.post("/api/operator/session").status_code in {404, 405} assert api_client.patch("/api/runtime/operator-auth", json={"required": False}).status_code in {404, 405} def test_admin_exports_are_secret_safe(api_client): secret = "EXPORT_SECRET_VALUE" api_client.patch( "/api/settings/connectors/portainer", json={ "url": "http://portainer.local", "endpointId": "1", "apiKey": secret, "verifyTls": False, }, ) for path in [ "/api/admin/export/config", "/api/admin/export/policies", "/api/admin/export/inventory", "/api/admin/export/audit-summary", ]: response = api_client.get(path) assert response.status_code == 200 assert secret not in response.text assert "API_KEY" not in response.text or "hasSecret" in response.text def test_sqlite_online_backup_creates_integrity_checked_copy(api_client): secret = "BACKUP_SECRET_VALUE" api_client.patch( "/api/settings/connectors/portainer", json={ "url": "http://portainer.local", "endpointId": "1", "apiKey": secret, "verifyTls": False, }, ) response = api_client.post("/api/admin/backups/sqlite") payload = response.json() assert response.status_code == 200 assert payload["success"] is True assert payload["integrityCheck"] == "ok" assert payload["restoreMode"] == "manual_stop_container" assert payload["backup"]["name"].startswith("appops-db-") assert Path(payload["backup"]["path"]).exists() assert payload["backup"]["manifestAvailable"] is True manifest_path = Path(payload["backup"]["path"]).with_suffix(".manifest.json") assert manifest_path.exists() manifest = json.loads(manifest_path.read_text(encoding="utf-8")) assert manifest["databaseFile"] == payload["backup"]["name"] assert manifest["sha256"] == payload["backup"]["sha256"] assert secret not in response.text list_response = api_client.get("/api/admin/backups/sqlite") listed = list_response.json() assert list_response.status_code == 200 assert listed["backupDir"].endswith("backups") assert listed["backups"][0]["name"] == payload["backup"]["name"] assert secret not in list_response.text backup_name = payload["backup"]["name"] verified = api_client.post(f"/api/admin/backups/sqlite/{backup_name}/verify") assert verified.status_code == 200 assert verified.json()["success"] is True assert verified.json()["integrityCheck"] == "ok" assert verified.json()["manifestCheck"] == "ok" assert verified.json()["restoreMode"] == "manual_stop_container" downloaded = api_client.get(f"/api/admin/backups/sqlite/{backup_name}/download") assert downloaded.status_code == 200 assert downloaded.headers["content-type"] == "application/vnd.sqlite3" assert "no-store" in downloaded.headers["cache-control"] assert len(downloaded.content) == payload["backup"]["sizeBytes"] def test_sqlite_backup_download_rejects_traversal(api_client): assert ( api_client.get("/api/admin/backups/sqlite/not-a-backup/download").status_code == 404 ) assert ( api_client.post("/api/admin/backups/sqlite/not-a-backup/verify").status_code == 404 )