meeting-lab/tests/test_extraction_protocol.py

import json
import tempfile
import unittest
from pathlib import Path

from src.meeting_lab.extraction.extract_chunks import (
    EXTRACTION_CATEGORIES,
    EXTRACTION_TASK_PROMPT_NAMES,
    build_prompt,
    extraction_path_for_chunk,
    normalize_current_schema,
    parse_json_response,
)
from src.meeting_lab.models.meeting_context import load_meeting_context
from src.meeting_lab.protocol.build_protocol import build_protocol
from scripts import run_gold_test


class ExtractionProtocolTests(unittest.TestCase):
    def test_extraction_path_drops_normalized_suffix(self) -> None:
        path = Path("chunks/chunk_01_normalized.txt")

        self.assertEqual(
            extraction_path_for_chunk(path),
            Path("chunks/chunk_01_extraction.json"),
        )

    def test_normalize_current_schema_maps_legacy_response(self) -> None:
        result = normalize_current_schema(
            {
                "facts": [
                    {
                        "speaker": "A",
                        "statement": "Fact one",
                        "status": "clear",
                        "evidence": "Fact",
                    }
                ],
                "decisions": [
                    {
                        "decision": "Decision one",
                        "evidence": "Decision",
                    }
                ],
                "todos": [
                    {
                        "task": "Todo one",
                        "responsible": "B",
                        "deadline": None,
                        "evidence": "Todo",
                    }
                ],
                "open_questions": [
                    {
                        "question": "Question one",
                        "evidence": "Question",
                    }
                ],
                "technical_details": [
                    {
                        "subject": "System",
                        "statement": "Technical one",
                        "status": "clear",
                        "evidence": "Technical",
                    }
                ],
            }
        )

        self.assertEqual(set(result), set(EXTRACTION_CATEGORIES))
        self.assertEqual(result["positions"], [])
        self.assertIn("Fact one", result["facts"][0])
        self.assertIn("Decision one", result["decisions"][0])
        self.assertIn("Todo one", result["todos"][0])
        self.assertIn("Question one", result["questions"][0])
        self.assertIn("Technical one", result["technical"][0])

    def test_parse_json_response_uses_final_object_after_thinking(self) -> None:
        text = """
Thinking:
I will reason about the transcript first.
{"facts": ["draft"], "decisions": []}

Final answer:
{
  "facts": ["final"],
  "decisions": [],
  "todos": [],
  "questions": [],
  "positions": [],
  "technical": []
}
"""

        parsed = parse_json_response(text)

        self.assertEqual(parsed["facts"], ["final"])
        self.assertEqual(set(parsed), set(EXTRACTION_CATEGORIES))

    def test_build_prompt_includes_common_and_decision_prompt_files(self) -> None:
        prompt = build_prompt("transcript.txt", "Anna: Agreed.")

        self.assertIn("Du extrahierst Informationen aus Meeting-Transkripten.", prompt)
        self.assertIn("Classification contract for Decisions", prompt)
        self.assertIn("A Decision requires evidence", prompt)
        self.assertIn("Anna: Agreed.", prompt)

    def test_build_prompt_includes_todo_prompt_file(self) -> None:
        prompt = build_prompt("transcript.txt", "Nina: I will update it.")

        self.assertEqual(
            EXTRACTION_TASK_PROMPT_NAMES,
            ("classification.md",),
        )
        self.assertIn("Action Item:", prompt)
        self.assertIn(
            "person may be named only when explicitly assigned",
            prompt,
        )

    def test_gold_runner_uses_shared_production_prompt_assembly(self) -> None:
        self.assertIs(run_gold_test.build_prompt, build_prompt)

    def test_no_context_and_meeting_context_prompts_use_same_task_prompt_set(self) -> None:
        context = load_meeting_context(
            Path("samples/real_live/project_process_meeting/meeting_context.yaml")
        )

        no_context_prompt = build_prompt("chunk_01_normalized.txt", "Anna: Agreed.")
        context_prompt = build_prompt(
            "chunk_01_normalized.txt",
            "Anna: Agreed.",
            meeting_context=context,
        )

        for prompt in (no_context_prompt, context_prompt):
            self.assertIn("Classification contract for Decisions", prompt)
            self.assertIn("Action Item:", prompt)
            self.assertIn("Open Question:", prompt)

    def test_prompt_assembly_is_deterministic(self) -> None:
        first = build_prompt("transcript.txt", "Nina: I will update it.")
        second = build_prompt("transcript.txt", "Nina: I will update it.")

        self.assertEqual(first, second)

    def test_classification_contracts_follow_transcript(self) -> None:
        prompt = build_prompt("transcript.txt", "SOURCE_SENTINEL")

        transcript_end = prompt.index("--- ENDE TRANSKRIPT ---")
        self.assertGreater(prompt.index("Classification contract", transcript_end), transcript_end)
        self.assertGreater(
            prompt.index("Action Item:", transcript_end),
            transcript_end,
        )
        self.assertGreater(
            prompt.index("Open Question:", transcript_end),
            transcript_end,
        )

    def test_build_protocol_groups_extraction_items(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            input_dir = Path(directory)
            (input_dir / "chunk_01_extraction.json").write_text(
                json.dumps(
                    {
                        "facts": ["Fact one"],
                        "decisions": ["Decision one"],
                        "todos": ["Todo one"],
                        "questions": ["Question one"],
                        "positions": [],
                        "technical": [],
                    }
                ),
                encoding="utf-8",
            )

            output_path = build_protocol(input_dir)

            markdown = output_path.read_text(encoding="utf-8")
            self.assertIn("# Meeting Protocol", markdown)
            self.assertIn("## Facts\n- Fact one", markdown)
            self.assertIn("## Decisions\n- Decision one", markdown)
            self.assertIn("## Open Questions\n- Question one", markdown)
            self.assertIn("## Action Items\n- Todo one", markdown)


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