278 lines
12 KiB
Python
278 lines
12 KiB
Python
"""Appointment policy must survive repository, launcher and lifecycle boundaries."""
|
|
import logging
|
|
from types import MethodType, SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from doctor_workstation.core.models import CallTicket
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.video.launcher import normalize_backend_ticket
|
|
from doctor_workstation.video.lifecycle import OrderedCallLifecycle
|
|
|
|
|
|
def ticket(**policy):
|
|
return CallTicket.from_dict({
|
|
"sdkAppId": 1400123456, "userId": "doctor_1", "userSig": "test-ticket",
|
|
"patientUserId": "patient_8", "diagnosis_id": 123, "patient_id": 8,
|
|
"appointment_id": 456, "appointment_type": "text", **policy,
|
|
})
|
|
|
|
|
|
@pytest.mark.parametrize("value", [None, False, 0, 1, "true", "1", []])
|
|
def test_only_actual_server_true_authorizes_calls(value):
|
|
request = normalize_backend_ticket(ticket(can_video_call=value, can_audio_call=value))
|
|
assert request.appointment_id == 456
|
|
assert request.appointment_type == "text"
|
|
assert request.patient_id == 8
|
|
assert request.to_web_config()["can_video_call"] is False
|
|
assert request.to_web_config()["can_audio_call"] is False
|
|
|
|
|
|
def test_raw_policy_preserved_and_missing_policy_denies():
|
|
request = normalize_backend_ticket(ticket())
|
|
assert not request.can_video_call and not request.can_audio_call
|
|
request = normalize_backend_ticket(ticket(appointment_type="phone", can_audio_call=True))
|
|
assert request.can_audio_call and not request.can_video_call
|
|
|
|
|
|
def test_repository_forwards_exact_appointment_and_media_type():
|
|
class Client:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def post(self, endpoint, payload):
|
|
self.calls.append((endpoint, payload))
|
|
return {"call_record_id": 99} if endpoint.endswith("startCall") else ticket().raw
|
|
|
|
client = Client()
|
|
repository = RemoteDoctorRepository(client)
|
|
assert repository.get_call_ticket(8, 123, appointment_id=456).raw["appointment_type"] == "text"
|
|
repository.start_call(123, 8, call_type=1, appointment_id=456)
|
|
assert client.calls == [
|
|
("tcm.diagnosis/getCallSignature", {"patient_id": 8, "diagnosis_id": 123, "appointment_id": 456}),
|
|
("tcm.diagnosis/startCall", {"patient_id": 8, "diagnosis_id": 123, "appointment_id": 456, "call_type": 1}),
|
|
]
|
|
|
|
|
|
def test_lifecycle_refreshes_exact_identity_and_uses_actual_media():
|
|
class Repository:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.appointment_id = 456
|
|
|
|
def get_call_ticket(self, **payload):
|
|
self.calls.append(payload)
|
|
return ticket(appointment_id=self.appointment_id, can_audio_call=True)
|
|
|
|
def start_call(self, **payload):
|
|
self.calls.append(payload)
|
|
return {"call_record_id": 99}
|
|
|
|
def end_call(self, **payload):
|
|
return {}
|
|
|
|
repository = Repository()
|
|
lifecycle = OrderedCallLifecycle(normalize_backend_ticket(ticket()), repository, logging.getLogger(__name__))
|
|
try:
|
|
assert lifecycle.refresh_call_policy().result(timeout=2)["can_audio_call"] is True
|
|
assert repository.calls[-1] == {"patient_id": 8, "diagnosis_id": 123, "appointment_id": 456}
|
|
assert lifecycle.start(call_type=1).result(timeout=2)
|
|
assert repository.calls[-1]["call_type"] == 1
|
|
assert repository.calls[-1]["appointment_id"] == 456
|
|
repository.appointment_id = 457
|
|
with pytest.raises(ValueError, match="当前问诊已变更"):
|
|
lifecycle.refresh_call_policy().result(timeout=2)
|
|
finally:
|
|
lifecycle.end("test").result(timeout=2)
|
|
|
|
|
|
@pytest.fixture
|
|
def chat_controller(monkeypatch):
|
|
"""Exercise the real controller callbacks without launching Qt or Tencent."""
|
|
from doctor_workstation import app as app_module
|
|
|
|
queued = []
|
|
launched = []
|
|
previews = []
|
|
diagnoses = []
|
|
dialog = object()
|
|
|
|
def get_call_ticket(patient_id, diagnosis_id, *, appointment_id):
|
|
return ticket(patient_id=patient_id, diagnosis_id=diagnosis_id, appointment_id=appointment_id)
|
|
|
|
def launch_video_call(_ticket, **kwargs):
|
|
window = SimpleNamespace(
|
|
show=lambda: None,
|
|
raise_=lambda: None,
|
|
activateWindow=lambda: None,
|
|
destroyed=SimpleNamespace(connect=lambda _callback: None),
|
|
)
|
|
call = SimpleNamespace(open_im=kwargs["open_im"], qt_window=window, close=lambda: None)
|
|
launched.append((call, kwargs))
|
|
return call
|
|
|
|
def open_diagnosis_by_id(diagnosis_id, *, modeless):
|
|
diagnoses.append((diagnosis_id, modeless))
|
|
return dialog
|
|
|
|
monkeypatch.setattr(app_module, "run_async", lambda function, **callbacks: queued.append((function, callbacks)))
|
|
monkeypatch.setattr(app_module, "launch_video_call", launch_video_call)
|
|
monkeypatch.setattr(app_module, "show_toast", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(app_module, "_build_video_patient_case", lambda *_args, **_kwargs: {})
|
|
monkeypatch.setattr(app_module, "WEBENGINE_AVAILABLE", True)
|
|
monkeypatch.setattr(app_module, "QTimer", SimpleNamespace(singleShot=lambda _delay, callback: callback()))
|
|
controller = SimpleNamespace(
|
|
shell_window=SimpleNamespace(open_diagnosis_by_id=open_diagnosis_by_id),
|
|
current_repository=SimpleNamespace(get_call_ticket=get_call_ticket),
|
|
current_demo_mode=False,
|
|
video_calls={}, video_pending={}, demo_video_dialogs={}, _pending_im_request=None,
|
|
config=SimpleNamespace(video_mode="embedded", video_web_url=""),
|
|
_show_video_preview=lambda window, diagnosis_dialog: previews.append((window, diagnosis_dialog)),
|
|
)
|
|
for method in ("_request_video", "_launch_video", "_video_ticket_error", "_open_video_diagnosis"):
|
|
setattr(controller, method, MethodType(getattr(app_module.ApplicationController, method), controller))
|
|
|
|
def request(appointment_id, diagnosis_id=123):
|
|
controller._request_video({
|
|
"patient_id": 8, "diagnosis_id": diagnosis_id,
|
|
"appointment_id": appointment_id, "mode": "im",
|
|
})
|
|
|
|
def complete(index):
|
|
function, callbacks = queued[index]
|
|
callbacks["on_success"](function())
|
|
|
|
return SimpleNamespace(
|
|
controller=controller, request=request, complete=complete, queued=queued,
|
|
launched=launched, previews=previews, diagnoses=diagnoses, dialog=dialog,
|
|
)
|
|
|
|
|
|
def test_diagnosis_callback_previews_exact_appointment_session(chat_controller):
|
|
case = chat_controller
|
|
case.request(456)
|
|
case.complete(0)
|
|
call, launch_args = case.launched[0]
|
|
case.controller.video_calls["123:457"] = SimpleNamespace(qt_window=object())
|
|
|
|
launch_args["on_open_diagnosis"]()
|
|
assert case.diagnoses == [(123, True)]
|
|
assert case.previews == [(call.qt_window, case.dialog)]
|
|
|
|
case.controller.video_calls.pop("123:456")
|
|
launch_args["on_open_diagnosis"]()
|
|
assert len(case.previews) == 1 # A closed session cannot borrow another appointment's video.
|
|
|
|
|
|
@pytest.mark.parametrize("second_diagnosis", [123, 124])
|
|
@pytest.mark.parametrize("completion_order", [(0, 1), (1, 0)])
|
|
def test_latest_im_selection_retires_pending_callbacks(chat_controller, second_diagnosis, completion_order):
|
|
case = chat_controller
|
|
case.request(456)
|
|
case.request(457, diagnosis_id=second_diagnosis)
|
|
assert list(case.controller.video_pending) == [f"{second_diagnosis}:457"]
|
|
|
|
for index in completion_order:
|
|
case.complete(index)
|
|
|
|
assert list(case.controller.video_calls) == [f"{second_diagnosis}:457"]
|
|
assert len(case.launched) == 1
|
|
assert case.controller.video_pending == {}
|
|
assert case.controller._pending_im_request is None
|
|
|
|
|
|
def test_reselecting_open_im_also_retires_other_pending_selection(chat_controller):
|
|
case = chat_controller
|
|
case.request(456)
|
|
case.complete(0)
|
|
original_call = case.launched[0][0]
|
|
case.request(457)
|
|
# The previous page may still be finishing its native shutdown. Treat a
|
|
# reactivated current page like any other current selection.
|
|
case.controller.video_calls["123:456"] = original_call
|
|
case.request(456)
|
|
case.complete(1)
|
|
assert list(case.controller.video_calls) == ["123:456"]
|
|
assert case.controller.video_pending == {}
|
|
assert len(case.launched) == 1
|
|
|
|
|
|
def test_repeated_pending_im_selection_does_not_duplicate_request(chat_controller):
|
|
case = chat_controller
|
|
case.request(456)
|
|
case.request(456)
|
|
assert len(case.queued) == 1
|
|
case.complete(0)
|
|
assert list(case.controller.video_calls) == ["123:456"]
|
|
|
|
|
|
def legacy_signature_repository(**changes):
|
|
response = {
|
|
"sdkAppId": 1400123456, "userId": "doctor_1", "userSig": "test-ticket",
|
|
"patientUserId": "patient_8", "diagnosis_id": 123, "patient_id": 8,
|
|
**changes,
|
|
}
|
|
repository = RemoteDoctorRepository(SimpleNamespace(post=lambda *_args: response))
|
|
repository.patient_detail = lambda _diagnosis_id: {}
|
|
return repository, response
|
|
|
|
|
|
def test_legacy_signature_opens_im_with_requested_context_and_no_media(chat_controller):
|
|
case = chat_controller
|
|
repository, response = legacy_signature_repository()
|
|
case.controller.current_repository = repository
|
|
case.request(456)
|
|
case.complete(0)
|
|
assert list(case.controller.video_calls) == ["123:456"]
|
|
assert case.launched[0][1]["open_im"] is True
|
|
request = normalize_backend_ticket(repository.get_call_ticket(8, 123, appointment_id=456))
|
|
assert request.appointment_id == 456
|
|
assert request.target_user_id == "patient_8"
|
|
assert request.can_video_call is False and request.can_audio_call is False
|
|
assert request.call_disabled_reason
|
|
assert "appointment_id" not in response # Never mutate the HTTP response/shared cache.
|
|
lifecycle = OrderedCallLifecycle(request, repository, logging.getLogger(__name__))
|
|
try:
|
|
refreshed = lifecycle.refresh_call_policy().result(timeout=2)
|
|
assert refreshed["appointmentId"] == 456
|
|
assert refreshed["can_video_call"] is False and refreshed["can_audio_call"] is False
|
|
finally:
|
|
lifecycle.end("test").result(timeout=2)
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"diagnosis_id": 124}, {"patient_id": 9}, {"patientUserId": "patient_9"},
|
|
{"diagnosis_id": None}, {"patient_id": None}, {"patientUserId": None},
|
|
])
|
|
def test_legacy_signature_requires_full_matching_identity(changes):
|
|
repository, _ = legacy_signature_repository(**changes)
|
|
with pytest.raises(ValueError, match="患者或诊单不匹配"):
|
|
repository.get_call_ticket(8, 123, appointment_id=456)
|
|
|
|
|
|
@pytest.mark.parametrize("returned_id", [457, 0, None, "", True, -1, 456.9, 456.0, "bad"])
|
|
def test_explicit_wrong_or_invalid_appointment_is_never_replaced(returned_id):
|
|
repository, _ = legacy_signature_repository(appointment_id=returned_id)
|
|
with pytest.raises(ValueError, match="本次挂号不匹配"):
|
|
repository.get_call_ticket(8, 123, appointment_id=456)
|
|
|
|
|
|
@pytest.mark.parametrize("fragment", [
|
|
{"can_video_call": True}, {"can_audio_call": True}, {"appointment_type": "text"},
|
|
{"appointmentId": 457}, {"call_disabled_reason": ""},
|
|
])
|
|
def test_partial_policy_is_not_treated_as_legacy_signature(fragment):
|
|
repository, _ = legacy_signature_repository(**fragment)
|
|
with pytest.raises(ValueError, match="本次挂号不匹配"):
|
|
repository.get_call_ticket(8, 123, appointment_id=456)
|
|
|
|
|
|
@pytest.mark.parametrize("returned_id", [456, "456"])
|
|
def test_modern_exact_appointment_policy_is_preserved(returned_id):
|
|
repository, _ = legacy_signature_repository(
|
|
appointment_id=returned_id, appointment_type="video", can_video_call=True, can_audio_call=True,
|
|
)
|
|
request = normalize_backend_ticket(repository.get_call_ticket(8, 123, appointment_id=456))
|
|
assert request.appointment_id == 456
|
|
assert request.can_video_call is True and request.can_audio_call is True
|