sales-trainer/backend/tests/test_sprint1_review_findings.py

"""Regression tests for the final Sprint 1 reviewer findings."""
from __future__ import annotations

import io
import multiprocessing as mp
from pathlib import Path
import sys

import jwt

from app.api.chat_routes import _safe_judge_debrief, serialize_session
from app.api.group_routes import serialize_group, strip_secret_fields
from app.auth.users import AuthError, UserStore
from app.config import Config
from app.services.analyzer import Analyzer
from app.services.file_parser import ParseError
from app.services.groups import GroupStore
from app.services.sessions import SessionStore
from app.storage.store import JsonStore
from scripts.mock_llm import MockLLM


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _setup_super_admin(user_store, login) -> str:
    user_store.complete_setup(
        "admin",
        "admin@example.com",
        "admin-ready-password",
        accepted_terms=True,
        accepted_terms_at="2026-08-13T00:00:00Z",
    )
    return login("admin", "admin-ready-password")["token"]


def _update_org_in_process(
    data_dir: str,
    fields: dict[str, object],
    read_event,
    release_read,
    result_queue,
) -> None:
    """Pause after reading one org so cross-process read-modify-write races are deterministic."""
    import app.storage.store as store_module

    original_read_json = store_module._read_json

    def synchronized_read(path):
        value = original_read_json(path)
        if path.name == "org-1.json":
            read_event.set()
            release_read.wait(timeout=10)
        return value

    store_module._read_json = synchronized_read
    try:
        JsonStore(Path(data_dir)).update("org-1", **fields)
    except BaseException as exc:  # pragma: no cover - child diagnostic
        result_queue.put(("error", type(exc).__name__))
    else:
        result_queue.put(("ok", fields))


def test_org_updates_preserve_concurrent_fields_across_processes(tmp_path):
    """Organization PATCH read-modify-write updates must not lose another worker's field."""
    store = JsonStore(tmp_path / "orgs")
    store.create(
        {
            "id": "org-1",
            "name": "Organization One",
            "plan": "trial",
            "seats": 5,
            "active": True,
        },
        key="org-1",
    )

    context = mp.get_context("fork")
    first_read = context.Event()
    release_first = context.Event()
    second_read = context.Event()
    release_second = context.Event()
    result_queue = context.Queue()
    first = context.Process(
        target=_update_org_in_process,
        args=(str(store.root), {"active": False}, first_read, release_first, result_queue),
    )
    second = context.Process(
        target=_update_org_in_process,
        args=(str(store.root), {"seats": 10}, second_read, release_second, result_queue),
    )

    first.start()
    second.start()
    try:
        # Forked workers can take longer to import the Flask app under a full
        # suite on a loaded CI runner; keep the synchronization deterministic
        # without changing the race it is testing.
        assert first_read.wait(timeout=15)
        second_read_before_first_write = second_read.wait(timeout=2)
        consumed_results = 0
        if second_read_before_first_write:
            # Without a process lock, let the second worker commit its stale read first.
            release_second.set()
            assert result_queue.get(timeout=5)[0] == "ok"
            consumed_results = 1
        release_first.set()
        assert second_read.wait(timeout=5)
        release_second.set()

        first.join(timeout=5)
        second.join(timeout=5)
        assert first.exitcode == 0
        assert second.exitcode == 0
        remaining_results = [result_queue.get(timeout=5)[0] for _ in range(2 - consumed_results)]
        assert remaining_results == ["ok"] * (2 - consumed_results)
    finally:
        release_first.set()
        release_second.set()
        for process in (first, second):
            process.join(timeout=5)
            if process.is_alive():  # pragma: no cover - cleanup for failed child
                process.terminate()
                process.join(timeout=5)

    updated = store.get("org-1")
    assert updated["active"] is False
    assert updated["seats"] == 10


