344 lines
13 KiB
Python
344 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Callable, Iterator
|
|
from types import MethodType, SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PySide6.QtCore import QRect, QSize, Qt
|
|
from PySide6.QtGui import QCloseEvent
|
|
from PySide6.QtWidgets import QApplication, QDialog, QMainWindow
|
|
|
|
from doctor_workstation.app import ApplicationController
|
|
from doctor_workstation.video import window as video_window
|
|
|
|
|
|
class _CallWindow(QMainWindow):
|
|
"""Exercise production Qt close handling without starting WebEngine or media."""
|
|
|
|
closeEvent = video_window._EmbeddedVideoWindow.closeEvent
|
|
set_compact_close_handler = video_window._EmbeddedVideoWindow.set_compact_close_handler
|
|
hangup = video_window._EmbeddedVideoWindow.hangup
|
|
_request_companion_shutdown = video_window._EmbeddedVideoWindow._request_companion_shutdown
|
|
_close_from_companion = video_window._EmbeddedVideoWindow._close_from_companion
|
|
_force_requested_shutdown = video_window._EmbeddedVideoWindow._force_requested_shutdown
|
|
_begin_shutdown = video_window._EmbeddedVideoWindow._begin_shutdown
|
|
_handle_bridge_message = video_window._EmbeddedVideoWindow._handle_bridge_message
|
|
_on_load_finished = video_window._EmbeddedVideoWindow._on_load_finished
|
|
_after_injection = video_window._EmbeddedVideoWindow._after_injection
|
|
|
|
def __init__(self, **changes: Any) -> None:
|
|
super().__init__()
|
|
self._on_compact_close: Callable[[], None] | None = None
|
|
self._closing = False
|
|
self._released = False
|
|
self._shutdown_requested = False
|
|
self._companion_ended = False
|
|
self._injected = True
|
|
self._start_requested = True
|
|
self._call_cycle_closed = False
|
|
self._media_active = True
|
|
self._close_reason = "window-closed"
|
|
self.open_im = False
|
|
self.lifecycle = SimpleNamespace(end=Mock())
|
|
self._page = SimpleNamespace(runJavaScript=Mock())
|
|
self._shutdown_timer = SimpleNamespace(start=Mock(), stop=Mock())
|
|
self._abort_local_audio_recording = Mock()
|
|
self._append_local_audio_chunk = Mock()
|
|
self._release_webengine = Mock(side_effect=self._mark_released)
|
|
self.restore_screenshot_preview = Mock()
|
|
self.call_ended = SimpleNamespace(emit=Mock())
|
|
self.call_error = SimpleNamespace(emit=Mock())
|
|
self.logger = Mock()
|
|
self.request = SimpleNamespace(safe_log_context=lambda: {})
|
|
for key, value in changes.items():
|
|
setattr(self, key, value)
|
|
self.setProperty("doctorSkipOpenGeometry", True)
|
|
self.setMinimumSize(620, 420)
|
|
self.setGeometry(40, 50, 680, 480)
|
|
|
|
def _mark_released(self) -> None:
|
|
self._released = True
|
|
|
|
|
|
@pytest.fixture
|
|
def qapp() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def make_window(qapp: QApplication) -> Iterator[Callable[..., _CallWindow]]:
|
|
windows: list[_CallWindow] = []
|
|
|
|
def create(**changes: Any) -> _CallWindow:
|
|
window = _CallWindow(**changes)
|
|
windows.append(window)
|
|
return window
|
|
|
|
yield create
|
|
for window in windows:
|
|
window.set_compact_close_handler(None)
|
|
window._companion_ended = True
|
|
window.close()
|
|
window.deleteLater()
|
|
qapp.processEvents()
|
|
|
|
|
|
def _controller() -> SimpleNamespace:
|
|
controller = SimpleNamespace(_video_preview_state=None, _video_preview_generation=0)
|
|
controller._show_video_preview = MethodType(
|
|
ApplicationController._show_video_preview, controller
|
|
)
|
|
controller._restore_video_preview = MethodType(
|
|
ApplicationController._restore_video_preview, controller
|
|
)
|
|
return controller
|
|
|
|
|
|
def test_native_compact_close_restores_visible_window_and_next_close_ends_call(
|
|
qapp: QApplication, make_window: Callable[..., _CallWindow]
|
|
) -> None:
|
|
window = make_window()
|
|
window.show()
|
|
qapp.processEvents()
|
|
original_geometry = QRect(window.geometry())
|
|
original_minimum = QSize(window.minimumSize())
|
|
controller = _controller()
|
|
dialog = QDialog()
|
|
|
|
controller._show_video_preview(window, dialog)
|
|
qapp.processEvents()
|
|
|
|
assert window.isVisible()
|
|
assert window.minimumSize() == QSize(440, 300)
|
|
assert window.geometry() != original_geometry
|
|
assert window.windowFlags() & Qt.WindowType.WindowStaysOnTopHint
|
|
assert window._on_compact_close is not None
|
|
|
|
assert window.close() is False
|
|
qapp.processEvents()
|
|
|
|
assert window.isVisible()
|
|
assert window.geometry() == original_geometry
|
|
assert window.minimumSize() == original_minimum
|
|
assert not window.windowFlags() & Qt.WindowType.WindowStaysOnTopHint
|
|
assert window.property("doctorSkipOpenGeometry") is False
|
|
assert controller._video_preview_state is None
|
|
assert window._on_compact_close is None
|
|
assert window._shutdown_requested is False
|
|
window._page.runJavaScript.assert_not_called()
|
|
window.lifecycle.end.assert_not_called()
|
|
window._abort_local_audio_recording.assert_not_called()
|
|
window._release_webengine.assert_not_called()
|
|
window.restore_screenshot_preview.assert_called_once_with()
|
|
|
|
assert window.close() is False
|
|
|
|
assert window._shutdown_requested is True
|
|
assert window.isVisible()
|
|
assert window._closing is False
|
|
window._page.runJavaScript.assert_called_once()
|
|
window._release_webengine.assert_not_called()
|
|
|
|
window._handle_bridge_message({"event": "hangup", "status": "local-hangup"})
|
|
qapp.processEvents()
|
|
|
|
assert not window.isVisible()
|
|
assert window._closing is True
|
|
window._release_webengine.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize("action", ["hangup", "facade-close"])
|
|
@pytest.mark.parametrize("wait_for_recording", [False, True], ids=["idle", "recording"])
|
|
def test_explicit_shutdown_bypasses_compact_restore_and_drains_recording(
|
|
make_window: Callable[..., _CallWindow], action: str, wait_for_recording: bool
|
|
) -> None:
|
|
window = make_window(_start_requested=wait_for_recording)
|
|
restore = Mock()
|
|
window.set_compact_close_handler(restore)
|
|
window.show()
|
|
|
|
def end_call() -> None:
|
|
if action == "hangup":
|
|
window.hangup()
|
|
else:
|
|
video_window.VideoCallWindow.close(SimpleNamespace(_session=window))
|
|
|
|
end_call()
|
|
|
|
restore.assert_not_called()
|
|
assert window._shutdown_requested is True
|
|
if wait_for_recording:
|
|
assert window.isVisible()
|
|
assert window._closing is False
|
|
window.lifecycle.end.assert_not_called()
|
|
window._release_webengine.assert_not_called()
|
|
window._abort_local_audio_recording.assert_not_called()
|
|
window._shutdown_timer.start.assert_called_once_with(190_000)
|
|
window._page.runJavaScript.assert_called_once()
|
|
assert "doctorConsultation?.close?.()" in window._page.runJavaScript.call_args.args[0]
|
|
|
|
end_call()
|
|
window.close()
|
|
window._page.runJavaScript.assert_called_once()
|
|
window._handle_bridge_message(
|
|
{"event": "local-audio-chunk", "sessionId": "recording", "sequence": 7, "data": "YQ=="}
|
|
)
|
|
window._append_local_audio_chunk.assert_called_once_with("recording", 7, "YQ==")
|
|
window._release_webengine.assert_not_called()
|
|
window._handle_bridge_message({"event": "hangup", "status": "local-hangup"})
|
|
else:
|
|
window._page.runJavaScript.assert_not_called()
|
|
window._shutdown_timer.start.assert_not_called()
|
|
reason = "desktop-hangup" if action == "hangup" else "window-closed"
|
|
window.lifecycle.end.assert_called_once_with(reason)
|
|
|
|
restore.assert_not_called()
|
|
assert not window.isVisible()
|
|
assert window._on_compact_close is None
|
|
assert window._closing is True
|
|
assert window._released is True
|
|
assert window._media_active is False
|
|
window._shutdown_timer.stop.assert_called_once_with()
|
|
window._abort_local_audio_recording.assert_called_once_with("")
|
|
window._release_webengine.assert_called_once_with()
|
|
window._begin_shutdown()
|
|
window._release_webengine.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"trigger",
|
|
["remote-hangup", "remote-error", "timeout", "load-failed", "load-failed-active", "bridge-missing"],
|
|
)
|
|
def test_terminal_paths_close_compact_window_without_restoring(
|
|
make_window: Callable[..., _CallWindow], trigger: str
|
|
) -> None:
|
|
window = make_window()
|
|
restore = Mock()
|
|
window.set_compact_close_handler(restore)
|
|
window.show()
|
|
|
|
if trigger == "remote-hangup":
|
|
window._handle_bridge_message({"event": "hangup", "status": "remote-hangup"})
|
|
window.call_ended.emit.assert_called_once_with("remote-hangup")
|
|
elif trigger == "remote-error":
|
|
window._handle_bridge_message({"event": "error", "message": "connection lost"})
|
|
window.call_error.emit.assert_called_once_with("connection lost")
|
|
elif trigger == "timeout":
|
|
window.hangup()
|
|
window._force_requested_shutdown()
|
|
window.lifecycle.end.assert_any_call("desktop-hangup-timeout")
|
|
elif trigger.startswith("load-failed"):
|
|
if trigger == "load-failed":
|
|
window._injected = False
|
|
window._start_requested = False
|
|
window._on_load_finished(False)
|
|
assert window._close_reason == "page-load-failed"
|
|
if trigger == "load-failed-active":
|
|
assert window._closing is False
|
|
window._release_webengine.assert_not_called()
|
|
restore.assert_not_called()
|
|
window._force_requested_shutdown()
|
|
window.lifecycle.end.assert_any_call("page-load-failed-timeout")
|
|
else:
|
|
window._after_injection(False)
|
|
assert window._close_reason == "bridge-api-missing"
|
|
|
|
restore.assert_not_called()
|
|
assert not window.isVisible()
|
|
assert window._closing is True
|
|
assert window._released is True
|
|
assert window._on_compact_close is None
|
|
window._release_webengine.assert_called_once_with()
|
|
window._shutdown_timer.stop.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"state", ["_shutdown_requested", "_companion_ended", "_closing", "_released"]
|
|
)
|
|
def test_close_event_never_restores_a_stopping_or_released_call(
|
|
make_window: Callable[..., _CallWindow], state: str
|
|
) -> None:
|
|
window = make_window(_start_requested=False, **{state: True})
|
|
restore = Mock()
|
|
window.set_compact_close_handler(restore)
|
|
event = QCloseEvent()
|
|
|
|
window.closeEvent(event)
|
|
|
|
restore.assert_not_called()
|
|
assert event.isAccepted()
|
|
|
|
|
|
def test_repeated_preview_refreshes_handler_and_ignores_old_dialog_generation(
|
|
make_window: Callable[..., _CallWindow],
|
|
) -> None:
|
|
window = make_window()
|
|
controller = _controller()
|
|
first_dialog, second_dialog, third_dialog = QDialog(), QDialog(), QDialog()
|
|
controller._show_video_preview(window, first_dialog)
|
|
first_handler = window._on_compact_close
|
|
original_state = controller._video_preview_state
|
|
|
|
controller._show_video_preview(window, second_dialog)
|
|
second_handler = window._on_compact_close
|
|
assert first_handler is not None
|
|
assert second_handler is not None
|
|
assert first_handler is not second_handler
|
|
assert controller._video_preview_state is original_state
|
|
first_dialog.finished.emit(0)
|
|
first_handler()
|
|
assert controller._video_preview_state is original_state
|
|
assert window._on_compact_close is second_handler
|
|
|
|
second_handler()
|
|
assert controller._video_preview_state is None
|
|
assert window._on_compact_close is None
|
|
|
|
controller._show_video_preview(window, third_dialog)
|
|
current_handler = window._on_compact_close
|
|
second_dialog.finished.emit(0)
|
|
second_handler()
|
|
assert controller._video_preview_state is not None
|
|
assert window._on_compact_close is current_handler
|
|
|
|
third_dialog.finished.emit(0)
|
|
assert controller._video_preview_state is None
|
|
assert window._on_compact_close is None
|
|
window.lifecycle.end.assert_not_called()
|
|
window._page.runJavaScript.assert_not_called()
|
|
|
|
|
|
def test_switching_preview_restores_old_window_and_isolates_old_callbacks(
|
|
make_window: Callable[..., _CallWindow],
|
|
) -> None:
|
|
first, second = make_window(), make_window()
|
|
controller = _controller()
|
|
first_dialog, second_dialog = QDialog(), QDialog()
|
|
first_geometry = QRect(first.geometry())
|
|
controller._show_video_preview(first, first_dialog)
|
|
old_handler = first._on_compact_close
|
|
|
|
controller._show_video_preview(second, second_dialog)
|
|
current_handler = second._on_compact_close
|
|
|
|
assert first._on_compact_close is None
|
|
assert first.geometry() == first_geometry
|
|
assert controller._video_preview_state["window"] is second
|
|
assert old_handler is not None
|
|
old_handler()
|
|
first_dialog.finished.emit(0)
|
|
assert controller._video_preview_state["window"] is second
|
|
assert second._on_compact_close is current_handler
|
|
|
|
second_dialog.finished.emit(0)
|
|
assert controller._video_preview_state is None
|
|
assert second._on_compact_close is None
|
|
first.lifecycle.end.assert_not_called()
|
|
second.lifecycle.end.assert_not_called()
|