from __future__ import annotations import logging import sys import threading import time from pathlib import Path from types import SimpleNamespace 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_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) -> None: events.append(("bind", diagnosis_id, room_id)) 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__)) 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 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 events == [ ("start", 123, 8, 2), ("bind", 123, "456789"), ("end", 123), ] 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_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:///")