201 lines
9.1 KiB
Python
201 lines
9.1 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
from src.meeting_lab.controlled_semantic_derivation.experiment_collective import (
|
|
DerivationValidationError,
|
|
build_prompt,
|
|
derive_collective_action,
|
|
evaluate_case,
|
|
load_gold_cases,
|
|
validate_recognition,
|
|
)
|
|
|
|
|
|
GOLD_PATH = Path("tests/gold/collective_commitment_v0/cases.json")
|
|
|
|
|
|
def recognition_for(case):
|
|
form = case["expected_recognition"]["commitment_form"]
|
|
action = "20 Meter testen"
|
|
if case["case_id"] == "CC-08":
|
|
action = "20 Meter testen, aber nur im Technikum"
|
|
return {
|
|
"observation_id": "obs_1",
|
|
"commitment_form": form,
|
|
"normalized_action_text": action,
|
|
}
|
|
|
|
|
|
class CollectiveCommitmentGoldExperimentTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.cases = load_gold_cases(GOLD_PATH)
|
|
cls.by_id = {case["case_id"]: case for case in cls.cases}
|
|
|
|
def test_fixture_contains_exactly_cc_01_through_cc_10(self):
|
|
self.assertEqual(list(self.by_id), [f"CC-{number:02d}" for number in range(1, 11)])
|
|
|
|
def test_all_cases_are_single_minimal_v3_style_observations(self):
|
|
keys = {"observation_id", "evidence_id", "content", "speaker", "named_person", "addressee"}
|
|
for case in self.cases:
|
|
with self.subTest(case=case["case_id"]):
|
|
self.assertEqual(len(case["observations"]), 1)
|
|
self.assertEqual(set(case["observations"][0]), keys)
|
|
|
|
def test_cc_01_establishes_collective_action_without_person_and_with_due(self):
|
|
case = self.by_id["CC-01"]
|
|
gates, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertTrue(all(gates.values()))
|
|
self.assertEqual(result["status"], "established")
|
|
self.assertEqual(result["commitment_scope"], "collective")
|
|
self.assertIsNone(result["responsible_person"])
|
|
self.assertEqual(result["due"], "nächste Woche")
|
|
|
|
def test_individual_commitment_routes_out_of_collective_path(self):
|
|
self._assert_unestablished("CC-02", "collective_commitment_form")
|
|
|
|
def test_tentative_suggestion_impersonal_and_passive_remain_unestablished(self):
|
|
for case_id in ("CC-03", "CC-04", "CC-05", "CC-06"):
|
|
with self.subTest(case=case_id):
|
|
self._assert_unestablished(case_id, "collective_commitment_form")
|
|
|
|
def test_rejection_remains_unestablished_and_negation_gate_is_negative(self):
|
|
case = self.by_id["CC-07"]
|
|
gates, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertFalse(gates["collective_commitment_form"])
|
|
self.assertFalse(gates["no_explicit_negation"])
|
|
self.assertIsNone(result)
|
|
|
|
def test_qualifier_case_establishes_preserves_limit_and_has_no_due(self):
|
|
case = self.by_id["CC-08"]
|
|
_, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertIsNotNone(result)
|
|
self.assertIn("nur im Technikum", result["content"])
|
|
self.assertIsNone(result["due"])
|
|
self.assertIsNone(result["responsible_person"])
|
|
|
|
def test_collective_without_deadline_establishes_with_null_due(self):
|
|
case = self.by_id["CC-09"]
|
|
_, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertIsNotNone(result)
|
|
self.assertIsNone(result["due"])
|
|
|
|
def test_speaker_ownership_trap_never_assigns_martin(self):
|
|
case = self.by_id["CC-10"]
|
|
_, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(case["observations"][0]["speaker"], "Martin")
|
|
self.assertIsNone(result["responsible_person"])
|
|
|
|
def test_changing_only_speaker_cannot_create_individual_owner(self):
|
|
case = deepcopy(self.by_id["CC-01"])
|
|
for speaker in ("Martin", "Clara", "Antonius"):
|
|
case["observations"][0]["speaker"] = speaker
|
|
_, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
with self.subTest(speaker=speaker):
|
|
self.assertIsNotNone(result)
|
|
self.assertIsNone(result["responsible_person"])
|
|
|
|
def test_none_and_individual_forms_never_establish(self):
|
|
case = self.by_id["CC-01"]
|
|
for form in ("none", "individual_first_person"):
|
|
recognition = recognition_for(case)
|
|
recognition["commitment_form"] = form
|
|
_, result = derive_collective_action(case["observations"], recognition)
|
|
with self.subTest(form=form):
|
|
self.assertIsNone(result)
|
|
|
|
def test_non_none_commitment_requires_action_text(self):
|
|
case = self.by_id["CC-01"]
|
|
recognition = recognition_for(case)
|
|
recognition["normalized_action_text"] = None
|
|
with self.assertRaisesRegex(DerivationValidationError, "requires normalized_action_text"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_none_commitment_allows_null_action_text_but_never_establishes(self):
|
|
case = self.by_id["CC-03"]
|
|
recognition = recognition_for(case)
|
|
recognition["normalized_action_text"] = None
|
|
_, result = derive_collective_action(case["observations"], recognition)
|
|
self.assertIsNone(result)
|
|
|
|
def test_unknown_observation_id_is_rejected(self):
|
|
case = self.by_id["CC-01"]
|
|
recognition = recognition_for(case)
|
|
recognition["observation_id"] = "obs_99"
|
|
with self.assertRaisesRegex(DerivationValidationError, "unknown observation"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_inconsistent_duplicate_evidence_provenance_is_rejected(self):
|
|
fixture = json.loads(GOLD_PATH.read_text())
|
|
fixture["cases"][0]["observations"].append({
|
|
"observation_id": "obs_2", "evidence_id": "e1", "content": "Martin: Zusatz.",
|
|
"speaker": "Martin", "named_person": None, "addressee": None,
|
|
})
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "cases.json"
|
|
path.write_text(json.dumps(fixture), encoding="utf-8")
|
|
with self.assertRaisesRegex(DerivationValidationError, "provenance must be unique"):
|
|
load_gold_cases(path)
|
|
|
|
def test_conflicting_deadlines_prevent_establishment(self):
|
|
case = deepcopy(self.by_id["CC-01"])
|
|
case["observations"][0]["content"] += " Bis Mittwoch."
|
|
gates, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertFalse(gates["deadline_supported_and_consistent"])
|
|
self.assertIsNone(result)
|
|
|
|
def test_forbidden_fields_are_rejected_recursively(self):
|
|
case = self.by_id["CC-01"]
|
|
forbidden = (
|
|
"responsible_person", "responsibility", "responsibility_scope",
|
|
"requested_actor", "owner", "ownership", "assignee", "status",
|
|
"established", "action_item", "protocol_category", "decision",
|
|
"unresolved_issue", "confidence", "relation", "relations", "graph",
|
|
)
|
|
for field in forbidden:
|
|
recognition = recognition_for(case)
|
|
recognition["wrapper"] = {field: "forbidden"}
|
|
with self.subTest(field=field), self.assertRaisesRegex(DerivationValidationError, "forbidden semantic keys"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_unknown_schema_field_is_rejected(self):
|
|
case = self.by_id["CC-01"]
|
|
recognition = recognition_for(case)
|
|
recognition["explanation"] = "extra"
|
|
with self.assertRaisesRegex(DerivationValidationError, "unknown keys"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_provenance_survives_and_successes_always_have_null_person(self):
|
|
for case_id in ("CC-01", "CC-08", "CC-09", "CC-10"):
|
|
case = self.by_id[case_id]
|
|
_, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
with self.subTest(case=case_id):
|
|
self.assertEqual(result["support"]["commitment"], {"observation_id": "obs_1", "evidence_id": "e1"})
|
|
self.assertIsNone(result["responsible_person"])
|
|
|
|
def test_all_expected_recognitions_have_correct_final_outcome(self):
|
|
for case in self.cases:
|
|
evaluation = evaluate_case(case, recognition_for(case))
|
|
with self.subTest(case=case["case_id"]):
|
|
self.assertEqual(evaluation["classification"], "PASS")
|
|
|
|
def test_prompt_is_fixed_narrow_and_contains_no_gold_expectation(self):
|
|
prompt = build_prompt(self.by_id["CC-01"]["observations"])
|
|
self.assertIn("commitment_form", prompt)
|
|
self.assertNotIn("expected_result", prompt)
|
|
self.assertNotIn("Who is responsible", prompt)
|
|
|
|
def _assert_unestablished(self, case_id, failed_gate):
|
|
case = self.by_id[case_id]
|
|
gates, result = derive_collective_action(case["observations"], recognition_for(case))
|
|
self.assertFalse(gates[failed_gate])
|
|
self.assertIsNone(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|