def test_judge_debrief_uses_closed_allowlist():
    debrief = _safe_judge_debrief(
        {
            "outcome": "won",
            "score": 101,
            "pain": "customer pain",
            "why": "good fit",
            "failurePoints": ["missed discovery", {"provider_path": "/secret"}],
            "coaching": ["ask one more question"],
            "painProgress": {"main": 120, "internal_secret": "not-a-number"},
            "password_hash": "must-not-leak",
            "provider_path": "/private/provider/path",
        },
        "won",
        {"pains": [{"description": "customer pain"}], "internal_secret": "must-not-leak"},
        {
            "why": "good fit",
            "failurePoints": ["missed discovery", {"provider_path": "/secret"}],
            "coaching": ["ask one more question"],
        },
    )

    assert set(debrief) == {
        "outcome", "score", "why", "failurePoints", "coaching", "revealed_persona",
    }
    assert debrief["score"] == 100
    assert debrief["failurePoints"] == ["missed discovery"]
    # IP protection: pain (prose + list) and painProgress are stripped from the debrief.
    assert "pain" not in debrief
    assert "painProgress" not in debrief
    assert "pains" not in debrief["revealed_persona"]
    assert "password_hash" not in str(debrief)
    assert "provider_path" not in str(debrief)


def test_group_serializer_fails_closed_on_malformed_internal_shapes():
    view = serialize_group(
        {
            "id": "group-1",
            "status": "failed",
            "error": "/private/provider/path",
            "input": "/private/source.txt",
            "personas": [{"id": "p1", "name": "Safe"}, "not-a-persona"],
        },
        {"role": "user"},
    )

    assert view["error"] == "analysis_failed"
    assert view["input"] == {}
    assert view["personas"] == []
    assert "/private/provider/path" not in str(view)


def test_session_serializer_omits_hidden_internal_state():
    view = serialize_session(
        {
            "id": "session-1",
            "user_id": "user-1",
            "group_id": "group-1",
            "persona_id": "persona-1",
            "persona_name": "Customer",
            "status": "active",
            "outcome": None,
            "persona_meta": {
                "scenario": "social",
                "locale": "th",
                "provider_path": "/private/provider/path",
            },
            "messages": [{"role": "customer", "text": "hello", "internal": "drop"}],
            "internal": {"last_reason": "hidden judge reasoning", "provider_path": "/private"},
            "debrief": None,
        }
    )

    assert "user_id" not in view
    assert "internal" not in view
    assert view["persona_meta"] == {"scenario": "social", "locale": "th"}
    assert view["messages"] == [{
        "role": "customer",
        "text": "ลูกค้าเริ่มต้นบทสนทนาแล้ว ลองทักและค้นหาความต้องการดูครับ",
    }]
    assert "/private" not in str(view)


def test_session_serializer_keeps_only_public_scene_notes():
    view = serialize_session(
        {
            "id": "session-2",
            "group_id": "group-1",
            "persona_id": "persona-1",
            "persona_name": "Customer",
            "status": "active",
            "messages": [
                {"role": "seller", "text": "hello"},
                {"role": "assistant", "text": "hidden judge output"},
                {"role": "system", "text": "hidden system prompt"},
                {"role": "system", "text": "⏳ ผ่านไป 2-3 สัปดาห์ ... ลูกค้าที่เคยสอบถามไปเงียบไประยะหนึ่ง ตอนนี้กลับมาติดต่ออีกครั้ง (พร้อมตัดสินใจมากขึ้น)"},
            ],
        }
    )

    assert view["messages"] == [
        {"role": "seller", "text": "hello"},
        {"role": "system", "text": "⏳ ผ่านไป 2-3 สัปดาห์ ... ลูกค้าที่เคยสอบถามไปเงียบไประยะหนึ่ง ตอนนี้กลับมาติดต่ออีกครั้ง (พร้อมตัดสินใจมากขึ้น)"},
    ]
    assert "hidden judge output" not in str(view)
    assert "hidden system prompt" not in str(view)


