sales-trainer/backend/tests/test_json_import.py

"""Tests for the fail-closed JSON-to-relational importer."""
from __future__ import annotations

import json
import hashlib
from pathlib import Path

import pytest
from sqlalchemy.orm import Session

from app.db import Base, create_db_engine
from app.models import Group, Message, Organization, Persona, TrainingSession, User
from scripts import migrate_json_to_postgres as importer
from scripts.migrate_json_to_postgres import ImportValidationError, run_import


def _write_json(root: Path, collection: str, key: str, value: dict) -> None:
    path = root / collection / f"{key}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")


def _seed_source(root: Path) -> None:
    _write_json(
        root,
        "orgs",
        "org-a",
        {
            "id": "org-a",
            "name": "Alpha",
            "plan": "trial",
            "seats": 5,
            "active": True,
            "created_at": "2026-08-15T00:00:00Z",
        },
    )
    _write_json(
        root,
        "users",
        "user-a",
        {
            "id": "user-a",
            "org_id": "org-a",
            "username": "alice",
            "email": "alice@example.com",
            "name": "Alice",
            "password_hash": "pbkdf2:sha256:source-hash",
            "role": "user",
            "active": True,
            "must_setup": False,
            "accepted_terms": True,
            "accepted_terms_at": "2026-08-15T00:00:00Z",
            "auth_version": 2,
            "created_at": "2026-08-15T00:00:00Z",
        },
    )
    _write_json(
        root,
        "groups",
        "group-a",
        {
            "id": "group-a",
            "org_id": "org-a",
            "creator_id": "user-a",
            "title": "Alpha Group",
            "status": "ready",
            "input": {"product": "A"},
            "sales_kit": {"productName": "Product A"},
            "report": {"summary": "Report A"},
            "personas": [
                {
                    "id": "persona-a",
                    "name": "Customer A",
                    "tier": "A",
                    "channel": "facebook",
                    "initiation_mode": "customer",
                    "profession": "Owner",
                    "pains": [{"name": "time"}],
                }
            ],
            "created_at": "2026-08-15T00:00:00Z",
            "updated_at": "2026-08-15T00:00:01Z",
        },
    )
    _write_json(
        root,
        "sessions",
        "session-a",
        {
            "id": "session-a",
            "org_id": "org-a",
            "user_id": "user-a",
            "group_id": "group-a",
            "persona_id": "persona-a",
            "persona_name": "Customer A",
            "persona_meta": {"tier": "A"},
            "scenario": "social",
            "locale": "th",
            "mode": "trainee",
            "status": "finished",
            "outcome": "won",
            "messages": [
                {"role": "customer", "text": "Hello", "ts": "2026-08-15T00:00:02Z"},
                {"role": "trainee", "text": "Hi", "ts": "2026-08-15T00:00:03Z"},
            ],
            "internal": {"trust": 80},
            "debrief": {"score": 90},
            "created_at": "2026-08-15T00:00:00Z",
            "updated_at": "2026-08-15T00:00:04Z",
        },
    )
    _write_json(
        root,
        "my_personas",
        "user-a__persona-private",
        {
            "key": "user-a__persona-private",
            "user_id": "user-a",
            "persona": {
                "id": "persona-private",
                "name": "Private Customer",
                "tier": "B",
                "channel": "line",
                "initiation_mode": "customer",
            },
            "created_at": "2026-08-15T00:00:05Z",
        },
    )


def _prepare_target(path: Path):
    engine = create_db_engine(f"sqlite:///{path}")
    Base.metadata.create_all(engine)
    return engine


