588 lines
18 KiB
Python
588 lines
18 KiB
Python
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:///")
|