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