chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持 USB 设备转 Wi-Fi ADB (#25)

Browse Source
This commit is contained in:
chengma
2026-08-07 14:42:00 +08:00
parent bf81bad741
commit d478e690b6
5 changed files with 741 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/test/test_android_device_service.py
+132
View File
@@ -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()
client/test/test_settings_ui_event.py
+171
View File
@@ -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()