test(services+interface+middleware): coverage for dashboard / server cmd handlers / runtime / audit / read-only / safe-deserialize

- DashboardSummaryOrchestrator: happy / empty / sub-orchestrator-raises
  paths, recent_activity cap, ISO helpers on None, count-by-status fast
  paths
- server_command_handlers: start/stop/status/restart/reload/logs all
  unit-tested with the daemon module mocked; reload covers loopback
  success, loopback failure -> SIGHUP fallback, ValueError fallback,
  bearer-token header presence vs absence
- server_runtime: signal handler installation, SIGHUP reload semantics,
  embedded entrypoint wiring of cwd + env + signal forwarding
- audit_log_middleware: AUDIT_ALWAYS_PREFIXES gating, latency capture,
  identity capture
- read_only_middleware: dispatch matrix across enabled flag + method +
  path-allowlist
- _safe_deserialize_iter: skip-on-error semantics, ERROR-level log,
  per-entity skip counter exposure

Author: fgogolli

tests/unit/api/middleware/__init__.py

@@ -0,0 +1,222 @@
+"""Unit tests for AuditLogMiddleware."""
+
+from __future__ import annotations
+
+import logging
+import time
+from unittest.mock import patch
+
+import pytest
+from starlette.testclient import TestClient
+from fastapi import FastAPI
+
+from orb.api.middleware.audit_log_middleware import AuditLogMiddleware
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+def _make_app_with_middleware():
+    """Return a minimal FastAPI app with AuditLogMiddleware attached."""
+    app = FastAPI()
+    app.add_middleware(AuditLogMiddleware)
+
+    @app.get("/api/v1/machines")
+    async def machines():
+        return {"ok": True}
+
+    @app.post("/api/v1/machines")
+    async def create_machine():
+        return {"created": True}
+
+    @app.get("/api/v1/config/settings")
+    async def config_settings():
+        return {"cfg": True}
+
+    @app.get("/api/v1/admin/status")
+    async def admin_status():
+        return {"admin": True}
+
+    @app.get("/api/v1/me")
+    async def me():
+        return {"user": "self"}
+
+    @app.get("/health")
+    async def health():
+        return {"ok": True}
+
+    @app.post("/api/v1/requests")
+    async def create_request():
+        return {"req": True}
+
+    return app
+
+
+# ---------------------------------------------------------------------------
+# dispatch matrix tests
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestAuditLogMiddlewareDispatch:
+
+    def test_get_non_audit_prefix_not_audited(self, caplog):
+        """A plain GET not matching AUDIT_ALWAYS_PREFIXES is skipped."""
+        app = _make_app_with_middleware()
+        client = TestClient(app, raise_server_exceptions=True)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            resp = client.get("/api/v1/machines")
+
+        assert resp.status_code == 200
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert audit_records == [], "GET /api/v1/machines should not be audited"
+
+    def test_get_config_prefix_is_audited(self, caplog):
+        """GET /api/v1/config/... must be audited even though it's a safe verb."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            resp = client.get("/api/v1/config/settings")
+
+        assert resp.status_code == 200
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert len(audit_records) == 1
+
+    def test_get_admin_prefix_is_audited(self, caplog):
+        """GET /api/v1/admin/... must be audited."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.get("/api/v1/admin/status")
+
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert len(audit_records) == 1
+
+    def test_get_me_prefix_is_audited(self, caplog):
+        """GET /api/v1/me must be audited."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.get("/api/v1/me")
+
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert len(audit_records) == 1
+
+    def test_non_get_mutating_request_is_audited(self, caplog):
+        """POST /api/v1/requests must always be audited."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/requests")
+
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert len(audit_records) == 1
+
+    def test_health_path_not_audited(self, caplog):
+        """GET /health is in SAFE_PATHS and should never be audited."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.get("/health")
+
+        audit_records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert audit_records == []
+
+
+# ---------------------------------------------------------------------------
+# audit log fields
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestAuditLogFields:
+
+    def test_latency_ms_present_in_log(self, caplog):
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/machines")
+
+        records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert len(records) == 1
+        rec = records[0]
+        latency = rec.__dict__.get("latency_ms")
+        assert latency is not None
+        assert isinstance(latency, float)
+        assert latency >= 0
+
+    def test_client_ip_captured(self, caplog):
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/machines")
+
+        records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert records[0].__dict__.get("client_ip") is not None
+
+    def test_user_id_defaults_to_anonymous(self, caplog):
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/machines")
+
+        records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert records[0].__dict__.get("user_id") == "anonymous"
+
+    def test_status_code_recorded(self, caplog):
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/machines")
+
+        records = [r for r in caplog.records if r.name == "orb.audit"]
+        assert records[0].__dict__.get("status_code") == 200
+
+    def test_method_and_path_recorded(self, caplog):
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            client.post("/api/v1/requests")
+
+        records = [r for r in caplog.records if r.name == "orb.audit"]
+        rec = records[0]
+        assert rec.__dict__.get("method") == "POST"
+        assert rec.__dict__.get("path") == "/api/v1/requests"
+
+
+# ---------------------------------------------------------------------------
+# latency monotonic semantics
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestAuditLogLatencySemantics:
+
+    def test_latency_uses_monotonic_clock(self, caplog):
+        """Latency must be measured with time.monotonic, not wall clock."""
+        app = _make_app_with_middleware()
+        client = TestClient(app)
+
+        call_times: list[float] = []
+        original_monotonic = time.monotonic
+
+        def patched_monotonic():
+            t = original_monotonic()
+            call_times.append(t)
+            return t
+
+        with caplog.at_level(logging.INFO, logger="orb.audit"):
+            with patch("orb.api.middleware.audit_log_middleware.time.monotonic", patched_monotonic):
+                client.post("/api/v1/machines")
+
+        # monotonic must have been called at least twice (start + end)
+        assert len(call_times) >= 2

tests/unit/api/middleware/test_audit_log_middleware.py

@@ -0,0 +1,148 @@
+"""Unit tests for ReadOnlyMiddleware."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import FastAPI
+from starlette.testclient import TestClient
+
+from orb.api.middleware.read_only_middleware import ReadOnlyMiddleware, _ALLOWED_PATHS
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+def _make_app(enabled: bool):
+    """Return a FastAPI app with ReadOnlyMiddleware configured."""
+    app = FastAPI()
+    app.add_middleware(ReadOnlyMiddleware, enabled=enabled)
+
+    @app.get("/api/v1/machines")
+    async def get_machines():
+        return {"machines": []}
+
+    @app.post("/api/v1/machines")
+    async def create_machine():
+        return {"created": True}
+
+    @app.post("/api/v1/requests")
+    async def create_request():
+        return {"req": True}
+
+    @app.post("/_event/some-event")
+    async def reflex_event():
+        return {"ok": True}
+
+    @app.post("/_upload/file")
+    async def reflex_upload():
+        return {"ok": True}
+
+    @app.post("/health")
+    async def health_post():
+        return {"ok": True}
+
+    return app
+
+
+# ---------------------------------------------------------------------------
+# Enabled + mutating → 403
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestReadOnlyMiddlewareEnabled:
+
+    def test_post_is_blocked_when_enabled(self):
+        client = TestClient(_make_app(enabled=True), raise_server_exceptions=True)
+        resp = client.post("/api/v1/machines")
+        assert resp.status_code == 403
+
+    def test_403_body_contains_read_only_mode_code(self):
+        client = TestClient(_make_app(enabled=True))
+        resp = client.post("/api/v1/machines")
+        body = resp.json()
+        assert body["error"]["code"] == "READ_ONLY_MODE"
+        assert body["success"] is False
+
+    def test_post_to_request_blocked(self):
+        client = TestClient(_make_app(enabled=True))
+        resp = client.post("/api/v1/requests")
+        assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# Enabled + safe verb → passes
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestReadOnlyMiddlewareSafeVerbs:
+
+    def test_get_passes_when_enabled(self):
+        client = TestClient(_make_app(enabled=True))
+        resp = client.get("/api/v1/machines")
+        assert resp.status_code == 200
+
+    def test_get_returns_normal_body(self):
+        client = TestClient(_make_app(enabled=True))
+        resp = client.get("/api/v1/machines")
+        assert resp.json() == {"machines": []}
+
+
+# ---------------------------------------------------------------------------
+# Enabled + allowed path → passes
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestReadOnlyMiddlewareAllowedPaths:
+
+    def test_post_health_passes_when_enabled(self):
+        client = TestClient(_make_app(enabled=True))
+        resp = client.post("/health")
+        assert resp.status_code == 200
+
+    def test_post_reflex_event_prefix_passes(self):
+        """/_event/... is an allowed path prefix — Reflex websocket sub-paths."""
+        client = TestClient(_make_app(enabled=True))
+        resp = client.post("/_event/some-event")
+        assert resp.status_code == 200
+
+    def test_post_reflex_upload_prefix_passes(self):
+        """/_upload/... is an allowed path prefix."""
+        client = TestClient(_make_app(enabled=True))
+        resp = client.post("/_upload/file")
+        assert resp.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Disabled + mutating → passes
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestReadOnlyMiddlewareDisabled:
+
+    def test_post_passes_when_disabled(self):
+        client = TestClient(_make_app(enabled=False))
+        resp = client.post("/api/v1/machines")
+        assert resp.status_code == 200
+
+    def test_get_passes_when_disabled(self):
+        client = TestClient(_make_app(enabled=False))
+        resp = client.get("/api/v1/machines")
+        assert resp.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Allowed path constants
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit
+class TestAllowedPathConstants:
+
+    def test_event_path_in_allowed_paths(self):
+        assert "/_event" in _ALLOWED_PATHS
+
+    def test_upload_path_in_allowed_paths(self):
+        assert "/_upload" in _ALLOWED_PATHS
+
+    def test_health_in_allowed_paths(self):
+        assert "/health" in _ALLOWED_PATHS