def test_dry_run_validates_without_writing_source_or_target(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    target = tmp_path / "target.db"
    _prepare_target(target)
    before = sorted(path.relative_to(source).as_posix() for path in source.rglob("*.json"))

    report = run_import(source, f"sqlite:///{target}")

    assert report["mode"] == "dry_run"
    assert report["source_checksum"]
    assert report["counts"]["organizations"]["would_create"] == 1
    assert report["counts"]["users"]["would_create"] == 1
    assert report["counts"]["groups"]["would_create"] == 2
    assert report["counts"]["personas"]["would_create"] == 2
    assert report["counts"]["sessions"]["would_create"] == 1
    assert report["counts"]["messages"]["would_create"] == 2
    assert sorted(path.relative_to(source).as_posix() for path in source.rglob("*.json")) == before

    with Session(create_db_engine(f"sqlite:///{target}")) as db:
        assert db.query(Organization).count() == 0


def test_apply_is_duplicate_safe_and_redacts_payloads(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    target = tmp_path / "target.db"
    _prepare_target(target)

    first = run_import(
        source,
        f"sqlite:///{target}",
        apply=True,
        backup_dir=tmp_path / "backup-1",
    )
    second = run_import(
        source,
        f"sqlite:///{target}",
        apply=True,
        backup_dir=tmp_path / "backup-2",
    )

    assert first["counts"]["users"]["created"] == 1
    assert second["counts"]["users"]["created"] == 0
    assert second["counts"]["users"]["unchanged"] == 1
    assert "source-hash" not in json.dumps(first)
    assert (tmp_path / "backup-1" / "users" / "user-a.json").exists()

    with Session(create_db_engine(f"sqlite:///{target}")) as db:
        assert db.query(Organization).count() == 1
        assert db.query(User).count() == 1
        assert db.query(Group).count() == 2
        assert db.query(Persona).count() == 2
        assert db.query(TrainingSession).count() == 1
        assert db.query(Message).count() == 2


def test_cross_tenant_reference_fails_closed_before_apply(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    group_path = source / "groups" / "group-a.json"
    group = json.loads(group_path.read_text(encoding="utf-8"))
    group["org_id"] = "org-missing"
    group_path.write_text(json.dumps(group), encoding="utf-8")
    target = tmp_path / "target.db"
    _prepare_target(target)

    with pytest.raises(ImportValidationError, match="cross-tenant"):
        run_import(source, f"sqlite:///{target}")

    with Session(create_db_engine(f"sqlite:///{target}")) as db:
        assert db.query(Organization).count() == 0


def test_target_conflict_aborts_without_overwrite(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    target = tmp_path / "target.db"
    _prepare_target(target)
    run_import(source, f"sqlite:///{target}", apply=True, backup_dir=tmp_path / "backup-1")

    group_path = source / "groups" / "group-a.json"
    group = json.loads(group_path.read_text(encoding="utf-8"))
    group["title"] = "Changed after import"
    group_path.write_text(json.dumps(group), encoding="utf-8")

    with pytest.raises(ImportValidationError, match="target conflict"):
        run_import(source, f"sqlite:///{target}", apply=True, backup_dir=tmp_path / "backup-2")

    with Session(create_db_engine(f"sqlite:///{target}")) as db:
        imported = db.get(Group, "group-a")
        assert imported is not None
        assert imported.name == "Alpha Group"


def test_nonempty_audit_jsonl_is_not_silently_dropped(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    audit = source / "audit" / "audit.jsonl"
    audit.parent.mkdir(parents=True, exist_ok=True)
    audit.write_text('{"action":"login"}\n', encoding="utf-8")

    with pytest.raises(ImportValidationError, match="audit"):
        run_import(source, None)


def test_malformed_scalar_fields_fail_closed(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    user_path = source / "users" / "user-a.json"
    user = json.loads(user_path.read_text(encoding="utf-8"))
    user["active"] = "yes"
    user_path.write_text(json.dumps(user), encoding="utf-8")

    with pytest.raises(ImportValidationError, match="active"):
        run_import(source, None)


def test_source_global_byte_budget_is_enforced(tmp_path: Path, monkeypatch):
    source = tmp_path / "source"
    _seed_source(source)
    monkeypatch.setattr(importer, "MAX_SOURCE_TOTAL_BYTES", 100, raising=False)

    with pytest.raises(ImportValidationError, match="total source size"):
        run_import(source, None)


def test_source_file_budget_is_enforced_before_collection_materialization(tmp_path: Path, monkeypatch):
    source = tmp_path / "source"
    _seed_source(source)
    monkeypatch.setattr(importer, "MAX_SOURCE_FILES", 1, raising=False)

    with pytest.raises(ImportValidationError, match="source file limit"):
        run_import(source, None)


def test_source_message_budget_is_enforced(tmp_path: Path, monkeypatch):
    source = tmp_path / "source"
    _seed_source(source)
    monkeypatch.setattr(importer, "MAX_SOURCE_MESSAGES", 1, raising=False)

    with pytest.raises(ImportValidationError, match="message limit"):
        run_import(source, None)


def test_source_row_budget_is_enforced_while_queuing_personas(tmp_path: Path, monkeypatch):
    source = tmp_path / "source"
    _seed_source(source)
    monkeypatch.setattr(importer, "MAX_SOURCE_ROWS", 3, raising=False)

    with pytest.raises(ImportValidationError, match="row limit"):
        run_import(source, None)


def test_memory_error_is_converted_to_fail_closed_import_error(monkeypatch, tmp_path: Path):
    def _raise_memory_error(_source):
        raise MemoryError

    monkeypatch.setattr(importer, "load_source", _raise_memory_error)

    with pytest.raises(ImportValidationError, match="available memory"):
        run_import(tmp_path / "source", None)


def test_deep_source_json_fails_as_bounded_validation_error(tmp_path: Path):
    source = tmp_path / "source"
    path = source / "orgs" / "org-deep.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    nested = '{"id":"org-deep","name":' + '{"nested":' * 1_100 + "null" + "}" * 1_100 + "}"
    path.write_text(nested, encoding="utf-8")

    with pytest.raises(ImportValidationError, match="JSON nesting limit"):
        run_import(source, None)


def test_generated_private_group_id_collision_fails_closed(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    private_group_id = "private-" + hashlib.sha256(b"user-a").hexdigest()[:24]
    _write_json(
        source,
        "orgs",
        "org-b",
        {"id": "org-b", "name": "Beta", "plan": "trial", "seats": 5, "active": True},
    )
    _write_json(
        source,
        "users",
        "user-b",
        {
            "id": "user-b",
            "org_id": "org-b",
            "username": "bob",
            "email": "bob@example.com",
            "name": "Bob",
            "password_hash": "pbkdf2:sha256:source-hash-b",
            "role": "user",
            "active": True,
            "must_setup": False,
            "accepted_terms": True,
            "auth_version": 0,
        },
    )
    _write_json(
        source,
        "groups",
        private_group_id,
        {
            "id": private_group_id,
            "org_id": "org-b",
            "creator_id": "user-b",
            "owner_user_id": "user-b",
            "title": "Conflicting private group",
            "status": "ready",
            "personas": [],
        },
    )

    with pytest.raises(ImportValidationError, match="private group id collision"):
        run_import(source, None)


def test_multiple_private_personas_share_one_generated_group(tmp_path: Path):
    source = tmp_path / "source"
    _seed_source(source)
    _write_json(
        source,
        "my_personas",
        "user-a__persona-private-2",
        {
            "key": "user-a__persona-private-2",
            "user_id": "user-a",
            "persona": {
                "id": "persona-private-2",
                "name": "Private Customer 2",
                "tier": "C",
                "channel": "email",
                "initiation_mode": "seller",
            },
        },
    )

    report = run_import(source, None)

    assert report["counts"]["groups"]["would_create"] == 2
    assert report["counts"]["personas"]["would_create"] == 3