from __future__ import annotations import hashlib import io import json import math import struct import tempfile import unittest import wave from contextlib import asynccontextmanager from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, patch from uuid import UUID, uuid4 from fastapi import HTTPException from pydantic import SecretStr, ValidationError from .contracts.multimodal_alliance import VoiceInteractionEvent from .deps import Principal, Role from .routes import multimodal_alliance as multimodal_routes from .routes import voice as voice_routes from .services import multimodal_alliance_store from .services.multimodal_alliance import align_voice_timeline SESSION_ID = UUID("71000000-0000-4000-8000-000000000001") LEARNER_ID = UUID("71000000-0000-4000-8000-000000000002") CONSENT_ID = UUID("71000000-0000-4000-8000-000000000003") def _principal(role: Role = Role.LEARNER) -> Principal: return Principal( user_id=str(LEARNER_ID), role=role, cohort_ids=["g7-cohort"], consent_at=1.0, profile_completed_at=1.0, ) def _synthetic_wav(duration_ms: int = 1000) -> bytes: sample_rate = 16_000 frame_count = sample_rate * duration_ms // 1000 payload = io.BytesIO() with wave.open(payload, "wb") as writer: writer.setnchannels(1) writer.setsampwidth(2) writer.setframerate(sample_rate) frames = bytearray() for index in range(frame_count): sample = int(4000 * math.sin(2 * math.pi * 220 * index / sample_rate)) frames.extend(struct.pack(" None: self.content_hash = content_hash self.execute_calls: list[tuple[str, tuple[object, ...]]] = [] async def execute(self, query: str, *args: object) -> str: self.execute_calls.append((query, args)) return "SELECT 1" async def fetchrow(self, query: str, *args: object): if "FROM app.sessions" in query: return {"id": SESSION_ID, "learner_id": LEARNER_ID} if "FROM app.multimodal_consent_snapshot" in query: return { "consent_snapshot_id": CONSENT_ID, "learner_id": LEARNER_ID, "sequence_no": 1, "consent_status": "granted", "retain_audio": True, "retain_derived_features": True, "transcript_retained": True, "retention_days": 30, "policy_version": "g7-test-v1", } if "FROM app.multimodal_ingestion_request" in query: return { "request_kind": "timeline", "content_hash": self.content_hash, "session_id": SESSION_ID, "learner_id": LEARNER_ID, "result_id": UUID("71000000-0000-4000-8000-000000000004"), } raise AssertionError(query) class MultimodalRouteSecurityTests(unittest.IsolatedAsyncioTestCase): async def test_internal_token_fails_before_db_provider(self) -> None: provider_called = False async def provider(): nonlocal provider_called provider_called = True yield object() with ( patch.object(multimodal_routes, "_evaluator_db_provider", provider), ): dependency = multimodal_routes.multimodal_internal_evaluator_db( settings=SimpleNamespace( multimodal_alliance_internal_token=SecretStr("too-short") ), presented_token="too-short" ) with self.assertRaises(HTTPException) as captured: await anext(dependency) self.assertEqual(captured.exception.status_code, 503) self.assertFalse(provider_called) async def test_independent_strong_token_enters_evaluator_db(self) -> None: sentinel = object() async def provider(): yield sentinel token = "g7-internal-token-0123456789abcdef" with ( patch.object(multimodal_routes, "_evaluator_db_provider", provider), ): dependency = multimodal_routes.multimodal_internal_evaluator_db( settings=SimpleNamespace( multimodal_alliance_internal_token=SecretStr(token) ), presented_token=token ) self.assertIs(await anext(dependency), sentinel) await dependency.aclose() async def test_teacher_cannot_read_raw_audio_asset(self) -> None: with self.assertRaises(multimodal_alliance_store.MultimodalAllianceStateError): await multimodal_alliance_store.read_raw_audio_asset( principal=_principal(Role.TEACHER), session_id=SESSION_ID, audio_asset_id=UUID("71000000-0000-4000-8000-000000000099"), ) async def test_role_safe_playback_resolves_private_file_without_exposing_handle(self) -> None: audio_asset_id = UUID("71000000-0000-4000-8000-000000000099") with tempfile.TemporaryDirectory() as temporary_directory: audio_root = Path(temporary_directory) / "multimodal-audio" audio_root.mkdir(parents=True) audio_path = audio_root / "scene.wav" audio_path.write_bytes(_synthetic_wav()) asset = { "audio_asset_id": audio_asset_id, "audio_ref": "private://scene.wav", "media_type": "audio/wav", } with patch.object( multimodal_alliance_store, "read_raw_audio_asset", AsyncMock(return_value=asset), ): response = await multimodal_routes.play_multimodal_raw_audio( session_id=SESSION_ID, audio_asset_id=audio_asset_id, settings=SimpleNamespace(user_upload_dir=temporary_directory), principal=_principal(), ) self.assertEqual(Path(response.path), audio_path) self.assertEqual(response.media_type, "audio/wav") self.assertEqual(response.headers["cache-control"], "private, no-store") self.assertNotIn("private://", str(response.headers)) def test_private_audio_ref_cannot_escape_storage_root(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: with self.assertRaises(HTTPException) as captured: multimodal_routes._resolve_private_audio_ref( SimpleNamespace(user_upload_dir=temporary_directory), "private://../outside.wav", ) self.assertEqual(captured.exception.status_code, 404) def test_raw_audio_listing_never_serializes_private_storage_handle(self) -> None: payload = multimodal_routes.RawAudioAccessResponse.model_validate( { "items": [ { "audio_asset_id": "71000000-0000-4000-8000-000000000099", "session_id": str(SESSION_ID), "learner_id": str(LEARNER_ID), "audio_ref": "private://must-not-cross-http.wav", "audio_sha256": "d" * 64, "media_type": "audio/wav", "byte_size": 32044, "duration_ms": 1000, "retained_until": "2026-09-06T00:00:00Z", "created_at": "2026-08-07T00:00:00Z", } ] } ).model_dump(mode="json") item = payload["items"][0] self.assertNotIn("audio_ref", item) self.assertNotIn("audio_sha256", item) self.assertNotIn("learner_id", item) class MultimodalTimelinePersistenceTests(unittest.IsolatedAsyncioTestCase): async def test_synthetic_wav_and_all_required_events_share_one_clock(self) -> None: audio = _synthetic_wav() with wave.open(io.BytesIO(audio), "rb") as reader: duration_ms = round(reader.getnframes() / reader.getframerate() * 1000) timeline = _timeline() request = multimodal_routes.MultimodalTimelineRequest( submission_id=uuid4(), timeline=timeline, audio_asset={ "audio_ref": "g7-synthetic://one-second.wav", "audio_sha256": hashlib.sha256(audio).hexdigest(), "media_type": "audio/wav", "byte_size": len(audio), }, ) self.assertEqual(duration_ms, request.timeline.audio_duration_ms) self.assertEqual( {event.event_type for event in request.timeline.events}, {"silence", "overlap", "interruption", "prosody"}, ) self.assertTrue(all(event.end_ms <= duration_ms for event in timeline.events)) async def test_stable_timeline_submission_replays_same_result(self) -> None: timeline = _timeline() audio_asset = { "audio_ref": "g7-synthetic://one-second.wav", "audio_sha256": "c" * 64, "media_type": "audio/wav", "byte_size": 32044, } payload = { "session_id": str(SESSION_ID), "timeline": timeline.model_dump(mode="json"), "audio_asset": audio_asset, "consent_snapshot_id": str(CONSENT_ID), } conn = _ReplayConnection( content_hash=multimodal_alliance_store._canonical_hash(payload) ) response = await multimodal_alliance_store.append_timeline( conn=conn, # type: ignore[arg-type] session_id=SESSION_ID, submission_id=uuid4(), timeline=timeline, audio_asset=audio_asset, ) self.assertTrue(response["idempotent_replay"]) self.assertFalse(any("INSERT INTO" in query for query, _ in conn.execute_calls)) lock_keys = [ args[0] for query, args in conn.execute_calls if "pg_advisory_xact_lock" in query ] self.assertEqual( lock_keys, [ f"multimodal-consent:{SESSION_ID}", f"multimodal-timeline:{SESSION_ID}", ], ) async def test_changed_timeline_reuse_is_conflict(self) -> None: conn = _ReplayConnection(content_hash="0" * 64) with self.assertRaisesRegex( multimodal_alliance_store.MultimodalAllianceConflictError, "different multimodal content", ): await multimodal_alliance_store.append_timeline( conn=conn, # type: ignore[arg-type] session_id=SESSION_ID, submission_id=uuid4(), timeline=_timeline(), audio_asset={ "audio_ref": "g7-synthetic://changed.wav", "audio_sha256": "d" * 64, "media_type": "audio/wav", "byte_size": 32044, }, ) class MultimodalPrivacyBoundaryTests(unittest.IsolatedAsyncioTestCase): async def test_non_dev_missing_consent_store_fails_closed(self) -> None: @asynccontextmanager async def unavailable_store(**_kwargs): raise RuntimeError("database unavailable") yield # pragma: no cover with ( patch.object(multimodal_alliance_store.db, "acquire", unavailable_store), patch.object(multimodal_alliance_store.settings, "environment", "prod"), ): with self.assertRaisesRegex( multimodal_alliance_store.MultimodalAllianceStateError, "voice processing is blocked", ): await multimodal_alliance_store.assert_voice_processing_allowed( principal=_principal(), session_id=str(SESSION_ID), ) async def test_dev_missing_consent_store_preserves_legacy_voice_fallback(self) -> None: @asynccontextmanager async def unavailable_store(**_kwargs): raise RuntimeError("database unavailable") yield # pragma: no cover with ( patch.object(multimodal_alliance_store.db, "acquire", unavailable_store), patch.object(multimodal_alliance_store.settings, "environment", "dev"), ): result = await multimodal_alliance_store.assert_voice_processing_allowed( principal=_principal(), session_id=str(SESSION_ID), ) self.assertIsNone(result) async def test_withdrawal_blocks_before_voice_processing(self) -> None: websocket = SimpleNamespace() sent: list[dict[str, object]] = [] async def send_json(_websocket, payload): sent.append(payload) with ( patch.object( multimodal_alliance_store, "assert_voice_processing_allowed", AsyncMock( side_effect=multimodal_alliance_store.MultimodalConsentWithdrawnError( "withdrawn" ) ), ), patch.object(voice_routes, "_safe_send_json", send_json), ): allowed = await voice_routes._multimodal_voice_processing_allowed( websocket, # type: ignore[arg-type] session_id=str(SESSION_ID), principal=_principal(), ) self.assertFalse(allowed) self.assertEqual(sent[0]["code"], "multimodal_consent_withdrawn") self.assertEqual(sent[-1], {"type": "state", "state": "idle"}) def test_emotion_certainty_is_rejected_but_observable_prosody_is_allowed( self, ) -> None: base = { "event_id": "oas-g7-event-prosody-boundary", "event_type": "prosody", "start_ms": 100, "end_ms": 300, "actor": "client", "uncertainty": 0.2, "source": "observed_audio_runtime", } with self.assertRaisesRegex(ValidationError, "clinical condition"): VoiceInteractionEvent( **base, observed_feature="내담자의 감정이 슬픔으로 확정됨", ) event = VoiceInteractionEvent( **base, observed_feature="median pitch decreased by 12Hz", ) self.assertFalse(event.clinical_claim_allowed) self.assertEqual(event.claim_scope, "interaction_signal") def test_request_json_contains_hashes_not_raw_transcript_tokens(self) -> None: body = multimodal_routes.MultimodalTimelineRequest( submission_id=uuid4(), timeline=_timeline(), audio_asset=None, ).model_dump(mode="json") serialized = json.dumps(body, ensure_ascii=False) self.assertIn("token_hash", serialized) self.assertNotIn('"token":', serialized) self.assertNotIn('"transcript":', serialized) if __name__ == "__main__": unittest.main()