From 45fde81b8b9b199282691dc67d750085450f9efe Mon Sep 17 00:00:00 2001 From: chengma Date: Fri, 7 Aug 2026 11:53:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20Android=20?= =?UTF-8?q?=E8=AE=BE=E5=A4=87=E6=90=9C=E7=B4=A2=20(#21)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/android_device_service.py | 159 ++++++++++++++++++ client/src/settings_ui.py | 3 + client/src/settings_ui_event.py | 169 +++++++++++++++++-- client/test/test_android_device_service.py | 109 +++++++++++++ client/test/test_settings_ui_event.py | 181 ++++++++++++++++++++- 5 files changed, 608 insertions(+), 13 deletions(-) create mode 100644 client/src/android_device_service.py create mode 100644 client/test/test_android_device_service.py diff --git a/client/src/android_device_service.py b/client/src/android_device_service.py new file mode 100644 index 0000000..0b663d3 --- /dev/null +++ b/client/src/android_device_service.py @@ -0,0 +1,159 @@ +"""通过 ADB 查询当前电脑已经识别的 Android 设备。""" + +from dataclasses import dataclass +import subprocess +from typing import Callable, List, Optional, Sequence + + +DEFAULT_ADB_TIMEOUT_SECONDS = 5.0 + + +class AndroidDeviceSearchError(RuntimeError): + """ADB 无法完成设备搜索时抛出的可读错误。""" + + +@dataclass(frozen=True) +class AndroidDevice: + """一台由 ``adb devices -l`` 返回的设备。""" + + serial: str + connection_type: str + model: str = "—" + android_version: str = "—" + status: str = "device" + + +CommandRunner = Callable[[Sequence[str], float], subprocess.CompletedProcess] + + +def run_adb_command( + command: Sequence[str], timeout_seconds: float +) -> subprocess.CompletedProcess: + """执行一条隐藏窗口的 ADB 命令，返回标准输出和错误输出。""" + + return subprocess.run( + list(command), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout_seconds, + check=False, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + + +class AndroidDeviceService: + """查询 ADB 设备列表；本类不依赖 Qt，也不创建 uiautomator2 设备。""" + + def __init__( + self, + command_runner: CommandRunner = run_adb_command, + timeout_seconds: float = DEFAULT_ADB_TIMEOUT_SECONDS, + ): + if timeout_seconds <= 0: + raise ValueError("ADB 超时时间必须大于 0") + self._run_command = command_runner + self._timeout_seconds = timeout_seconds + + def search( + self, is_cancelled: Optional[Callable[[], bool]] = None + ) -> List[AndroidDevice]: + """返回当前 ADB 设备；取消后停止补充设备属性。""" + + result = self._run(["adb", "devices", "-l"], "搜索设备") + devices = self.parse_devices(result.stdout or "") + + enriched = [] + for device in devices: + if is_cancelled is not None and is_cancelled(): + break + if device.status != "device": + enriched.append(device) + continue + + model = device.model + if model == "—": + model = self._read_property(device.serial, "ro.product.model") + if is_cancelled is not None and is_cancelled(): + break + android_version = self._read_property( + device.serial, "ro.build.version.release" + ) + enriched.append( + AndroidDevice( + serial=device.serial, + connection_type=device.connection_type, + model=model, + android_version=android_version, + status=device.status, + ) + ) + return enriched + + @staticmethod + def parse_devices(output: str) -> List[AndroidDevice]: + """解析 ``adb devices -l`` 输出，不依赖操作系统换行格式。""" + + devices = [] + for raw_line in output.splitlines(): + line = raw_line.strip() + if not line or line.startswith("List of devices") or line.startswith("*"): + continue + + parts = line.split() + if len(parts) < 2: + continue + + serial, status = parts[0], parts[1] + fields = {} + for token in parts[2:]: + if ":" not in token: + continue + key, value = token.split(":", 1) + fields[key] = value + + model = fields.get("model", "—").replace("_", " ") or "—" + devices.append( + AndroidDevice( + serial=serial, + connection_type="Wi-Fi" if ":" in serial else "USB", + model=model, + status=status, + ) + ) + return devices + + def _read_property(self, serial: str, property_name: str) -> str: + """读取单个属性；失败只返回占位符，不丢掉已经发现的设备。""" + + try: + result = self._run( + ["adb", "-s", serial, "shell", "getprop", property_name], + "读取设备信息", + ) + except AndroidDeviceSearchError: + return "—" + value = (result.stdout or "").strip() + return value or "—" + + def _run(self, command: Sequence[str], action: str): + try: + result = self._run_command(command, self._timeout_seconds) + except FileNotFoundError as exc: + raise AndroidDeviceSearchError( + "未找到 adb，请安装 Android platform-tools 并把 adb 加入 PATH 后重试" + ) from exc + except subprocess.TimeoutExpired as exc: + raise AndroidDeviceSearchError( + f"ADB {action}超时，请检查设备连接或重启 ADB 后重试" + ) from exc + except OSError as exc: + raise AndroidDeviceSearchError( + f"无法启动 adb：{str(exc) or '系统拒绝执行'}" + ) from exc + + if result.returncode != 0: + detail = (result.stderr or result.stdout or "未知错误").strip() + raise AndroidDeviceSearchError(f"ADB {action}失败：{detail}") + return result diff --git a/client/src/settings_ui.py b/client/src/settings_ui.py index f619d95..f59c00b 100644 --- a/client/src/settings_ui.py +++ b/client/src/settings_ui.py @@ -206,6 +206,7 @@ class SettingsPage(QWidget): parent=None, settings_repository=None, admin_gateway=None, + android_device_service=None, ): super().__init__(parent) self.setObjectName("settingsPage") @@ -218,6 +219,7 @@ class SettingsPage(QWidget): self, settings_repository=settings_repository, admin_gateway=admin_gateway, + android_device_service=android_device_service, ) def _build_ui(self) -> None: @@ -277,6 +279,7 @@ class SettingsPage(QWidget): header.setSectionResizeMode(5, QHeaderView.ResizeToContents) self.deviceStatusLabel = CaptionLabel("当前使用设备：未选择", self) + self.deviceStatusLabel.setAccessibleName("Android 设备搜索和选择状态") self.androidDeviceCard = self._build_android_device_card() content = QWidget(self) diff --git a/client/src/settings_ui_event.py b/client/src/settings_ui_event.py index 58b88a8..9e5a02c 100644 --- a/client/src/settings_ui_event.py +++ b/client/src/settings_ui_event.py @@ -1,13 +1,17 @@ -"""设置页事件、当前 Client 本地保存和后台登记。 +"""设置页事件、ADB 搜索、当前 Client 本地保存和后台登记。 -本文件不直接写 SQL。SQLite 保存和 Admin HTTP 请求由 Worker 在线程中执行， -后台结果只通过信号返回主线程更新页面。 +本文件不直接写 SQL。ADB、SQLite 保存和 Admin HTTP 请求由 Worker 在线程中 +执行，后台结果只通过信号返回主线程更新页面。 """ from typing import Optional from PyQt5.QtCore import QCoreApplication, QObject, QThread, pyqtSignal, pyqtSlot +from .android_device_service import ( + AndroidDeviceSearchError, + AndroidDeviceService, +) from .admin_gateway import ( AdminGatewayError, AndroidDeviceInfo, @@ -21,6 +25,7 @@ from .current_client_service import ( ) from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway from .settings_repository import SettingsRepository +from .settings_ui import AndroidDeviceRow DEVICE_ID_PLACEHOLDER = "待生成" @@ -89,6 +94,41 @@ class CurrentClientSaveWorker(QObject): self.completed.emit() +class AndroidDeviceSearchWorker(QObject): + """在后台线程查询 ADB，结果只通过信号返回主线程。""" + + succeeded = pyqtSignal(object) + failed = pyqtSignal(str) + completed = pyqtSignal() + + def __init__(self, service: AndroidDeviceService): + super().__init__() + self._service = service + self._cancelled = False + + def cancel(self) -> None: + """停止后续属性查询；正在执行的命令等待自身超时结束。""" + + self._cancelled = True + + @pyqtSlot() + def run(self) -> None: + try: + try: + devices = self._service.search(lambda: self._cancelled) + except AndroidDeviceSearchError as exc: + if not self._cancelled: + self.failed.emit(str(exc)) + except Exception as exc: + if not self._cancelled: + self.failed.emit(str(exc) or "无法搜索 Android 设备") + else: + if not self._cancelled: + self.succeeded.emit(devices) + finally: + self.completed.emit() + + class SettingsPageEventBinder(QObject): """绑定设备管理控件，并向应用层发出稳定事件。""" @@ -103,6 +143,7 @@ class SettingsPageEventBinder(QObject): page, settings_repository: Optional[SettingsRepository] = None, admin_gateway: Optional[ClientRegistrationGateway] = None, + android_device_service: Optional[AndroidDeviceService] = None, ): super().__init__(page) self._page = page @@ -111,6 +152,12 @@ class SettingsPageEventBinder(QObject): self._closing = False self._thread: Optional[QThread] = None self._worker: Optional[CurrentClientSaveWorker] = None + self._search_busy = False + self._search_thread: Optional[QThread] = None + self._search_worker: Optional[AndroidDeviceSearchWorker] = None + self._android_device_service = ( + android_device_service or AndroidDeviceService() + ) repository = settings_repository or SettingsRepository() self._client_service = CurrentClientService(repository) @@ -135,6 +182,7 @@ class SettingsPageEventBinder(QObject): self._start_save_current_device ) page.searchButton.clicked.connect(self._request_search) + self.searchRequested.connect(self._start_search) page.connectButton.clicked.connect(self._request_connect) page.saveButton.clicked.connect(self._request_save) page.deleteButton.clicked.connect(self._request_delete) @@ -150,7 +198,7 @@ class SettingsPageEventBinder(QObject): @pyqtSlot() def _request_save_current_device(self) -> None: - if self._busy or self._current_device_busy: + if self._busy or self._current_device_busy or self._search_busy: return device_id = self._page.deviceIdInput.text().strip() @@ -162,7 +210,7 @@ class SettingsPageEventBinder(QObject): def _start_save_current_device( self, _displayed_device_id: str, device_name: str ) -> None: - if self._closing or self._current_device_busy: + if self._closing or self._current_device_busy or self._search_busy: return try: @@ -202,10 +250,40 @@ class SettingsPageEventBinder(QObject): @pyqtSlot() def _request_search(self) -> None: - if self._busy: + if self._busy or self._current_device_busy or self._search_busy: return self.searchRequested.emit() + @pyqtSlot() + def _start_search(self) -> None: + if ( + self._closing + or self._busy + or self._current_device_busy + or self._search_busy + ): + return + + self._search_busy = True + self._sync_button_state() + self._page.deviceStatusLabel.setText("正在搜索 Android 设备…") + + thread = QThread(self) + worker = AndroidDeviceSearchWorker(self._android_device_service) + worker.moveToThread(thread) + + thread.started.connect(worker.run) + worker.succeeded.connect(self._on_search_succeeded) + worker.failed.connect(self._on_search_failed) + worker.completed.connect(thread.quit) + worker.completed.connect(worker.deleteLater) + thread.finished.connect(self._on_search_thread_finished) + thread.finished.connect(thread.deleteLater) + + self._search_thread = thread + self._search_worker = worker + thread.start() + @pyqtSlot() def _request_connect(self) -> None: if self._busy: @@ -251,12 +329,23 @@ class SettingsPageEventBinder(QObject): serial = self._page.deviceTableModel.checked_serial has_address = bool(self._page.addressInput.text().strip()) self._page.currentDeviceSaveButton.setEnabled( - not self._busy and not self._current_device_busy + not self._busy + and not self._current_device_busy + and not self._search_busy + ) + device_commands_enabled = not self._busy and not self._search_busy + self._page.searchButton.setEnabled( + device_commands_enabled and not self._current_device_busy + ) + self._page.connectButton.setEnabled( + device_commands_enabled and has_address + ) + self._page.saveButton.setEnabled( + device_commands_enabled and bool(serial) + ) + self._page.deleteButton.setEnabled( + device_commands_enabled and bool(serial) ) - self._page.searchButton.setEnabled(not self._busy) - self._page.connectButton.setEnabled(not self._busy and has_address) - self._page.saveButton.setEnabled(not self._busy and bool(serial)) - self._page.deleteButton.setEnabled(not self._busy and bool(serial)) def set_busy(self, busy: bool, message: str = "") -> None: """切换界面忙碌状态，防止用户重复提交设备命令。""" @@ -285,6 +374,47 @@ class SettingsPageEventBinder(QObject): serial = self._page.deviceTableModel.checked_serial.strip() return AndroidDeviceInfo(serial) if serial else None + @pyqtSlot(object) + def _on_search_succeeded(self, devices) -> None: + if self._closing: + return + + rows = [ + AndroidDeviceRow( + serial=device.serial, + connection_type=device.connection_type, + model=device.model, + android_version=device.android_version, + status=device.status, + ) + for device in devices + ] + self._page.set_android_devices(rows) + if rows: + self._page.deviceStatusLabel.setText( + f"搜索完成：找到 {len(rows)} 台 Android 设备" + ) + else: + self._page.deviceStatusLabel.setText( + "未找到 Android 设备，请检查 USB 调试或无线连接后重试" + ) + + @pyqtSlot(str) + def _on_search_failed(self, message: str) -> None: + if self._closing: + return + self._page.deviceStatusLabel.setText( + f"搜索失败：{message}；设备列表未更改" + ) + + @pyqtSlot() + def _on_search_thread_finished(self) -> None: + self._search_worker = None + self._search_thread = None + self._search_busy = False + if not self._closing: + self._sync_button_state() + @pyqtSlot(str, str) def _on_local_saved(self, client_id: str, client_name: str) -> None: if self._closing: @@ -348,3 +478,20 @@ class SettingsPageEventBinder(QObject): if thread is not None and thread.isRunning(): thread.quit() thread.wait(4000) + + search_worker = self._search_worker + search_thread = self._search_thread + if search_worker is not None: + search_worker.cancel() + for signal, slot in ( + (search_worker.succeeded, self._on_search_succeeded), + (search_worker.failed, self._on_search_failed), + ): + try: + signal.disconnect(slot) + except TypeError: + pass + + if search_thread is not None and search_thread.isRunning(): + search_thread.quit() + search_thread.wait(6000) diff --git a/client/test/test_android_device_service.py b/client/test/test_android_device_service.py new file mode 100644 index 0000000..fcef745 --- /dev/null +++ b/client/test/test_android_device_service.py @@ -0,0 +1,109 @@ +"""ADB Android 设备搜索服务测试。""" + +import subprocess +import unittest + +from src.android_device_service import ( + AndroidDeviceSearchError, + AndroidDeviceService, +) + + +def completed(command, stdout="", stderr="", returncode=0): + """创建一条假的 ADB 命令结果。""" + + return subprocess.CompletedProcess(command, returncode, stdout, stderr) + + +class AndroidDeviceServiceTest(unittest.TestCase): + def test_parse_usb_wifi_and_unavailable_devices(self): + output = """List of devices attached +USB-001 device product:p model:Pixel_8 device:husky transport_id:1 +192.168.0.173:5555 offline transport_id:2 +USB-002 unauthorized usb:1-2 transport_id:3 +""" + + devices = AndroidDeviceService.parse_devices(output) + + self.assertEqual(len(devices), 3) + self.assertEqual(devices[0].connection_type, "USB") + self.assertEqual(devices[0].model, "Pixel 8") + self.assertEqual(devices[1].connection_type, "Wi-Fi") + self.assertEqual(devices[1].status, "offline") + self.assertEqual(devices[2].status, "unauthorized") + + def test_search_reads_version_only_for_connected_device(self): + commands = [] + + def runner(command, _timeout): + commands.append(list(command)) + if command == ["adb", "devices", "-l"]: + return completed( + command, + "List of devices attached\n" + "USB-001 device model:Pixel_8 transport_id:1\n" + "192.168.0.173:5555 offline transport_id:2\n", + ) + return completed(command, "14\n") + + devices = AndroidDeviceService(runner).search() + + self.assertEqual(devices[0].android_version, "14") + self.assertEqual(devices[1].android_version, "—") + self.assertEqual(len(commands), 2) + self.assertIn("ro.build.version.release", commands[1]) + + def test_missing_model_is_read_from_device(self): + def runner(command, _timeout): + if command == ["adb", "devices", "-l"]: + return completed( + command, + "List of devices attached\nUSB-001 device transport_id:1\n", + ) + if command[-1] == "ro.product.model": + return completed(command, "Redmi K70\n") + return completed(command, "15\n") + + device = AndroidDeviceService(runner).search()[0] + + self.assertEqual(device.model, "Redmi K70") + self.assertEqual(device.android_version, "15") + + def test_property_failure_keeps_discovered_device(self): + def runner(command, _timeout): + if command == ["adb", "devices", "-l"]: + return completed( + command, + "List of devices attached\nUSB-001 device model:Phone\n", + ) + return completed(command, stderr="device disconnected", returncode=1) + + device = AndroidDeviceService(runner).search()[0] + + self.assertEqual(device.serial, "USB-001") + self.assertEqual(device.android_version, "—") + + def test_missing_adb_has_recovery_message(self): + def runner(_command, _timeout): + raise FileNotFoundError("adb") + + with self.assertRaisesRegex(AndroidDeviceSearchError, "未找到 adb"): + AndroidDeviceService(runner).search() + + def test_timeout_has_recovery_message(self): + def runner(command, timeout): + raise subprocess.TimeoutExpired(command, timeout) + + with self.assertRaisesRegex(AndroidDeviceSearchError, "搜索设备超时"): + AndroidDeviceService(runner).search() + + def test_nonzero_exit_includes_adb_error(self): + def runner(command, _timeout): + return completed(command, stderr="ADB server didn't ACK", returncode=1) + + with self.assertRaisesRegex(AndroidDeviceSearchError, "didn't ACK"): + AndroidDeviceService(runner).search() + + +if __name__ == "__main__": + unittest.main() diff --git a/client/test/test_settings_ui_event.py b/client/test/test_settings_ui_event.py index 4eddc9f..72aab08 100644 --- a/client/test/test_settings_ui_event.py +++ b/client/test/test_settings_ui_event.py @@ -8,13 +8,17 @@ from pathlib import Path os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") -from PyQt5.QtCore import QTimer +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 SettingsPage +from src.settings_ui import AndroidDeviceRow, SettingsPage class SlowMockAdminGateway(MockAdminGateway): @@ -25,6 +29,34 @@ class SlowMockAdminGateway(MockAdminGateway): 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): @@ -144,6 +176,151 @@ class SettingsPageEventTest(unittest.TestCase): 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()