150 lines
5.2 KiB
Python
150 lines
5.2 KiB
Python
"""设置页保存当前 Client 并后台登记的离屏测试。"""
|
|
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PyQt5.QtCore import QTimer
|
|
from PyQt5.QtTest import QTest
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
from src.mock_admin_gateway import MockAdminGateway
|
|
from src.settings_repository import SettingsRepository
|
|
from src.settings_ui import SettingsPage
|
|
|
|
|
|
class SlowMockAdminGateway(MockAdminGateway):
|
|
"""让登记停留一小段时间，用来证明主线程仍能处理事件。"""
|
|
|
|
def register_client(self, client, capabilities):
|
|
time.sleep(0.08)
|
|
return super().register_client(client, capabilities)
|
|
|
|
|
|
class SettingsPageEventTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self):
|
|
self.temp_directory = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.temp_directory.name) / "client.db"
|
|
self.repository = SettingsRepository(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.temp_directory.cleanup()
|
|
|
|
def _wait_until(self, predicate, timeout_ms=2000):
|
|
elapsed = 0
|
|
while not predicate() and elapsed < timeout_ms:
|
|
QTest.qWait(10)
|
|
elapsed += 10
|
|
self.assertTrue(predicate(), "等待异步操作超时")
|
|
|
|
def test_existing_client_info_is_restored_when_page_opens(self):
|
|
self.repository.set_many(
|
|
{
|
|
"admin.client_id": "CLIENT-EXISTING",
|
|
"admin.client_name": "办公室电脑",
|
|
}
|
|
)
|
|
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
)
|
|
|
|
self.assertEqual(page.deviceIdInput.text(), "CLIENT-EXISTING")
|
|
self.assertEqual(page.deviceNameInput.text(), "办公室电脑")
|
|
self.assertEqual(page.currentDeviceStatusLabel.text(), "本地设备信息已加载")
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_save_persists_locally_and_registers_with_mock(self):
|
|
gateway = MockAdminGateway()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=gateway,
|
|
)
|
|
page.deviceNameInput.setText(" 办公室电脑 ")
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self.assertFalse(page.currentDeviceSaveButton.isEnabled())
|
|
self._wait_until(lambda: gateway.registration_count == 1)
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
client_id = self.repository.get("admin.client_id")
|
|
self.assertTrue(client_id.startswith("CLIENT-"))
|
|
self.assertEqual(
|
|
self.repository.get("admin.client_name"), "办公室电脑"
|
|
)
|
|
self.assertEqual(page.deviceIdInput.text(), client_id)
|
|
self.assertEqual(
|
|
page.currentDeviceStatusLabel.text(),
|
|
"本地已保存，已登记到 Admin",
|
|
)
|
|
self.assertTrue(page.currentDeviceSaveButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_admin_failure_keeps_local_values_and_allows_retry(self):
|
|
gateway = MockAdminGateway()
|
|
gateway.fail_next_call_temporarily()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=gateway,
|
|
)
|
|
page.deviceNameInput.setText("仓库电脑")
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
self.assertTrue(self.repository.get("admin.client_id"))
|
|
self.assertEqual(self.repository.get("admin.client_name"), "仓库电脑")
|
|
self.assertIn("本地已保存，Admin 登记失败", page.currentDeviceStatusLabel.text())
|
|
self.assertIn("可再次点击保存重试", page.currentDeviceStatusLabel.text())
|
|
self.assertTrue(page.currentDeviceSaveButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_slow_registration_does_not_block_main_event_loop(self):
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=SlowMockAdminGateway(),
|
|
)
|
|
timer_fired = []
|
|
QTimer.singleShot(10, lambda: timer_fired.append(True))
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
self.assertTrue(timer_fired)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_shutdown_ignores_late_registration_result(self):
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=SlowMockAdminGateway(),
|
|
)
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(
|
|
lambda: "正在登记" in page.currentDeviceStatusLabel.text()
|
|
)
|
|
status_before_close = page.currentDeviceStatusLabel.text()
|
|
|
|
page.eventBinder.shutdown()
|
|
QTest.qWait(120)
|
|
|
|
self.assertEqual(page.currentDeviceStatusLabel.text(), status_before_close)
|
|
page.deleteLater()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|