"""设置页保存当前 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, AndroidDeviceConversionCancelled, AndroidDeviceSearchError, AndroidWifiConversionResult, ) from src.mock_admin_gateway import MockAdminGateway from src.selected_android_device_service import SELECTED_ANDROID_SERIAL_KEY from src.settings_repository import SettingsRepository from src.settings_ui import AndroidDeviceRow, SettingsPage from src.task_models import TaskType 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 self.restore_serials = [] 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 def restore_saved_device(self, serial, is_cancelled=None): self.restore_serials.append(serial) return self.search(is_cancelled) def is_package_installed(self, _serial, _package_name): return True 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")] def restore_saved_device(self, serial, is_cancelled=None): self.restore_serial = serial return self.search(is_cancelled) def is_package_installed(self, _serial, _package_name): return True class ControlledWifiConversionService: """模拟 USB 转 Wi-Fi 的进度、成功、失败和延迟。""" def __init__(self, delay=0.0, error=None): self.delay = delay self.error = error self.call_count = 0 self.search_count = 0 def search(self, _is_cancelled=None): self.search_count += 1 return [] def restore_saved_device(self, _serial, is_cancelled=None): return self.search(is_cancelled) def is_package_installed(self, _serial, _package_name): return True def convert_usb_to_wifi( self, usb_serial, is_cancelled=None, on_progress=None, ): self.call_count += 1 if on_progress is not None: on_progress("正在连接 Wi-Fi 设备 192.168.0.173:5555…") if self.delay: time.sleep(self.delay) if is_cancelled is not None and is_cancelled(): raise AndroidDeviceConversionCancelled() if self.error is not None: raise self.error self.last_usb_serial = usb_serial wifi_serial = "192.168.0.173:5555" return AndroidWifiConversionResult( wifi_serial, [ AndroidDevice(usb_serial, "USB", "Phone", "14"), AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14"), ], ) class ControlledPackageCheckService: """控制 PDD 包检测结果，其他 ADB 行为返回空列表。""" def __init__(self, installed=True, delay=0.0, error=None): self.installed = installed self.delay = delay self.error = error self.checked_serials = [] self.search_count = 0 def search(self, _is_cancelled=None): self.search_count += 1 return [] def restore_saved_device(self, _serial, _is_cancelled=None): return self.search(_is_cancelled) def is_package_installed(self, serial, _package_name): self.checked_serials.append(serial) if self.delay: time.sleep(self.delay) if self.error is not None: raise self.error if isinstance(self.installed, dict): return self.installed.get(serial, False) return self.installed class ControlledSettingsRepository(SettingsRepository): """控制 Android 设备设置写入速度和失败，其他设置保持正常。""" def __init__(self, db_path, delay=0.0): super().__init__(db_path) self.delay = delay self.fail_selected_write = False self.selected_write_count = 0 def set(self, setting_key, value, updated_at=None): if setting_key == SELECTED_ANDROID_SERIAL_KEY: self.selected_write_count += 1 if self.delay: time.sleep(self.delay) if self.fail_selected_write: raise OSError("模拟数据库写入失败") return super().set(setting_key, value, updated_at) def delete(self, setting_key): if setting_key == SELECTED_ANDROID_SERIAL_KEY: self.selected_write_count += 1 if self.delay: time.sleep(self.delay) if self.fail_selected_write: raise OSError("模拟数据库删除失败") return super().delete(setting_key) 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_settings_page_has_no_manual_live_purchase_authorization(self): page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), live_purchase_adapter_ready=True, ) self.assertFalse(hasattr(page, "livePurchaseCard")) self.assertFalse(hasattr(page, "livePurchaseEnableButton")) self.assertFalse(hasattr(page, "livePurchaseConfirmationInput")) page.eventBinder.shutdown() page.deleteLater() def test_ready_client_automatically_registers_live_capability(self): self.repository.set_many( { "admin.client_id": "CLIENT-LIVE", "admin.client_name": "采购电脑", SELECTED_ANDROID_SERIAL_KEY: "USB-LIVE", } ) gateway = MockAdminGateway() service = SequenceAndroidDeviceService([[]]) page = SettingsPage( settings_repository=self.repository, admin_gateway=gateway, android_device_service=service, live_purchase_adapter_ready=True, ) self._wait_until(lambda: page.eventBinder._search_thread is None) page.currentDeviceSaveButton.click() self._wait_until(lambda: gateway.registration_count == 1) self._wait_until(lambda: page.eventBinder._thread is None) registered = gateway.registered_client("CLIENT-LIVE") self.assertEqual(registered[1].purchase_mode, "live") self.assertIn(TaskType.PURCHASE, registered[1].supported_types) 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() def test_selected_android_device_is_saved_to_sqlite(self): service = ControlledPackageCheckService() page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices( [AndroidDeviceRow("USB-001", "USB", "Pixel 8", "14")] ) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) changed_serials = [] page.eventBinder.androidDeviceConfigurationChanged.connect( changed_serials.append ) page.saveButton.click() self.assertFalse(page.saveButton.isEnabled()) self._wait_until( lambda: self.repository.get(SELECTED_ANDROID_SERIAL_KEY) == "USB-001" and page.eventBinder._android_setting_thread is None ) self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-001" ) self.assertEqual( page.deviceStatusLabel.text(), "当前使用设备：USB-001（已保存）", ) self.assertTrue(page.deleteButton.isEnabled()) self.assertEqual(changed_serials, ["USB-001"]) page.eventBinder.shutdown() page.deleteLater() def test_startup_without_saved_device_does_not_search(self): service = SequenceAndroidDeviceService([[]]) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) QTest.qWait(30) self.assertEqual(service.call_count, 0) page.eventBinder.shutdown() page.deleteLater() def test_saved_usb_device_is_automatically_searched_on_startup(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001") service = SequenceAndroidDeviceService( [[AndroidDevice("USB-001", "USB", "Pixel 8", "14")]] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual(page.eventBinder._saved_android_serial, "USB-001") self.assertEqual(service.restore_serials, ["USB-001"]) self.assertEqual(page.deviceTableModel.checked_serial, "USB-001") self.assertIn("恢复完成", page.deviceStatusLabel.text()) self.assertTrue(page.deleteButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_saved_wifi_device_is_reconnected_and_selected_on_startup(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [[AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14")]] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual(service.restore_serials, [wifi_serial]) self.assertEqual(page.deviceTableModel.checked_serial, wifi_serial) self.assertIn("恢复完成", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_manual_search_reconnects_saved_wifi_before_refresh(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [ [AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14")], [AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14")], ] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.searchButton.click() self._wait_until(lambda: service.call_count == 2) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual(service.restore_serials, [wifi_serial, wifi_serial]) self.assertEqual(page.deviceTableModel.checked_serial, wifi_serial) self.assertIn("搜索完成", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_manual_wifi_reconnect_failure_still_lists_usb_device(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [ [AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14")], AndroidDeviceSearchError("目标设备拒绝连接"), [AndroidDevice("USB-001", "USB", "Pixel 8", "14")], ] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.searchButton.click() self._wait_until(lambda: service.call_count == 3) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), wifi_serial, ) self.assertIsNotNone( page.deviceTableModel.device_for_serial("USB-001") ) self.assertEqual(page.deviceTableModel.checked_serial, "") self.assertIn("找到 1 台 Android 设备", page.deviceStatusLabel.text()) self.assertIn("恢复失败", page.deviceStatusLabel.text()) self.assertIn("配置已保留", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_startup_restore_failure_keeps_saved_configuration(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [AndroidDeviceSearchError("目标设备拒绝连接"), []] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 2) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), wifi_serial, ) self.assertIn("自动恢复未完成", page.deviceStatusLabel.text()) self.assertIn("未找到其他 Android 设备", page.deviceStatusLabel.text()) self.assertIn("配置已保留", page.deviceStatusLabel.text()) self.assertTrue(page.searchButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_search_failure_after_restore_failure_preserves_previous_rows(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [ [AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14")], AndroidDeviceSearchError("目标设备拒绝连接"), AndroidDeviceSearchError("adb devices 执行失败"), ] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.searchButton.click() self._wait_until(lambda: service.call_count == 3) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertIsNotNone( page.deviceTableModel.device_for_serial(wifi_serial) ) self.assertEqual(page.deviceTableModel.checked_serial, wifi_serial) self.assertIn("搜索失败", page.deviceStatusLabel.text()) self.assertIn("adb devices 执行失败", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_saved_wifi_waiting_for_authorization_has_clear_message(self): wifi_serial = "192.168.0.173:5555" self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial) service = SequenceAndroidDeviceService( [[AndroidDevice(wifi_serial, "Wi-Fi", status="unauthorized")]] ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.call_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) self.assertEqual(page.deviceTableModel.checked_serial, "") self.assertIn("尚未授权", page.deviceStatusLabel.text()) self.assertIn("请在手机上允许调试", page.deviceStatusLabel.text()) self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), wifi_serial, ) page.eventBinder.shutdown() page.deleteLater() def test_slow_startup_restore_keeps_main_thread_responsive(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001") service = SlowAndroidDeviceService() timer_fired = [] QTimer.singleShot(10, lambda: timer_fired.append(True)) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) 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.restore_serial, "USB-001") self.assertEqual(page.deviceTableModel.checked_serial, "USB-001") page.eventBinder.shutdown() page.deleteLater() def test_shutdown_ignores_late_startup_restore_result(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001") service = SlowAndroidDeviceService() page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) 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() def test_search_auto_selects_saved_online_device(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001") service = SequenceAndroidDeviceService( [[AndroidDevice("USB-001", "USB", "Pixel 8", "14")]] ) 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.assertEqual(page.deviceTableModel.checked_serial, "USB-001") self.assertIn("已自动选择", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_search_keeps_saved_device_when_it_is_offline(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001") 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.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-001" ) self.assertEqual(page.deviceTableModel.checked_serial, "") self.assertIn("已保存设备当前未连接", page.deviceStatusLabel.text()) self.assertIn("配置已保留", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_selected_device_shows_installed_pdd_package(self): service = ControlledPackageCheckService(installed=True) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) self.assertEqual(service.checked_serials, ["USB-001"]) self.assertEqual( page.pddAppStatusLabel.text(), "com.xunmeng.pinduoduo（已安装）", ) self.assertFalse(hasattr(page, "addressInput")) self.assertFalse(hasattr(page, "connectButton")) page.eventBinder.shutdown() page.deleteLater() def test_save_rechecks_pdd_and_blocks_device_without_app(self): service = ControlledPackageCheckService(installed=False) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.saveButton.click() self._wait_until( lambda: len(service.checked_serials) == 2 and page.eventBinder._pdd_check_thread is None ) self.assertIsNone( self.repository.get(SELECTED_ANDROID_SERIAL_KEY) ) self.assertIn("未安装", page.pddAppStatusLabel.text()) self.assertIn("保存失败", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_save_blocks_when_pdd_check_fails(self): service = ControlledPackageCheckService( error=AndroidDeviceSearchError("device offline") ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.saveButton.click() self._wait_until( lambda: len(service.checked_serials) == 2 and page.eventBinder._pdd_check_thread is None ) self.assertIsNone( self.repository.get(SELECTED_ANDROID_SERIAL_KEY) ) self.assertIn("检测失败", page.pddAppStatusLabel.text()) self.assertIn("原配置未更改", page.deviceStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_device_change_ignores_stale_pdd_check_result(self): service = ControlledPackageCheckService( installed={"USB-OLD": True, "USB-NEW": False}, delay=0.05, ) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices( [ AndroidDeviceRow("USB-OLD", "USB"), AndroidDeviceRow("USB-NEW", "USB"), ] ) page.deviceTableModel.set_checked_serial("USB-OLD") self._wait_until(lambda: page.eventBinder._pdd_check_busy) page.deviceTableModel.set_checked_serial("USB-NEW") self._wait_until( lambda: service.checked_serials == ["USB-OLD", "USB-NEW"] and page.eventBinder._pdd_check_thread is None ) self.assertEqual(page.deviceTableModel.checked_serial, "USB-NEW") self.assertIn("未安装", page.pddAppStatusLabel.text()) page.eventBinder.shutdown() page.deleteLater() def test_selected_android_device_save_failure_keeps_old_configuration(self): repository = ControlledSettingsRepository(self.db_path) repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD") repository.fail_selected_write = True service = ControlledPackageCheckService() page = SettingsPage( settings_repository=repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.search_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.set_android_devices( [ AndroidDeviceRow("USB-OLD", "USB"), AndroidDeviceRow("USB-NEW", "USB"), ] ) page.deviceTableModel.set_checked_serial("USB-NEW") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.saveButton.click() self._wait_until( lambda: "原设备配置未更改" in page.deviceStatusLabel.text() and page.eventBinder._android_setting_thread is None ) self.assertEqual( repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-OLD" ) self.assertEqual(page.deviceTableModel.checked_serial, "USB-NEW") self.assertIn("原设备配置未更改", page.deviceStatusLabel.text()) self.assertTrue(page.saveButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_delete_clears_saved_offline_device_only(self): self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OFFLINE") self.repository.set("android.pdd_package", "com.xunmeng.pinduoduo") page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), ) page.deleteButton.click() self._wait_until( lambda: page.eventBinder._android_setting_thread is None ) self.assertIsNone( self.repository.get(SELECTED_ANDROID_SERIAL_KEY) ) self.assertEqual( self.repository.get("android.pdd_package"), "com.xunmeng.pinduoduo", ) self.assertIn("本地配置已删除", page.deviceStatusLabel.text()) self.assertFalse(page.deleteButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_slow_android_device_save_does_not_block_or_start_twice(self): repository = ControlledSettingsRepository(self.db_path, delay=0.08) service = ControlledPackageCheckService() page = SettingsPage( settings_repository=repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) timer_fired = [] QTimer.singleShot(10, lambda: timer_fired.append(True)) page.saveButton.click() page.eventBinder._request_save() self._wait_until(lambda: bool(timer_fired), timeout_ms=500) self._wait_until( lambda: repository.selected_write_count == 1 and page.eventBinder._android_setting_thread is None ) self.assertTrue(timer_fired) self.assertEqual(repository.selected_write_count, 1) page.eventBinder.shutdown() page.deleteLater() def test_shutdown_ignores_late_android_device_save_result(self): repository = ControlledSettingsRepository(self.db_path, delay=0.08) service = ControlledPackageCheckService() page = SettingsPage( settings_repository=repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.saveButton.click() self._wait_until( lambda: page.eventBinder._android_setting_thread is not None ) status_before_close = page.deviceStatusLabel.text() page.eventBinder.shutdown() QTest.qWait(120) self.assertEqual(page.deviceStatusLabel.text(), status_before_close) page.deleteLater() def test_convert_wifi_button_requires_connected_usb_device(self): service = ControlledPackageCheckService() page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self.assertFalse(page.convertWifiButton.isEnabled()) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) self.assertTrue(page.convertWifiButton.isEnabled()) page.set_android_devices( [AndroidDeviceRow("192.168.0.173:5555", "Wi-Fi")] ) page.deviceTableModel.set_checked_serial("192.168.0.173:5555") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) self.assertFalse(page.convertWifiButton.isEnabled()) page.set_android_devices( [AndroidDeviceRow("USB-OFFLINE", "USB", status="offline")] ) page.deviceTableModel.set_checked_serial("USB-OFFLINE") self.assertFalse(page.convertWifiButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_convert_wifi_refreshes_and_checks_wifi_without_saving(self): service = ControlledWifiConversionService() self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD") page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.search_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.convertWifiButton.click() self._wait_until( lambda: page.eventBinder._wifi_conversion_thread is None ) self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) self.assertEqual(service.last_usb_serial, "USB-001") self.assertEqual( page.deviceTableModel.checked_serial, "192.168.0.173:5555", ) self.assertIsNotNone( page.deviceTableModel.device_for_serial("USB-001") ) self.assertIsNotNone( page.deviceTableModel.device_for_serial( "192.168.0.173:5555" ) ) self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-OLD", ) self.assertIn("已连接并勾选", page.deviceStatusLabel.text()) self.assertIn("保存，再拔掉 USB", page.deviceStatusLabel.text()) self.assertTrue(page.saveButton.isEnabled()) self.assertFalse(page.convertWifiButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_convert_wifi_failure_preserves_list_and_saved_configuration(self): service = ControlledWifiConversionService( error=AndroidDeviceSearchError("Wi-Fi 连接验证失败") ) self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD") page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) self._wait_until(lambda: service.search_count == 1) self._wait_until(lambda: page.eventBinder._search_thread is None) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.convertWifiButton.click() self._wait_until( lambda: page.eventBinder._wifi_conversion_thread is None ) self.assertIsNotNone( page.deviceTableModel.device_for_serial("USB-001") ) self.assertEqual(page.deviceTableModel.checked_serial, "USB-001") self.assertEqual( self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-OLD", ) self.assertIn("转换失败", page.deviceStatusLabel.text()) self.assertIn("可重试", page.deviceStatusLabel.text()) self.assertTrue(page.convertWifiButton.isEnabled()) page.eventBinder.shutdown() page.deleteLater() def test_slow_convert_wifi_does_not_block_or_start_twice(self): service = ControlledWifiConversionService(delay=0.08) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) timer_fired = [] QTimer.singleShot(10, lambda: timer_fired.append(True)) page.convertWifiButton.click() page.eventBinder._request_convert_wifi() self._wait_until(lambda: bool(timer_fired), timeout_ms=500) self._wait_until( lambda: page.eventBinder._wifi_conversion_thread is None ) self.assertTrue(timer_fired) self.assertEqual(service.call_count, 1) page.eventBinder.shutdown() page.deleteLater() def test_shutdown_ignores_late_wifi_conversion_result(self): service = ControlledWifiConversionService(delay=0.08) page = SettingsPage( settings_repository=self.repository, admin_gateway=MockAdminGateway(), android_device_service=service, ) page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) page.deviceTableModel.set_checked_serial("USB-001") self._wait_until(lambda: page.eventBinder._pdd_check_thread is None) page.convertWifiButton.click() self._wait_until( lambda: page.eventBinder._wifi_conversion_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.assertEqual(page.deviceTableModel.checked_serial, "USB-001") page.deleteLater() if __name__ == "__main__": unittest.main()