76 lines
5.5 KiB
Python
76 lines
5.5 KiB
Python
import argparse, copy, json, tempfile, unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
from src.meeting_lab.controlled_semantic_derivation.experiment_h import DerivationValidationError
|
|
import src.meeting_lab.controlled_semantic_derivation.experiment_target_resolution as module
|
|
|
|
GOLD=Path("tests/gold/target_resolution_v0/cases.json")
|
|
CASES=module.load_cases(GOLD); BY_ID={c["case_id"]:c for c in CASES}
|
|
|
|
def target(case, target_id=None, text="konkrete Zielhandlung"):
|
|
return {"candidate_observation_id":case["negative_act"]["observation_id"],"target_observation_id":target_id if target_id is not None else case["expected"]["target_observation_id"],"normalized_target_text":text}
|
|
|
|
class TargetResolutionTests(unittest.TestCase):
|
|
def test_eligibility_enum_boundary(self):
|
|
for cid in ("TR-01","TR-02","TR-03","TR-04"):
|
|
self.assertTrue(module.eligibility(BY_ID[cid]["negative_act"],BY_ID[cid]["observations"])["eligible_for_target_resolution"])
|
|
for cid in ("TR-05","TR-06","TR-07","TR-08"):
|
|
gate=module.eligibility(BY_ID[cid]["negative_act"],BY_ID[cid]["observations"])
|
|
self.assertFalse(gate["eligible_for_target_resolution"]); self.assertEqual(gate["reason"],"negative_act_form_not_explicit_non_pursuit")
|
|
|
|
def test_ineligible_cases_never_call_resolver_and_record_skip(self):
|
|
fixture={"schema_version":module.SCHEMA_VERSION,"cases":[BY_ID[x] for x in ("TR-05","TR-06","TR-07","TR-08")]}; resolver=Mock()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp); cases=root/"cases.json"; cases.write_text(json.dumps(fixture)); out=root/"out"
|
|
summary=module.run(argparse.Namespace(cases=cases,output=out,endpoint="x",model="qwen3.5:9B",timeout=1,num_ctx=16384,num_predict=1024),resolver)
|
|
self.assertEqual(summary["target_resolution_llm_call_count"],0); resolver.assert_not_called()
|
|
for cid in ("tr-05","tr-06","tr-07","tr-08"):
|
|
skipped=json.loads((out/cid/"target_resolution_skipped.json").read_text()); self.assertFalse(skipped["call_made"])
|
|
|
|
def test_self_contained_target_equals_candidate(self):
|
|
c=BY_ID["TR-01"]; self.assertEqual(module.validate_target(target(c,text="Zusammenarbeit mit Dr. Schlummer fortsetzen"),c["observations"],"obs_1")["target_observation_id"],"obs_1")
|
|
|
|
def test_paired_target_precedes_candidate(self):
|
|
c=BY_ID["TR-02"]; self.assertEqual(module.validate_target(target(c,text="externe Lösung weiterverfolgen"),c["observations"],"obs_2")["target_observation_id"],"obs_1")
|
|
|
|
def test_target_after_candidate_rejected(self):
|
|
c=copy.deepcopy(BY_ID["TR-02"]); data={"candidate_observation_id":"obs_1","target_observation_id":"obs_2","normalized_target_text":"x"}
|
|
with self.assertRaises(DerivationValidationError): module.validate_target(data,c["observations"],"obs_1")
|
|
|
|
def test_unknown_candidate_and_target_rejected(self):
|
|
c=BY_ID["TR-02"]
|
|
with self.assertRaises(DerivationValidationError): module.validate_target({"candidate_observation_id":"missing","target_observation_id":"obs_1","normalized_target_text":"x"},c["observations"],"missing")
|
|
with self.assertRaises(DerivationValidationError): module.validate_target({"candidate_observation_id":"obs_2","target_observation_id":"missing","normalized_target_text":"x"},c["observations"],"obs_2")
|
|
|
|
def test_duplicate_observation_and_evidence_ids_rejected(self):
|
|
for field in ("observation_id","evidence_id"):
|
|
obs=copy.deepcopy(BY_ID["TR-02"]["observations"]); obs[1][field]=obs[0][field]
|
|
with self.assertRaises(DerivationValidationError): module.validate_observations(obs)
|
|
|
|
def test_null_and_non_null_text_constraints(self):
|
|
c=BY_ID["TR-02"]
|
|
valid={"candidate_observation_id":"obs_2","target_observation_id":None,"normalized_target_text":None}; self.assertEqual(module.validate_target(valid,c["observations"],"obs_2"),valid)
|
|
for bad in ({"candidate_observation_id":"obs_2","target_observation_id":None,"normalized_target_text":"x"},{"candidate_observation_id":"obs_2","target_observation_id":"obs_1","normalized_target_text":""}):
|
|
with self.assertRaises(DerivationValidationError): module.validate_target(bad,c["observations"],"obs_2")
|
|
|
|
def test_unknown_and_recursive_forbidden_fields_rejected(self):
|
|
c=BY_ID["TR-02"]
|
|
for extra in ({"extra":1},{"nested":{"status":"rejected"}}):
|
|
data=target(c,text="x"); data.update(extra)
|
|
with self.assertRaises(DerivationValidationError): module.validate_target(data,c["observations"],"obs_2")
|
|
|
|
def test_self_contained_prompt_fixes_linkage_deterministically(self):
|
|
prompt=module.build_prompt(BY_ID["TR-01"]); self.assertIn("deterministically fixed to obs_1",prompt); self.assertIn("only normalize",prompt)
|
|
|
|
def test_ineligible_prompt_is_impossible(self):
|
|
with self.assertRaises(DerivationValidationError): module.build_prompt(BY_ID["TR-05"])
|
|
|
|
def test_experiment_has_no_rejection_derivation(self):
|
|
self.assertFalse(hasattr(module,"derive")); self.assertNotIn("explicitly_rejected",module.TARGET_KEYS); self.assertNotIn("status",module.TARGET_KEYS)
|
|
|
|
def test_evaluation_preserves_scope_and_alternative_contract(self):
|
|
c=BY_ID["TR-04"]; gate=module.eligibility(c["negative_act"],c["observations"]); result=module.evaluate(c,gate,True,target(c,text="Versuch in der realen Anlage durchführen")); self.assertEqual(result["classification"],"PASS"); self.assertTrue(result["alternative_isolation"])
|
|
|
|
if __name__=="__main__": unittest.main()
|