sales-trainer/backend/tests/test_wave2_review_regressions.py

"""Regressions for confirmed Wave-2 Chat/session and auth review findings."""
from __future__ import annotations

import datetime

import jwt
import pytest

from app.api.chat_routes import serialize_session
from app.auth.users import AuthError
from app.config import Config
from app.services import oauth as oauth_svc
from app.services.groups import is_ready_group
from app.services.sessions import SessionStore


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


def _enable_google(monkeypatch, *, org_id: str = "org-default") -> None:
    monkeypatch.setattr(Config, "OAUTH_DEFAULT_ORG", org_id)
    monkeypatch.setattr(Config, "OAUTH_GOOGLE_CLIENT_ID", "google-client")
    monkeypatch.setattr(Config, "OAUTH_GOOGLE_CLIENT_SECRET", "google-secret")


def _oauth(client):
    return client.post("/api/auth/oauth", json={"provider": "google", "token": "verified-token"})


def test_serializer_redacts_every_customer_message_before_first_seller_turn():
    public = serialize_session(
        {
            "id": "session-legacy",
            "persona_meta": {"locale": "en"},
            "messages": [
                {"role": "customer", "text": "SECRET_OPENER_ONE"},
                {"role": "customer", "text": "SECRET_OPENER_TWO"},
                {"role": "seller", "text": "Hello"},
                {"role": "customer", "text": "Public reply"},
            ],
        }
    )

    texts = [row["text"] for row in public["messages"]]
    assert "SECRET_OPENER_ONE" not in texts
    assert "SECRET_OPENER_TWO" not in texts
    assert texts[:2] == [
        "Hi, a customer has started the conversation.",
        "Hi, a customer has started the conversation.",
    ]
    assert texts[-1] == "Public reply"


@pytest.mark.parametrize(
    "field,value",
    [("name", {"secret": "nested"}), ("profession", ["nested"])],
)
def test_ready_group_rejects_non_scalar_revealable_persona_fields(field, value):
    persona = {"id": "persona-1", "name": "Buyer", field: value}
    group = {
        "status": "ready",
        "sales_kit": {"product": "CRM"},
        "report": {"summary": "ready"},
        "personas": [persona],
    }

    assert is_ready_group(group) is False


def test_session_start_fails_closed_beside_malformed_same_scope_status(tmp_path):
    store = SessionStore(tmp_path)
    corrupt = store.create(
        org_id="org-1",
        user_id="user-1",
        group_id="group-1",
        persona_id="persona-1",
        persona_name="Buyer",
    )
    store.sessions.update(corrupt["id"], status="corrupt")

    with pytest.raises(ValueError, match="invalid persisted session state"):
        store.start(
            org_id="org-1",
            user_id="user-1",
            group_id="group-1",
            persona_id="persona-1",
            persona_name="Buyer",
        )

    assert len(store.sessions.all()) == 1


def test_public_registration_cannot_rearm_super_admin_after_store_depletion(
    client, user_store
):
    user_store.users.delete("admin")

    first = client.post(
        "/api/auth/register",
        json={
            "username": "replacement-one",
            "password": "replacement-one-password",
            "email": "replacement-one@example.com",
            "accepted_terms": True,
        },
    )
    assert first.status_code == 201, first.get_json()
    assert first.get_json()["user"]["role"] == "user"

    user_store.users.delete("replacement-one")
    second = client.post(
        "/api/auth/register",
        json={
            "username": "replacement-two",
            "password": "replacement-two-password",
            "email": "replacement-two@example.com",
            "accepted_terms": True,
        },
    )
    assert second.status_code == 201, second.get_json()
    assert second.get_json()["user"]["role"] == "user"


def test_decode_token_rejects_lifetime_longer_than_configured(user_store, monkeypatch):
    monkeypatch.setattr(Config, "JWT_EXPIRES_HOURS", 1)
    now = datetime.datetime.now(datetime.timezone.utc)
    forged = jwt.encode(
        {
            "sub": "admin",
            "org_id": "org-default",
            "role": "super_admin",
            "auth_version": 0,
            "iat": now,
            "exp": now + datetime.timedelta(hours=720),
        },
        Config.SECRET_KEY,
        algorithm=Config.JWT_ALGO,
    )

    with pytest.raises(AuthError, match="invalid or expired token"):
        user_store.decode_token(forged)


@pytest.mark.parametrize(
    "secret,bootstrap",
    [("x" * 32, "strong-bootstrap-password"), ("strong-jwt-secret-0123456789abcdef", "z" * 12)],
)
def test_production_rejects_low_entropy_repeated_secrets(monkeypatch, secret, bootstrap):
    monkeypatch.setattr(Config, "APP_ENV", "production")
    monkeypatch.setattr(Config, "FLASK_DEBUG", False)
    monkeypatch.setattr(Config, "SECRET_KEY", secret)
    monkeypatch.setattr(Config, "BOOTSTRAP_ADMIN_PASSWORD", bootstrap)

    with pytest.raises(RuntimeError):
        Config.validate_runtime_security(require_bootstrap=True)


