52 lines
4.5 KiB
Python
52 lines
4.5 KiB
Python
import argparse,copy,json,tempfile,unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
import src.meeting_lab.controlled_semantic_derivation.experiment_target_normalization as module
|
|
from src.meeting_lab.controlled_semantic_derivation.experiment_h import DerivationValidationError
|
|
|
|
CASES=module.load_cases(Path("tests/gold/target_normalization_v0/cases.json")); BY={c["case_id"]:c for c in CASES}
|
|
def output(case,text=None):
|
|
link=case["fixed_linkage"]; return {"candidate_observation_id":link["candidate_observation_id"],"target_observation_id":link["target_observation_id"],"normalized_target_text":text or case["expected"]["normalized_target_text"]}
|
|
|
|
class TargetNormalizationTests(unittest.TestCase):
|
|
def test_exact_id_copying_accepted(self):
|
|
for case in CASES: self.assertEqual(module.validate_output(output(case),case),output(case))
|
|
def test_changed_candidate_rejected(self):
|
|
case=BY["TN-02"]; data=output(case); data["candidate_observation_id"]="obs_1"
|
|
with self.assertRaises(DerivationValidationError): module.validate_output(data,case)
|
|
def test_changed_target_rejected(self):
|
|
case=BY["TN-02"]; data=output(case); data["target_observation_id"]="obs_2"
|
|
with self.assertRaises(DerivationValidationError): module.validate_output(data,case)
|
|
def test_empty_and_null_text_rejected(self):
|
|
case=BY["TN-01"]
|
|
for value in ("",None):
|
|
data=output(case); data["normalized_target_text"]=value
|
|
with self.assertRaises(DerivationValidationError): module.validate_output(data,case)
|
|
def test_unknown_and_forbidden_fields_rejected(self):
|
|
case=BY["TN-01"]
|
|
for extra in ({"extra":1},{"status":"x"},{"nested":{"decision":True}}):
|
|
data=output(case); data.update(extra)
|
|
with self.assertRaises(DerivationValidationError): module.validate_output(data,case)
|
|
def test_true_schema_fixes_both_ids_and_disallows_null(self):
|
|
case=BY["TN-02"]; schema=module.output_schema(case); self.assertEqual(schema["properties"]["candidate_observation_id"]["const"],"obs_2"); self.assertEqual(schema["properties"]["target_observation_id"]["const"],"obs_1"); self.assertEqual(schema["properties"]["normalized_target_text"]["type"],"string"); self.assertFalse(schema["additionalProperties"])
|
|
def test_payload_uses_schema_object(self):
|
|
schema=module.output_schema(BY["TN-01"]); payload=module.build_payload("qwen3.5:9B","p",schema,16384,1024); self.assertIs(payload["format"],schema); self.assertIsInstance(payload["format"],dict)
|
|
def test_duplicate_ids_and_evidence_rejected(self):
|
|
for field in ("observation_id","evidence_id"):
|
|
case=copy.deepcopy(BY["TN-02"]); case["observations"][1][field]=case["observations"][0][field]
|
|
with self.assertRaises(DerivationValidationError): module.validate_linkage(case)
|
|
def test_prompt_is_fixed_normalization_only(self):
|
|
first=module.build_prompt(BY["TN-01"]); second=module.build_prompt(BY["TN-02"]); self.assertIn("do not perform target selection",first); self.assertIn("concrete POSITIVE action",first); self.assertEqual(first.split("Fixed candidate_observation_id:")[0],second.split("Fixed candidate_observation_id:")[0])
|
|
def test_no_target_selection_or_rejection_derivation_exists(self):
|
|
self.assertFalse(hasattr(module,"select_target")); self.assertFalse(hasattr(module,"derive")); self.assertNotIn("explicitly_rejected",module.OUTPUT_KEYS); self.assertNotIn("status",module.OUTPUT_KEYS)
|
|
def test_artifacts_preserve_fixed_linkage(self):
|
|
case=BY["TN-01"]; fixture={"schema_version":module.SCHEMA_VERSION,"cases":[case]}; caller=Mock(return_value=(json.dumps(output(case)),{"model":"qwen3.5:9B"}))
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp); path=root/"cases.json"; path.write_text(json.dumps(fixture)); out=root/"out"; args=argparse.Namespace(cases=path,output=out,endpoint="x",model="qwen3.5:9B",timeout=1,num_ctx=16384,num_predict=1024); summary=module.run(args,caller); self.assertEqual(summary["llm_call_count"],1); self.assertEqual(json.loads((out/"tn-01"/"fixed_linkage.json").read_text()),case["fixed_linkage"]); self.assertTrue((out/"tn-01"/"normalized_target_result.json").exists())
|
|
def test_negative_polarity_fails_evaluation(self):
|
|
case=BY["TN-01"]; self.assertEqual(module.evaluate(case,output(case,"Mit Dr. Schlummer arbeiten wir nicht weiter"))["classification"],"FAIL")
|
|
def test_material_scope_and_alternative_contract(self):
|
|
self.assertEqual(module.evaluate(BY["TN-03"],output(BY["TN-03"]))["classification"],"PASS"); self.assertEqual(module.evaluate(BY["TN-04"],output(BY["TN-04"]))["classification"],"PASS")
|
|
|
|
if __name__=="__main__": unittest.main()
|