vignette/apps/api/app/test_dataset_export.py

import importlib.util
import sys
import tempfile
import unittest
from pathlib import Path

from app.services.dataset_export import (
    APPROVED_EXPORT_STATUS,
    DRY_RUN_EXPORT_STATUS,
    ExportKeyMaps,
    build_dataset_record,
    build_manifest,
    cohen_kappa,
    intraclass_correlation,
    scan_for_pii,
    sha256_file,
    validate_manifest_gate,
    write_jsonl,
)


REPO_ROOT = Path(__file__).resolve().parents[3]
EXPORT_SCRIPT_PATH = REPO_ROOT / "scripts" / "export-recursive-dataset.py"


def load_export_script():
    spec = importlib.util.spec_from_file_location("export_recursive_dataset", EXPORT_SCRIPT_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


class DatasetExportTests(unittest.TestCase):
    def test_exporter_query_requires_consent_and_client_visible_turns(self) -> None:
        exporter = load_export_script()
        query = exporter.TURN_QUERY

        self.assertIn("u.consent_at IS NOT NULL", query)
        self.assertIn("'client' = ANY(t.visible_to)", query)
        self.assertNotRegex(query, r"\bt\.text\s*(?:,|AS\b)")
        self.assertNotIn("sc.text", query)

    def test_build_dataset_record_uses_masked_text_and_pseudonymous_keys(self) -> None:
        keys = ExportKeyMaps()
        record = build_dataset_record(
            {
                "session_id": "11111111-1111-1111-1111-111111111111",
                "learner_id": "22222222-2222-2222-2222-222222222222",
                "turn_id": "33333333-3333-3333-3333-333333333333",
                "persona_code": "P1",
                "stage": "rapport",
                "speaker": "counselor",
                "text": "raw text should never be exported",
                "text_masked": "안녕하세요, [NAME]님.",
                "feedback_scores": [{"dimension": "공감", "score": 4}],
            },
            item_index=1,
            export_manifest_id="phase3-rl-seed-test",
            keys=keys,
        )

        self.assertEqual(record["participant_key"], "PX-0001")
        self.assertEqual(record["session_key"], "SX-0001")
        self.assertEqual(record["turn_key"], "TX-000001")
        self.assertEqual(record["text_masked"], "안녕하세요, [NAME]님.")
        blob = str(record)
        self.assertNotIn("11111111-1111-1111-1111-111111111111", blob)
        self.assertNotIn("22222222-2222-2222-2222-222222222222", blob)
        self.assertNotIn("raw text should never be exported", blob)

    def test_build_dataset_record_drops_supervisor_comment_text(self) -> None:
        record = build_dataset_record(
            {
                "session_id": "11111111-1111-1111-1111-111111111111",
                "learner_id": "22222222-2222-2222-2222-222222222222",
                "persona_code": "P1",
                "stage": "rapport",
                "speaker": "client",
                "text_masked": "마스킹된 발화입니다.",
                "supervisor_comments": [
                    {
                        "kind": "note",
                        "text": "김서연 010-1234-5678 raw supervisor note",
                        "intent_deviation": {"severity": "low"},
                    }
                ],
            },
            item_index=1,
            export_manifest_id="phase3-rl-seed-test",
            keys=ExportKeyMaps(),
        )

        self.assertEqual(record["supervisor_comments"], [{"kind": "note", "intent_deviation": {"severity": "low"}}])
        self.assertNotIn("김서연", str(record))
        self.assertNotIn("010-1234-5678", str(record))

    def test_pii_scan_detects_identifiers_without_raw_samples(self) -> None:
        findings = scan_for_pii(
            {
                "text_masked": "메일 learner@hs.ac.kr, 전화 010-1234-5678, 주민 990101-1234567",
                "api_key": "sk-should-not-appear",
            }
        )

        kinds = {finding["kind"] for finding in findings}
        self.assertIn("email", kinds)
        self.assertIn("phone", kinds)
        self.assertIn("national_id", kinds)
        self.assertIn("blocked_field", kinds)
        self.assertFalse(any("learner@hs.ac.kr" in finding["sample"] for finding in findings))
        self.assertFalse(any("010-1234-5678" in finding["sample"] for finding in findings))

    def test_agreement_metrics(self) -> None:
        annotations = [
            {"item_id": "1", "labels": {"appropriateness": "good", "rapport_signal": 4.0}},
            {"item_id": "1", "labels": {"appropriateness": "good", "rapport_signal": 4.1}},
            {"item_id": "2", "labels": {"appropriateness": "bad", "rapport_signal": 2.0}},
            {"item_id": "2", "labels": {"appropriateness": "bad", "rapport_signal": 2.1}},
            {"item_id": "3", "labels": {"appropriateness": "good", "rapport_signal": 5.0}},
            {"item_id": "3", "labels": {"appropriateness": "good", "rapport_signal": 5.0}},
        ]

        self.assertEqual(cohen_kappa(annotations, "appropriateness"), 1.0)
        self.assertGreater(intraclass_correlation(annotations, "rapport_signal") or 0, 0.9)

    def test_manifest_gate_blocks_unapproved_approved_status(self) -> None:
        with self.assertRaisesRegex(ValueError, "PII scan must pass"):
            build_manifest(
                export_id="phase3-rl-seed-test",
                dataset_name="vignette_phase3_recursive_learning_seed",
                export_status=APPROVED_EXPORT_STATUS,
                purpose="test",
                records=[],
                jsonl_path="03-export/anonymized_dataset.jsonl",
                jsonl_sha256="",
                pii_findings=[{"kind": "email", "path": "$.text_masked", "sample": "<email:hs.ac.kr>"}],
                participants_included=0,
                agreement={"kappa": 0.59, "icc": 0.74, "gold_status": "not_gold"},
            )

    def test_manifest_gate_requires_withdrawal_and_recursive_consent_scope(self) -> None:
        manifest = {
            "export_status": APPROVED_EXPORT_STATUS,
            "pii_scan": {"status": "pass"},
            "agreement": {"kappa": 0.70, "icc": 0.75},
            "selection_criteria": {"include_withdrawn": True},
            "consent_scope": {"allowed_uses": ["education_quality_review"]},
            "approvals": {
                "data_steward": "steward",
                "legal_or_privacy_reviewer": "privacy",
                "technical_operator": "operator",
                "approved_at": "2026-07-01T00:00:00Z",
            },
        }

        with self.assertRaisesRegex(
            ValueError,
            "include_withdrawn must be false; recursive_learning_seed consent scope is required",
        ):
            validate_manifest_gate(manifest)

    def test_jsonl_hash_manifest_dry_run(self) -> None:
        record = {
            "schema": "phase3_dataset_item_v1",
            "item_id": "DI-000001",
            "source_refs": {"session_started_at": "2026-06-27T00:00:00Z"},
        }
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "03-export" / "anonymized_dataset.jsonl"
            write_jsonl([record], path)
            digest = sha256_file(path)
            manifest = build_manifest(
                export_id="phase3-rl-seed-test",
                dataset_name="vignette_phase3_recursive_learning_seed",
                export_status=DRY_RUN_EXPORT_STATUS,
                purpose="test",
                records=[record],
                jsonl_path="03-export/anonymized_dataset.jsonl",
                jsonl_sha256=digest,
                pii_findings=[],
                participants_included=1,
            )
        self.assertEqual(manifest["files"][0]["rows"], 1)
        self.assertEqual(manifest["files"][0]["sha256"], digest)
        self.assertEqual(manifest["pii_scan"]["status"], "pass")


if __name__ == "__main__":
    unittest.main()