def test_oauth_same_provider_subject_cannot_create_second_account(
    client, user_store, monkeypatch
):
    _enable_google(monkeypatch)
    monkeypatch.setattr(
        oauth_svc,
        "validate_google_token",
        lambda _token: ("first-oauth@example.com", "stable-subject", "First"),
    )
    first = _oauth(client)
    assert first.status_code == 200, first.get_json()
    first_id = first.get_json()["user"]["id"]

    monkeypatch.setattr(
        oauth_svc,
        "validate_google_token",
        lambda _token: ("changed-oauth@example.com", "stable-subject", "Changed"),
    )
    second = _oauth(client)

    assert second.status_code == 200, second.get_json()
    assert second.get_json()["user"]["id"] == first_id
    assert len([u for u in user_store.users.all() if u["id"].startswith("g_")]) == 1


def test_oauth_refuses_implicit_email_link_without_provider_binding(
    client, user_store, monkeypatch
):
    _enable_google(monkeypatch)
    user_store.create_user(
        org_id="org-default",
        username="password-user",
        password="password-user-password",
        name="Password User",
        role="user",
        email="linked@example.com",
        must_setup=False,
    )
    monkeypatch.setattr(
        oauth_svc,
        "validate_google_token",
        lambda _token: ("linked@example.com", "new-subject", "Password User"),
    )

    response = _oauth(client)

    assert response.status_code == 401


def test_oauth_does_not_create_missing_default_org(client, user_store, monkeypatch):
    _enable_google(monkeypatch, org_id="org-missing")
    monkeypatch.setattr(
        oauth_svc,
        "validate_google_token",
        lambda _token: ("missing-org@example.com", "missing-org-sub", "Missing"),
    )

    response = _oauth(client)

    assert response.status_code == 401
    assert user_store.get_org_or_none("org-missing") is None


def test_malformed_consent_state_revokes_existing_token(client, user_store, login):
    user_store.complete_setup(
        "admin",
        "admin-wave2@example.com",
        "admin-wave2-password",
        accepted_terms=True,
    )
    token = login("admin", "admin-wave2-password")["token"]
    user_store.users.update("admin", accepted_terms_at="malformed")

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

    assert response.status_code == 401


@pytest.mark.parametrize(
    "field,value",
    [("tier", {"invalid": "mapping"}), ("recontact", "false")],
)
def test_ready_group_rejects_explicit_malformed_behavior_traits(field, value):
    persona = {"id": "persona-1", "name": "Buyer", field: value}
    group = {
        "status": "ready",
        "sales_kit": {"product": "CRM"},
        "report": {"summary": "ready"},
        "personas": [persona],
    }

    assert is_ready_group(group) is False


def test_ready_group_rejects_malformed_intent_tier_alongside_valid_tier():
    group = {
        "status": "ready",
        "sales_kit": {"product": "CRM"},
        "report": {"summary": "ready"},
        "personas": [
            {
                "id": "persona-1",
                "name": "Buyer",
                "tier": "A",
                "intent_tier": {"invalid": "mapping"},
            }
        ],
    }

    assert is_ready_group(group) is False


def test_oauth_rejects_duplicate_binding_within_same_user(client, user_store, monkeypatch):
    _enable_google(monkeypatch)
    monkeypatch.setattr(
        oauth_svc,
        "validate_google_token",
        lambda _token: ("duplicate@example.com", "duplicate-sub", "Duplicate"),
    )
    first = _oauth(client)
    assert first.status_code == 200, first.get_json()
    user_id = first.get_json()["user"]["id"]
    binding = {"provider": "google", "subject": "duplicate-sub"}
    user_store.users.update(user_id, oauth_identities=[binding, dict(binding)])

    response = _oauth(client)

    assert response.status_code == 401


@pytest.mark.parametrize(
    "secret,bootstrap",
    [
        ("abcd" * 8, "strong-bootstrap-password"),
        ("strong-jwt-secret-0123456789abcdef", "abcd" * 3),
    ],
)
def test_production_rejects_low_period_repeated_secrets(
    monkeypatch, secret, bootstrap
):
    monkeypatch.setattr(Config, "APP_ENV", "production")
    monkeypatch.setattr(Config, "FLASK_DEBUG", False)
    monkeypatch.setattr(Config, "SECRET_KEY", secret)
    monkeypatch.setattr(Config, "BOOTSTRAP_ADMIN_PASSWORD", bootstrap)

    with pytest.raises(RuntimeError):
        Config.validate_runtime_security(require_bootstrap=True)


def test_unaccepted_terms_rejects_nonempty_consent_timestamp(
    client, user_store, login
):
    token = login("admin", "pytest-bootstrap-password")["token"]
    user_store.users.update("admin", accepted_terms_at="malformed")

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

    assert response.status_code == 401