from __future__ import annotations import unittest from pathlib import Path from pydantic import ValidationError from .contracts.multimodal_alliance import ( AlignedVoiceTimeline, AudioRetentionRecord, FusionCalibration, ModalityAxisMeasurement, VoiceInteractionEvent, ) from .services.multimodal_alliance import ( align_voice_timeline, build_calibrated_axis_read_model, evaluate_multimodal_benchmark, load_multimodal_benchmark, ) BENCHMARK_PATH = ( Path(__file__).resolve().parent / "data" / "multimodal_alliance_benchmark_g7.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 VoiceTimelineContractTests(unittest.TestCase): def test_words_and_interaction_events_align_to_one_audio_clock(self) -> None: timeline = align_voice_timeline( audio_duration_ms=3000, words=( { "word_index": 1, "start_ms": 1200, "end_ms": 1500, "speaker": "client", "token_hash": "b" * 64, }, { "word_index": 0, "start_ms": 300, "end_ms": 700, "speaker": "learner", "token_hash": "a" * 64, }, ), events=( VoiceInteractionEvent( event_id="oas-g7-event-silence-test", event_type="silence", start_ms=700, end_ms=1200, actor="both", observed_feature="500ms turn transition silence", uncertainty=0.1, source="observed_audio_runtime", ), ), ) self.assertEqual([item.word_index for item in timeline.words], [0, 1]) self.assertEqual(timeline.events[0].start_ms, 700) def test_out_of_bounds_timestamp_is_rejected(self) -> None: with self.assertRaisesRegex(ValidationError, "exceeds audio duration"): AlignedVoiceTimeline( audio_duration_ms=1000, words=( { "word_index": 0, "start_ms": 900, "end_ms": 1100, "speaker": "learner", "token_hash": "a" * 64, }, ), events=(), ) def test_voice_event_rejects_clinical_inference(self) -> None: with self.assertRaisesRegex(ValidationError, "clinical condition"): VoiceInteractionEvent( event_id="oas-g7-event-invalid-clinical", event_type="prosody", start_ms=100, end_ms=500, actor="client", observed_feature="우울증 진단 신호", uncertainty=0.4, source="observed_audio_runtime", ) def test_voice_event_is_observational_and_nonclinical(self) -> None: event = VoiceInteractionEvent( event_id="oas-g7-event-prosody-observed", event_type="prosody", start_ms=100, end_ms=500, actor="client", observed_feature="말끝의 음량이 앞 구간보다 낮게 관찰됨", uncertainty=0.3, source="observed_audio_runtime", ) self.assertEqual(event.claim_scope, "interaction_signal") self.assertFalse(event.clinical_claim_allowed) class CalibratedFusionTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_multimodal_benchmark(BENCHMARK_PATH) def test_verified_incremental_gain_enables_axis_specific_fusion(self) -> None: case = self.pack.cases[0] result = build_calibrated_axis_read_model( text=case.text_measurement, voice=case.voice_measurement, calibration=case.calibration, ) self.assertTrue(result.fusion_applied) self.assertEqual(set(result.modalities_used), {"text", "voice"}) self.assertEqual(result.fusion_calibration_id, case.calibration.calibration_id) self.assertEqual(len(result.measurement_ids), 2) def test_no_incremental_gain_keeps_text_only(self) -> None: case = self.pack.cases[1] result = build_calibrated_axis_read_model( text=case.text_measurement, voice=case.voice_measurement, calibration=case.calibration, ) self.assertFalse(result.fusion_applied) self.assertEqual(result.modalities_used, ("text",)) self.assertEqual(result.value, case.text_measurement.value) self.assertIn("voice_incremental_gain_not_demonstrated", result.counterevidence) def test_voice_error_is_not_imputed_as_normal_score(self) -> None: case = self.pack.cases[2] result = build_calibrated_axis_read_model( text=case.text_measurement, voice=case.voice_measurement, calibration=case.calibration, ) self.assertFalse(result.fusion_applied) self.assertEqual(result.value, case.text_measurement.value) self.assertNotIn(case.voice_measurement.measurement_id, result.measurement_ids) self.assertIn("voice_measurement_not_ready", result.counterevidence) def test_fusion_rejects_cross_axis_mixing(self) -> None: text = self.pack.cases[0].text_measurement voice = self.pack.cases[1].voice_measurement calibration = self.pack.cases[0].calibration with self.assertRaisesRegex(ValueError, "one axis"): build_calibrated_axis_read_model( text=text, voice=voice, calibration=calibration ) def test_weights_must_sum_to_one(self) -> None: payload = self.pack.cases[0].calibration.model_dump(mode="json") payload.update(text_weight=0.8, voice_weight=0.4) with self.assertRaisesRegex(ValidationError, "sum to one"): FusionCalibration.model_validate(payload) def test_ready_measurement_requires_model_run_and_evidence(self) -> None: payload = self.pack.cases[0].voice_measurement.model_dump(mode="json") payload["model_run_id"] = None with self.assertRaisesRegex(ValidationError, "requires model and evidence"): ModalityAxisMeasurement.model_validate(payload) class AudioRetentionTests(unittest.TestCase): def test_granted_audio_has_hash_and_expiry(self) -> None: record = AudioRetentionRecord( session_id="session-audio", consent_status="granted", audio_ref="voice:session-audio:sha256", audio_sha256="a" * 64, retained_until_sequence=100, transcript_retained=True, ) self.assertIsNotNone(record.audio_ref) def test_withdrawn_audio_requires_deletion_and_retains_no_audio(self) -> None: record = AudioRetentionRecord( session_id="session-withdrawn", consent_status="withdrawn", deletion_event_id="delete-audio-001", transcript_retained=False, ) self.assertIsNone(record.audio_ref) payload = record.model_dump(mode="json") payload["audio_ref"] = "voice:should-not-remain" with self.assertRaisesRegex(ValidationError, "must not retain"): AudioRetentionRecord.model_validate(payload) class MultimodalBenchmarkTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_multimodal_benchmark(BENCHMARK_PATH) def test_benchmark_fusion_decisions_all_match(self) -> None: report = evaluate_multimodal_benchmark(self.pack) self.assertEqual(report["fusion_decision_accuracy"], 1.0) self.assertFalse(report["clinical_claim_allowed"]) def test_benchmark_measures_gain_against_independent_targets(self) -> None: report = evaluate_multimodal_benchmark(self.pack) gain = report["voice_gain_benchmark"] self.assertEqual(gain["metric"], "one_minus_mean_absolute_error") self.assertEqual(gain["observations"], 3) self.assertAlmostEqual(gain["text_only_accuracy"], 0.9533333333) self.assertAlmostEqual(gain["calibrated_multimodal_accuracy"], 0.9673333333) self.assertAlmostEqual(gain["measured_incremental_gain"], 0.014) self.assertTrue(gain["voice_gain_demonstrated"]) def test_benchmark_gain_changes_when_gold_targets_change(self) -> None: text_is_gold = tuple( case.model_copy(update={"target_value": case.text_measurement.value}) for case in self.pack.cases ) report = evaluate_multimodal_benchmark( self.pack.model_copy(update={"cases": text_is_gold}) ) gain = report["voice_gain_benchmark"] self.assertLess(gain["measured_incremental_gain"], 0.0) self.assertFalse(gain["voice_gain_demonstrated"]) def test_benchmark_outputs_have_no_total_score(self) -> None: report = evaluate_multimodal_benchmark(self.pack) self.assertTrue( {"total", "total_score", "overall_score", "clinical_score"}.isdisjoint( _all_keys(report) ) ) if __name__ == "__main__": unittest.main()