158 lines
6.4 KiB
Python
158 lines
6.4 KiB
Python
"""Exercise bridge trust and JSON delivery without starting WebEngine or HTTP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
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/im/index.html"
|
|
|
|
|
|
def host(**changes):
|
|
item = SimpleNamespace(
|
|
_closing=False, _released=False, _shutdown_requested=False, _injected=True,
|
|
_policy=TrustedDocumentPolicy.from_url(TRUSTED, is_local=False),
|
|
_page=Mock(), _im_ai=Mock(), _im_ai_timer=Mock(),
|
|
)
|
|
item._page.url.return_value.toString.return_value = TRUSTED
|
|
item._im_ai_context_is_trusted = lambda: window._EmbeddedVideoWindow._im_ai_context_is_trusted(item)
|
|
for key, value in changes.items():
|
|
setattr(item, key, value)
|
|
return item
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"_closing": True}, {"_released": True}, {"_shutdown_requested": True}, {"_injected": False},
|
|
])
|
|
def test_inactive_or_closing_page_cannot_submit_or_receive_ai(changes):
|
|
item = host(**changes)
|
|
window._EmbeddedVideoWindow._handle_im_ai_message(item, {"event": "im-ai-request"})
|
|
window._EmbeddedVideoWindow._flush_im_ai(item)
|
|
item._im_ai.handle.assert_not_called()
|
|
item._im_ai.take_result.assert_not_called()
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["about:blank", "https://evil.example.test/im/index.html", TRUSTED + "?other=1"])
|
|
def test_document_navigation_cannot_reuse_authenticated_ai_bridge(url):
|
|
item = host()
|
|
item._page.url.return_value.toString.return_value = url
|
|
window._EmbeddedVideoWindow._handle_im_ai_message(item, {"event": "im-ai-request"})
|
|
window._EmbeddedVideoWindow._flush_im_ai(item)
|
|
item._im_ai.handle.assert_not_called()
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_trusted_bridge_routes_only_to_identity_checked_controller():
|
|
item = host()
|
|
message = {"source": "doctor-call", "event": "im-ai-request", "diagnosisId": 501}
|
|
window._EmbeddedVideoWindow._handle_im_ai_message(item, message)
|
|
item._im_ai.handle.assert_called_once_with(message)
|
|
|
|
|
|
def test_result_content_is_json_encoded_and_no_javascript_is_executed():
|
|
item = host()
|
|
payload = {"status": "success", "text": '\");window.bad=true;//\n</script>\u2028', "warnings": ["合成提示"]}
|
|
item._im_ai.take_result.return_value = payload
|
|
window._EmbeddedVideoWindow._flush_im_ai(item)
|
|
script = item._page.runJavaScript.call_args.args[0]
|
|
prefix, suffix = "window.doctorConsultation?.imAiResult?.(", ");"
|
|
assert script.startswith(prefix) and script.endswith(suffix)
|
|
assert json.loads(script[len(prefix):-len(suffix)]) == payload
|
|
assert "\\u2028" in script
|
|
|
|
|
|
def test_host_acknowledges_initial_summary_only_after_request_gate_opens():
|
|
item = host(_injected=False)
|
|
item._im_ai.session_id = 'synthetic-session-\\"\u2028'
|
|
|
|
def verify_gate(script):
|
|
assert item._im_ai_context_is_trusted() is True
|
|
prefix, suffix = "window.doctorConsultation?.imAiReady?.(", ");"
|
|
assert script.startswith(prefix) and script.endswith(suffix)
|
|
assert json.loads(script[len(prefix):-len(suffix)]) == item._im_ai.session_id
|
|
|
|
item._page.runJavaScript.side_effect = verify_gate
|
|
window._EmbeddedVideoWindow._after_injection(item, True)
|
|
item._page.runJavaScript.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"_closing": True}, {"_released": True}, {"_shutdown_requested": True},
|
|
])
|
|
def test_late_injection_cannot_acknowledge_a_closed_ai_context(changes):
|
|
item = host(_injected=False, **changes)
|
|
window._EmbeddedVideoWindow._after_injection(item, True)
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_navigation_cannot_receive_ai_readiness_acknowledgement():
|
|
item = host(_injected=False)
|
|
item._page.url.return_value.toString.return_value = "https://evil.example.test/im/index.html"
|
|
window._EmbeddedVideoWindow._after_injection(item, True)
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_failed_injection_closes_without_ai_readiness_acknowledgement():
|
|
item = host(_injected=False, _request_companion_shutdown=Mock())
|
|
window._EmbeddedVideoWindow._after_injection(item, False)
|
|
item._request_companion_shutdown.assert_called_once_with("bridge-api-missing")
|
|
item._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_shutdown_cancels_ai_before_waiting_for_media_or_destroying_page():
|
|
item = host(
|
|
_start_requested=True, _call_cycle_closed=False, _companion_ended=False,
|
|
_shutdown_timer=Mock(), close=Mock(),
|
|
)
|
|
window._EmbeddedVideoWindow._request_companion_shutdown(item, "window-closed")
|
|
item._im_ai.close.assert_called_once()
|
|
item._im_ai_timer.stop.assert_called_once()
|
|
assert item._shutdown_requested is True
|
|
item.close.assert_not_called() # Recording shutdown still gets its existing grace period.
|
|
item._shutdown_timer.start.assert_called_once_with(190_000)
|
|
|
|
|
|
def test_qt_notify_rejects_oversized_or_wrong_source_payloads():
|
|
callback = Mock()
|
|
bridge = window._QtVideoBridge(callback, Mock())
|
|
bridge.notify("x" * 16_385)
|
|
bridge.notify(json.dumps({"source": "evil", "event": "im-ai-request"}))
|
|
bridge.notify("[]")
|
|
callback.assert_not_called()
|
|
message = {"source": "doctor-call", "event": "im-ai-cancel"}
|
|
bridge.notify(json.dumps(message))
|
|
callback.assert_called_once_with(message)
|
|
|
|
|
|
def test_video_module_remains_importable_without_optional_qt():
|
|
source = Path(__file__).resolve().parents[1] / "src"
|
|
script = """
|
|
import builtins
|
|
original_import = builtins.__import__
|
|
def without_qt(name, *args, **kwargs):
|
|
if name == 'PySide6' or name.startswith('PySide6.'):
|
|
raise ImportError('synthetic core-only installation')
|
|
return original_import(name, *args, **kwargs)
|
|
builtins.__import__ = without_qt
|
|
from doctor_workstation.video.window import WEBENGINE_AVAILABLE, VideoCallWindow
|
|
from doctor_workstation.video.im_ai import ImAiController
|
|
assert WEBENGINE_AVAILABLE is False
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script], env={**os.environ, "PYTHONPATH": str(source)},
|
|
capture_output=True, text=True, timeout=10, check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|