"""Admin / governance endpoints — owner-only. The `require_owner` gate is the security-critical contract here. These tests exercise: * audit-log filter combinations (action / entity_type / since / until / limit) * export shape — every section is present, no embedding bytes leak through * cross-tenant isolation on both endpoints * role gate — a `creator`-role membership is 403'd """ import json from app.db import SessionLocal from app.models import Membership from app.roles import CREATOR from tests._helpers import ( auth_headers, create_creator, create_persona, enable_widget, seed_approved_clip, signup, ) def _force_role(email: str, role: str) -> None: """Hack-route to set the user's membership role for the role-gate test — we don't have an invite/role-management endpoint yet (Phase 6+).""" from app.models import User db = SessionLocal() try: user = db.query(User).filter(User.email == email).first() m = db.query(Membership).filter(Membership.user_id == user.id).first() m.role = role db.commit() finally: db.close() # ─── /audit-logs ──────────────────────────────────────────────────────────── def test_audit_logs_returns_recent_owner_actions(client, fake_storage, no_redis): token = signup(client, "audit-r1@example.com") creator = create_creator(client, token) persona = create_persona(client, token, creator["id"]) enable_widget(client, token, persona["id"]) r = client.get("/audit-logs", headers=auth_headers(token)) assert r.status_code == 200, r.text actions = {row["action"] for row in r.json()} # We expect at least these three from the flow above assert {"persona.created", "consent.accepted", "persona.widget_enabled"} <= actions def test_audit_logs_action_filter(client, fake_storage, no_redis): token = signup(client, "audit-r2@example.com") creator = create_creator(client, token) create_persona(client, token, creator["id"]) r = client.get( "/audit-logs?action=persona.created", headers=auth_headers(token) ) assert r.status_code == 200 rows = r.json() assert all(row["action"] == "persona.created" for row in rows) assert len(rows) >= 1 def test_audit_logs_limit_capped(client): token = signup(client, "audit-r3@example.com") r = client.get("/audit-logs?limit=1000", headers=auth_headers(token)) # 422 from Pydantic on limit > 500 assert r.status_code == 422 def test_audit_logs_cross_tenant_invisible(client, fake_storage, no_redis): token_a = signup(client, "alice-au@example.com", org="OrgAuA") token_b = signup(client, "bob-au@example.com", org="OrgAuB") creator_a = create_creator(client, token_a) create_persona(client, token_a, creator_a["id"]) rows_b = client.get("/audit-logs", headers=auth_headers(token_b)).json() # Bob sees only his own org's audit rows (his signup wrote none — the # audit list should only contain entries scoped to org B) assert all(row["organization_id"] != "alice-au-org-anything" for row in rows_b) a_actions = {row["action"] for row in rows_b} assert "persona.created" not in a_actions # Alice's persona didn't show up def test_audit_logs_role_gate(client, fake_storage, no_redis): """A user whose membership.role == 'creator' is 403'd from /audit-logs.""" token = signup(client, "creator-role@example.com") _force_role("creator-role@example.com", CREATOR) r = client.get("/audit-logs", headers=auth_headers(token)) assert r.status_code == 403 assert "owner" in r.json()["detail"] # ─── /export ──────────────────────────────────────────────────────────────── def test_export_shape_includes_all_sections(client, fake_storage, no_redis): token = signup(client, "exp-1@example.com") creator = create_creator(client, token) persona = create_persona(client, token, creator["id"]) enable_widget(client, token, persona["id"]) seed_approved_clip(client, token, persona["id"], intent_tag="who_you_are") r = client.get("/export", headers=auth_headers(token)) assert r.status_code == 200 assert r.headers["content-type"].startswith("application/json") assert "personadesk-export.json" in r.headers["content-disposition"] payload = json.loads(r.text) expected_keys = { "export_version", "exported_at", "organization", "creators", "personas", "consents", "media", "clips", "sessions", "leads", "audit_logs", } assert set(payload.keys()) == expected_keys assert payload["export_version"] == 1 assert payload["organization"]["name"] assert len(payload["creators"]) == 1 assert len(payload["personas"]) == 1 assert len(payload["clips"]) == 1 def test_export_clip_drops_embedding_keeps_transcript(client, fake_storage, no_redis): """Invariant: the export contains useful clip fields but never the raw transcript_embedding float vector (huge, useless to a human, would explode the payload).""" token = signup(client, "exp-2@example.com") creator = create_creator(client, token) persona = create_persona(client, token, creator["id"]) seed_approved_clip( client, token, persona["id"], intent_tag="pricing", transcript="המחיר X", ) payload = json.loads(client.get("/export", headers=auth_headers(token)).text) clip = payload["clips"][0] assert "transcript_embedding" not in clip assert clip["transcript"] == "המחיר X" assert clip["intent_tags"] == ["pricing"] def test_export_cross_tenant_isolation(client, fake_storage, no_redis): token_a = signup(client, "exp-A@example.com", org="OrgExpA") token_b = signup(client, "exp-B@example.com", org="OrgExpB") creator_a = create_creator(client, token_a, display_name="Alice's") create_persona(client, token_a, creator_a["id"], display_name="Alice's Persona") # Bob's export shouldn't mention Alice payload_b = json.loads(client.get("/export", headers=auth_headers(token_b)).text) names_b = {c["display_name"] for c in payload_b["creators"]} assert "Alice's" not in names_b assert payload_b["personas"] == [] def test_export_role_gate(client): token = signup(client, "exp-role@example.com") _force_role("exp-role@example.com", CREATOR) assert client.get("/export", headers=auth_headers(token)).status_code == 403