def test_super_admin_manages_admin_and_super_admin_roles(
    client, user_store, login
):
    """Per UX/SAAS plan (P0.1): a super_admin MAY promote/manage other
    super_admins. Only a regular admin is locked out of admin/super_admin status
    changes (granting super_admin/admin is super_admin-only)."""
    token = _setup_super_admin(user_store, login)
    # Room for the admin, a peer, a promoted user, a reg admin and a target.
    with user_store.orgs.record_lock("org-default"):
        user_store.orgs.update("org-default", seats=50)
    peer = user_store.create_user(
        org_id="org-default",
        username="peer-super-admin",
        password="peer-super-password",
        name="Peer Super Admin",
        role="super_admin",
        must_setup=False,
    )

    # super_admin may provision another super_admin directly
    create_response = client.post(
        "/api/admin/users",
        json={
            "username": "new-super-admin",
            "password": "new-super-password",
            "role": "super_admin",
        },
        headers=_headers(token),
    )
    assert create_response.status_code == 201, create_response.get_json()
    assert user_store.get_user("new-super-admin")["role"] == "super_admin"

    # super_admin may demote/change a peer super_admin and manage their status
    peer_role_response = client.put(
        f"/api/admin/users/{peer['username']}",
        json={"role": "admin"},
        headers=_headers(token),
    )
    assert peer_role_response.status_code == 200, peer_role_response.get_json()
    assert user_store.get_user(peer["username"])["role"] == "admin"

    peer_active_response = client.put(
        f"/api/admin/users/{peer['username']}",
        json={"active": False},
        headers=_headers(token),
    )
    assert peer_active_response.status_code == 200, peer_active_response.get_json()

    # super_admin may promote a plain user to super_admin
    normal = user_store.create_user(
        org_id="org-default",
        username="normal-user",
        password="normal-password",
        name="Normal User",
        role="user",
        must_setup=False,
    )
    promote_response = client.put(
        f"/api/admin/users/{normal['username']}",
        json={"role": "super_admin"},
        headers=_headers(token),
    )
    assert promote_response.status_code == 200, promote_response.get_json()
    assert user_store.get_user(normal["username"])["role"] == "super_admin"

    # A REGULAR admin still cannot grant/change admin or super_admin status.
    reg_admin = user_store.create_user(
        org_id="org-default",
        username="tenant-admin-creator",
        password="tenant-admin-password",
        name="Tenant Admin",
        role="admin",
        must_setup=False,
    )
    reg_admin_token = login(reg_admin["username"], "tenant-admin-password")["token"]
    target = user_store.create_user(
        org_id="org-default",
        username="ten-ant-target",
        password="tenant-target-password",
        name="Target",
        role="user",
        must_setup=False,
    )
    reg_role_response = client.put(
        f"/api/admin/users/{target['username']}",
        json={"role": "admin"},
        headers=_headers(reg_admin_token),
    )
    assert reg_role_response.status_code == 403, reg_role_response.get_json()
    reg_super_response = client.put(
        f"/api/admin/users/{target['username']}",
        json={"role": "super_admin"},
        headers=_headers(reg_admin_token),
    )
    assert reg_super_response.status_code == 403, reg_super_response.get_json()
    assert user_store.get_user(target["username"])["role"] == "user"


def test_admin_invites_enforce_email_uniqueness(
    client, user_store, login
):
    token = _setup_super_admin(user_store, login)
    first = client.post(
        "/api/admin/users",
        json={
            "username": "invited-email-one",
            "password": "invited-email-password-one",
            "email": "shared-invite@example.com",
            "role": "user",
        },
        headers=_headers(token),
    )
    second = client.post(
        "/api/admin/users",
        json={
            "username": "invited-email-two",
            "password": "invited-email-password-two",
            "email": "shared-invite@example.com",
            "role": "user",
        },
        headers=_headers(token),
    )

    assert first.status_code == 201, first.get_json()
    assert first.get_json()["user"]["email"] == "shared-invite@example.com"
    assert second.status_code == 400, second.get_json()
    assert len(
        [u for u in user_store.users.all() if u.get("email") == "shared-invite@example.com"]
    ) == 1


