327 lines
11 KiB
Python
327 lines
11 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 Qt, QTimer
|
|
from PyQt5.QtTest import QTest
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
from src.android_device_service import (
|
|
AndroidDevice,
|
|
AndroidDeviceSearchError,
|
|
)
|
|
from src.mock_admin_gateway import MockAdminGateway
|
|
from src.settings_repository import SettingsRepository
|
|
from src.settings_ui import AndroidDeviceRow, SettingsPage
|
|
|
|
|
|
class SlowMockAdminGateway(MockAdminGateway):
|
|
"""让登记停留一小段时间，用来证明主线程仍能处理事件。"""
|
|
|
|
def register_client(self, client, capabilities):
|
|
time.sleep(0.08)
|
|
return super().register_client(client, capabilities)
|
|
|
|
|
|
class SequenceAndroidDeviceService:
|
|
"""按顺序返回搜索结果，便于测试刷新行为。"""
|
|
|
|
def __init__(self, results):
|
|
self._results = list(results)
|
|
self.call_count = 0
|
|
|
|
def search(self, _is_cancelled=None):
|
|
result = self._results[min(self.call_count, len(self._results) - 1)]
|
|
self.call_count += 1
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return result
|
|
|
|
|
|
class SlowAndroidDeviceService:
|
|
"""模拟较慢的 ADB 搜索。"""
|
|
|
|
def __init__(self, delay=0.08):
|
|
self.delay = delay
|
|
self.call_count = 0
|
|
|
|
def search(self, _is_cancelled=None):
|
|
self.call_count += 1
|
|
time.sleep(self.delay)
|
|
return [AndroidDevice("USB-001", "USB", "Pixel 8", "14")]
|
|
|
|
|
|
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()
|
|
|
|
def test_search_updates_table_and_device_status(self):
|
|
service = SequenceAndroidDeviceService(
|
|
[[
|
|
AndroidDevice("USB-001", "USB", "Pixel 8", "14"),
|
|
AndroidDevice(
|
|
"192.168.0.173:5555",
|
|
"Wi-Fi",
|
|
status="offline",
|
|
),
|
|
]]
|
|
)
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self.assertFalse(page.searchButton.isEnabled())
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
model = page.deviceTableModel
|
|
self.assertEqual(model.rowCount(), 2)
|
|
self.assertEqual(model.device_for_serial("USB-001").model, "Pixel 8")
|
|
self.assertEqual(
|
|
model.device_for_serial("192.168.0.173:5555").status,
|
|
"offline",
|
|
)
|
|
offline_index = model.index(
|
|
model.row_for_serial("192.168.0.173:5555"), 0
|
|
)
|
|
self.assertFalse(
|
|
model.setData(offline_index, Qt.Checked, Qt.CheckStateRole)
|
|
)
|
|
self.assertEqual(
|
|
page.deviceStatusLabel.text(),
|
|
"搜索完成：找到 2 台 Android 设备",
|
|
)
|
|
self.assertTrue(page.searchButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_empty_search_keeps_placeholder_rows(self):
|
|
service = SequenceAndroidDeviceService([[]])
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertGreaterEqual(page.deviceTableModel.rowCount(), 1)
|
|
self.assertIn("未找到 Android 设备", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_search_failure_preserves_previous_rows(self):
|
|
service = SequenceAndroidDeviceService(
|
|
[AndroidDeviceSearchError("未找到 adb，请检查 PATH 后重试")]
|
|
)
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
page.set_android_devices(
|
|
[AndroidDeviceRow("USB-OLD", "USB", "Old Phone", "12")]
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial("USB-OLD")
|
|
)
|
|
self.assertIn("搜索失败：未找到 adb", page.deviceStatusLabel.text())
|
|
self.assertIn("设备列表未更改", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_search_preserves_or_clears_checked_device_by_serial(self):
|
|
connected = AndroidDevice("USB-001", "USB", "Pixel 8", "14")
|
|
service = SequenceAndroidDeviceService([[connected], [connected], []])
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
page.deviceTableModel.set_checked_serial("USB-001")
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, "USB-001")
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, "")
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_slow_search_does_not_block_or_start_twice(self):
|
|
service = SlowAndroidDeviceService()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
timer_fired = []
|
|
QTimer.singleShot(10, lambda: timer_fired.append(True))
|
|
|
|
page.searchButton.click()
|
|
page.eventBinder._request_search()
|
|
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertTrue(timer_fired)
|
|
self.assertEqual(service.call_count, 1)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_shutdown_ignores_late_search_result(self):
|
|
service = SlowAndroidDeviceService()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is not None)
|
|
status_before_close = page.deviceStatusLabel.text()
|
|
page.eventBinder.shutdown()
|
|
QTest.qWait(120)
|
|
|
|
self.assertEqual(page.deviceStatusLabel.text(), status_before_close)
|
|
self.assertIsNone(
|
|
page.deviceTableModel.device_for_serial("USB-001")
|
|
)
|
|
page.deleteLater()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|