270 lines
10 KiB
Python
270 lines
10 KiB
Python
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"])
|
|
|
|
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()
|