136 lines
6.5 KiB
Python
136 lines
6.5 KiB
Python
"""Trusted native PCM ingestion and exact transcript/AI identity handoff."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from doctor_workstation.video import window
|
|
from doctor_workstation.video.security import TrustedDocumentPolicy
|
|
|
|
pytestmark = pytest.mark.skipif(not window.WEBENGINE_AVAILABLE, reason="optional QtWebEngine")
|
|
TRUSTED = "https://video.example.test/call/index.html"
|
|
|
|
|
|
def host(**changes):
|
|
item = SimpleNamespace(
|
|
_closing=False, _released=False, _shutdown_requested=False, _injected=True,
|
|
_start_requested=True, _call_cycle_closed=False,
|
|
_policy=TrustedDocumentPolicy.from_url(TRUSTED, is_local=False),
|
|
_page=Mock(), _local_transcription=Mock(), _live_ai=Mock(),
|
|
_video_ai_timer=Mock(), _asr_session_id="tr-local", _asr_ready=True,
|
|
_asr_partial=False, _transcription_final_status="completed",
|
|
lifecycle=SimpleNamespace(call_record_id=71, transcription_session_id="tr-local"),
|
|
request=SimpleNamespace(user_id="doctor_1", target_user_id="patient_9"),
|
|
_save_transcript_segment=Mock(),
|
|
)
|
|
item._page.url.return_value.toString.return_value = TRUSTED
|
|
for key, value in changes.items():
|
|
setattr(item, key, value)
|
|
item._local_transcription_context_is_trusted = lambda: window._EmbeddedVideoWindow._local_transcription_context_is_trusted(item)
|
|
item._on_local_transcription = lambda event: window._EmbeddedVideoWindow._on_local_transcription(item, event)
|
|
return item
|
|
|
|
|
|
def chunk(**changes):
|
|
return {"event": "local-transcription-chunk", "sessionId": "tr-local", "role": "patient",
|
|
"sequence": 3, "timestamp": 1793332800000.0, "data": "AQACAA==", **changes}
|
|
|
|
|
|
def test_pcm_uses_bound_session_and_double_timestamp_is_converted_safely():
|
|
item = host()
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, chunk())
|
|
item._local_transcription.submit.assert_called_once_with("tr-local", "patient", 3, 1793332800000, "AQACAA==")
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [{"sessionId": "other"}, {"timestamp": float("nan")},
|
|
{"timestamp": float("inf")}, {"timestamp": -1}, {"timestamp": True}])
|
|
def test_wrong_session_or_unbounded_pcm_metadata_rejected(changes):
|
|
item = host()
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, chunk(**changes))
|
|
item._local_transcription.submit.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [{"_closing": True}, {"_released": True}, {"_injected": False}])
|
|
def test_untrusted_lifecycle_cannot_ingest_audio(changes):
|
|
item = host(**changes)
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, chunk())
|
|
item._local_transcription.submit.assert_not_called()
|
|
|
|
|
|
def test_navigation_blocks_pcm_and_analysis_commands():
|
|
item = host()
|
|
item._page.url.return_value.toString.return_value = "https://another.example.test/"
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, chunk())
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, {"event": "video-ai-action", "action": "refresh"})
|
|
item._local_transcription.submit.assert_not_called()
|
|
item._live_ai.handle.assert_not_called()
|
|
|
|
|
|
def test_pending_shutdown_drains_pcm_but_cannot_start_new_inference_session():
|
|
item = host(_shutdown_requested=True)
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, chunk())
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, {"event": "local-transcription-start", "sessionId": "tr-next"})
|
|
window._EmbeddedVideoWindow._handle_video_assistance_message(item, {"event": "video-ai-action", "action": "refresh"})
|
|
item._local_transcription.submit.assert_called_once()
|
|
item._local_transcription.start.assert_not_called()
|
|
item._live_ai.handle.assert_not_called()
|
|
|
|
|
|
def test_asr_segment_is_persisted_with_host_speaker_before_json_display():
|
|
item = host()
|
|
event = {"type": "segment", "sessionId": "tr-local", "role": "patient", "segmentId": "tr-local:patient:3",
|
|
"text": '没有；0.5 mg，不是 5 mg。</script>\u2028', "timestamp": 1793332800000, "isCompleted": True}
|
|
item._on_local_transcription(event)
|
|
item._save_transcript_segment.assert_called_once()
|
|
saved = item._save_transcript_segment.call_args.args[0]
|
|
assert saved["segment"]["text"] == event["text"]
|
|
assert saved["segment"]["speaker_user_id"] == "patient_9"
|
|
script = item._page.runJavaScript.call_args.args[0]
|
|
assert json.loads(script.split("?.(", 1)[1][:-2]) == event
|
|
item._live_ai.changed.assert_not_called() # waits for persistence ACK
|
|
|
|
|
|
def test_late_asr_segment_cannot_enter_new_call():
|
|
item = host(_asr_session_id="tr-next")
|
|
item._on_local_transcription({"type": "segment", "sessionId": "tr-local", "role": "patient"})
|
|
item._save_transcript_segment.assert_not_called()
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_persistence_ack_is_the_only_ai_dirty_trigger():
|
|
item = host()
|
|
window._EmbeddedVideoWindow._on_transcription_completed(item, "segment", "other", "s1", True, "")
|
|
item._live_ai.changed.assert_not_called()
|
|
window._EmbeddedVideoWindow._on_transcription_completed(item, "segment", "tr-local", "s1", True, "")
|
|
item._live_ai.changed.assert_called_once_with("tr-local")
|
|
window._EmbeddedVideoWindow._on_transcription_completed(item, "segment", "tr-local", "s2", False, "")
|
|
assert item._asr_partial
|
|
assert item._live_ai.changed.call_count == 1
|
|
|
|
|
|
def test_model_missing_is_failed_and_partial_chunk_is_not_false_ready():
|
|
item = host(_asr_ready=False)
|
|
item._on_local_transcription({"type": "error", "sessionId": "tr-local", "message": "本地模型未安装"})
|
|
item._live_ai.transcription_error.assert_called_once_with("本地模型未安装", status="failed")
|
|
|
|
|
|
def test_shutdown_keeps_asr_for_drain_then_closes_it():
|
|
item = host()
|
|
window._EmbeddedVideoWindow._close_video_assistance(item, close_asr=False)
|
|
item._live_ai.close.assert_called_once()
|
|
item._local_transcription.close.assert_not_called()
|
|
window._EmbeddedVideoWindow._close_video_assistance(item)
|
|
item._local_transcription.close.assert_called_once()
|
|
|
|
|
|
def test_special_pcm_bridge_bounds_data_and_preserves_epoch():
|
|
callback = Mock()
|
|
bridge = window._QtVideoBridge(callback, Mock())
|
|
bridge.appendLocalTranscriptionChunk("tr-local", "doctor", 0, 1793332800000.0, "x" * 512001)
|
|
callback.assert_not_called()
|
|
bridge.appendLocalTranscriptionChunk("tr-local", "doctor", 0, 1793332800000.0, "AQACAA==")
|
|
assert callback.call_args.args[0]["timestamp"] == 1793332800000.0
|