meeting-lab/tests/test_canonicalize.py

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": [],
}


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)


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