From d478e690b6921dcc7ea6a4c400783f223a89f8a1 Mon Sep 17 00:00:00 2001 From: chengma Date: Fri, 7 Aug 2026 14:42:00 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=20USB=20=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E8=BD=AC=20Wi-Fi=20ADB=20(#25)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/android_device_service.py | 211 ++++++++++++++++++++ client/src/settings_ui.py | 9 + client/src/settings_ui_event.py | 221 ++++++++++++++++++++- client/test/test_android_device_service.py | 132 ++++++++++++ client/test/test_settings_ui_event.py | 171 ++++++++++++++++ 5 files changed, 741 insertions(+), 3 deletions(-) diff --git a/client/src/android_device_service.py b/client/src/android_device_service.py index 0b663d3..7bf8d6f 100644 --- a/client/src/android_device_service.py +++ b/client/src/android_device_service.py @@ -1,17 +1,25 @@ """通过 ADB 查询当前电脑已经识别的 Android 设备。""" from dataclasses import dataclass +import ipaddress import subprocess +import time from typing import Callable, List, Optional, Sequence DEFAULT_ADB_TIMEOUT_SECONDS = 5.0 +DEFAULT_USB_DISCONNECT_TIMEOUT_SECONDS = 60.0 +DEFAULT_USB_POLL_INTERVAL_SECONDS = 0.5 class AndroidDeviceSearchError(RuntimeError): """ADB 无法完成设备搜索时抛出的可读错误。""" +class AndroidDeviceConversionCancelled(RuntimeError): + """页面关闭后停止 USB 转 Wi-Fi 的后续步骤。""" + + @dataclass(frozen=True) class AndroidDevice: """一台由 ``adb devices -l`` 返回的设备。""" @@ -23,7 +31,17 @@ class AndroidDevice: status: str = "device" +@dataclass(frozen=True) +class AndroidWifiConversionResult: + """USB 转 Wi-Fi 成功后的目标设备号和最新设备列表。""" + + wifi_serial: str + devices: List[AndroidDevice] + + CommandRunner = Callable[[Sequence[str], float], subprocess.CompletedProcess] +Sleeper = Callable[[float], None] +ProgressCallback = Callable[[str], None] def run_adb_command( @@ -50,11 +68,23 @@ class AndroidDeviceService: self, command_runner: CommandRunner = run_adb_command, timeout_seconds: float = DEFAULT_ADB_TIMEOUT_SECONDS, + usb_disconnect_timeout_seconds: float = ( + DEFAULT_USB_DISCONNECT_TIMEOUT_SECONDS + ), + usb_poll_interval_seconds: float = DEFAULT_USB_POLL_INTERVAL_SECONDS, + sleeper: Sleeper = time.sleep, ): if timeout_seconds <= 0: raise ValueError("ADB 超时时间必须大于 0") + if usb_disconnect_timeout_seconds <= 0: + raise ValueError("等待拔出 USB 的超时时间必须大于 0") + if usb_poll_interval_seconds <= 0: + raise ValueError("USB 检查间隔必须大于 0") self._run_command = command_runner self._timeout_seconds = timeout_seconds + self._usb_disconnect_timeout_seconds = usb_disconnect_timeout_seconds + self._usb_poll_interval_seconds = usb_poll_interval_seconds + self._sleep = sleeper def search( self, is_cancelled: Optional[Callable[[], bool]] = None @@ -91,6 +121,110 @@ class AndroidDeviceService: ) return enriched + def convert_usb_to_wifi( + self, + usb_serial: str, + is_cancelled: Optional[Callable[[], bool]] = None, + on_progress: Optional[ProgressCallback] = None, + ) -> AndroidWifiConversionResult: + """开启 USB 设备的 5555 端口，等拔线后重连并返回最新列表。""" + + self._validate_usb_serial(usb_serial) + self._check_cancelled(is_cancelled) + self._report(on_progress, "正在读取勾选设备的 Wi-Fi 地址…") + route = self._run( + ["adb", "-s", usb_serial, "shell", "ip", "route"], + "读取 Wi-Fi 地址", + ) + ip_address = self.parse_wifi_ipv4(route.stdout or "") + wifi_serial = f"{ip_address}:5555" + + self._check_cancelled(is_cancelled) + self._report(on_progress, "正在开启 ADB 5555 端口…") + self._run( + ["adb", "-s", usb_serial, "tcpip", "5555"], + "开启 Wi-Fi 调试", + ) + + # 部分手机插线时就能建立 Wi-Fi 通道；该连接仅用于提前准备， + # 真正的结果仍以拔线后再次连接和验证为准。 + try: + self._connect_wifi(wifi_serial) + except AndroidDeviceSearchError: + pass + + self._report( + on_progress, + f"已开启 {wifi_serial}，请拔掉 USB 数据线…", + ) + self._wait_until_usb_disconnected(usb_serial, is_cancelled) + + self._check_cancelled(is_cancelled) + self._report(on_progress, "已检测到 USB 拔出，正在重新连接 Wi-Fi…") + self._connect_wifi(wifi_serial) + + self._check_cancelled(is_cancelled) + devices = self._list_devices() + wifi_device = next( + (device for device in devices if device.serial == wifi_serial), + None, + ) + if wifi_device is None or wifi_device.status != "device": + status = wifi_device.status if wifi_device is not None else "未发现" + raise AndroidDeviceSearchError( + f"Wi-Fi 连接验证失败：{wifi_serial} 状态为 {status}；" + "请确认手机和电脑在同一网络后重试" + ) + + self._report(on_progress, "Wi-Fi ADB 已连接，正在刷新设备列表…") + return AndroidWifiConversionResult(wifi_serial, self.search(is_cancelled)) + + @staticmethod + def parse_wifi_ipv4(route_output: str) -> str: + """从 Android ``ip route`` 输出的 ``src`` 字段读取有效 IPv4。""" + + route_tokens = [line.split() for line in route_output.splitlines()] + default_interfaces = [] + for tokens in route_tokens: + if not tokens or tokens[0] != "default" or "dev" not in tokens: + continue + device_index = tokens.index("dev") + if device_index + 1 < len(tokens): + default_interfaces.append(tokens[device_index + 1]) + + preferred_routes = [] + for interface in default_interfaces: + preferred_routes.extend( + tokens + for tokens in route_tokens + if "dev" in tokens + and tokens.index("dev") + 1 < len(tokens) + and tokens[tokens.index("dev") + 1] == interface + ) + preferred_routes.extend( + tokens for tokens in route_tokens if tokens not in preferred_routes + ) + + for tokens in preferred_routes: + for index, token in enumerate(tokens[:-1]): + if token != "src": + continue + candidate = tokens[index + 1] + try: + address = ipaddress.ip_address(candidate) + except ValueError: + continue + if ( + address.version == 4 + and not address.is_loopback + and not address.is_unspecified + and not address.is_multicast + ): + return str(address) + raise AndroidDeviceSearchError( + "无法读取手机 Wi-Fi IPv4；请先让手机连接与电脑相同的 Wi-Fi" + ) + @staticmethod def parse_devices(output: str) -> List[AndroidDevice]: """解析 ``adb devices -l`` 输出，不依赖操作系统换行格式。""" @@ -137,6 +271,83 @@ class AndroidDeviceService: value = (result.stdout or "").strip() return value or "—" + def _wait_until_usb_disconnected( + self, + usb_serial: str, + is_cancelled: Optional[Callable[[], bool]], + ) -> None: + attempts = max( + 1, + int( + self._usb_disconnect_timeout_seconds + / self._usb_poll_interval_seconds + ), + ) + missing_checks = 0 + for attempt in range(attempts): + self._check_cancelled(is_cancelled) + devices = self._list_devices() + if all(device.serial != usb_serial for device in devices): + missing_checks += 1 + if missing_checks >= 2: + return + else: + missing_checks = 0 + if attempt < attempts - 1: + self._sleep(self._usb_poll_interval_seconds) + + raise AndroidDeviceSearchError( + "等待拔出 USB 超时；5555 已开启，请拔线后重新点击转换" + ) + + def _list_devices(self) -> List[AndroidDevice]: + result = self._run(["adb", "devices", "-l"], "读取设备列表") + return self.parse_devices(result.stdout or "") + + def _connect_wifi(self, wifi_serial: str) -> None: + result = self._run( + ["adb", "connect", wifi_serial], + "连接 Wi-Fi 设备", + ) + detail = " ".join( + part.strip() for part in (result.stdout, result.stderr) if part + ) + normalized_detail = detail.lower() + if ( + "connected to" not in normalized_detail + and "already connected to" not in normalized_detail + ): + raise AndroidDeviceSearchError( + f"ADB 连接 {wifi_serial} 失败：" + f"{detail or '没有返回成功信息'}" + ) + + @staticmethod + def _validate_usb_serial(serial: str) -> None: + if not isinstance(serial, str) or not serial.strip(): + raise AndroidDeviceSearchError("请先勾选一台已连接的 USB 设备") + contains_whitespace = any( + character.isspace() for character in serial + ) + if serial != serial.strip() or contains_whitespace: + raise AndroidDeviceSearchError("USB 设备号格式不正确") + if ":" in serial: + raise AndroidDeviceSearchError("当前设备已经是 Wi-Fi 连接") + + @staticmethod + def _check_cancelled( + is_cancelled: Optional[Callable[[], bool]], + ) -> None: + if is_cancelled is not None and is_cancelled(): + raise AndroidDeviceConversionCancelled() + + @staticmethod + def _report( + callback: Optional[ProgressCallback], message: str + ) -> None: + if callback is not None: + callback(message) + def _run(self, command: Sequence[str], action: str): try: result = self._run_command(command, self._timeout_seconds) diff --git a/client/src/settings_ui.py b/client/src/settings_ui.py index f59c00b..0d34fe0 100644 --- a/client/src/settings_ui.py +++ b/client/src/settings_ui.py @@ -250,8 +250,16 @@ class SettingsPage(QWidget): self.searchButton = PushButton(FIF.SEARCH, "搜索", self) self.connectButton = PushButton(FIF.LINK, "连接", self) + self.convertWifiButton = PushButton(FIF.WIFI, "转为 Wi-Fi", self) + self.convertWifiButton.setAccessibleName( + "将勾选的 USB Android 设备转为 Wi-Fi 连接" + ) + self.convertWifiButton.setToolTip( + "开启 ADB 5555，并在拔掉 USB 后重新连接" + ) self.saveButton = PushButton(FIF.SAVE, "保存", self) self.deleteButton = PushButton(FIF.DELETE, "删除", self) + self.convertWifiButton.setEnabled(False) self.saveButton.setEnabled(False) self.deleteButton.setEnabled(False) @@ -360,6 +368,7 @@ class SettingsPage(QWidget): commandLayout = QHBoxLayout() commandLayout.setSpacing(10) commandLayout.addWidget(self.searchButton) + commandLayout.addWidget(self.convertWifiButton) commandLayout.addStretch(1) commandLayout.addWidget(self.saveButton) commandLayout.addWidget(self.deleteButton) diff --git a/client/src/settings_ui_event.py b/client/src/settings_ui_event.py index f364588..63329f3 100644 --- a/client/src/settings_ui_event.py +++ b/client/src/settings_ui_event.py @@ -9,6 +9,7 @@ from typing import Optional from PyQt5.QtCore import QCoreApplication, QObject, QThread, pyqtSignal, pyqtSlot from .android_device_service import ( + AndroidDeviceConversionCancelled, AndroidDeviceSearchError, AndroidDeviceService, ) @@ -130,6 +131,49 @@ class AndroidDeviceSearchWorker(QObject): self.completed.emit() +class AndroidWifiConversionWorker(QObject): + """在后台引导一台 USB 设备切换到 Wi-Fi ADB。""" + + progressChanged = pyqtSignal(str) + succeeded = pyqtSignal(object) + failed = pyqtSignal(str) + completed = pyqtSignal() + + def __init__(self, service: AndroidDeviceService, usb_serial: str): + super().__init__() + self._service = service + self._usb_serial = usb_serial + self._cancelled = False + + def cancel(self) -> None: + """停止轮询和后续命令；正在执行的 ADB 命令等待自身超时。""" + + self._cancelled = True + + @pyqtSlot() + def run(self) -> None: + try: + try: + result = self._service.convert_usb_to_wifi( + self._usb_serial, + lambda: self._cancelled, + self.progressChanged.emit, + ) + except AndroidDeviceConversionCancelled: + return + 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 "无法将 USB 设备转为 Wi-Fi") + else: + if not self._cancelled: + self.succeeded.emit(result) + finally: + self.completed.emit() + + class AndroidDeviceSettingWorker(QObject): """在后台保存或删除当前使用的 Android 设备号。""" @@ -183,6 +227,7 @@ class SettingsPageEventBinder(QObject): searchRequested = pyqtSignal() connectRequested = pyqtSignal(str) + convertWifiRequested = pyqtSignal(str) currentDeviceSaveRequested = pyqtSignal(str, str) saveRequested = pyqtSignal(str, str) deleteRequested = pyqtSignal(str) @@ -204,6 +249,11 @@ class SettingsPageEventBinder(QObject): self._search_busy = False self._search_thread: Optional[QThread] = None self._search_worker: Optional[AndroidDeviceSearchWorker] = None + self._wifi_conversion_busy = False + self._wifi_conversion_thread: Optional[QThread] = None + self._wifi_conversion_worker: Optional[ + AndroidWifiConversionWorker + ] = None self._android_setting_busy = False self._android_setting_thread: Optional[QThread] = None self._android_setting_worker: Optional[AndroidDeviceSettingWorker] = None @@ -240,6 +290,8 @@ class SettingsPageEventBinder(QObject): page.searchButton.clicked.connect(self._request_search) self.searchRequested.connect(self._start_search) page.connectButton.clicked.connect(self._request_connect) + page.convertWifiButton.clicked.connect(self._request_convert_wifi) + self.convertWifiRequested.connect(self._start_convert_wifi) page.saveButton.clicked.connect(self._request_save) self.saveRequested.connect(self._start_save_android_device) page.deleteButton.clicked.connect(self._request_delete) @@ -261,6 +313,7 @@ class SettingsPageEventBinder(QObject): self._busy or self._current_device_busy or self._search_busy + or self._wifi_conversion_busy or self._android_setting_busy ): return @@ -278,6 +331,7 @@ class SettingsPageEventBinder(QObject): self._closing or self._current_device_busy or self._search_busy + or self._wifi_conversion_busy or self._android_setting_busy ): return @@ -323,6 +377,7 @@ class SettingsPageEventBinder(QObject): self._busy or self._current_device_busy or self._search_busy + or self._wifi_conversion_busy or self._android_setting_busy ): return @@ -335,6 +390,7 @@ class SettingsPageEventBinder(QObject): or self._busy or self._current_device_busy or self._search_busy + or self._wifi_conversion_busy or self._android_setting_busy ): return @@ -361,7 +417,13 @@ class SettingsPageEventBinder(QObject): @pyqtSlot() def _request_connect(self) -> None: - if self._busy: + if ( + self._busy + or self._current_device_busy + or self._search_busy + or self._wifi_conversion_busy + or self._android_setting_busy + ): return address = self._page.addressInput.text().strip() @@ -373,9 +435,76 @@ class SettingsPageEventBinder(QObject): return self.connectRequested.emit(address) + @pyqtSlot() + def _request_convert_wifi(self) -> None: + if ( + self._busy + or self._current_device_busy + or self._search_busy + or self._wifi_conversion_busy + or self._android_setting_busy + ): + return + + serial = self._page.deviceTableModel.checked_serial + device = self._page.deviceTableModel.device_for_serial(serial) + if device is None or device.status != "device": + self._page.deviceStatusLabel.setText( + "转换失败：请先勾选一台已连接的 USB 设备" + ) + self._page.deviceTable.setFocus() + return + if device.connection_type != "USB": + self._page.deviceStatusLabel.setText( + "转换失败：当前设备已经是 Wi-Fi 连接" + ) + return + self.convertWifiRequested.emit(serial) + + @pyqtSlot(str) + def _start_convert_wifi(self, usb_serial: str) -> None: + if ( + self._closing + or self._busy + or self._current_device_busy + or self._search_busy + or self._wifi_conversion_busy + or self._android_setting_busy + ): + return + + self._wifi_conversion_busy = True + self._sync_button_state() + self._page.deviceStatusLabel.setText("正在准备 USB 转 Wi-Fi…") + + thread = QThread(self) + worker = AndroidWifiConversionWorker( + self._android_device_service, + usb_serial, + ) + worker.moveToThread(thread) + + thread.started.connect(worker.run) + worker.progressChanged.connect(self._on_wifi_conversion_progress) + worker.succeeded.connect(self._on_wifi_conversion_succeeded) + worker.failed.connect(self._on_wifi_conversion_failed) + worker.completed.connect(thread.quit) + worker.completed.connect(worker.deleteLater) + thread.finished.connect(self._on_wifi_conversion_finished) + thread.finished.connect(thread.deleteLater) + + self._wifi_conversion_thread = thread + self._wifi_conversion_worker = worker + thread.start() + @pyqtSlot() def _request_save(self) -> None: - if self._busy or self._search_busy or self._android_setting_busy: + if ( + self._busy + or self._search_busy + or self._wifi_conversion_busy + or self._android_setting_busy + ): return serial = self._page.deviceTableModel.checked_serial @@ -394,7 +523,12 @@ class SettingsPageEventBinder(QObject): @pyqtSlot() def _request_delete(self) -> None: - if self._busy or self._search_busy or self._android_setting_busy: + if ( + self._busy + or self._search_busy + or self._wifi_conversion_busy + or self._android_setting_busy + ): return serial = self._saved_android_serial @@ -414,6 +548,7 @@ class SettingsPageEventBinder(QObject): self._closing or self._busy or self._search_busy + or self._wifi_conversion_busy or self._android_setting_busy ): return @@ -450,16 +585,19 @@ class SettingsPageEventBinder(QObject): @pyqtSlot() def _sync_button_state(self, *_args) -> None: serial = self._page.deviceTableModel.checked_serial + selected_device = self._page.deviceTableModel.device_for_serial(serial) has_address = bool(self._page.addressInput.text().strip()) self._page.currentDeviceSaveButton.setEnabled( not self._busy and not self._current_device_busy and not self._search_busy + and not self._wifi_conversion_busy and not self._android_setting_busy ) device_commands_enabled = ( not self._busy and not self._search_busy + and not self._wifi_conversion_busy and not self._android_setting_busy ) self._page.searchButton.setEnabled( @@ -471,6 +609,13 @@ class SettingsPageEventBinder(QObject): self._page.saveButton.setEnabled( device_commands_enabled and bool(serial) ) + self._page.convertWifiButton.setEnabled( + device_commands_enabled + and not self._current_device_busy + and selected_device is not None + and selected_device.status == "device" + and selected_device.connection_type == "USB" + ) self._page.deleteButton.setEnabled( device_commands_enabled and bool(self._saved_android_serial) ) @@ -576,6 +721,49 @@ class SettingsPageEventBinder(QObject): if not self._closing: self._sync_button_state() + @pyqtSlot(str) + def _on_wifi_conversion_progress(self, message: str) -> None: + if not self._closing: + self._page.deviceStatusLabel.setText(message) + + @pyqtSlot(object) + def _on_wifi_conversion_succeeded(self, result) -> 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 result.devices + ] + self._page.set_android_devices(rows) + self._page.deviceTableModel.set_checked_serial(result.wifi_serial) + self._page.deviceStatusLabel.setText( + f"Wi-Fi 设备 {result.wifi_serial} 已连接并勾选；" + "请确认后点击保存，仅在可信网络使用" + ) + + @pyqtSlot(str) + def _on_wifi_conversion_failed(self, message: str) -> None: + if self._closing: + return + self._page.deviceStatusLabel.setText( + f"转换失败：{message}；设备列表和已保存配置未更改，可重试" + ) + + @pyqtSlot() + def _on_wifi_conversion_finished(self) -> None: + self._wifi_conversion_worker = None + self._wifi_conversion_thread = None + self._wifi_conversion_busy = False + if not self._closing: + self._sync_button_state() + @pyqtSlot(str, str) def _on_android_device_setting_succeeded( self, action: str, serial: str @@ -695,6 +883,33 @@ class SettingsPageEventBinder(QObject): search_thread.quit() search_thread.wait(6000) + conversion_worker = self._wifi_conversion_worker + conversion_thread = self._wifi_conversion_thread + if conversion_worker is not None: + conversion_worker.cancel() + for signal, slot in ( + ( + conversion_worker.progressChanged, + self._on_wifi_conversion_progress, + ), + ( + conversion_worker.succeeded, + self._on_wifi_conversion_succeeded, + ), + ( + conversion_worker.failed, + self._on_wifi_conversion_failed, + ), + ): + try: + signal.disconnect(slot) + except TypeError: + pass + + if conversion_thread is not None and conversion_thread.isRunning(): + conversion_thread.quit() + conversion_thread.wait(6000) + setting_worker = self._android_setting_worker setting_thread = self._android_setting_thread if setting_worker is not None: diff --git a/client/test/test_android_device_service.py b/client/test/test_android_device_service.py index fcef745..03ecbb8 100644 --- a/client/test/test_android_device_service.py +++ b/client/test/test_android_device_service.py @@ -16,6 +16,25 @@ def completed(command, stdout="", stderr="", returncode=0): class AndroidDeviceServiceTest(unittest.TestCase): + def test_parse_wifi_ipv4_from_route(self): + output = ( + "10.20.0.0/16 dev rmnet_data0 scope link src 10.20.1.2\n" + "default via 192.168.0.1 dev wlan0 proto dhcp\n" + "192.168.0.0/24 dev wlan0 proto kernel scope link " + "src 192.168.0.173\n" + ) + + self.assertEqual( + AndroidDeviceService.parse_wifi_ipv4(output), + "192.168.0.173", + ) + + def test_missing_wifi_ipv4_has_recovery_message(self): + with self.assertRaisesRegex(AndroidDeviceSearchError, "连接.*Wi-Fi"): + AndroidDeviceService.parse_wifi_ipv4( + "default via 10.0.0.1 dev wlan0\n" + ) + 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 @@ -104,6 +123,119 @@ USB-002 unauthorized usb:1-2 transport_id:3 with self.assertRaisesRegex(AndroidDeviceSearchError, "didn't ACK"): AndroidDeviceService(runner).search() + def test_convert_usb_to_wifi_reconnects_after_usb_is_removed(self): + commands = [] + device_list_count = 0 + + def runner(command, _timeout): + nonlocal device_list_count + command = list(command) + commands.append(command) + if command[-3:] == ["shell", "ip", "route"]: + return completed( + command, + "192.168.0.0/24 dev wlan0 scope link " + "src 192.168.0.173\n", + ) + if command[:2] == ["adb", "connect"]: + return completed(command, "connected to 192.168.0.173:5555\n") + if command == ["adb", "devices", "-l"]: + device_list_count += 1 + usb_line = ( + "USB-001 device model:Phone\n" + if device_list_count == 1 + else "" + ) + return completed( + command, + "List of devices attached\n" + f"{usb_line}" + "192.168.0.173:5555 device model:Phone\n", + ) + if command[-1] == "ro.build.version.release": + return completed(command, "14\n") + return completed(command, "restarting in TCP mode port: 5555\n") + + progress = [] + service = AndroidDeviceService( + runner, + usb_disconnect_timeout_seconds=0.03, + usb_poll_interval_seconds=0.01, + sleeper=lambda _seconds: None, + ) + + result = service.convert_usb_to_wifi( + "USB-001", + on_progress=progress.append, + ) + + self.assertEqual(result.wifi_serial, "192.168.0.173:5555") + self.assertEqual(result.devices[0].connection_type, "Wi-Fi") + self.assertIn( + ["adb", "-s", "USB-001", "tcpip", "5555"], + commands, + ) + self.assertEqual( + commands.count(["adb", "connect", "192.168.0.173:5555"]), + 2, + ) + self.assertTrue(any("拔掉 USB" in message for message in progress)) + + def test_convert_usb_to_wifi_times_out_without_unplug(self): + def runner(command, _timeout): + command = list(command) + if command[-3:] == ["shell", "ip", "route"]: + return completed(command, "local src 192.168.0.173\n") + if command[:2] == ["adb", "connect"]: + return completed(command, "connected to 192.168.0.173:5555\n") + if command == ["adb", "devices", "-l"]: + return completed( + command, + "List of devices attached\nUSB-001 device\n", + ) + return completed(command, "restarting in TCP mode port: 5555\n") + + service = AndroidDeviceService( + runner, + usb_disconnect_timeout_seconds=0.03, + usb_poll_interval_seconds=0.01, + sleeper=lambda _seconds: None, + ) + + with self.assertRaisesRegex(AndroidDeviceSearchError, "拔出 USB 超时"): + service.convert_usb_to_wifi("USB-001") + + def test_convert_usb_to_wifi_rejects_wifi_serial(self): + with self.assertRaisesRegex(AndroidDeviceSearchError, "已经是 Wi-Fi"): + AndroidDeviceService().convert_usb_to_wifi( + "192.168.0.173:5555" + ) + + def test_convert_usb_to_wifi_requires_connected_wifi_status(self): + def runner(command, _timeout): + command = list(command) + if command[-3:] == ["shell", "ip", "route"]: + return completed(command, "local src 192.168.0.173\n") + if command[:2] == ["adb", "connect"]: + return completed(command, "connected to 192.168.0.173:5555\n") + if command == ["adb", "devices", "-l"]: + return completed( + command, + "List of devices attached\n" + "192.168.0.173:5555 offline\n", + ) + return completed(command, "restarting in TCP mode port: 5555\n") + + service = AndroidDeviceService( + runner, + usb_disconnect_timeout_seconds=0.03, + usb_poll_interval_seconds=0.01, + sleeper=lambda _seconds: None, + ) + + with self.assertRaisesRegex(AndroidDeviceSearchError, "状态为 offline"): + service.convert_usb_to_wifi("USB-001") + if __name__ == "__main__": unittest.main() diff --git a/client/test/test_settings_ui_event.py b/client/test/test_settings_ui_event.py index 7ce6e59..e6d66ff 100644 --- a/client/test/test_settings_ui_event.py +++ b/client/test/test_settings_ui_event.py @@ -14,7 +14,9 @@ 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 @@ -58,6 +60,40 @@ class SlowAndroidDeviceService: return [AndroidDevice("USB-001", "USB", "Pixel 8", "14")] +class ControlledWifiConversionService: + """模拟 USB 转 Wi-Fi 的进度、成功、失败和延迟。""" + + def __init__(self, delay=0.0, error=None): + self.delay = delay + self.error = error + self.call_count = 0 + + def search(self, _is_cancelled=None): + return [] + + 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("已开启 192.168.0.173:5555，请拔掉 USB 数据线…") + 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(wifi_serial, "Wi-Fi", "Phone", "14")], + ) + + class ControlledSettingsRepository(SettingsRepository): """控制 Android 设备设置写入速度和失败，其他设置保持正常。""" @@ -530,6 +566,141 @@ class SettingsPageEventTest(unittest.TestCase): self.assertEqual(page.deviceStatusLabel.text(), status_before_close) page.deleteLater() + def test_convert_wifi_button_requires_connected_usb_device(self): + page = SettingsPage( + settings_repository=self.repository, + admin_gateway=MockAdminGateway(), + ) + + self.assertFalse(page.convertWifiButton.isEnabled()) + + page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) + page.deviceTableModel.set_checked_serial("USB-001") + 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.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, + ) + page.set_android_devices([AndroidDeviceRow("USB-001", "USB")]) + page.deviceTableModel.set_checked_serial("USB-001") + + page.convertWifiButton.click() + self._wait_until( + lambda: page.eventBinder._wifi_conversion_thread is None + ) + + self.assertEqual(service.last_usb_serial, "USB-001") + self.assertEqual( + page.deviceTableModel.checked_serial, + "192.168.0.173:5555", + ) + self.assertEqual( + self.repository.get(SELECTED_ANDROID_SERIAL_KEY), + "USB-OLD", + ) + self.assertIn("已连接并勾选", 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("等待拔出 USB 超时") + ) + self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD") + 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") + + 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") + 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") + + 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()