from __future__ import annotations import logging import sys import threading import time from pathlib import Path from types import MethodType, SimpleNamespace from typing import Any import pytest PROJECT_ROOT = Path(__file__).resolve().parents[1] SOURCE_ROOT = PROJECT_ROOT / "src" if str(SOURCE_ROOT) not in sys.path: sys.path.insert(0, str(SOURCE_ROOT)) from doctor_workstation.video.launcher import ( # noqa: E402 BackendMode, VideoCallLauncher, VideoCallRequest, VideoTicketError, normalize_backend_ticket, ) from doctor_workstation.video.lifecycle import OrderedCallLifecycle # noqa: E402 from doctor_workstation.video.security import ( # noqa: E402 TrustedDocumentError, TrustedDocumentPolicy, ) def test_companion_uses_legacy_safe_transcription_session_identity() -> None: """Generated session IDs stay below 32 chars so upgraded databases cannot truncate.""" source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text( encoding="utf-8" ) function_source = source.split("function newTranscriptionSessionId", 1)[1].split( "function requestTranscriptionStart", 1 )[0] assert "replaceAll('-', '')" in function_source assert ".slice(0, 28)" in function_source assert "return `tr-${" in function_source def test_companion_archives_cloud_video_local_mixed_audio_and_transcript() -> None: """A connected call starts three independent artifacts before hangup.""" source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text( encoding="utf-8" ) assert "context.createMediaStreamDestination()" in source assert "cloud.getAudioTrack({ processed: true })" in source assert "attachPatientAudioTrack(cloud, activeConfig.targetUserId)" in source assert "new MediaRecorder(destination.stream" in source assert "recorder.start(1000)" in source assert "bridge.startLocalAudioRecording(sessionId, mimeType)" in source assert "bridge.appendLocalAudioChunk(" in source assert "bridge.finishLocalAudioRecording(sessionId, totalBytes)" in source assert "operations.push(stopLocalRecording())" in source assert "operations.push(stopTranscription('completed'))" in source assert "Promise.allSettled(operations)" in source assert "腾讯云混流视频、本机语音录音和实时转写均已启动" in source def test_companion_watches_room_id_for_the_entire_call_cycle() -> None: """A slowly-created TRTC room must still bind to the exact call record.""" source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text( encoding="utf-8" ) room_source = source.split("function readRoomId", 1)[1].split( "function handleStatusChanged", 1 )[0] assert "TUIStore.watch(StoreName.CALL, roomIdWatchOptions)" in room_source assert "[NAME.ROOM_ID]: handleRoomIdChanged" in room_source assert "cycle === callCycleGeneration" in room_source assert "while (activeConfig && !endNotified" in room_source assert "attempt < 40" not in room_source assert "diagnosisId: activeConfig.diagnosisId" in room_source def test_room_binding_is_acknowledged_and_transcriber_room_is_a_fallback() -> None: companion_source = ( PROJECT_ROOT / "video_companion" / "src" / "main.ts" ).read_text(encoding="utf-8") window_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py" ).read_text(encoding="utf-8") assert "observeRoomId(roomId)" in companion_source assert "onRealtimeTranscriberStarted: (roomId)" in companion_source assert "function roomBindingResult(" in companion_source assert "if (boundRoomId) return" in companion_source assert "roomBindingResult?.(" in window_source def test_im_conversation_renders_deduplicated_video_call_status_timeline() -> None: """Video lifecycle feedback belongs in the IM timeline as local status events.""" companion_root = PROJECT_ROOT / "video_companion" / "src" source = (companion_root / "main.ts").read_text(encoding="utf-8") app_source = (companion_root / "App.vue").read_text(encoding="utf-8") styles = (companion_root / "style.css").read_text(encoding="utf-8") timeline_source = source.split("function appendVideoCallStatus", 1)[1].split( "function onMessageReceived", 1 )[0] assert "activeConfig.mode !== 'chat'" in timeline_source assert "local-video-call-${callCycleGeneration}-${callStatus}" in timeline_source assert "findIndex((item) => item.id === id)" in timeline_source assert "appendVideoCallStatus('starting', '正在创建安全视频通话')" in source assert "appendVideoCallStatus('dialing', '正在呼叫患者')" in source assert "appendVideoCallStatus('connected', '视频通话已接通')" in source assert "appendVideoCallStatus('ended', '视频通话已结束')" in source assert "appendVideoCallStatus('failed', `视频通话发起失败：${message}`)" in source assert "message.type === 'call-status'" in app_source assert 'class="call-status-event"' in app_source assert 'role="status"' in app_source assert "IM 已连接 · ${props.statusText.value}" in app_source assert ".message-row--call-status" in styles assert ".call-status-event--connected" in styles assert ".call-status-event--failed" in styles def test_companion_local_recording_waits_for_real_audio_and_has_runtime_fallbacks() -> None: source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text( encoding="utf-8" ) assert "querySelectorAll('video, audio')" in source assert "stream.getAudioTracks()" in source assert "navigator.mediaDevices.getUserMedia" in source assert "await waitForCallAudioTracks(cloud, sessionId)" in source assert "if (!attached)" in source assert "cloud.getAudioTrack(userId)" in source assert "event.sourceTrack" in source assert "cloud.on('remote-audio-available'" in source assert "localAudioCloud.off('remote-audio-available'" in source assert "!event.userId || event.userId === activeConfig?.userID" in source assert "localRecordingAttachedSourceCount <= 0" in source assert "localRecordingBytes < 1024" in source assert "已阻止上传空文件" in source def test_qt_close_waits_for_local_audio_finish_before_destroying_webengine() -> None: """A title-bar/desktop hangup must keep accepting bridge chunks until COS ack.""" source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py" ).read_text(encoding="utf-8") request_shutdown = source.split( "def _request_companion_shutdown", 1 )[1].split("def _force_requested_shutdown", 1)[0] begin_shutdown = source.split("def _begin_shutdown", 1)[1].split( "def wait_for_lifecycles", 1 )[0] close_event = source.split("def closeEvent", 1)[1].split( "else:", 1 )[0] assert "window.doctorConsultation?.close?.()" in request_shutdown assert "self._shutdown_requested = True" in request_shutdown assert "self._closing = True" not in request_shutdown assert "event.ignore()" in close_event assert "self._request_companion_shutdown" in close_event assert "self._shutdown_timer.stop()" in begin_shutdown assert "window.doctorConsultation?.close?.()" not in begin_shutdown assert "if not self.open_im or self._shutdown_requested" in source def test_local_audio_capture_keeps_and_persists_its_own_call_room_identity() -> None: """A later IM call cycle must not relabel an earlier recording.""" source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py" ).read_text(encoding="utf-8") assert "lifecycle: OrderedCallLifecycle" in source assert "lifecycle = capture.lifecycle" in source assert "call_record_id=call_record_id" in source assert 'room_id=lifecycle.current_room_id or ""' in source assert "store.bind_identity(" in source def test_companion_shows_incremental_subtitles_but_only_persists_final_segments() -> None: main_source = ( PROJECT_ROOT / "video_companion" / "src" / "main.ts" ).read_text(encoding="utf-8") component_source = ( PROJECT_ROOT / "video_companion" / "src" / "App.vue" ).read_text(encoding="utf-8") handler = main_source.split("function handleTranscriberMessage", 1)[1].split( "function subscribeTranscriber", 1 )[0] assert "showLiveCaption(message)" in handler assert "if (message.isCompleted !== true) return" in handler assert handler.index("showLiveCaption(message)") < handler.index( "if (message.isCompleted !== true) return" ) assert 'aria-label="实时语音字幕"' in component_source assert "liveCaptions.value" in component_source assert "caption.speaker" in component_source assert "caption.text" in component_source def test_companion_keeps_transcript_and_patient_case_visible_in_a_side_rail() -> None: companion_root = PROJECT_ROOT / "video_companion" / "src" main_source = (companion_root / "main.ts").read_text(encoding="utf-8") component_source = (companion_root / "App.vue").read_text(encoding="utf-8") styles = (companion_root / "style.css").read_text(encoding="utf-8") window_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py" ).read_text(encoding="utf-8") assert "liveCaptions.value = [...previous, caption].slice(-120)" in main_source assert "liveCaptionClearTimer" not in main_source assert 'aria-label="患者病例与实时对话"' in component_source assert 'id="patient-case-title"' in component_source assert 'id="live-transcript-title"' in component_source assert 'aria-label="打开完整诊单"' in component_source assert "runAction(onOpenDiagnosis)" in component_source assert "detail.clinicalDiagnosis" in component_source assert 'v-for="field in caseFields"' in component_source assert "{{ field.value }}" in component_source assert "{{ caption.time }}" in component_source assert "'caption-entry--partial': !caption.completed" in component_source assert ':allowed-full-screen="false"' in component_source assert ".video-layer--with-rail" in styles assert ".consultation-rail" in styles assert '"patientCase": self.patient_case' in window_source assert 'event == "open-diagnosis-request"' in window_source assert "QTimer.singleShot(0, self._open_diagnosis_safely)" in window_source stop_source = main_source.split("async function stopTranscription", 1)[1].split( "function transcriptionResult", 1 )[0] assert "clearLiveCaptions()" not in stop_source def test_video_diagnosis_entry_uses_existing_permission_scoped_drawer() -> None: app_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "app.py" ).read_text(encoding="utf-8") launcher_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "launcher.py" ).read_text(encoding="utf-8") main_source = ( PROJECT_ROOT / "video_companion" / "src" / "main.ts" ).read_text(encoding="utf-8") shell_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "ui" / "shell.py" ).read_text(encoding="utf-8") diagnosis_source = ( PROJECT_ROOT / "src" / "doctor_workstation" / "ui" / "dialogs" / "diagnosis.py" ).read_text(encoding="utf-8") styles = ( PROJECT_ROOT / "video_companion" / "src" / "style.css" ).read_text(encoding="utf-8") assert "shell.open_diagnosis_by_id(diagnosis_id, modeless=True)" in app_source assert "self._show_video_preview(video_window, dialog)" in app_source assert "WindowStaysOnTopHint" in app_source assert "dialog.finished.connect" in app_source assert "modeless=modeless" in shell_source assert "not modeless and not self._standalone_readonly" in diagnosis_source compact_styles = styles.split("@media (max-width: 700px)", 1)[1] assert ".consultation-rail { display: none; }" in compact_styles assert ".capture-button { display: none; }" in compact_styles assert "on_open_diagnosis=on_open_diagnosis" in launcher_source assert "event: 'open-diagnosis-request'" in main_source def test_video_preview_is_compact_and_restores_after_diagnosis_closes() -> None: from PySide6.QtCore import Qt from doctor_workstation.app import ApplicationController class _Rect: def x(self) -> int: return 0 def y(self) -> int: return 0 def width(self) -> int: return 1920 def height(self) -> int: return 1040 class _Screen: def availableGeometry(self) -> _Rect: # noqa: N802 - Qt-compatible double return _Rect() class _Window: def __init__(self) -> None: self.original_geometry = object() self.original_minimum = object() self.minimum = self.original_minimum self.geometry_value = self.original_geometry self.size = (900, 600) self.position = (30, 40) self.stays_on_top = False self.activated = 0 def geometry(self) -> object: return self.geometry_value def minimumSize(self) -> object: # noqa: N802 - Qt-compatible double return self.minimum def isMaximized(self) -> bool: # noqa: N802 - Qt-compatible double return False def isFullScreen(self) -> bool: # noqa: N802 - Qt-compatible double return False def windowFlags(self) -> Qt.WindowType: # noqa: N802 - Qt-compatible double return Qt.WindowType.Window def screen(self) -> _Screen: return _Screen() def showNormal(self) -> None: # noqa: N802 - Qt-compatible double return None def showMaximized(self) -> None: # noqa: N802 - Qt-compatible double return None def showFullScreen(self) -> None: # noqa: N802 - Qt-compatible double return None def setMinimumSize(self, *value: object) -> None: # noqa: N802 self.minimum = value[0] if len(value) == 1 else value def setWindowFlag(self, _flag: Any, enabled: bool) -> None: # noqa: N802 self.stays_on_top = enabled def resize(self, width: int, height: int) -> None: self.size = (width, height) def move(self, x: int, y: int) -> None: self.position = (x, y) def setGeometry(self, geometry: object) -> None: # noqa: N802 self.geometry_value = geometry def show(self) -> None: return None def raise_(self) -> None: return None def activateWindow(self) -> None: # noqa: N802 - Qt-compatible double self.activated += 1 class _Signal: def __init__(self) -> None: self.callbacks: list[Any] = [] def connect(self, callback: Any) -> None: self.callbacks.append(callback) class _Dialog: def __init__(self) -> None: self.finished = _Signal() def raise_(self) -> None: return None def activateWindow(self) -> None: # noqa: N802 - Qt-compatible double return None controller = SimpleNamespace( _video_preview_state=None, _video_preview_generation=0, ) controller._restore_video_preview = MethodType( # type: ignore[attr-defined] ApplicationController._restore_video_preview, controller, ) window = _Window() dialog = _Dialog() ApplicationController._show_video_preview(controller, window, dialog) assert window.minimum == (440, 300) assert window.size == (540, 356) assert window.position == (1362, 18) assert window.stays_on_top is True assert len(dialog.finished.callbacks) == 1 dialog.finished.callbacks[0](0) assert window.minimum is window.original_minimum assert window.geometry_value is window.original_geometry assert window.stays_on_top is False assert window.activated == 1 def test_video_patient_case_snapshot_is_bounded_and_clinically_useful() -> None: from doctor_workstation.app import _build_video_patient_case summary = _build_video_patient_case( { "diagnosis": { "patient_name": "张三", "id": 8279, "source_patient_id": 42, "gender": 1, "age": 47, "chief_complaint": "反复口渴三个月", "present_illness": "近期空腹血糖偏高", "allergy_history": "青霉素", "current_medicine": ["二甲双胍", "阿卡波糖"], "clinical_diagnosis": "2 型糖尿病", }, "appointment": {"appointment_date": "2026-08-26"}, "internal_audit": {"token": "must-not-cross-the-bridge"}, }, {}, diagnosis_id=8279, patient_id=42, patient_name="患者", ) assert summary["diagnosisId"] == "8279" assert summary["name"] == "张三" assert summary["gender"] == "男" assert summary["age"] == "47" assert summary["chiefComplaint"] == "反复口渴三个月" assert summary["currentMedication"] == "二甲双胍、阿卡波糖" assert summary["allergyHistory"] == "青霉素" assert "internal_audit" not in summary assert set(summary) == { "diagnosisId", "name", "gender", "age", "height", "weight", "diagnosisDate", "appointmentDate", "clinicalDiagnosis", "chiefComplaint", "presentIllness", "pastHistory", "allergyHistory", "personalHistory", "familyHistory", "currentMedication", "tongue", "pulse", "prescriptionOpinion", "remark", } bounded = _build_video_patient_case( {}, { "diagnosis_id": 8279, "patient_id": 42, "patient_name": "患" * 180, "age": "4" * 40, "remark": "病" * 2400, }, diagnosis_id=8279, patient_id=42, patient_name="患者", ) assert len(bounded["name"]) == 120 assert len(bounded["age"]) == 20 assert len(bounded["remark"]) == 2000 @pytest.mark.parametrize( "detail", [ { "id": 8279, "source_patient_id": 42, "patient_name": "张三", "chief_complaint": "口渴", }, { "data": { "id": 8279, "source_patient_id": 42, "patient_name": "张三", "chief_complaint": "口渴", } }, { "diagnosis": { "id": 8279, "source_patient_id": 42, "patient_name": "张三", "chief_complaint": "口渴", } }, ], ) def test_video_patient_case_accepts_supported_readonly_detail_shapes(detail: object) -> None: from doctor_workstation.app import _build_video_patient_case summary = _build_video_patient_case( detail, {}, diagnosis_id=8279, patient_id=42, patient_name="患者", ) assert summary["name"] == "张三" assert summary["chiefComplaint"] == "口渴" def test_video_patient_case_fails_closed_on_identity_mismatch() -> None: from doctor_workstation.app import _build_video_patient_case summary = _build_video_patient_case( { "diagnosis": { "id": 9001, "source_patient_id": 7, "patient_name": "其他患者", "chief_complaint": "不得展示", "allergy_history": "不得展示", } }, { "diagnosis_id": 9001, "patient_id": 7, "chief_complaint": "也不得展示", }, diagnosis_id=8279, patient_id=42, patient_name="张三", ) assert summary["name"] == "张三" assert summary["chiefComplaint"] == "" assert summary["allergyHistory"] == "" def test_built_video_companion_contains_the_patient_case_rail() -> None: dist_root = PROJECT_ROOT / "video_companion" / "dist" styles = "\n".join( path.read_text(encoding="utf-8") for path in (dist_root / "assets").glob("*.css") ) scripts = "\n".join( path.read_text(encoding="utf-8") for path in (dist_root / "assets").glob("*.js") ) assert ".consultation-rail" in styles assert ".video-layer--with-rail" in styles assert "patientCase" in scripts assert "患者病例与实时对话" in scripts def test_companion_screenshot_requires_doctor_confirmation_before_upload() -> None: source = (PROJECT_ROOT / "video_companion" / "src" / "App.vue").read_text( encoding="utf-8" ) capture = source.split("async function captureScreenshot", 1)[1].split( "function discardScreenshot", 1 )[0] confirm = source.split("async function confirmScreenshot", 1)[1].split( "watch(", 1 )[0] assert "screenshotPreview.value = canvas.toDataURL" in capture assert "onSaveScreenshot" not in capture assert "await props.onSaveScreenshot(screenshotPreview.value)" in confirm assert "确认画面后再保存到患者资料" in source assert "确认并上传" in source assert "取消" in source def test_companion_loads_im_history_without_an_empty_first_page_cursor() -> None: source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text( encoding="utf-8" ) load_messages = source.split("async function loadMessages", 1)[1].split( "async function sendText", 1 )[0] assert "nextReqMessageID: prepend ? nextReqMessageID : ''" not in load_messages assert "nextReqMessageID?: string" in load_messages assert ( "if (prepend && nextReqMessageID) request.nextReqMessageID = nextReqMessageID" in load_messages ) assert "chat.getMessageList(request)" in load_messages def test_companion_preserves_im_scroll_position_for_live_and_older_messages() -> None: source = (PROJECT_ROOT / "video_companion" / "src" / "App.vue").read_text( encoding="utf-8" ) watcher = source.split("() => props.messages.value.length", 1)[1].split( "watch(", 1 )[0] load_earlier = source.split("async function loadEarlierMessages", 1)[1].split( "async function runAction", 1 )[0] assert "if (!stickToMessageBottom.value) return" in watcher assert "stickToMessageBottom.value = false" in load_earlier assert "container.scrollHeight - previousHeight" in load_earlier assert '@scroll="handleMessageScroll"' in source assert '@click="loadEarlierMessages"' in source def test_normalizes_admin_ticket_aliases_to_companion_contract() -> None: request = normalize_backend_ticket( { "sdkAppId": "1400123456", "userId": " doctor_42 ", "userSig": "short-lived-ticket", "patientUserId": " patient_8 ", "diagnosisId": 123, "patientId": 8, } ) assert request == VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) assert request.to_web_config() == { "SDKAppID": 1400123456, "userID": "doctor_42", "userSig": "short-lived-ticket", "targetUserId": "patient_8", "diagnosisId": 123, } def test_accepts_uppercase_aliases_and_nested_backend_envelope() -> None: request = VideoCallRequest.from_backend_ticket( { "data": { "SDKAppID": 1400123456, "userID": "doctor_42", "userSig": "ticket-value", "targetUserId": "patient_8", } }, diagnosis_id="diagnosis-123", patient_id=8, backend_mode="embedded", ) assert request.diagnosis_id == "diagnosis-123" assert request.backend_mode is BackendMode.EMBEDDED def test_accepts_repository_call_ticket_object_without_importing_core_models() -> None: ticket = SimpleNamespace( sdk_app_id=1400123456, user_id="doctor_42", user_sig="ticket-value", patient_user_id="patient_8", diagnosis_id=123, raw={"sdkAppId": 1400123456}, ) request = normalize_backend_ticket(ticket, patient_id=8) assert request.patient_id == 8 assert request.to_web_config()["targetUserId"] == "patient_8" def test_secret_is_excluded_from_repr_and_safe_log_context() -> None: request = normalize_backend_ticket( { "sdkAppId": 1400123456, "userId": "doctor_42", "userSig": "never-write-this-value", "patientUserId": "patient_8", }, diagnosis_id=123, ) assert "never-write-this-value" not in repr(request) assert "never-write-this-value" not in str(request.safe_log_context()) assert "user_sig" not in request.safe_log_context() @pytest.mark.parametrize("forbidden_key", ["SDKSecretKey", "sdk_secret_key", "secretKey"]) def test_rejects_server_side_secret_material(forbidden_key: str) -> None: with pytest.raises(VideoTicketError, match="forbidden server-side secret"): normalize_backend_ticket( { "sdkAppId": 1400123456, "userId": "doctor_42", "userSig": "ticket-value", "patientUserId": "patient_8", "diagnosisId": 123, forbidden_key: "must-never-reach-a-client", } ) @pytest.mark.parametrize( ("field", "value"), [ ("sdkAppId", 0), ("userId", ""), ("userSig", ""), ("patientUserId", " "), ("diagnosisId", None), ], ) def test_rejects_incomplete_or_invalid_ticket(field: str, value: object) -> None: ticket: dict[str, object] = { "sdkAppId": 1400123456, "userId": "doctor_42", "userSig": "ticket-value", "patientUserId": "patient_8", "diagnosisId": 123, } ticket[field] = value with pytest.raises(VideoTicketError): normalize_backend_ticket(ticket) def test_rejects_conflicting_aliases_and_modes() -> None: with pytest.raises(VideoTicketError, match="conflicting SDKAppID aliases"): normalize_backend_ticket( { "SDKAppID": 1400123456, "sdkAppId": 1400654321, "userID": "doctor_42", "userSig": "ticket-value", "targetUserId": "patient_8", "diagnosisId": 123, } ) with pytest.raises(VideoTicketError, match="backend mode"): BackendMode.parse("native") def test_launcher_rejects_browser_before_importing_window_or_writing_repository() -> None: class Repository: def start_call(self, **payload: object) -> None: raise AssertionError(f"unexpected repository write: {payload}") launcher = VideoCallLauncher(repository=Repository(), backend_mode="browser") with pytest.raises(VideoTicketError, match="one-time handoff"): launcher.prepare( { "sdkAppId": 1400123456, "userId": "doctor_42", "userSig": "ticket-value", "patientUserId": "patient_8", "diagnosisId": 123, } ) def test_call_lifecycle_is_fifo_daemon_and_never_blocks_caller() -> None: events: list[tuple[object, ...]] = [] start_entered = threading.Event() release_start = threading.Event() class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, int]: start_entered.set() assert release_start.wait(2) events.append(("start", diagnosis_id, patient_id, call_type)) return {"call_record_id": 900} def bind_call_room( self, diagnosis_id: int, room_id: str, *, call_record_id: int ) -> None: events.append(("bind", diagnosis_id, room_id, call_record_id)) def end_call(self, diagnosis_id: int, *, call_record_id: int) -> None: events.append(("end", diagnosis_id, call_record_id)) request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) started_at = time.monotonic() start_future = lifecycle.start() bind_future = lifecycle.bind_room("456789") duplicate_bind = lifecycle.bind_room("456789") changed_bind = lifecycle.bind_room("another-room") end_future = lifecycle.end("test") elapsed = time.monotonic() - started_at assert start_entered.wait(1) assert elapsed < 0.2 assert lifecycle.worker_is_daemon is True assert lifecycle.wait(0.01) is False assert duplicate_bind is bind_future assert changed_bind.result(timeout=0) is False assert lifecycle.current_room_id == "456789" release_start.set() assert start_future.result(timeout=2) is True assert bind_future.result(timeout=2) is True assert end_future.result(timeout=2) is True assert lifecycle.wait(1) is True assert lifecycle.current_room_id == "456789" assert events == [ ("start", 123, 8, 2), ("bind", 123, "456789", 900), ("end", 123, 900), ] def test_transcription_lifecycle_uses_start_record_id_and_remains_fifo() -> None: events: list[tuple[object, ...]] = [] class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, object]: events.append(("start", diagnosis_id, patient_id, call_type)) return {"data": {"callRecordId": 901}} def start_call_transcription( self, diagnosis_id: int, call_record_id: int, transcription_session_id: str, *, language: str, ) -> None: events.append( ( "transcription-start", diagnosis_id, call_record_id, transcription_session_id, language, ) ) def upsert_call_transcript_segments( self, diagnosis_id: int, call_record_id: int, transcription_session_id: str, segments: list[dict[str, object]], ) -> None: events.append( ( "segment", diagnosis_id, call_record_id, transcription_session_id, segments, ) ) def finish_call_transcription( self, diagnosis_id: int, call_record_id: int, transcription_session_id: str, expected_segment_count: int, *, status: str, ) -> None: events.append( ( "finish", diagnosis_id, call_record_id, transcription_session_id, expected_segment_count, status, ) ) def end_call(self, diagnosis_id: int) -> None: events.append(("end", diagnosis_id)) request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) start = lifecycle.start() transcription_start = lifecycle.start_transcription("session-1") segment = lifecycle.save_transcript_segment( { "segment_id": "seg-1", "speaker_user_id": "patient_8", "speaker_role": "patient", "timestamp": 1200, "text": "patient words", } ) duplicate = lifecycle.save_transcript_segment( {"segment_id": "seg-1", "text": "must not produce another write"} ) finish = lifecycle.finish_transcription(status="completed") end = lifecycle.end("test") assert duplicate is segment assert start.result(timeout=2) is True assert transcription_start.result(timeout=2) is True assert segment.result(timeout=2) is True assert finish.result(timeout=2) is True assert end.result(timeout=2) is True assert lifecycle.wait(1) is True assert lifecycle.call_record_id == 901 assert events == [ ("start", 123, 8, 2), ("transcription-start", 123, 901, "session-1", "zh-CN"), ( "segment", 123, 901, "session-1", [ { "segment_id": "seg-1", "transcription_session_id": "session-1", "speaker_user_id": "patient_8", "speaker_role": "patient", "timestamp": 1200, "text": "patient words", } ], ), ("finish", 123, 901, "session-1", 1, "completed"), ("end", 123), ] def test_end_auto_finishes_active_transcription_as_partial() -> None: events: list[tuple[object, ...]] = [] class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, int]: events.append(("start", diagnosis_id, patient_id, call_type)) return {"call_record_id": 902} def start_call_transcription( self, diagnosis_id: int, call_record_id: int, transcription_session_id: str, *, language: str, ) -> None: events.append( ( "transcription-start", diagnosis_id, call_record_id, transcription_session_id, language, ) ) def finish_call_transcription( self, diagnosis_id: int, call_record_id: int, transcription_session_id: str, expected_segment_count: int, *, status: str, ) -> None: events.append( ( "finish", diagnosis_id, call_record_id, transcription_session_id, expected_segment_count, status, ) ) def end_call(self, diagnosis_id: int) -> None: events.append(("end", diagnosis_id)) request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) lifecycle.start() lifecycle.start_transcription("session-auto-partial") ended = lifecycle.end("window-closed") assert ended.result(timeout=2) is True assert lifecycle.wait(1) is True assert events == [ ("start", 123, 8, 2), ("transcription-start", 123, 902, "session-auto-partial", "zh-CN"), ("finish", 123, 902, "session-auto-partial", 0, "partial"), ("end", 123), ] def test_failed_start_prevents_bind_and_end_writes() -> None: events: list[str] = [] class Repository: def start_call(self, diagnosis_id: int, *, call_type: int) -> None: del diagnosis_id, call_type events.append("start") raise RuntimeError("backend unavailable") def bind_call_room(self, diagnosis_id: int, room_id: str) -> None: del diagnosis_id, room_id events.append("bind") def end_call(self, diagnosis_id: int) -> None: del diagnosis_id events.append("end") request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) start_future = lifecycle.start() bind_future = lifecycle.bind_room("456789") end_future = lifecycle.end("test") with pytest.raises(RuntimeError, match="backend unavailable"): start_future.result(timeout=1) assert bind_future.result(timeout=1) is False assert end_future.result(timeout=1) is False assert lifecycle.wait(1) is True assert events == ["start"] def test_explicit_cos_recording_failure_fails_room_binding_without_losing_call_identity() -> None: class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, int]: del diagnosis_id, patient_id, call_type return {"call_record_id": 904} def bind_call_room( self, diagnosis_id: int, room_id: str, *, call_record_id: int ) -> dict[str, object]: assert (diagnosis_id, room_id, call_record_id) == (123, "456789", 904) return { "call_record_id": 904, "cloud_recording": { "started": False, "message": "COS bucket is unavailable", }, } request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) with pytest.raises(RuntimeError, match="COS bucket is unavailable"): lifecycle.bind_room("456789").result(timeout=2) assert lifecycle.call_record_id == 904 assert lifecycle.bound_room_id is None assert lifecycle.wait(1) is True def test_failed_room_binding_releases_claim_and_can_retry_same_room() -> None: bind_attempts = 0 class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, int]: del diagnosis_id, patient_id, call_type return {"call_record_id": 906} def bind_call_room( self, diagnosis_id: int, room_id: str, *, call_record_id: int ) -> dict[str, object]: nonlocal bind_attempts assert (diagnosis_id, room_id, call_record_id) == (123, "456789", 906) bind_attempts += 1 if bind_attempts == 1: raise RuntimeError("temporary bind failure") return { "call_record_id": 906, "cloud_recording": {"started": True}, } request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) with pytest.raises(RuntimeError, match="temporary bind failure"): lifecycle.bind_room("456789").result(timeout=2) assert lifecycle.current_room_id is None assert lifecycle.bind_room("456789").result(timeout=2) is True assert lifecycle.bound_room_id == "456789" assert lifecycle.current_room_id == "456789" assert bind_attempts == 2 assert lifecycle.wait(1) is True def test_local_audio_upload_uses_exact_started_record_and_precedes_end( tmp_path: Path, ) -> None: events: list[tuple[object, ...]] = [] class Repository: def start_call( self, diagnosis_id: int, patient_id: int, *, call_type: int ) -> dict[str, int]: events.append(("start", diagnosis_id, patient_id, call_type)) return {"call_record_id": 905} def upload_call_recording( self, path: Path, diagnosis_id: int, *, call_record_id: int, mime_type: str, ) -> dict[str, object]: events.append( ( "local-audio", path.read_bytes(), diagnosis_id, call_record_id, mime_type, ) ) return { "completed": True, "call_record_id": call_record_id, "media_kind": "local_audio", } def end_call(self, diagnosis_id: int, *, call_record_id: int) -> None: events.append(("end", diagnosis_id, call_record_id)) recording = tmp_path / "call-audio.webm" recording.write_bytes(b"opus-webm-audio") request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, patient_id=8, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) lifecycle.start() uploaded = lifecycle.save_local_audio_recording( recording, mime_type="audio/webm;codecs=opus", ) ended = lifecycle.end("doctor-hangup") assert uploaded.result(timeout=2) is True assert ended.result(timeout=2) is True assert lifecycle.wait(1) is True assert events == [ ("start", 123, 8, 2), ("local-audio", b"opus-webm-audio", 123, 905, "audio/webm;codecs=opus"), ("end", 123, 905), ] def test_video_screenshot_is_uploaded_and_appended_to_patient_tongue_images() -> None: events: list[tuple[object, ...]] = [] class Repository: def start_call(self, diagnosis_id: int, *, call_type: int) -> dict[str, int]: events.append(("start", diagnosis_id, call_type)) return {"call_record_id": 903} def upload_material_bytes( self, content: bytes, filename: str, material_type: str, cid: int = 0, ) -> str: events.append(("upload", content, filename, material_type, cid)) return "/uploads/image/callshot-123.jpg" def add_doctor_note( self, diagnosis_id: int, content: str, tongue_images: list[str], ) -> None: events.append(("note", diagnosis_id, content, tongue_images)) def end_call(self, diagnosis_id: int) -> None: events.append(("end", diagnosis_id)) request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) lifecycle.start() screenshot = lifecycle.save_screenshot(b"jpeg-frame", "callshot-123.jpg") lifecycle.end("test") assert screenshot.result(timeout=2) == "/uploads/image/callshot-123.jpg" assert lifecycle.wait(1) is True assert events == [ ("start", 123, 2), ("upload", b"jpeg-frame", "callshot-123.jpg", "image", 0), ("note", 123, "", ["/uploads/image/callshot-123.jpg"]), ("end", 123), ] def test_start_rejects_missing_record_id_without_using_ticket_fallback() -> None: class Repository: def start_call(self, diagnosis_id: int, *, call_type: int) -> dict[str, object]: del diagnosis_id, call_type return {} request = VideoCallRequest( sdk_app_id=1400123456, user_id="doctor_42", user_sig="short-lived-ticket", target_user_id="patient_8", diagnosis_id=123, call_record_id=77, ) lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__)) with pytest.raises(ValueError, match="startCall response did not include"): lifecycle.start().result(timeout=2) assert lifecycle.started is False def test_https_document_policy_is_exact_and_origin_scoped() -> None: policy = TrustedDocumentPolicy.from_url( "https://RTC.Example.com/doctor-call/index.html?tenant=a#boot", is_local=False, ) assert policy.allows_main_document( "https://rtc.example.com:443/doctor-call/index.html?tenant=a#ready" ) assert not policy.allows_main_document( "https://rtc.example.com/doctor-call/index.html?tenant=b" ) assert not policy.allows_main_document("https://rtc.example.com/other/index.html?tenant=a") assert policy.allows_origin("https://rtc.example.com") assert not policy.allows_origin("https://sub.rtc.example.com") assert not policy.allows_origin("http://rtc.example.com") with pytest.raises(TrustedDocumentError, match="HTTPS"): TrustedDocumentPolicy.from_url("http://rtc.example.com/doctor-call", is_local=False) def test_local_document_policy_rejects_sibling_files(tmp_path: Path) -> None: index = tmp_path / "dist" / "index.html" index.parent.mkdir() index.touch() sibling = index.with_name("other.html") sibling.touch() policy = TrustedDocumentPolicy.from_url(index.as_uri(), is_local=True) assert policy.allows_main_document(index.as_uri()) assert not policy.allows_main_document(sibling.as_uri()) assert policy.allows_origin("file:///")