| """Tests for the DMHY annotation pipeline interchange validator.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
|
|
|
|
| class DmhyAnnotationPipelineValidatorTests(unittest.TestCase): |
| def write_jsonl(self, path: Path, rows: list[dict]) -> None: |
| path.write_text( |
| "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), |
| encoding="utf-8", |
| ) |
|
|
| def unit(self, terminal_ids: list[str] | None = None) -> dict: |
| return { |
| "unit_id": "u-1", |
| "source_kind": "prefix_tree", |
| "source_id": "t-1", |
| "terminal_ids": terminal_ids or ["t-1", "t-2"], |
| "weight": 2, |
| "context": { |
| "prefixes": ["Show - 01"], |
| "digit_skeletons": ["Show - ##"], |
| "edge_labels": [" [1080p]"], |
| "notes": None, |
| }, |
| "examples": { |
| "values": ["Show - 01 [1080p].mkv"], |
| "suffixes": [" [1080p]"], |
| }, |
| "expected_output": {"schema_version": "dmhy-annotation-v1"}, |
| } |
|
|
| def patch(self, terminal_ids: list[str] | None = None, status: str = "ok") -> dict: |
| return { |
| "unit_id": "u-1", |
| "terminal_ids": terminal_ids or ["t-1", "t-2"], |
| "annotation": { |
| "episode_title_suffixes": [], |
| "media_suffixes": ["[1080p]"], |
| "title_candidates": [], |
| "llm_label": None, |
| "notes": "clean", |
| }, |
| "status": status, |
| "errors": [], |
| } |
|
|
| def run_cli(self, *args: str) -> subprocess.CompletedProcess[str]: |
| return subprocess.run( |
| [sys.executable, "-m", "tools.validate_dmhy_annotation_pipeline", *args], |
| check=False, |
| capture_output=True, |
| text=True, |
| ) |
|
|
| def test_validate_units_and_manifest(self) -> None: |
| with tempfile.TemporaryDirectory() as tmpdir: |
| tmp = Path(tmpdir) |
| units = tmp / "units.jsonl" |
| manifest = tmp / "manifest.json" |
| self.write_jsonl(units, [self.unit()]) |
|
|
| result = self.run_cli("validate-units", str(units), "--manifest-output", str(manifest)) |
|
|
| self.assertEqual(result.returncode, 0, result.stderr) |
| report = json.loads(manifest.read_text(encoding="utf-8")) |
| self.assertEqual(report["total_rows"], 1) |
| self.assertEqual(report["error_count"], 0) |
| self.assertEqual(report["terminal_coverage_count"], 2) |
|
|
| def test_validate_patches_reports_terminal_mismatch(self) -> None: |
| with tempfile.TemporaryDirectory() as tmpdir: |
| tmp = Path(tmpdir) |
| units = tmp / "units.jsonl" |
| patches = tmp / "patches.jsonl" |
| self.write_jsonl(units, [self.unit(["t-1", "t-2"])]) |
| self.write_jsonl(patches, [self.patch(["t-1"])]) |
|
|
| result = self.run_cli("validate-patches", str(patches), "--units", str(units)) |
|
|
| self.assertEqual(result.returncode, 1) |
| report = json.loads(result.stdout) |
| self.assertEqual(report["terminal_id_mismatch_count"], 1) |
| self.assertEqual(report["common_field_errors"]["terminal_ids.mismatch"], 1) |
|
|
| def test_compare_patches_counts_statuses_intersection_and_empty_rates(self) -> None: |
| with tempfile.TemporaryDirectory() as tmpdir: |
| tmp = Path(tmpdir) |
| left = tmp / "tree.jsonl" |
| right = tmp / "dag.jsonl" |
| left_rows = [ |
| self.patch(["t-1", "t-2"], "ok"), |
| { |
| **self.patch(["t-3"], "fallback"), |
| "unit_id": "u-2", |
| "annotation": { |
| "episode_title_suffixes": [], |
| "media_suffixes": [], |
| "title_candidates": [], |
| "llm_label": None, |
| "notes": "empty fallback", |
| }, |
| }, |
| ] |
| right_rows = [ |
| self.patch(["t-2", "t-4"], "failed"), |
| ] |
| right_rows[0]["errors"] = ["timeout"] |
| self.write_jsonl(left, left_rows) |
| self.write_jsonl(right, right_rows) |
|
|
| result = self.run_cli( |
| "compare-patches", |
| str(left), |
| str(right), |
| "--left-label", |
| "prefix_tree", |
| "--right-label", |
| "prefix_dag", |
| ) |
|
|
| self.assertEqual(result.returncode, 0, result.stderr) |
| report = json.loads(result.stdout) |
| self.assertEqual(report["left"]["status_counts"]["ok"], 1) |
| self.assertEqual(report["left"]["status_counts"]["fallback"], 1) |
| self.assertEqual(report["right"]["status_counts"]["failed"], 1) |
| self.assertEqual(report["terminal_intersection_count"], 1) |
| self.assertEqual(report["terminal_union_count"], 4) |
| self.assertEqual(report["left"]["annotation_empty_rates"]["all_annotation_arrays"], 0.5) |
| self.assertEqual(report["right"]["text_summary"]["errors"][0]["text"], "timeout") |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
