492 lines
25 KiB
Python
492 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""患者查询错误分类、ZYT 会话保活节流与绑定窗口提示；只用合成 HTTP 和内存配置。
|
|
|
|
源码运行时 application_data_dir() 就是源码目录，请在临时副本里运行本测试。
|
|
"""
|
|
|
|
import copy
|
|
import io
|
|
import json
|
|
import os
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
from unittest import mock
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import backend_client as client # noqa: E402
|
|
|
|
ZYT_SESSION_PATH = "/adminapi/desktop/session"
|
|
|
|
|
|
class PatientRequestErrorMapping(unittest.TestCase):
|
|
def request_error(self, status, body, path=None):
|
|
error = urllib.error.HTTPError(
|
|
"https://synthetic.invalid/path", status, "synthetic", {},
|
|
io.BytesIO(json.dumps(body, ensure_ascii=False).encode("utf-8")))
|
|
with mock.patch.object(client.urllib.request, "urlopen", side_effect=error):
|
|
try:
|
|
client._request("GET", "https://synthetic.invalid", path or client.DESKTOP_PATIENT_SEARCH_PATH)
|
|
except client.BackendError as exc:
|
|
return exc
|
|
self.fail("expected synthetic HTTP error")
|
|
|
|
def test_ip_restricted_is_query_rejected_not_expiry_or_logout(self):
|
|
for path in (client.DESKTOP_PATIENT_SEARCH_PATH + "?keyword=synthetic", client.DESKTOP_PATIENT_BINDING_PATH,
|
|
client.DESKTOP_PATIENT_CONTEXT_PATH, client.DESKTOP_PATIENT_ORDER_SUMMARY_PATH):
|
|
with self.subTest(path=path):
|
|
error = self.request_error(
|
|
409, {"detail": {"code": "zyt_ip_restricted", "message": "synthetic ip rejected"}}, path)
|
|
self.assertIsInstance(error, client.PatientQueryRejected)
|
|
self.assertNotIsInstance(error, client.PatientSessionExpired)
|
|
self.assertNotIsInstance(error, client.AuthenticationError)
|
|
self.assertEqual(error.code, "zyt_ip_restricted")
|
|
self.assertEqual(error.status, 409)
|
|
self.assertEqual(str(error), "synthetic ip rejected")
|
|
|
|
def test_ip_restricted_without_message_uses_readable_default(self):
|
|
error = self.request_error(409, {"detail": {"code": "zyt_ip_restricted"}})
|
|
self.assertIsInstance(error, client.PatientQueryRejected)
|
|
self.assertNotIn("HTTP 409", str(error))
|
|
self.assertIn("登录 IP 校验", str(error))
|
|
self.assertNotIn("过期", str(error))
|
|
|
|
def test_order_routes_show_staff_message_instead_of_model_text(self):
|
|
# 订单接口的 message 是给模型的中性口径，客服界面要显示 staff_message 里的说明。
|
|
for code, kind in (("zyt_ip_restricted", client.PatientQueryRejected),
|
|
("zyt_session_expired", client.PatientSessionExpired)):
|
|
with self.subTest(code=code):
|
|
error = self.request_error(409, {"detail": {
|
|
"code": code, "message": "synthetic neutral", "staff_message": "synthetic staff"}},
|
|
client.DESKTOP_PATIENT_ORDER_SUMMARY_PATH)
|
|
self.assertIsInstance(error, kind)
|
|
self.assertEqual(str(error), "synthetic staff")
|
|
|
|
def test_session_expired_mapping_unchanged(self):
|
|
error = self.request_error(409, {"detail": {"code": "zyt_session_expired", "message": "synthetic expired"}})
|
|
self.assertIsInstance(error, client.PatientSessionExpired)
|
|
self.assertNotIsInstance(error, client.PatientQueryRejected)
|
|
self.assertEqual(error.code, "zyt_session_expired")
|
|
self.assertEqual(str(error), "synthetic expired")
|
|
|
|
def test_legacy_expiry_text_still_expired(self):
|
|
error = self.request_error(502, {"detail": "登录超时，请重新登录"})
|
|
self.assertIsInstance(error, client.PatientSessionExpired)
|
|
|
|
def test_patient_permission_denied_stays_generic_backend_error(self):
|
|
error = self.request_error(403, {"detail": {"code": "permission_denied", "message": "synthetic denied"}})
|
|
self.assertEqual(type(error), client.BackendError)
|
|
self.assertEqual(error.code, "permission_denied")
|
|
self.assertEqual(error.status, 403)
|
|
|
|
def test_desktop_revocation_elsewhere_stays_authentication_error(self):
|
|
for status in (401, 403):
|
|
with self.subTest(status=status):
|
|
error = self.request_error(status, {"detail": "synthetic revoked"}, client.DESKTOP_ME_PATH)
|
|
self.assertIsInstance(error, client.AuthenticationError)
|
|
self.assertNotIsInstance(error, client.PatientQueryRejected)
|
|
self.assertIsInstance(self.request_error(401, {"detail": "synthetic revoked"}), client.AuthenticationError)
|
|
|
|
|
|
class PatientKeepaliveThrottle(unittest.TestCase):
|
|
def setUp(self):
|
|
self.store = {
|
|
"server_url": "https://synthetic.invalid", "zyt_api_url": "https://identity.invalid",
|
|
"zyt_account": "synthetic-admin", "desktop_token_protected": "sealed:desktop-original",
|
|
"zyt_token_protected": "sealed:zyt-old", "desktop_expires_at": 5000,
|
|
"desktop_account": {"id": 7}, "last_error": "",
|
|
}
|
|
|
|
def save(settings):
|
|
self.store = copy.deepcopy(settings)
|
|
|
|
self.enterContext(mock.patch.object(client, "load_settings", side_effect=lambda: copy.deepcopy(self.store)))
|
|
self.enterContext(mock.patch.object(client, "save_settings", side_effect=save))
|
|
self.enterContext(mock.patch.object(client, "_protect_secret", side_effect=lambda value: "sealed:" + value))
|
|
self.enterContext(mock.patch.object(
|
|
client, "_unprotect_secret", side_effect=lambda value: str(value or "").removeprefix("sealed:")))
|
|
self.enterContext(mock.patch.object(client, "is_frozen", return_value=False))
|
|
self.enterContext(mock.patch.object(
|
|
client.urllib.request, "urlopen", side_effect=AssertionError("real network forbidden")))
|
|
self.request = self.enterContext(mock.patch.object(client, "_request", side_effect=self.response))
|
|
self.login = self.enterContext(mock.patch.object(client, "_zyt_login", side_effect=AssertionError("login forbidden")))
|
|
self.exchange = self.enterContext(mock.patch.object(
|
|
client, "_exchange_desktop_token", side_effect=AssertionError("exchange forbidden")))
|
|
self.logout = self.enterContext(mock.patch.object(client, "desktop_logout", side_effect=AssertionError("logout forbidden")))
|
|
self.clock = [1000.0]
|
|
self.enterContext(mock.patch.object(client.time, "monotonic", side_effect=lambda: self.clock[0]))
|
|
self.enterContext(mock.patch.object(client.time, "time", return_value=1000))
|
|
original = dict(client._PATIENT_KEEPALIVE_LAST)
|
|
client._PATIENT_KEEPALIVE_LAST.clear()
|
|
self.addCleanup(lambda: (client._PATIENT_KEEPALIVE_LAST.clear(), client._PATIENT_KEEPALIVE_LAST.update(original)))
|
|
self.me_failure = None
|
|
self.zyt_failure = None
|
|
self.zyt_result = {"code": 1, "data": {"status": "active"}}
|
|
self.seen_before_zyt_call = []
|
|
|
|
def response(self, method, server, path, **kwargs):
|
|
if path == client.DESKTOP_ME_PATH:
|
|
if self.me_failure:
|
|
raise self.me_failure
|
|
return 200, {"id": 7, "expires_at": 9000}
|
|
if path == ZYT_SESSION_PATH:
|
|
self.seen_before_zyt_call.append(dict(client._PATIENT_KEEPALIVE_LAST))
|
|
if self.zyt_failure:
|
|
raise self.zyt_failure
|
|
return 200, copy.deepcopy(self.zyt_result)
|
|
raise AssertionError("unexpected synthetic route: " + path)
|
|
|
|
def zyt_calls(self):
|
|
return sum(call.args[2] == ZYT_SESSION_PATH for call in self.request.call_args_list)
|
|
|
|
def advance(self, seconds):
|
|
self.clock[0] += seconds
|
|
|
|
def test_interval_is_shorter_than_the_thirty_minute_gui_timer(self):
|
|
self.assertLess(client.PATIENT_KEEPALIVE_INTERVAL_SECONDS, 30 * 60)
|
|
self.assertLess(client.PATIENT_KEEPALIVE_RETRY_SECONDS, client.PATIENT_KEEPALIVE_INTERVAL_SECONDS)
|
|
|
|
def test_every_thirty_minute_tick_really_calls_zyt(self):
|
|
for _ in range(4):
|
|
self.assertEqual(client.keepalive_zyt_session(), {"status": "active"})
|
|
self.advance(30 * 60)
|
|
self.assertEqual(self.zyt_calls(), 4)
|
|
self.login.assert_not_called()
|
|
self.exchange.assert_not_called()
|
|
self.logout.assert_not_called()
|
|
self.assertEqual(self.store["zyt_token_protected"], "sealed:zyt-old")
|
|
|
|
def test_extra_trigger_inside_interval_is_throttled(self):
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.advance(client.PATIENT_KEEPALIVE_INTERVAL_SECONDS - 1)
|
|
self.assertEqual(client.keepalive_zyt_session(), {"status": "skipped", "reason": "throttled"})
|
|
self.advance(1)
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.assertEqual(self.zyt_calls(), 2)
|
|
|
|
def test_throttle_is_recorded_only_after_the_call_finishes(self):
|
|
client.keepalive_zyt_session()
|
|
self.assertEqual(self.seen_before_zyt_call, [{}])
|
|
self.assertEqual(client._PATIENT_KEEPALIVE_LAST["wait"], client.PATIENT_KEEPALIVE_INTERVAL_SECONDS)
|
|
|
|
def test_zyt_network_failure_allows_retry_after_five_minutes(self):
|
|
self.zyt_failure = client.BackendError("synthetic offline")
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "unreachable")
|
|
self.advance(client.PATIENT_KEEPALIVE_RETRY_SECONDS - 1)
|
|
self.assertEqual(client.keepalive_zyt_session()["reason"], "throttled")
|
|
self.advance(1)
|
|
self.zyt_failure = None
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.assertEqual(self.zyt_calls(), 2)
|
|
self.advance(client.PATIENT_KEEPALIVE_RETRY_SECONDS)
|
|
self.assertEqual(client.keepalive_zyt_session()["reason"], "throttled")
|
|
|
|
def test_unclear_zyt_answer_is_retried_soon(self):
|
|
self.zyt_result = {"code": 0, "msg": "synthetic unknown"}
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "unreachable")
|
|
self.assertEqual(client._PATIENT_KEEPALIVE_LAST["wait"], client.PATIENT_KEEPALIVE_RETRY_SECONDS)
|
|
|
|
def test_backend_unreachable_before_zyt_allows_retry_after_five_minutes(self):
|
|
self.me_failure = client.BackendError("synthetic backend offline")
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "unreachable")
|
|
self.assertEqual(self.zyt_calls(), 0)
|
|
self.advance(client.PATIENT_KEEPALIVE_RETRY_SECONDS - 1)
|
|
self.assertEqual(client.keepalive_zyt_session()["reason"], "throttled")
|
|
self.advance(1)
|
|
self.me_failure = None
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
|
|
def test_definite_zyt_answers_use_full_interval(self):
|
|
cases = [({"code": -1, "msg": "登录超时，请重新登录"}, None, "expired"),
|
|
(None, client.AuthenticationError("synthetic denied", status=403), "forbidden"),
|
|
(None, client.BackendError("synthetic not deployed", status=404), "unsupported")]
|
|
for answer, failure, expected in cases:
|
|
with self.subTest(expected=expected):
|
|
client._PATIENT_KEEPALIVE_LAST.clear()
|
|
self.zyt_result, self.zyt_failure = answer, failure
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], expected)
|
|
self.advance(client.PATIENT_KEEPALIVE_RETRY_SECONDS)
|
|
self.assertEqual(client.keepalive_zyt_session()["reason"], "throttled")
|
|
self.logout.assert_not_called()
|
|
self.login.assert_not_called()
|
|
|
|
def test_desktop_invalid_does_not_consume_the_interval(self):
|
|
self.me_failure = client.AuthenticationError("synthetic revoked", status=401)
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "desktop_invalid")
|
|
self.me_failure = None
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.logout.assert_not_called()
|
|
|
|
def test_new_patient_token_is_not_throttled_by_old_one(self):
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.store["zyt_token_protected"] = "sealed:zyt-new"
|
|
self.assertEqual(client.keepalive_zyt_session()["status"], "active")
|
|
self.assertEqual(self.zyt_calls(), 2)
|
|
|
|
def test_session_change_during_call_is_skipped_and_not_recorded(self):
|
|
def response(method, server, path, **kwargs):
|
|
if path == ZYT_SESSION_PATH:
|
|
self.store["zyt_token_protected"] = "sealed:zyt-concurrently-new"
|
|
return 200, {"code": -1, "msg": "登录超时，请重新登录"}
|
|
return self.response(method, server, path, **kwargs)
|
|
self.request.side_effect = response
|
|
self.assertEqual(client.keepalive_zyt_session(), {"status": "skipped", "reason": "session_changed"})
|
|
self.assertEqual(client._PATIENT_KEEPALIVE_LAST, {})
|
|
self.assertEqual(self.store["zyt_token_protected"], "sealed:zyt-concurrently-new")
|
|
|
|
|
|
try:
|
|
from PySide6.QtWidgets import QApplication
|
|
import wechat_gui_qt as gui
|
|
except Exception as exc: # pragma: no cover - 没有 Qt 运行库时只跑上面的用例
|
|
gui = None
|
|
GUI_IMPORT_ERROR = repr(exc)
|
|
else:
|
|
GUI_IMPORT_ERROR = ""
|
|
|
|
|
|
def _context():
|
|
return {
|
|
"permissions": {"can_search": True, "can_bind": True, "can_unbind": True},
|
|
"conversation": {"name": "synthetic chat"},
|
|
"person": {"id": "person-1"},
|
|
"people": [{"id": "person-1", "display_name": "synthetic person", "patient_bindings": []}],
|
|
}
|
|
|
|
|
|
@unittest.skipIf(gui is None, f"PySide6 GUI unavailable: {GUI_IMPORT_ERROR}")
|
|
class PatientBindingDialogStates(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self):
|
|
self.dialog = gui.PatientBindingDialog(_context(), "account-1", "conversation-1")
|
|
self.addCleanup(self.dialog.deleteLater)
|
|
self.dialog.search_input.setText("synthetic keyword")
|
|
|
|
def button_shown(self):
|
|
return not self.dialog.reauthenticate_button.isHidden()
|
|
|
|
def wait_idle(self, seconds=5.0):
|
|
deadline = time.monotonic() + seconds
|
|
while time.monotonic() < deadline:
|
|
self.app.processEvents()
|
|
if not self.dialog._busy:
|
|
return
|
|
time.sleep(0.01)
|
|
self.fail("dialog stayed busy")
|
|
|
|
def search_through_worker(self, failure):
|
|
with mock.patch.object(client, "search_desktop_patients", side_effect=failure):
|
|
self.dialog._search()
|
|
self.wait_idle()
|
|
|
|
def test_expired_offers_reverification(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.assertTrue(self.button_shown())
|
|
self.assertIn("登录已过期", self.dialog.hint_label.text())
|
|
self.assertFalse(self.dialog.search_button.isEnabled())
|
|
self.assertFalse(self.dialog.bind_button.isEnabled())
|
|
|
|
def test_rejected_shows_backend_message_without_reverification(self):
|
|
message = "患者信息暂时查不到：synthetic IP 校验说明"
|
|
self.search_through_worker(client.PatientQueryRejected(message, status=409))
|
|
self.assertFalse(self.button_shown())
|
|
self.assertEqual(self.dialog.hint_label.text(), message)
|
|
self.assertNotIn("登录已过期", self.dialog.hint_label.text())
|
|
self.assertTrue(self.dialog.search_button.isEnabled())
|
|
|
|
def test_rejected_without_message_uses_fallback(self):
|
|
self.dialog._search_finished({"ok": False, "error": "", "patient_query_rejected": True})
|
|
self.assertIn("重新验证无法解决", self.dialog.hint_label.text())
|
|
self.assertFalse(self.button_shown())
|
|
|
|
def test_bind_rejected_is_not_expiry(self):
|
|
self.dialog._operation_finished({"ok": False, "action": "bind", "error": "synthetic rejected",
|
|
"patient_session_expired": False, "patient_query_rejected": True})
|
|
self.assertEqual(self.dialog.hint_label.text(), "synthetic rejected")
|
|
self.assertFalse(self.button_shown())
|
|
self.assertFalse(self.dialog.changed)
|
|
|
|
def test_expired_again_after_reverification_does_not_loop(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.assertTrue(self.button_shown())
|
|
with mock.patch.object(client, "search_desktop_patients",
|
|
side_effect=client.PatientSessionExpired(status=409)) as search:
|
|
self.dialog._reauthentication_finished({"ok": True, "context": _context()})
|
|
self.wait_idle()
|
|
search.assert_called_once_with("synthetic keyword")
|
|
text = self.dialog.hint_label.text()
|
|
self.assertIn("已重新验证 ZYT 登录", text)
|
|
self.assertIn("仍被 ZYT 拒绝", text)
|
|
self.assertNotIn("登录已过期", text)
|
|
self.assertFalse(self.button_shown())
|
|
# 不把人困住：搜索仍可用，稍后可以直接重试。
|
|
self.assertTrue(self.dialog.search_button.isEnabled())
|
|
self.assertFalse(self.dialog._patient_session_expired)
|
|
|
|
def test_successful_search_after_reverification(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
result = {"items": [{"patient_id": 9, "patient_name": "synthetic patient"}], "total": 1}
|
|
with mock.patch.object(client, "search_desktop_patients", return_value=result):
|
|
self.dialog._reauthentication_finished({"ok": True, "context": _context()})
|
|
self.wait_idle()
|
|
self.assertEqual(self.dialog.result_table.rowCount(), 1)
|
|
self.assertIn("找到 1 名患者", self.dialog.hint_label.text())
|
|
self.assertFalse(self.button_shown())
|
|
|
|
def test_failed_password_keeps_reverification_available(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.dialog._reauthentication_finished({"ok": False, "error": "synthetic wrong password", "verified": False})
|
|
self.assertIn("验证失败", self.dialog.hint_label.text())
|
|
self.assertTrue(self.button_shown())
|
|
self.assertTrue(self.dialog.reauthenticate_button.isEnabled())
|
|
|
|
def test_verified_but_context_rejected_is_not_reported_as_failed_verification(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.dialog._reauthentication_finished({"ok": False, "verified": True, "error": "synthetic ip rejected",
|
|
"patient_query_rejected": True})
|
|
self.assertEqual(self.dialog.hint_label.text(), "synthetic ip rejected")
|
|
self.assertFalse(self.button_shown())
|
|
self.assertTrue(self.dialog.search_button.isEnabled())
|
|
|
|
def test_verified_but_context_expired_shows_still_rejected(self):
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.dialog._reauthentication_finished({"ok": False, "verified": True, "error": "synthetic expired",
|
|
"patient_session_expired": True})
|
|
self.assertIn("仍被 ZYT 拒绝", self.dialog.hint_label.text())
|
|
self.assertNotIn("验证失败", self.dialog.hint_label.text())
|
|
self.assertFalse(self.button_shown())
|
|
|
|
def reauthenticate_through_worker(self, context_failure):
|
|
"""点“重新验证”走真实的工作线程和信号；只替换密码框和两次后台调用。"""
|
|
self.search_through_worker(client.PatientSessionExpired(status=409))
|
|
self.assertTrue(self.button_shown())
|
|
with mock.patch.object(gui.QInputDialog, "getText", return_value=("synthetic password", True)), \
|
|
mock.patch.object(client, "load_settings", return_value={"zyt_account": "synthetic-admin"}), \
|
|
mock.patch.object(client, "reauthenticate_patient_session", return_value={}) as reauthenticate, \
|
|
mock.patch.object(client, "desktop_patient_context", side_effect=context_failure) as context:
|
|
self.dialog._reauthenticate()
|
|
self.assertTrue(self.dialog._busy)
|
|
self.wait_idle()
|
|
reauthenticate.assert_called_once_with("synthetic password")
|
|
context.assert_called_once_with("account-1", "conversation-1", person_id="person-1")
|
|
|
|
def test_reauthenticate_worker_context_still_expired_stops_the_loop(self):
|
|
self.reauthenticate_through_worker(client.PatientSessionExpired(status=409))
|
|
text = self.dialog.hint_label.text()
|
|
self.assertIn("已重新验证 ZYT 登录", text)
|
|
self.assertIn("仍被 ZYT 拒绝", text)
|
|
self.assertNotIn("验证失败", text)
|
|
self.assertNotIn("登录已过期", text)
|
|
self.assertFalse(self.button_shown())
|
|
self.assertTrue(self.dialog.search_button.isEnabled())
|
|
self.assertTrue(self.dialog._reauthenticated)
|
|
self.assertFalse(self.dialog._patient_session_expired)
|
|
|
|
def test_reauthenticate_worker_context_rejected_shows_backend_message(self):
|
|
message = "患者信息暂时查不到：synthetic IP 校验说明"
|
|
self.reauthenticate_through_worker(client.PatientQueryRejected(message, status=409))
|
|
self.assertEqual(self.dialog.hint_label.text(), message)
|
|
self.assertFalse(self.button_shown())
|
|
self.assertTrue(self.dialog.search_button.isEnabled())
|
|
self.assertTrue(self.dialog._reauthenticated)
|
|
self.assertFalse(self.dialog._patient_session_expired)
|
|
|
|
|
|
class _KeepaliveHost:
|
|
"""只带 _patient_session_checked 用到的状态。"""
|
|
|
|
def __init__(self):
|
|
from PySide6.QtCore import QTimer
|
|
|
|
self._patient_keepalive_running = True
|
|
self._patient_session_state = "unknown"
|
|
self.patient_session_timer = QTimer()
|
|
self.patient_session_timer.setInterval(30 * 60 * 1000)
|
|
self.patient_session_timer.start()
|
|
self.patient_session_retry_timer = QTimer()
|
|
self.patient_session_retry_timer.setSingleShot(True)
|
|
self.patient_session_retry_timer.setInterval(6 * 60 * 1000)
|
|
self.logs = []
|
|
|
|
def append_log(self, text, tag=""):
|
|
self.logs.append((text, tag))
|
|
|
|
def checked(self, result):
|
|
gui.MainWindow._patient_session_checked(self, result)
|
|
|
|
|
|
@unittest.skipIf(gui is None, f"PySide6 GUI unavailable: {GUI_IMPORT_ERROR}")
|
|
class PatientKeepaliveLogging(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def test_unreachable_logs_once_and_schedules_one_retry(self):
|
|
host = _KeepaliveHost()
|
|
host.checked({"status": "unreachable", "message": "synthetic offline"})
|
|
host.checked({"status": "unreachable", "message": "synthetic offline"})
|
|
host.checked({"status": "unreachable"})
|
|
self.assertEqual(len(host.logs), 1)
|
|
self.assertIn("synthetic offline", host.logs[0][0])
|
|
self.assertEqual(host.logs[0][1], "warn")
|
|
self.assertTrue(host.patient_session_retry_timer.isActive())
|
|
self.assertFalse(host._patient_keepalive_running)
|
|
|
|
def test_periodic_failure_while_retry_pending_restarts_the_retry(self):
|
|
# 补查还在等时周期检查又失败：旧补查会落进这次失败后的 5 分钟节流里被跳过，
|
|
# 所以必须从这次失败重新计时。
|
|
host = _KeepaliveHost()
|
|
host.checked({"status": "unreachable", "message": "synthetic offline"})
|
|
host.patient_session_retry_timer = mock.Mock()
|
|
host.patient_session_retry_timer.isActive.return_value = True
|
|
host.checked({"status": "unreachable", "message": "synthetic offline"})
|
|
host.patient_session_retry_timer.start.assert_called_once_with()
|
|
host.patient_session_retry_timer.stop.assert_not_called()
|
|
self.assertEqual(len(host.logs), 1)
|
|
|
|
def test_recovery_stops_retry_restarts_period_and_logs_once(self):
|
|
host = _KeepaliveHost()
|
|
host.patient_session_timer = mock.Mock()
|
|
host.patient_session_timer.isActive.return_value = True
|
|
host.checked({"status": "unreachable", "message": "synthetic offline"})
|
|
host.patient_session_timer.start.assert_not_called()
|
|
host.checked({"status": "active"})
|
|
self.assertFalse(host.patient_session_retry_timer.isActive())
|
|
# 从这次真实请求重新计 30 分钟，下一次到点不会被 25 分钟节流挡掉。
|
|
host.patient_session_timer.start.assert_called_once_with()
|
|
host.checked({"status": "active"})
|
|
self.assertEqual([tag for _, tag in host.logs], ["warn", "ok"])
|
|
self.assertIn("恢复正常", host.logs[1][0])
|
|
|
|
def test_forbidden_and_unsupported_log_on_change_only(self):
|
|
host = _KeepaliveHost()
|
|
for status in ("forbidden", "forbidden", "unsupported", "unsupported", "active", "active"):
|
|
host.checked({"status": status, "message": f"synthetic {status}"})
|
|
self.assertEqual(len(host.logs), 3)
|
|
self.assertIn("synthetic forbidden", host.logs[0][0])
|
|
self.assertIn("synthetic unsupported", host.logs[1][0])
|
|
self.assertFalse(host.patient_session_retry_timer.isActive())
|
|
|
|
def test_first_active_and_skipped_are_silent(self):
|
|
host = _KeepaliveHost()
|
|
host.checked({"status": "skipped", "reason": "throttled"})
|
|
host.checked({"status": "active"})
|
|
self.assertEqual(host.logs, [])
|
|
self.assertEqual(host._patient_session_state, "active")
|
|
|
|
def test_expired_still_logged_once(self):
|
|
host = _KeepaliveHost()
|
|
host.checked({"status": "expired"})
|
|
host.checked({"status": "expired"})
|
|
host.checked({"status": "active"})
|
|
self.assertEqual([tag for _, tag in host.logs], ["warn", "ok"])
|
|
self.assertIn("已恢复", host.logs[1][0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|