import json import tempfile import unittest from pathlib import Path from src.meeting_lab.consolidation.canonicalize import ( canonicalize_extractions, find_extraction_files, load_json_object, ) EMPTY_EXTRACTION = { "facts": [], "decisions": [], "todos": [], "questions": [], "positions": [], "technical": [], } PROGEO_CONTEXT = Path("samples/real_live/progeo_meeting/meeting_context.yaml") def write_extraction(directory: Path, name: str, data: dict) -> Path: path = directory / name path.write_text(json.dumps(data), encoding="utf-8") return path class CanonicalizeTests(unittest.TestCase): def test_stable_file_ordering(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) write_extraction(root, "chunk_10_extraction.json", EMPTY_EXTRACTION) write_extraction(root, "chunk_02_extraction.json", EMPTY_EXTRACTION) write_extraction(root, "chunk_01_extraction.json", EMPTY_EXTRACTION) self.assertEqual( [path.name for path in find_extraction_files(root)], [ "chunk_01_extraction.json", "chunk_02_extraction.json", "chunk_10_extraction.json", ], ) def test_required_category_validation(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data.pop("technical") write_extraction(root, "chunk_01_extraction.json", data) with self.assertRaisesRegex(ValueError, "technical"): canonicalize_extractions(root) def test_stable_item_ids_and_string_parsing(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["facts"] = ["Ada | Revenue increased | clear | Revenue increased"] data["decisions"] = ["Ship the patch | Agreed to ship"] data["todos"] = ["Update docs | Mira | Friday | I will update docs"] data["questions"] = ["Which plan? | Which plan should we use?"] data["positions"] = ["Kai | Prefer option B | I prefer option B"] data["technical"] = ["API | Uses v2 mapping | clear | API uses v2"] write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root) items = output["items"] self.assertEqual( [item["item_id"] for item in items], [ "fact_0001", "decision_0001", "action_item_0001", "open_question_0001", "position_0001", "technical_detail_0001", ], ) fact = items[0] self.assertEqual(fact["speaker"], "Ada") self.assertEqual(fact["text"], "Revenue increased") self.assertEqual(fact["status"], "clear") todo = items[2] self.assertEqual(todo["responsible"], "Mira") self.assertEqual(todo["deadline"], "Friday") self.assertEqual(todo["evidence"], "I will update docs") def test_source_reference_preservation(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["decisions"] = ["Decision one | Evidence one"] write_extraction(root, "chunk_01_extraction.json", data) item = canonicalize_extractions(root)["items"][0] self.assertEqual(item["source_file"], "chunk_01_extraction.json") self.assertEqual(item["source_index"], 0) self.assertEqual(item["original_value"], "Decision one | Evidence one") self.assertEqual( item["source_references"], [ { "source_file": "chunk_01_extraction.json", "source_index": 0, "evidence": "Evidence one", "original_value": "Decision one | Evidence one", } ], ) def test_exact_duplicate_handling(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) first = dict(EMPTY_EXTRACTION) second = dict(EMPTY_EXTRACTION) first["decisions"] = ["Same decision | Same evidence"] second["decisions"] = ["Same decision | Same evidence"] write_extraction(root, "chunk_01_extraction.json", first) write_extraction(root, "chunk_02_extraction.json", second) output = canonicalize_extractions(root) self.assertEqual(len(output["items"]), 1) self.assertEqual(output["stats"]["exact_duplicates_merged"], 1) self.assertEqual(output["items"][0]["duplicate_count"], 2) self.assertEqual(len(output["items"][0]["source_references"]), 2) def test_no_merging_of_merely_similar_statements(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["decisions"] = [ "Ship the patch Friday | Agreed to ship Friday", "Ship the patch next week | Agreed to ship next week", ] write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root) self.assertEqual(len(output["items"]), 2) self.assertEqual(output["stats"]["exact_duplicates_merged"], 0) def test_invalid_json_handling(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "chunk_01_extraction.json").write_text("{invalid", encoding="utf-8") with self.assertRaisesRegex(ValueError, "Invalid JSON"): load_json_object(root / "chunk_01_extraction.json") def test_empty_categories(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) write_extraction(root, "chunk_01_extraction.json", EMPTY_EXTRACTION) output = canonicalize_extractions(root) self.assertEqual(output["items"], []) self.assertEqual(output["stats"]["input_item_count"], 0) self.assertEqual(output["stats"]["output_item_count"], 0) def test_responsible_aliases_normalize_with_meeting_context(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["todos"] = [ {"task": "A", "responsible": "Martin", "deadline": None, "evidence": "e"}, {"task": "B", "responsible": "Herr Tazl", "deadline": None, "evidence": "e"}, {"task": "C", "responsible": "Marlene", "deadline": None, "evidence": "e"}, {"task": "D", "responsible": "Gothard", "deadline": None, "evidence": "e"}, {"task": "E", "responsible": "Henning", "deadline": None, "evidence": "e"}, ] write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root, meeting_context_path=PROGEO_CONTEXT) responsibles = [item["responsible"] for item in output["items"]] self.assertEqual( responsibles, ["Martin", "Martin", "Marleen", "Gotthard", "Henning"], ) self.assertEqual(output["stats"]["responsibility_rejection_count"], 0) self.assertEqual(output["stats"]["responsibility_normalization_count"], 3) def test_invalid_responsible_values_are_cleared_with_validation_report(self) -> None: invalid_values = [ "31. August", "15. oder 16. September", "27.8.", "am 31.", "morgen", "nächste Woche", "Frankfurt", "Progeo", "Secugrid", "8:30 Uhr", "Unbekannte Person", ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["todos"] = [ { "task": f"Task {index}", "responsible": value, "deadline": None, "evidence": "e", } for index, value in enumerate(invalid_values, start=1) ] data["todos"].append( { "task": "Null task", "responsible": None, "deadline": None, "evidence": "e", } ) write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root, meeting_context_path=PROGEO_CONTEXT) self.assertTrue(all(item["responsible"] is None for item in output["items"])) self.assertEqual( output["stats"]["responsibility_rejection_count"], len(invalid_values), ) rejected = output["responsibility_validations"] self.assertEqual([item["original_value"] for item in rejected], invalid_values) self.assertTrue(all(item["cleared_to_null"] for item in rejected)) def test_null_responsible_remains_valid_without_warning(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["todos"] = [ {"task": "Task", "responsible": None, "deadline": None, "evidence": "e"} ] write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root, meeting_context_path=PROGEO_CONTEXT) self.assertIsNone(output["items"][0]["responsible"]) self.assertEqual(output["responsibility_validations"], []) def test_responsible_validation_does_not_run_without_meeting_context(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) data = dict(EMPTY_EXTRACTION) data["todos"] = ["Update docs | Mira | Friday | I will update docs"] write_extraction(root, "chunk_01_extraction.json", data) output = canonicalize_extractions(root) self.assertEqual(output["items"][0]["responsible"], "Mira") self.assertEqual(output["responsibility_validations"], []) if __name__ == "__main__": unittest.main()