def test_admin_user_listing_uses_closed_user_allowlist(
    client, user_store, login
):
    token = _setup_super_admin(user_store, login)
    user = user_store.create_user(
        org_id="org-default",
        username="listed-user",
        password="listed-user-password",
        name="Listed User",
        role="user",
        email="listed-user@example.com",
        must_setup=False,
    )
    user_store.users.update(
        user["username"],
        password_reset_token="internal-reset-token",
        provider_secret="internal-provider-secret",
    )

    response = client.get("/api/admin/users", headers=_headers(token))

    assert response.status_code == 200, response.get_json()
    listed = next(item for item in response.get_json()["users"] if item["id"] == user["id"])
    assert listed["username"] == "listed-user"
    assert "password_hash" not in listed
    assert "password_reset_token" not in listed
    assert "provider_secret" not in listed


def test_missing_jwt_org_claim_is_rejected(client, user_store, login):
    user = user_store.create_user(
        org_id="org-default",
        username="missing-claim-user",
        password="missing-claim-password",
        name="Missing Claim User",
        role="user",
        must_setup=False,
    )
    token = login(user["username"], "missing-claim-password")["token"]
    payload = user_store.decode_token(token)
    payload.pop("org_id", None)
    token_without_org = jwt.encode(
        payload, Config.SECRET_KEY, algorithm=Config.JWT_ALGO
    )

    response = client.get(
        "/api/auth/me", headers=_headers(token_without_org)
    )

    assert response.status_code == 401, response.get_json()


def test_login_rejects_user_with_missing_organization(client, user_store):
    user_store.create_user(
        org_id="org-default",
        username="orphan-user",
        password="orphan-password",
        name="Orphan User",
        role="user",
        must_setup=False,
    )
    user_store.orgs.delete("org-default")

    response = client.post(
        "/api/auth/login",
        json={"username": "orphan-user", "password": "orphan-password"},
    )

    assert response.status_code == 401, response.get_json()


