agnes-the-ai-analyst/tests/test_admin_user_capabilities_ui.py
ZdenekSrotyr e9d7af3cce feat(rbac+marketplace): RBAC v13 + Claude Code marketplace + #81/#83/#44 hardening
This squashes 13 commits from ma/staging plus a small docstring translation
into a single coherent unit. Three workstreams.

== RBAC v13 redesign ==
- Drops core.viewer/analyst/km_admin/admin hierarchy and the
  internal_roles / group_mappings / user_role_grants / plugin_access tables.
- Replaced by user_group_members + resource_grants. Atomic v12→v13 backfill
  wrapped in BEGIN/COMMIT; ROLLBACK leaves schema_version at 12 for retry.
- Two authorization primitives in app.auth.access:
    require_admin                        — Admin-group god-mode
    require_resource_access(rt, "{path}") — entity-scoped grants
  Single DB lookup per request; no session cache; no implies BFS.
- /admin/access UI (single page) replaces /admin/role-mapping +
  /admin/plugin-access. CLI `da admin group/grant *` replaces
  `da admin role/mapping/grant-role/revoke-role/effective-roles`.
- ResourceType.TABLE listing-only — admins can record table grants,
  runtime enforcement still flows through legacy dataset_permissions
  (migration plan in docs/TODO-rbac-data-enforcement.md).

