from __future__ import annotations import contextlib import csv import importlib.util import io import json import sys import tempfile import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] API_ROOT = REPO_ROOT / "apps/api" if str(API_ROOT) not in sys.path: sys.path.insert(0, str(API_ROOT)) from app.contracts.g7_external_evidence import ( # noqa: E402 G7HumanVoiceGainEvidencePack, ) from app.test_g7_voice_gain_evidence import _valid_payload # noqa: E402 from scripts.test_g7_external_proof import human_pack # noqa: E402 SCRIPT_PATH = Path(__file__).with_name("prepare-g7-human-voice-gain-intake.py") SPEC = importlib.util.spec_from_file_location( "prepare_g7_human_voice_gain_intake", 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) def _write_csv( path: Path, fields: tuple[str, ...], rows: list[dict[str, object]] ) -> None: with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n") writer.writeheader() writer.writerows(rows) def _write_intake(directory: Path, payload: dict[str, object]) -> None: reliability = payload["reliability"] assert isinstance(reliability, dict) manifest = { "intake_schema_version": MODULE.INTAKE_SCHEMA_VERSION, "template_only": False, "provenance": payload["provenance"], "text_only_model": payload["text_only_model"], "voice_enabled_model": payload["voice_enabled_model"], "power_plan": payload["power_plan"], "reported_icc": reliability["reported_icc"], "reported_categorical_kappa": reliability["reported_categorical_kappa"], "reliability_report_sha256": reliability["report_sha256"], } (directory / MODULE.MANIFEST_NAME).write_text( json.dumps(manifest), encoding="utf-8" ) participants = payload["participants"] assert isinstance(participants, list) _write_csv( directory / MODULE.PARTICIPANTS_NAME, MODULE.PARTICIPANT_FIELDS, participants, ) attestations = payload["labeler_attestations"] assert isinstance(attestations, list) labeler_rows = [] for item in attestations: assert isinstance(item, dict) labeler_rows.append( { "labeler_key": item["labeler_key"], "blinded_to_model_condition": "true", "blinded_to_other_labelers": "true", "labeled_independently": "true", "attestation_sha256": item["attestation_sha256"], } ) _write_csv( directory / MODULE.LABELERS_NAME, MODULE.LABELER_FIELDS, labeler_rows, ) observations = payload["observations"] assert isinstance(observations, list) observation_rows = [] for observation in observations: assert isinstance(observation, dict) labels = observation["labels"] assert isinstance(labels, list) for label in labels: assert isinstance(label, dict) observation_rows.append( { "observation_id": observation["observation_id"], "participant_key": observation["participant_key"], "session_key": observation["session_key"], "axis": observation["axis"], "text_only_status": observation["text_only_status"], "text_only_score": ( "" if observation.get("text_only_score") is None else observation["text_only_score"] ), "voice_enabled_status": observation["voice_enabled_status"], "voice_enabled_score": ( "" if observation.get("voice_enabled_score") is None else observation["voice_enabled_score"] ), "labeler_key": label["labeler_key"], "label_score": label["score"], "label_category": label["category"], } ) _write_csv( directory / MODULE.OBSERVATIONS_NAME, MODULE.OBSERVATION_FIELDS, observation_rows, ) class G7HumanVoiceGainIntakeTests(unittest.TestCase): def test_empty_template_is_explicitly_not_evidence(self) -> None: with tempfile.TemporaryDirectory() as root: target = Path(root) / "intake" output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = MODULE.main(["--create-template", str(target)]) self.assertEqual(0, exit_code) self.assertEqual( { MODULE.MANIFEST_NAME, MODULE.PARTICIPANTS_NAME, MODULE.LABELERS_NAME, MODULE.OBSERVATIONS_NAME, MODULE.README_NAME, }, {item.name for item in target.iterdir()}, ) manifest = json.loads( (target / MODULE.MANIFEST_NAME).read_text(encoding="utf-8") ) self.assertIs(True, manifest["template_only"]) self.assertFalse(json.loads(output.getvalue())["template_is_evidence"]) readme = (target / MODULE.README_NAME).read_text(encoding="utf-8") for required_guidance in ( "## 작성 순서", "`registered_at`은 `held_out_labels_opened_at`보다 빨라야 한다", "`observed`, `missing`, `error`", "ICC(A,1) 0.75 이상", "categorical kappa 0.70 이상", "bootstrap 10,000회", "--compile --out ", ): self.assertIn(required_guidance, readme) report, pack = MODULE.compile_intake(target) self.assertIsNone(pack) self.assertEqual("template_cannot_compile", report["errors"][0]["code"]) def test_production_intake_compiles_and_checks_reported_reliability(self) -> None: with tempfile.TemporaryDirectory() as root: directory = Path(root) / "private-human-intake" directory.mkdir() payload = human_pack() _write_intake(directory, payload) output_path = Path(root) / "compiled.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = MODULE.main( ["--compile", str(directory), "--out", str(output_path)] ) self.assertEqual(0, exit_code) report = json.loads(stdout.getvalue()) self.assertTrue(report["passed"]) self.assertTrue(report["pack_written"]) self.assertEqual(30, report["result"]["held_out_participants"]) serialized_report = json.dumps(report, ensure_ascii=False) self.assertNotIn("private-human-intake", serialized_report) self.assertNotIn("held-000", serialized_report) self.assertNotIn("labeler-001", serialized_report) compiled = json.loads(output_path.read_text(encoding="utf-8")) pack = G7HumanVoiceGainEvidencePack.model_validate(compiled) self.assertEqual(1.0, pack.reliability.reported_icc) self.assertEqual(1.0, pack.reliability.reported_categorical_kappa) self.assertTrue( all(not item.raw_audio_included for item in pack.observations) ) self.assertTrue( all(not item.transcript_included for item in pack.observations) ) def test_underpowered_intake_never_writes_a_pack(self) -> None: with tempfile.TemporaryDirectory() as root: directory = Path(root) / "intake" directory.mkdir() _write_intake(directory, _valid_payload()) output_path = Path(root) / "compiled.json" report, pack = MODULE.compile_intake(directory) self.assertIsNone(pack) self.assertFalse(report["passed"]) self.assertIn( "production_participant_floor", report["result"]["failure_reasons"], ) self.assertFalse(output_path.exists()) def test_reported_reliability_must_match_rows(self) -> None: with tempfile.TemporaryDirectory() as root: directory = Path(root) / "intake" directory.mkdir() payload = human_pack() reliability = payload["reliability"] assert isinstance(reliability, dict) reliability["reported_icc"] = 0.8 _write_intake(directory, payload) report, pack = MODULE.compile_intake(directory) self.assertIsNone(pack) self.assertFalse(report["passed"]) self.assertIn( "reported_icc_matches_rows", report["result"]["failure_reasons"], ) def test_unexpected_raw_material_column_is_rejected_without_echo(self) -> None: with tempfile.TemporaryDirectory() as root: directory = Path(root) / "person-at-example.test" directory.mkdir() _write_intake(directory, human_pack()) path = directory / MODULE.OBSERVATIONS_NAME rows = list(csv.reader(path.read_text(encoding="utf-8").splitlines())) rows[0].append("raw_transcript") rows[1].append("private words") with path.open("w", encoding="utf-8", newline="") as handle: csv.writer(handle, lineterminator="\n").writerows(rows) report, pack = MODULE.compile_intake(directory) self.assertIsNone(pack) self.assertEqual("csv_headers_invalid", report["errors"][0]["code"]) serialized = json.dumps(report, ensure_ascii=False) self.assertNotIn("private words", serialized) self.assertNotIn("person-at-example.test", serialized) def test_existing_output_is_not_overwritten(self) -> None: with tempfile.TemporaryDirectory() as root: directory = Path(root) / "intake" directory.mkdir() _write_intake(directory, human_pack()) output_path = Path(root) / "compiled.json" output_path.write_text("preserve", encoding="utf-8") stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = MODULE.main( ["--compile", str(directory), "--out", str(output_path)] ) self.assertEqual(1, exit_code) self.assertEqual("preserve", output_path.read_text(encoding="utf-8")) report = json.loads(stdout.getvalue()) self.assertEqual("output_exists", report["errors"][0]["code"]) self.assertFalse(report["pack_written"]) if __name__ == "__main__": unittest.main()