179 lines
7.9 KiB
Python
179 lines
7.9 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
from src.meeting_lab.controlled_semantic_derivation.experiment_gold import (
|
|
RECOGNITION_SCHEMA_VERSION,
|
|
DerivationValidationError,
|
|
build_prompt,
|
|
derive_action,
|
|
evaluate_case,
|
|
load_gold_cases,
|
|
validate_recognition,
|
|
)
|
|
|
|
|
|
GOLD_PATH = Path("tests/gold/request_acceptance_v0/cases.json")
|
|
|
|
|
|
def recognition_for(case):
|
|
expected = case["expected_recognition"]
|
|
request = None
|
|
acceptance = None
|
|
if expected["request"]:
|
|
request = {
|
|
"observation_id": "obs_1",
|
|
"is_concrete_request": True,
|
|
"normalized_action_text": "Auswertung der Messwerte bis Dienstag",
|
|
}
|
|
if expected["commitment"]:
|
|
acceptance = {
|
|
"observation_id": "obs_2",
|
|
"is_explicit_commitment": True,
|
|
"same_requested_work": expected["same_work"],
|
|
"normalized_action_text": (
|
|
"Auswertung der Messwerte" if expected["same_work"] else "Präsentation"
|
|
),
|
|
}
|
|
return {
|
|
"schema_version": RECOGNITION_SCHEMA_VERSION,
|
|
"request": request,
|
|
"acceptance": acceptance,
|
|
}
|
|
|
|
|
|
class RequestAcceptanceGoldExperimentTests(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_has_exactly_required_ten_cases(self):
|
|
self.assertEqual(list(self.by_id), [f"RA-{number:02d}" for number in range(1, 11)])
|
|
|
|
def test_all_cases_use_minimal_v3_style_observations(self):
|
|
required = {"observation_id", "evidence_id", "content", "speaker", "named_person", "addressee"}
|
|
for case in self.cases:
|
|
with self.subTest(case=case["case_id"]):
|
|
self.assertGreaterEqual(len(case["observations"]), 1)
|
|
self.assertLessEqual(len(case["observations"]), 2)
|
|
self.assertTrue(all(set(observation) == required for observation in case["observations"]))
|
|
|
|
def test_positive_cases_establish_exact_action_person_due_and_provenance(self):
|
|
for case_id in ("RA-01", "RA-02"):
|
|
case = self.by_id[case_id]
|
|
gates, result = derive_action(case["observations"], recognition_for(case))
|
|
with self.subTest(case=case_id):
|
|
self.assertTrue(all(gates.values()))
|
|
self.assertEqual(result["content"], "Auswertung der Messwerte")
|
|
self.assertEqual(result["requested_actor"], "Clara")
|
|
self.assertEqual(result["responsible_person"], "Clara")
|
|
self.assertEqual(result["due"], "Dienstag")
|
|
self.assertEqual(result["support"]["request"], {"observation_id": "obs_1", "evidence_id": "e1"})
|
|
self.assertEqual(result["support"]["acceptance"], {"observation_id": "obs_2", "evidence_id": "e2"})
|
|
|
|
def test_paraphrase_does_not_require_lexical_identity(self):
|
|
case = self.by_id["RA-02"]
|
|
recognition = recognition_for(case)
|
|
recognition["acceptance"]["normalized_action_text"] = "darum kümmern und fertigstellen"
|
|
gates, result = derive_action(case["observations"], recognition)
|
|
self.assertTrue(gates["same_requested_work"])
|
|
self.assertIsNotNone(result)
|
|
|
|
def test_acknowledgement_is_unestablished(self):
|
|
self._assert_unestablished("RA-03", "acceptance_semantic_positive")
|
|
|
|
def test_tentative_response_is_unestablished(self):
|
|
self._assert_unestablished("RA-04", "acceptance_semantic_positive")
|
|
|
|
def test_different_responder_is_not_personal_acceptance(self):
|
|
self._assert_unestablished("RA-05", "acceptance_semantic_positive")
|
|
case = self.by_id["RA-05"]
|
|
recognition = recognition_for(self.by_id["RA-01"])
|
|
gates, result = derive_action(case["observations"], recognition)
|
|
self.assertFalse(gates["acceptance_speaker_matches_addressee"])
|
|
self.assertIsNone(result)
|
|
|
|
def test_different_work_fails_same_work_gate(self):
|
|
self._assert_unestablished("RA-06", "same_requested_work")
|
|
|
|
def test_request_without_response_is_unestablished(self):
|
|
self._assert_unestablished("RA-07", "acceptance_observation_exists")
|
|
|
|
def test_collective_impersonal_and_suggestion_controls_are_unestablished(self):
|
|
for case_id in ("RA-08", "RA-09", "RA-10"):
|
|
with self.subTest(case=case_id):
|
|
self._assert_unestablished(case_id, "request_semantic_positive")
|
|
|
|
def test_named_person_without_request_cannot_create_responsibility(self):
|
|
case = self.by_id["RA-10"]
|
|
self.assertEqual(case["observations"][0]["named_person"], "Dirk Textor")
|
|
_, result = derive_action(case["observations"], recognition_for(case))
|
|
self.assertIsNone(result)
|
|
|
|
def test_all_expected_recognitions_evaluate_as_pass(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_equivalent_translated_normalized_action_does_not_fail_structure(self):
|
|
case = self.by_id["RA-01"]
|
|
recognition = recognition_for(case)
|
|
recognition["request"]["normalized_action_text"] = "evaluate measurement values"
|
|
evaluation = evaluate_case(case, recognition)
|
|
self.assertEqual(evaluation["classification"], "PASS")
|
|
self.assertEqual(evaluation["result"]["content"], "evaluate measurement values")
|
|
|
|
def test_forbidden_llm_fields_are_rejected_recursively(self):
|
|
case = self.by_id["RA-01"]
|
|
for field in (
|
|
"responsible_person", "responsibility", "requested_actor", "status",
|
|
"established", "action_item", "protocol_category", "confidence", "graph",
|
|
):
|
|
recognition = recognition_for(case)
|
|
recognition["request"][field] = "forbidden"
|
|
with self.subTest(field=field), self.assertRaisesRegex(DerivationValidationError, "forbidden semantic keys"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_unknown_observation_reference_is_rejected(self):
|
|
case = self.by_id["RA-01"]
|
|
recognition = recognition_for(case)
|
|
recognition["acceptance"]["observation_id"] = "obs_99"
|
|
with self.assertRaisesRegex(DerivationValidationError, "unknown observation"):
|
|
validate_recognition(recognition, case["observations"])
|
|
|
|
def test_duplicate_evidence_provenance_is_rejected(self):
|
|
fixture = json.loads(GOLD_PATH.read_text())
|
|
fixture["cases"][0]["observations"][1]["evidence_id"] = "e1"
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "cases.json"
|
|
path.write_text(json.dumps(fixture), encoding="utf-8")
|
|
with self.assertRaisesRegex(DerivationValidationError, "must be unique"):
|
|
load_gold_cases(path)
|
|
|
|
def test_prompt_is_fixed_narrow_and_contains_observations_only(self):
|
|
prompt = build_prompt(self.by_id["RA-01"]["observations"])
|
|
self.assertIn("V3-style observations", prompt)
|
|
self.assertNotIn("expected_result", prompt)
|
|
self.assertNotIn("Who is responsible", prompt)
|
|
|
|
def test_conflicting_weekdays_fail_deadline_gate(self):
|
|
case = deepcopy(self.by_id["RA-01"])
|
|
case["observations"][1]["content"] = "Clara: Ja, ich übernehme die Auswertung bis Mittwoch."
|
|
gates, result = derive_action(case["observations"], recognition_for(case))
|
|
self.assertFalse(gates["deadline_consistent"])
|
|
self.assertIsNone(result)
|
|
|
|
def _assert_unestablished(self, case_id, failed_gate):
|
|
case = self.by_id[case_id]
|
|
gates, result = derive_action(case["observations"], recognition_for(case))
|
|
self.assertFalse(gates[failed_gate])
|
|
self.assertIsNone(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|