== Claude Code marketplace ==
- Aggregated /marketplace.zip + /marketplace.git/* (PAT-gated,
  RBAC-filtered, content-addressed cache via dulwich).
- Admin god-mode dropped on the marketplace surface — admins curate
  their own view via grants like everyone else.
- Bare-repo cache materializes per RBAC-filtered ETag; stale entries
  not pruned in this iteration (disclaimed in git_backend.py docstring).

== #81 #83 #44 security/ops hardening ==
- #81 Group A — orchestrator ATTACH allow-listing (extension/url/alias).
- #81 Group B — Keboola extractor 3-state exit codes:
    0 success / 1 total fail / 2 PARTIAL fail
  Sync API logs PARTIAL FAILURE alert on exit 2. Operators with binary
  alerting must teach it the new partial signal.
- #81 Group C — schema v10 view_ownership; rejects silent overwrite
  of a prior connector's view name on collision.
- #81 Group D — extractor-side identifier validation.
- #83 — Jira webhook fail-closed when JIRA_WEBHOOK_SECRET unset
  + path-traversal fix.
- #44 — entire /api/scripts/* surface is admin-only (planted-script +
  sandbox-bypass risk closed).

== Web UI polish + deploy fix ==
- /admin/access: live grant-count badges (no stale snapshot revert),
  shared-header CSS link added to /catalog and /admin/{tables,permissions},
  per-resource-type colored stripes.
- docker-compose.host-mount.yml: bind,rbind so dual-disk hosts don't
  silently shadow sub-mounts and write state to the wrong disk.

== OSS vendor-neutralization (waves 1+2) ==
- scripts/grpn/ → scripts/ops/. Customer-specific identifiers
  (project IDs, internal hostnames, dev/prod VM IPs, brand names)
  replaced with placeholders across code, docs, Terraform, Caddyfile,
  OAuth probe, and planning docs. Downstream infra repos that copied
  scripts/grpn/agnes-tls-rotate.sh or agnes-auto-upgrade.sh must
  update the path.

== Translation ==
- src/repositories/user_groups.py::ensure_system docstring translated
  from Czech to English for codebase consistency.

Co-authored-by: Mina Rustamyan <mina@keboola.com>
2026-04-28 14:25:04 +02:00

347 lines
11 KiB
Python

"""Tests for the per-user detail page (/admin/users/{user_id}).
v12 status: the legacy v9 capabilities UI (core-role dropdown / additional
capabilities checkboxes / effective-roles debug section) is gone — the
new /admin/users/{id} page surfaces group memberships and an effective
resource-access readout instead. The whole module asserts v9 markers and
hits removed REST endpoints (/api/role-grants, /api/admin/users/.../core-role)
so it's skipped en bloc until rewritten against the v12 layout.
"""
import tempfile
import uuid
import pytest
pytest.skip(
"v12: legacy capabilities UI replaced by group memberships + effective "
"access. Rewrite the module against templates/admin_user_detail.html "
"(v12) and /api/admin/users/{id}/memberships endpoints.",
allow_module_level=True,
)
@pytest.fixture
def fresh_db(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
monkeypatch.setenv("DATA_DIR", tmp)
monkeypatch.setenv("TESTING", "1")
monkeypatch.setenv("JWT_SECRET_KEY", "test-jwt-secret-key-minimum-32-chars!!")
yield tmp
def _make_user_and_session(conn, email: str, role: str):
from src.repositories.users import UserRepository
from app.auth.jwt import create_access_token
from tests.helpers.auth import grant_admin
uid = str(uuid.uuid4())
UserRepository(conn).create(id=uid, email=email, name=email.split("@")[0], role=role)
if role == "admin":
grant_admin(conn, uid)
token = create_access_token(user_id=uid, email=email, role=role)
return uid, token
# ── Auth gate ────────────────────────────────────────────────────────────
def test_admin_can_render_user_detail_page(fresh_db):
"""Admin GET /admin/users/{user_id}: 200 with all section markers."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
headers={"Accept": "text/html"},
cookies={"access_token": admin_sess},
)
assert resp.status_code == 200, resp.text
body = resp.text
# Page-shell markers
assert "victim@t" in body
assert target_uid in body
# All three sections are present
assert 'data-section="core-role"' in body
assert 'data-section="capabilities"' in body
assert 'data-section="effective-roles"' in body
def test_user_detail_page_renders_core_role_dropdown(fresh_db):
"""Section A: core role single-select dropdown is present."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
cookies={"access_token": admin_sess},
)
assert resp.status_code == 200
body = resp.text
# Section A markers
assert 'id="core-role-select"' in body
assert "Core role" in body
# JS endpoint references
assert "/api/admin/internal-roles" in body
assert "/api/admin/users/" in body
assert "/role-grants" in body
assert "/effective-roles" in body
def test_user_detail_page_renders_capabilities_list(fresh_db):
"""Section B: capabilities list element is in the rendered page."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
cookies={"access_token": admin_sess},
)
assert resp.status_code == 200
body = resp.text
# Section B markers — caps list container + loading state
assert 'id="caps-list"' in body
assert "Additional capabilities" in body
# Toggle handler presence
assert "toggleCapability" in body
def test_user_detail_page_renders_effective_roles_section(fresh_db):
"""Section C: effective-roles debug view with the three lists."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
cookies={"access_token": admin_sess},
)
assert resp.status_code == 200
body = resp.text
# Section C — three list containers for direct/groups/expanded
assert 'id="effective-direct"' in body
assert 'id="effective-groups"' in body
assert 'id="effective-expanded"' in body
# Section labels
assert "Direct grants" in body
assert "Group-derived grants" in body
assert "Expanded set" in body
def test_user_detail_page_unknown_user_returns_404(fresh_db):
"""GET /admin/users/{nonexistent}: 404 (UI surface check, not API)."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
"/admin/users/00000000-0000-0000-0000-000000000000",
cookies={"access_token": admin_sess},
)
assert resp.status_code == 404, resp.text
def test_non_admin_cannot_access_user_detail_page(fresh_db):
"""Non-admin GET /admin/users/{id}: 401/403."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
analyst_uid, sess = _make_user_and_session(conn, "user@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{analyst_uid}",
cookies={"access_token": sess},
follow_redirects=False,
)
assert resp.status_code in (302, 401, 403), resp.text
# ── Toggle and core-role-change JS contract ──────────────────────────────
def test_user_detail_page_toggle_capability_uses_role_grants_api(fresh_db):
"""The capability toggle must POST/DELETE against the role-grants endpoint
so the API agent's URL contract is honored."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
cookies={"access_token": admin_sess},
)
body = resp.text
# POST to role-grants with role_key body for grant
assert "role_key" in body
# DELETE to role-grants/{grant_id} for revoke
assert 'method: "POST"' in body
assert 'method: "DELETE"' in body
def test_user_detail_page_core_role_change_deletes_then_creates(fresh_db):
"""Section A flow: changing core role deletes old grants then POSTs the new one,
matching the brief's "delete+create dance from the client" requirement."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
target_uid, _ = _make_user_and_session(conn, "victim@t", "analyst")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
f"/admin/users/{target_uid}",
cookies={"access_token": admin_sess},
)
body = resp.text
# The changeCoreRole function name
assert "changeCoreRole" in body
# The JS comment / logic to filter core.* grants is present.
assert 'core.' in body or "core_role" in body
# ── Add-user form (Part 3 verification: writes to user_role_grants) ─────
def test_create_user_via_api_grants_core_role(fresh_db):
"""Verify the existing add-user flow auto-grants core.{role}.
The brief asks us to verify (not modify) that the Add-user modal's
POST hits an endpoint that routes through UserRepository.create(),
which inserts a user_role_grants row. This pins the contract: a
fresh user must end up with core.{role} in user_role_grants.
"""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.post(
"/api/users",
cookies={"access_token": admin_sess},
json={"email": "newbie@t", "name": "Newbie", "role": "analyst", "send_invite": False},
)
assert resp.status_code == 201, resp.text
new_user = resp.json()
new_uid = new_user["id"]
# Now check user_role_grants directly — confirm core.analyst is present.
conn = get_system_db()
try:
rows = conn.execute(
"""SELECT r.key
FROM user_role_grants g
JOIN internal_roles r ON g.internal_role_id = r.id
WHERE g.user_id = ?""",
[new_uid],
).fetchall()
keys = [r[0] for r in rows]
finally:
conn.close()
close_system_db()
assert "core.analyst" in keys, (
"UserRepository.create should auto-grant core.{role}; "
f"got {keys}"
)
def test_admin_users_page_renders_detail_link(fresh_db):
"""The /admin/users list now links to /admin/users/{id} — verify the
JS that builds row HTML emits the right href."""
from fastapi.testclient import TestClient
from src.db import get_system_db, close_system_db
from app.main import app
conn = get_system_db()
try:
_, admin_sess = _make_user_and_session(conn, "admin@t", "admin")
finally:
conn.close()
close_system_db()
client = TestClient(app)
resp = client.get(
"/admin/users",
cookies={"access_token": admin_sess},
)
assert resp.status_code == 200
body = resp.text
# The <a href> template literal in renderUsers includes /admin/users/${u.id}
assert "/admin/users/${encodeURIComponent(u.id)}" in body