from __future__ import annotations import unittest from datetime import UTC, datetime, timedelta from pathlib import Path from uuid import uuid4 from pydantic import ValidationError from .contracts.calibration_transfer import ( ActualTransferExecution, CalibrationBlockInput, CalibrationTransferBenchmarkPack, IndependentPerformanceObservation, LockedSelfPredictionHistory, NormalizedEvaluatorLabels, TransferVariation, TransferSuiteInput, ) from .services.calibration_transfer import ( assess_actual_transfer_executions, assess_calibration, assess_synthetic_subgroup_drift, assess_transfer, evaluate_calibration_transfer_benchmark, load_calibration_transfer_benchmark, prescribe_metacognitive_practice, render_calibration_transfer_benchmark_report, ) BENCHMARK_PATH = ( Path(__file__).resolve().parent / "data" / "calibration_transfer_benchmark_g5.v1.json" ) def _all_keys(value: object) -> set[str]: if isinstance(value, dict): children = set().union(*(_all_keys(item) for item in value.values())) return set(value) | children if isinstance(value, (list, tuple)): return set().union(*(_all_keys(item) for item in value)) if value else set() return set() class CalibrationTransferContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_calibration_transfer_benchmark(BENCHMARK_PATH) def test_benchmark_is_version_fixed_synthetic_and_adversarial(self) -> None: self.assertEqual(self.pack.version, "1.0.0") self.assertEqual(self.pack.data_classification, "synthetic_educational") self.assertFalse(self.pack.clinical_claim_allowed) tags = {tag for case in self.pack.cases for tag in case.tags} self.assertTrue( { "post_reveal_contamination", "memorized_phrase_transfer", "calibration_improvement", "synthetic_subgroup_drift", }.issubset(tags) ) def test_prediction_revision_after_external_reveal_is_rejected(self) -> None: history = self.pack.cases[0].calibration_blocks[0].prediction_history payload = history.model_dump(mode="json") previous = payload["revisions"][-1] payload["revisions"].append( { **previous, "prediction_id": "oas-g5-prediction-a1-contaminated", "revision_no": 2, "supersedes_prediction_id": previous["prediction_id"], "recorded_sequence": payload["external_reveal_sequence"], "revision_reason": "외부평가를 본 뒤 수정하려는 오염 시도", } ) payload["locked_sequence"] = payload["external_reveal_sequence"] with self.assertRaisesRegex(ValidationError, "external evaluation"): LockedSelfPredictionHistory.model_validate(payload) def test_prediction_history_preserves_pre_reveal_revision_chain(self) -> None: history = self.pack.cases[0].calibration_blocks[0].prediction_history payload = history.model_dump(mode="json") previous = payload["revisions"][-1] payload["external_reveal_sequence"] = 4 payload["revisions"].append( { **previous, "prediction_id": "oas-g5-prediction-a1-revised", "predicted_success_probability": 0.8, "revision_no": 2, "supersedes_prediction_id": previous["prediction_id"], "recorded_sequence": 2, "revision_reason": "외부평가 전 반대근거를 반영함", } ) payload["locked_sequence"] = 3 value = LockedSelfPredictionHistory.model_validate(payload) self.assertEqual(value.locked_prediction.revision_no, 2) self.assertEqual(len(value.revisions), 2) def test_model_observation_requires_model_run_provenance(self) -> None: payload = ( self.pack.cases[0].calibration_blocks[0].observation.model_dump(mode="json") ) payload.update( source_kind="model_inferred", perspective="independent_observer", model_run_id=None, ) with self.assertRaisesRegex(ValidationError, "requires model_run_id"): IndependentPerformanceObservation.model_validate(payload) payload["model_run_id"] = str(uuid4()) value = IndependentPerformanceObservation.model_validate(payload) self.assertIsNotNone(value.model_run_id) def test_calibration_block_rejects_prediction_observation_target_mismatch( self, ) -> None: block = self.pack.cases[0].calibration_blocks[0] payload = block.model_dump(mode="json") payload["observation"]["scenario_variant_id"] = "other-scene" with self.assertRaisesRegex(ValidationError, "scenario_variant_id"): CalibrationBlockInput.model_validate(payload) def test_insufficient_observation_cannot_carry_success_evidence(self) -> None: payload = ( self.pack.cases[0].calibration_blocks[0].observation.model_dump(mode="json") ) payload.update(status="insufficient_evidence", uncertainty=1.0) with self.assertRaisesRegex(ValidationError, "evidence-free"): IndependentPerformanceObservation.model_validate(payload) class CalibrationMirrorTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_calibration_transfer_benchmark(BENCHMARK_PATH) def test_repeated_blocks_reduce_calibration_error(self) -> None: assessments = assess_calibration(self.pack.cases[0].calibration_blocks) self.assertEqual(len(assessments), 1) result = assessments[0] self.assertEqual(result.competency_id, "competency.empathic-check") self.assertEqual(result.pair_count, 6) self.assertEqual(result.improvement, "improved") self.assertIsNotNone(result.baseline_error) self.assertIsNotNone(result.recent_error) assert result.baseline_error is not None and result.recent_error is not None self.assertLess(result.recent_error, result.baseline_error) self.assertIsNotNone(result.error_interval) def test_calibration_output_has_no_total_or_overall_score(self) -> None: payload = [ item.model_dump(mode="json") for item in assess_calibration(self.pack.cases[0].calibration_blocks) ] self.assertTrue( {"total", "total_score", "overall_score", "global_score"}.isdisjoint( _all_keys(payload) ) ) def test_insufficient_calibration_remains_scoreless(self) -> None: blocks = self.pack.cases[0].calibration_blocks[:2] result = assess_calibration(blocks)[0] self.assertEqual(result.bias, "insufficient_evidence") self.assertEqual(result.improvement, "insufficient_evidence") self.assertIsNone(result.mean_absolute_error) self.assertIsNone(result.error_interval) def test_metacognitive_prescription_matches_bias_without_reward_score(self) -> None: result = assess_calibration(self.pack.cases[0].calibration_blocks[:3])[0] self.assertEqual(result.bias, "overconfident") prescription = prescribe_metacognitive_practice(result) self.assertEqual(prescription.practice_mode, "counterevidence_forecast") self.assertIn("실패", prescription.instruction_ko) self.assertTrue(prescription.completion_evidence) def test_duplicate_practice_block_is_rejected(self) -> None: block = self.pack.cases[0].calibration_blocks[0] with self.assertRaisesRegex(ValueError, "duplicate calibration practice block"): assess_calibration((block, block)) class TransferSuiteTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_calibration_transfer_benchmark(BENCHMARK_PATH) cls.true_suite = cls.pack.cases[1].transfer_suite cls.memorized_suite = cls.pack.cases[2].transfer_suite cls.drift_suite = cls.pack.cases[3].transfer_suite assert cls.true_suite and cls.memorized_suite and cls.drift_suite def test_unseen_varied_cases_verify_transfer(self) -> None: result = assess_transfer(self.true_suite)[0] self.assertTrue(result.eligible) self.assertTrue(result.transfer_verified) self.assertEqual(result.success_rate, 1.0) self.assertGreaterEqual(result.coverage["contexts"], 2) self.assertGreaterEqual(result.coverage["relationship_styles"], 2) self.assertGreaterEqual(result.coverage["difficulty_levels"], 2) self.assertGreaterEqual(result.coverage["expression_variants"], 2) self.assertGreaterEqual(result.coverage["scenario_families"], 2) def test_memorized_phrase_never_verifies_transfer(self) -> None: result = assess_transfer(self.memorized_suite)[0] self.assertFalse(result.eligible) self.assertFalse(result.transfer_verified) self.assertIn("memorized_training_phrase_reused", result.blockers) def test_undercovered_suite_does_not_verify_transfer(self) -> None: payload = self.true_suite.model_dump(mode="json") payload["trials"] = payload["trials"][:2] result = assess_transfer(TransferSuiteInput.model_validate(payload))[0] self.assertFalse(result.transfer_verified) self.assertTrue( any( item.startswith("observed_trials_below_minimum") for item in result.blockers ) ) def test_subgroup_drift_requires_minimum_samples(self) -> None: report = assess_synthetic_subgroup_drift(self.true_suite)[0] self.assertEqual(report.status, "insufficient_evidence") self.assertIsNone(report.max_rate_gap) def test_synthetic_subgroup_drift_is_flagged_without_clinical_claim(self) -> None: report = assess_synthetic_subgroup_drift(self.drift_suite)[0] self.assertEqual(report.status, "drift_flagged") self.assertAlmostEqual(report.max_rate_gap or 0.0, 2 / 3) self.assertIn("교육용 합성", report.notice_ko) self.assertIn("실제 인구집단", report.notice_ko) def test_duplicate_transfer_trial_is_rejected(self) -> None: payload = self.true_suite.model_dump(mode="json") payload["trials"].append(payload["trials"][0]) with self.assertRaisesRegex(ValidationError, "trial ids must be unique"): TransferSuiteInput.model_validate(payload) def _actual_execution( index: int, *, phrase_family_id: str, passed: bool = True, training_phrase_collision: bool = False, ) -> ActualTransferExecution: return ActualTransferExecution( execution_event_id=uuid4(), original_transfer_trial_record_id=uuid4(), practice_session_id=uuid4(), competency_id="competency.empathic-check", scenario_variant_id=f"actual-scenario-{index}", variation=TransferVariation( context_variant=f"context-{index % 2}", relationship_style=("collaborative" if index % 2 else "withdrawn"), difficulty_level=2 + (index % 2), expression_variant=f"expression-{index % 2}", synthetic_subgroup=f"synthetic-group-{index % 2}", scenario_family_id=f"scenario-family-{index % 2}", phrase_family_id=phrase_family_id, ), status="passed" if passed else "failed", uncertainty=0.25 if passed else 0.4, evidence_turn_ids=(uuid4(), uuid4()), normalized_evaluator_labels=NormalizedEvaluatorLabels( technique_codes=("reflection",), client_state_codes=("affect_contact",), appropriateness=("pos",), intent_deviation_dimensions=(), evaluator_error_count=0, ), counterevidence=() if passed else ("target_technique_not_observed",), model_run_id=uuid4(), training_phrase_collision=training_phrase_collision, created_at=datetime(2026, 8, 7, tzinfo=UTC) + timedelta(minutes=index), ) class ActualTransferExecutionTests(unittest.TestCase): def test_repeated_phrase_family_cannot_inflate_actual_transfer_coverage( self, ) -> None: executions = tuple( _actual_execution(index, phrase_family_id="same-phrase-family") for index in range(4) ) result = assess_actual_transfer_executions(executions)[0] self.assertEqual(result.evidence_source, "actual_practice_execution") self.assertEqual(result.execution_count, 4) self.assertEqual(result.independent_execution_count, 1) self.assertEqual(result.phrase_family_collision_count, 3) self.assertEqual(result.coverage["phrase_families"], 1) self.assertEqual(result.actual_transfer_status, "insufficient_evidence") self.assertTrue( any( blocker.startswith("actual_independent_phrase_families_below_minimum") for blocker in result.blockers ) ) def test_four_diverse_actual_executions_can_verify_transfer(self) -> None: executions = tuple( _actual_execution(index, phrase_family_id=f"phrase-family-{index}") for index in range(4) ) result = assess_actual_transfer_executions(executions)[0] self.assertEqual(result.independent_execution_count, 4) self.assertEqual(result.observed_execution_count, 4) self.assertEqual(result.success_rate, 1.0) self.assertTrue(result.eligible) self.assertEqual(result.actual_transfer_status, "verified") def test_training_phrase_collision_is_recorded_but_never_verified(self) -> None: executions = tuple( _actual_execution( index, phrase_family_id=f"phrase-family-{index}", training_phrase_collision=index == 0, ) for index in range(5) ) result = assess_actual_transfer_executions(executions)[0] self.assertFalse(result.eligible) self.assertEqual(result.actual_transfer_status, "insufficient_evidence") self.assertIn("training_phrase_family_reused", result.blockers) def test_eligible_actual_evidence_below_target_is_not_verified(self) -> None: executions = tuple( _actual_execution( index, phrase_family_id=f"phrase-family-{index}", passed=index != 0, ) for index in range(4) ) result = assess_actual_transfer_executions(executions)[0] self.assertTrue(result.eligible) self.assertEqual(result.success_rate, 0.75) self.assertEqual(result.actual_transfer_status, "not_verified") self.assertTrue( any( blocker.startswith("actual_transfer_success_rate_below_target") for blocker in result.blockers ) ) class CalibrationTransferBenchmarkTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_calibration_transfer_benchmark(BENCHMARK_PATH) def test_benchmark_expectations_all_match(self) -> None: report = evaluate_calibration_transfer_benchmark(self.pack) self.assertEqual(report["expectation_accuracy"], 1.0) self.assertEqual(report["post_reveal_contamination_rejections"], 1) self.assertEqual(report["memorized_phrase_false_verifications"], 0) self.assertFalse(report["clinical_claim_allowed"]) def test_report_is_deterministic_and_json_serializable(self) -> None: first = evaluate_calibration_transfer_benchmark(self.pack) second = evaluate_calibration_transfer_benchmark(self.pack) self.assertEqual(first, second) rendered = render_calibration_transfer_benchmark_report(first) self.assertIn("vignette.calibration-transfer-benchmark-report.v1", rendered) def test_pack_rejects_missing_adversarial_coverage(self) -> None: payload = self.pack.model_dump(mode="json") payload["cases"][3]["tags"] = ["coverage-removed"] with self.assertRaisesRegex(ValidationError, "adversarial coverage"): CalibrationTransferBenchmarkPack.model_validate(payload) if __name__ == "__main__": unittest.main()