feat: admin dashboard

tests/api/test_admin_dashboard.py

@@ -0,0 +1,267 @@
+"""Endpoint tests for the admin dashboard (Phase 0 + Phase 1).
+
+Repos are mocked at the route layer (the real SQL is covered by
+tests/storage/db/repositories/test_admin_stats.py); these pin the route wiring,
+the @admin_required boundary, the audited mutations, and the safety guards.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import ExitStack, contextmanager
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+@pytest.fixture
+def client():
+    from application.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app.test_client()
+
+
+@contextmanager
+def _fake_conn():
+    yield Mock()
+
+
+@contextmanager
+def _admin(**route_patches):
+    """Authenticate as admin and patch application.api.admin.routes.* members."""
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch("application.app.handle_auth", return_value={"sub": "admin1"})
+        )
+        stack.enter_context(
+            patch("application.app.resolve_roles", return_value=["admin", "user"])
+        )
+        stack.enter_context(patch("application.api.admin.routes.db_readonly", _fake_conn))
+        stack.enter_context(patch("application.api.admin.routes.db_session", _fake_conn))
+        for name, value in route_patches.items():
+            stack.enter_context(patch(f"application.api.admin.routes.{name}", value))
+        yield
+
+
+def _body(resp):
+    return json.loads(resp.data)
+
+
+@pytest.mark.unit
+class TestGuard:
+    def test_non_admin_forbidden(self, client):
+        with patch("application.app.handle_auth", return_value={"sub": "u"}), patch(
+            "application.app.resolve_roles", return_value=["user"]
+        ):
+            assert client.get("/api/admin/overview").status_code == 403
+
+    def test_unauthenticated(self, client):
+        with patch("application.app.handle_auth", return_value=None):
+            assert client.get("/api/admin/overview").status_code == 401
+
+
+@pytest.mark.unit
+class TestOverview:
+    def test_ok(self, client):
+        stats = Mock()
+        stats.overview.return_value = {
+            "users": {"total": 3, "active": 3, "inactive": 0},
+            "admins": 1,
+            "agents": 0,
+            "sources": 0,
+            "conversations": 0,
+            "new_users_7d": 1,
+            "active_users_30d": 2,
+            "failed_logins_7d": 0,
+            "tokens_30d": 42,
+        }
+        with _admin(AdminStatsRepository=Mock(return_value=stats)):
+            resp = client.get("/api/admin/overview")
+        assert resp.status_code == 200
+        data = _body(resp)
+        assert data["users"]["total"] == 3
+        assert data["tokens_30d"] == 42
+
+
+@pytest.mark.unit
+class TestAdminsAndRoles:
+    def test_list_admins(self, client):
+        roles = Mock()
+        roles.list_admins.return_value = [
+            {"user_id": "a", "granted_at": None, "sources": ["manual"]}
+        ]
+        with _admin(UserRolesRepository=Mock(return_value=roles)):
+            resp = client.get("/api/admin/admins")
+        assert resp.status_code == 200
+        assert _body(resp)["admins"][0]["user_id"] == "a"
+
+    def test_grant_admin_audited(self, client):
+        roles = Mock()
+        roles.grant.return_value = True
+        events = Mock()
+        with _admin(
+            UserRolesRepository=Mock(return_value=roles),
+            AuthEventsRepository=Mock(return_value=events),
+        ):
+            resp = client.post("/api/admin/users/alice/role")
+        assert resp.status_code == 200
+        assert _body(resp)["granted"] is True
+        roles.grant.assert_called_once()
+        # audited as role_granted
+        assert events.insert.call_args.args[1] == "role_granted"
+
+    def test_revoke_blocks_last_admin(self, client):
+        roles = Mock()
+        roles.list_admins.return_value = [{"user_id": "alice"}]
+        with _admin(
+            UserRolesRepository=Mock(return_value=roles),
+            AuthEventsRepository=Mock(return_value=Mock()),
+        ):
+            resp = client.delete("/api/admin/users/alice/role")
+        assert resp.status_code == 409
+        roles.revoke.assert_not_called()
+
+    def test_revoke_ok_when_other_admins_exist(self, client):
+        roles = Mock()
+        roles.list_admins.return_value = [{"user_id": "alice"}, {"user_id": "bob"}]
+        roles.revoke.return_value = True
+        events = Mock()
+        with _admin(
+            UserRolesRepository=Mock(return_value=roles),
+            AuthEventsRepository=Mock(return_value=events),
+        ):
+            resp = client.delete("/api/admin/users/alice/role")
+        assert resp.status_code == 200
+        roles.revoke.assert_called_once()
+        assert events.insert.call_args.args[1] == "role_revoked"
+
+
+@pytest.mark.unit
+class TestUserLifecycle:
+    def test_self_deactivation_blocked(self, client):
+        with _admin(UsersRepository=Mock()):
+            resp = client.patch("/api/admin/users/admin1", json={"active": False})
+        assert resp.status_code == 409
+
+    def test_bad_body(self, client):
+        with _admin(UsersRepository=Mock()):
+            resp = client.patch("/api/admin/users/bob", json={})
+        assert resp.status_code == 400
+
+    def test_user_not_found(self, client):
+        users = Mock()
+        users.get.return_value = None
+        with _admin(
+            UsersRepository=Mock(return_value=users),
+            AuthEventsRepository=Mock(return_value=Mock()),
+        ):
+            resp = client.patch("/api/admin/users/ghost", json={"active": True})
+        assert resp.status_code == 404
+
+    def test_deactivate_revokes_sessions_and_audits(self, client):
+        users = Mock()
+        users.get.return_value = {
+            "id": "00000000-0000-0000-0000-000000000001",
+            "user_id": "bob",
+        }
+        users.set_active.return_value = {"active": False}
+        events = Mock()
+        with _admin(
+            UsersRepository=Mock(return_value=users),
+            AuthEventsRepository=Mock(return_value=events),
+        ), patch("application.api.admin.routes.denylist") as dl:
+            resp = client.patch("/api/admin/users/bob", json={"active": False})
+        assert resp.status_code == 200
+        dl.deny_user.assert_called_once_with("bob")
+        assert events.insert.call_args.args[1] == "admin_user_deactivated"
+
+    def test_force_logout(self, client):
+        events = Mock()
+        with _admin(AuthEventsRepository=Mock(return_value=events)), patch(
+            "application.api.admin.routes.denylist"
+        ) as dl:
+            dl.deny_user.return_value = True
+            resp = client.post("/api/admin/users/bob/revoke-sessions")
+        assert resp.status_code == 200
+        dl.deny_user.assert_called_once_with("bob")
+
+    def test_user_detail(self, client):
+        users = Mock()
+        users.get.return_value = {
+            "user_id": "bob",
+            "active": True,
+            "created_at": None,
+            "updated_at": None,
+        }
+        roles = Mock()
+        roles.role_names_for.return_value = ["admin"]
+        roles.list_for.return_value = []
+        events = Mock()
+        events.list_recent.return_value = []
+        stats = Mock()
+        stats.user_counts.return_value = {
+            "agents": 0,
+            "sources": 0,
+            "conversations": 0,
+            "tokens_30d": 0,
+        }
+        with _admin(
+            UsersRepository=Mock(return_value=users),
+            UserRolesRepository=Mock(return_value=roles),
+            AuthEventsRepository=Mock(return_value=events),
+            AdminStatsRepository=Mock(return_value=stats),
+        ):
+            resp = client.get("/api/admin/users/bob")
+        assert resp.status_code == 200
+        data = _body(resp)
+        assert data["user"]["user_id"] == "bob"
+        assert "admin" in data["roles"] and "user" in data["roles"]
+
+
+@pytest.mark.unit
+class TestUsageAndAudit:
+    def test_usage(self, client):
+        usage = Mock()
+        usage.bucketed_totals.return_value = [
+            {"bucket": "2026-06-14", "prompt_tokens": 5, "generated_tokens": 3}
+        ]
+        usage.sum_tokens_in_range.return_value = 8
+        stats = Mock()
+        stats.top_token_users.return_value = [{"user_id": "a", "tokens": 8}]
+        with _admin(
+            TokenUsageRepository=Mock(return_value=usage),
+            AdminStatsRepository=Mock(return_value=stats),
+        ):
+            resp = client.get("/api/admin/usage?days=7&bucket=day")
+        assert resp.status_code == 200
+        data = _body(resp)
+        assert data["total_tokens"] == 8
+        assert len(data["series"]) == 1
+        assert data["top_users"][0]["user_id"] == "a"
+
+    def test_usage_invalid_bucket(self, client):
+        with _admin():
+            resp = client.get("/api/admin/usage?bucket=year")
+        assert resp.status_code == 400
+
+    def test_audit_feed(self, client):
+        repo = Mock()
+        repo.count_all.return_value = 1
+        repo.list_all.return_value = [{"user_id": "a", "event": "oidc_login"}]
+        with _admin(AuthEventsRepository=Mock(return_value=repo)):
+            resp = client.get("/api/admin/audit?event=oidc_login&page=1&page_size=10")
+        assert resp.status_code == 200
+        data = _body(resp)
+        assert data["total"] == 1
+        assert data["events"][0]["event"] == "oidc_login"
+        repo.list_all.assert_called_once()
+
+    def test_device_audit_feed(self, client):
+        repo = Mock()
+        repo.count_global.return_value = 0
+        repo.list_global.return_value = []
+        with _admin(DeviceAuditLogRepository=Mock(return_value=repo)):
+            resp = client.get("/api/admin/devices/audit?decision=denied")
+        assert resp.status_code == 200
+        assert _body(resp)["invocations"] == []

tests/api/test_rbac_endpoints.py

@@ -64,66 +64,70 @@ class TestAdminUsersEndpoint:
 
     @staticmethod
     def _admin_get(client, repo, query=""):
+        # The list endpoint resolves users (with last_seen) via AdminStatsRepository.
         with patch("application.app.handle_auth", return_value={"sub": "a1"}), patch(
             "application.app.resolve_roles", return_value=["admin", "user"]
         ), patch("application.api.admin.routes.db_readonly", _fake_readonly), patch(
-            "application.api.admin.routes.UsersRepository", return_value=repo
+            "application.api.admin.routes.AdminStatsRepository", return_value=repo
         ):
             return client.get(f"/api/admin/users{query}")
 
     @staticmethod
     def _rows(n):
         return [
-            {"user_id": f"u{i}", "active": True, "created_at": "2026-01-01"}
+            {
+                "user_id": f"u{i}",
+                "active": True,
+                "created_at": "2026-01-01",
+                "last_seen": "2026-02-01",
+            }
             for i in range(n)
         ]
 
     def test_admin_gets_user_list(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (
-            1,
-            [{"user_id": "u1", "active": True, "created_at": "2026-01-01"}],
-        )
+        repo.list_users.return_value = (1, self._rows(1))
         resp = self._admin_get(client, repo)
         assert resp.status_code == 200
         data = json.loads(resp.data)
         assert data["success"] is True
         assert data["total"] == 1
         assert data["users"][0] == {
-            "user_id": "u1",
+            "user_id": "u0",
             "active": True,
             "created_at": "2026-01-01",
+            "last_seen": "2026-02-01",
         }
-        repo.list_paginated.assert_called_once_with(None, 0, 25)
+        repo.list_users.assert_called_once_with(None, 0, 25)
 
     def test_page_size_clamped_and_offset_computed(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (0, [])
+        repo.list_users.return_value = (0, [])
         self._admin_get(client, repo, "?page=2&page_size=999")
         # page_size clamps to _MAX_PAGE_SIZE=100, offset=(2-1)*100
-        repo.list_paginated.assert_called_once_with(None, 100, 100)
+        repo.list_users.assert_called_once_with(None, 100, 100)
 
     def test_non_int_params_fall_back_to_defaults(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (0, [])
+        repo.list_users.return_value = (0, [])
         self._admin_get(client, repo, "?page=abc&page_size=xyz")
-        repo.list_paginated.assert_called_once_with(None, 0, 25)
+        repo.list_users.assert_called_once_with(None, 0, 25)
 
     def test_user_id_filter_passed_first_positional(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (0, [])
+        repo.list_users.return_value = (0, [])
         self._admin_get(client, repo, "?user_id=foo")
-        repo.list_paginated.assert_called_once_with("foo", 0, 25)
+        repo.list_users.assert_called_once_with("foo", 0, 25)
 
     def test_has_more_true_when_more_pages_remain(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (5, self._rows(3))
+        repo.list_users.return_value = (5, self._rows(3))
         resp = self._admin_get(client, repo, "?page=1&page_size=3")
         assert json.loads(resp.data)["has_more"] is True
 
     def test_has_more_false_at_last_page(self, client):
         repo = Mock()
-        repo.list_paginated.return_value = (3, self._rows(3))
+        repo.list_users.return_value = (3, self._rows(3))
         resp = self._admin_get(client, repo, "?page=1&page_size=3")
         assert json.loads(resp.data)["has_more"] is False
 
@@ -188,7 +192,7 @@ class TestChokepointOverwritesForgedRoles:
         authz_repo = Mock()
         authz_repo.role_names_for.return_value = ["admin"]
         admin_repo = Mock()
-        admin_repo.list_paginated.return_value = (0, [])
+        admin_repo.list_users.return_value = (0, [])
         with patch(
             "application.app.handle_auth", return_value={"sub": "alice"}
         ), patch("application.app.oidc_session_denied", return_value=False), patch.object(
@@ -196,7 +200,7 @@ class TestChokepointOverwritesForgedRoles:
         ), patch.object(authz, "db_readonly", _fake_readonly), patch.object(
             authz, "UserRolesRepository", return_value=authz_repo
         ), patch("application.api.admin.routes.db_readonly", _fake_readonly), patch(
-            "application.api.admin.routes.UsersRepository", return_value=admin_repo
+            "application.api.admin.routes.AdminStatsRepository", return_value=admin_repo
         ):
             admin_resp = client.get("/api/admin/users")
             me_resp = client.get("/api/user/me")