def test_terms_acceptance_requires_boolean_true(client, user_store, login):
    user = user_store.create_user(
        org_id="org-default",
        username="terms-user",
        password="terms-password",
        name="Terms User",
        role="user",
        must_setup=True,
    )
    token = login(user["username"], "terms-password")["token"]

    response = client.post(
        "/api/auth/setup",
        json={
            "email": "terms@example.com",
            "password": "terms-new-password",
            "accepted_terms": "false",
        },
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert user_store.get_user(user["username"])["must_setup"] is True


def test_setup_does_not_return_lower_layer_exception_text(
    client, user_store, login, monkeypatch
):
    user = user_store.create_user(
        org_id="org-default",
        username="setup-error-user",
        password="setup-error-password",
        name="Setup Error User",
        role="user",
        must_setup=True,
    )
    token = login(user["username"], "setup-error-password")["token"]

    def explode(*args, **kwargs):
        raise AuthError("provider-key=/private/path/secret-token")

    monkeypatch.setattr(UserStore, "complete_setup", explode)
    response = client.post(
        "/api/auth/setup",
        json={
            "email": "setup-error@example.com",
            "password": "setup-error-new-password",
            "accepted_terms": True,
        },
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "setup could not be completed"}
    assert "secret-token" not in response.get_data(as_text=True)


def test_admin_persona_redaction_is_an_explicit_allowlist():
    persona = {
        "name": "Known Persona",
        "profession": "Engineer",
        "future_hidden_signal": {"secret": "do not return"},
        "future_nested_secret": "do not return",
    }

    redacted = strip_secret_fields(persona)

    assert redacted["name"] == "Known Persona"
    assert "future_hidden_signal" not in redacted
    assert "future_nested_secret" not in redacted


def test_analysis_failure_returns_generic_error_and_safe_group_error(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    client.application.extensions["llm"] = MockLLM()
    created = client.post(
        "/api/groups",
        json={"product": "CRM", "segment": "SME"},
        headers=_headers(token),
    )
    assert created.status_code == 201, created.get_json()
    gid = created.get_json()["group"]["id"]

    def explode(*args, **kwargs):
        raise RuntimeError("provider-key=/private/path/secret-token")

    monkeypatch.setattr(Analyzer, "analyze", explode)

    response = client.post(
        f"/api/groups/{gid}/analyze", headers=_headers(token)
    )

    assert response.status_code == 500, response.get_json()
    assert response.get_json() == {"error": "analysis failed"}
    assert "secret-token" not in response.get_data(as_text=True)
    stored = client.application.extensions["group_store"].get(gid)
    assert stored["status"] == "failed"
    assert stored["error"] == "analysis_failed"


def test_existing_group_error_is_sanitized_in_api_views(client, user_store, login):
    token = _setup_super_admin(user_store, login)
    group_store = client.application.extensions["group_store"]
    group = group_store.create(
        org_id="org-default", creator_id="admin", title="Failed group"
    )
    group_store.update(
        group["id"],
        status="failed",
        error="provider-key=/private/path/legacy-secret",
    )

    response = client.get(
        f"/api/groups/{group['id']}", headers=_headers(token)
    )

    assert response.status_code == 200, response.get_json()
    view = response.get_json()["group"]
    assert view["error"] == "analysis_failed"
    assert "legacy-secret" not in response.get_data(as_text=True)


def test_trainee_group_view_drops_unknown_top_level_group_fields(
    client, user_store, login
):
    admin_token = _setup_super_admin(user_store, login)
    group_store = client.application.extensions["group_store"]
    group = group_store.create(
        org_id="org-default", creator_id="admin", title="Safe group"
    )
    group_store.groups.update(
        group["id"],
        status="ready",
        sales_kit={"productName": "Safe product"},
        report={"summary": "Safe report"},
        personas=[{"id": "safe-persona", "name": "Safe Persona"}],
        internal_secret="provider-key=/private/path/secret-token",
    )
    user = user_store.create_user(
        org_id="org-default",
        username="group-envelope-trainee",
        password="group-envelope-password",
        name="Group Envelope Trainee",
        role="user",
        must_setup=False,
    )
    token = login(user["username"], "group-envelope-password")["token"]

    response = client.get(
        f"/api/groups/{group['id']}", headers=_headers(token)
    )

    assert response.status_code == 200, response.get_json()
    assert "internal_secret" not in response.get_json()["group"]
    assert "secret-token" not in response.get_data(as_text=True)


def test_private_persona_routes_use_trainee_redaction(client, user_store, login):
    user = user_store.create_user(
        org_id="org-default",
        username="private-persona-trainee",
        password="private-persona-password",
        name="Private Persona Trainee",
        role="user",
        must_setup=False,
    )
    token = login(user["username"], "private-persona-password")["token"]
    group_store = client.application.extensions["group_store"]
    private_group = group_store.get_or_create_private_group(
        org_id="org-default", owner_user_id=user["id"], owner_name=user["name"]
    )
    group_store.set_personas(
        private_group["id"],
        [{
            "id": "private-p1",
            "name": "Private Customer",
            "profession": "Designer",
            "pains": [{"description": "hidden pain"}],
            "income": "hidden income",
            "tolerance": 1,
        }],
    )

    response = client.get(
        "/api/me/personas", headers=_headers(token)
    )

    assert response.status_code == 200, response.get_json()
    persona = response.get_json()["personas"][0]
    assert persona["name"] == "Private Customer"
    assert "pains" not in persona
    assert "income" not in persona
    assert "tolerance" not in persona


def test_upload_request_limit_is_configured(client):
    assert client.application.config["MAX_CONTENT_LENGTH"] == (
        Config.UPLOAD_MAX_MB * 1024 * 1024
    )


def test_oversized_upload_is_rejected_and_partial_file_removed(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    monkeypatch.setattr(Config, "UPLOAD_MAX_MB", 1)
    payload = {
        "product": "CRM",
        "files": (io.BytesIO(b"x" * (1024 * 1024 + 1)), "brief.txt"),
    }

    response = client.post(
        "/api/groups", data=payload, content_type="multipart/form-data", headers=_headers(token)
    )

    assert response.status_code == 413, response.get_json()
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*brief.txt"))


def test_parse_failure_returns_generic_error_and_cleans_upload(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    client.application.extensions["llm"] = MockLLM()

    def explode(path):
        raise ParseError("cannot open /private/path/provider-secret")

    monkeypatch.setattr("app.services.file_parser.parse_document", explode)
    response = client.post(
        "/api/groups",
        data={"files": (io.BytesIO(b"not a real document"), "brief.txt")},
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "could not parse uploaded file"}
    assert "provider-secret" not in response.get_data(as_text=True)
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*brief.txt"))


def test_multi_file_parse_failure_cleans_every_saved_upload(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)

    def parse(path):
        if path.name.endswith("second.txt"):
            raise ParseError("cannot open /private/path/provider-secret")
        return "safe source text"

    monkeypatch.setattr("app.services.file_parser.parse_document", parse)
    response = client.post(
        "/api/groups",
        data={
            "files": [
                (io.BytesIO(b"first"), "first.txt"),
                (io.BytesIO(b"second"), "second.txt"),
            ]
        },
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "could not parse uploaded file"}
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*first.txt"))
    assert not list(upload_dir.glob("*second.txt"))


def test_missing_product_cleans_saved_upload(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    monkeypatch.setattr("app.services.file_parser.parse_document", lambda _path: "")

    response = client.post(
        "/api/groups",
        data={"files": (io.BytesIO(b"empty source"), "empty.txt")},
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {
        "error": "provide product info in the form or via file upload"
    }
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*empty.txt"))


def test_file_save_failure_cleans_registered_destination(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)

    def explode(*args, **kwargs):
        raise OSError("provider=/private/path/save-secret")

    monkeypatch.setattr("werkzeug.datastructures.FileStorage.save", explode)
    response = client.post(
        "/api/groups",
        data={"files": (io.BytesIO(b"source"), "save-failure.txt")},
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "could not save uploaded file"}
    assert "save-secret" not in response.get_data(as_text=True)
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*save-failure.txt"))


def test_group_persistence_failure_cleans_saved_upload(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    monkeypatch.setattr("app.services.file_parser.parse_document", lambda _path: "safe source")

    def explode(*args, **kwargs):
        raise OSError("storage=/private/path/persistence-secret")

    monkeypatch.setattr(GroupStore, "create", explode)
    response = client.post(
        "/api/groups",
        data={
            "product": "CRM",
            "files": (io.BytesIO(b"source"), "persistence.txt"),
        },
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 500, response.get_json()
    assert response.get_json() == {"error": "could not create group"}
    assert "persistence-secret" not in response.get_data(as_text=True)
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*persistence.txt"))


def test_parser_import_failure_returns_generic_error_and_cleans_upload(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    monkeypatch.setitem(sys.modules, "app.services.file_parser", None)

    response = client.post(
        "/api/groups",
        data={"files": (io.BytesIO(b"plain text"), "brief.txt")},
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "could not parse uploaded file"}
    assert "ModuleNotFoundError" not in response.get_data(as_text=True)
    upload_dir = Config.DATA_DIR / "uploads"
    assert not list(upload_dir.glob("*brief.txt"))


def test_persona_update_does_not_return_lower_layer_exception_text(
    client, user_store, login, monkeypatch
):
    token = _setup_super_admin(user_store, login)
    client.application.extensions["llm"] = MockLLM()
    created = client.post(
        "/api/groups",
        json={"product": "CRM", "segment": "SME"},
        headers=_headers(token),
    )
    gid = created.get_json()["group"]["id"]
    analyzed = client.post(f"/api/groups/{gid}/analyze", headers=_headers(token))
    pid = analyzed.get_json()["personas"][0]["id"]

    def explode(*args, **kwargs):
        raise ValueError("provider-key=/private/path/secret-token")

    monkeypatch.setattr(GroupStore, "update_persona", explode)
    response = client.put(
        f"/api/groups/{gid}/personas/{pid}",
        json={"name": "Updated"},
        headers=_headers(token),
    )

    assert response.status_code == 404, response.get_json()
    assert response.get_json() == {"error": "persona not found"}
    assert "secret-token" not in response.get_data(as_text=True)


def test_chat_start_does_not_return_lower_layer_exception_text(
    client, user_store, login, monkeypatch
):
    admin_token = _setup_super_admin(user_store, login)
    client.application.extensions["llm"] = MockLLM()
    created = client.post(
        "/api/groups",
        json={"product": "CRM", "segment": "SME"},
        headers=_headers(admin_token),
    )
    gid = created.get_json()["group"]["id"]
    analyzed = client.post(
        f"/api/groups/{gid}/analyze", headers=_headers(admin_token)
    )
    pid = analyzed.get_json()["personas"][0]["id"]
    trainee = user_store.create_user(
        org_id="org-default",
        username="chat-error-trainee",
        password="chat-error-password",
        name="Chat Error Trainee",
        role="user",
        must_setup=False,
    )
    trainee_token = login(trainee["username"], "chat-error-password")["token"]

    def explode(*args, **kwargs):
        raise ValueError("provider-key=/private/path/secret-token")

    monkeypatch.setattr(SessionStore, "create", explode)
    response = client.post(
        f"/api/chat/{gid}/personas/{pid}/chat/start",
        json={"scenario": "social"},
        headers=_headers(trainee_token),
    )

    assert response.status_code == 400, response.get_json()
    assert response.get_json() == {"error": "could not start session"}
    assert "secret-token" not in response.get_data(as_text=True)


def test_request_limit_rejects_oversized_multipart_body(client, user_store, login):
    token = _setup_super_admin(user_store, login)
    client.application.config["MAX_CONTENT_LENGTH"] = 64

    response = client.post(
        "/api/groups",
        data={"files": (io.BytesIO(b"x" * 256), "brief.txt")},
        content_type="multipart/form-data",
        headers=_headers(token),
    )

    assert response.status_code == 413, response.get_json()


def test_admin_active_fields_require_literal_boolean(
    client, user_store, login
):
    token = _setup_super_admin(user_store, login)
    target = user_store.create_user(
        org_id="org-default",
        username="strict-active-target",
        password="strict-active-password",
        name="Strict Active Target",
        role="user",
        must_setup=False,
    )

    org_response = client.patch(
        "/api/admin/orgs/org-default",
        json={"active": "false"},
        headers=_headers(token),
    )
    user_response = client.put(
        f"/api/admin/users/{target['username']}",
        json={"active": "false"},
        headers=_headers(token),
    )

    assert org_response.status_code == 400, org_response.get_json()
    assert user_response.status_code == 400, user_response.get_json()
    assert user_store.get_org("org-default")["active"] is True
    assert user_store.get_user(target["username"])["active"] is True


def test_admin_org_fields_reject_coerced_types(
    client, user_store, login
):
    token = _setup_super_admin(user_store, login)
    before = dict(user_store.get_org("org-default"))

    plan_response = client.patch(
        "/api/admin/orgs/org-default",
        json={"plan": 1},
        headers=_headers(token),
    )
    seats_response = client.patch(
        "/api/admin/orgs/org-default",
        json={"seats": "10"},
        headers=_headers(token),
    )
    bool_response = client.patch(
        "/api/admin/orgs/org-default",
        json={"seats": True},
        headers=_headers(token),
    )

    assert plan_response.status_code == 400, plan_response.get_json()
    assert seats_response.status_code == 400, seats_response.get_json()
    assert bool_response.status_code == 400, bool_response.get_json()
    assert user_store.get_org("org-default") == before


def test_admin_new_org_flag_requires_literal_boolean(
    client, user_store, login
):
    token = _setup_super_admin(user_store, login)

    response = client.post(
        "/api/admin/users",
        json={
            "username": "invalid-new-org-flag",
            "password": "invalid-new-org-password",
            "role": "admin",
            "new_org": "true",
        },
        headers=_headers(token),
    )

    assert response.status_code == 400, response.get_json()
    assert user_store.get_user_or_none("invalid-new-org-flag") is None