1336 lines
45 KiB
Python
1336 lines
45 KiB
Python
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<HTMLMediaElement>('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,
|
|
"patientId": 8,
|
|
"appointmentId": 0,
|
|
"appointment_type": None,
|
|
"appointment_type_desc": "",
|
|
"can_video_call": False,
|
|
"can_audio_call": False,
|
|
"call_disabled_reason": "",
|
|
}
|
|
|
|
|
|
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:///")
|