tests/test_admin.py

"""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