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

fix: 修复 Wi-Fi ADB 拔线后刷新丢失 (#27)

Browse Source
This commit is contained in:
chengma
2026-08-07 15:12:32 +08:00
parent ffaeb55a25
commit 57503978bd
4 changed files with 212 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/test/test_android_device_service.py
+39 -48
View File
@@ -101,6 +101,33 @@ USB-002 unauthorized usb:1-2 transport_id:3
["adb", "connect", "192.168.0.173:5555"],
)
def test_restore_saved_wifi_waits_for_authorization_handshake(self):
device_list_count = 0
sleeps = []
def runner(command, _timeout):
nonlocal device_list_count
command = list(command)
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
status = "unauthorized" if device_list_count == 1 else "device"
return completed(
command,
"List of devices attached\n"
f"192.168.0.173:5555 {status} model:Phone\n",
)
return completed(command, "14\n")
devices = AndroidDeviceService(
runner,
sleeper=sleeps.append,
).restore_saved_device("192.168.0.173:5555")
self.assertEqual(devices[0].status, "device")
self.assertEqual(sleeps, [0.25])
def test_restore_saved_usb_only_searches(self):
commands = []
@@ -175,12 +202,10 @@ 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):
def test_convert_usb_to_wifi_connects_while_usb_is_attached(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"]:
@@ -192,16 +217,10 @@ USB-002 unauthorized usb:1-2 transport_id:3
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}"
"USB-001 device model:Phone\n"
"192.168.0.173:5555 device model:Phone\n",
)
if command[-1] == "ro.build.version.release":
@@ -209,12 +228,7 @@ USB-002 unauthorized usb:1-2 transport_id:3
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,
)
service = AndroidDeviceService(runner)
result = service.convert_usb_to_wifi(
"USB-001",
@@ -222,41 +236,23 @@ USB-002 unauthorized usb:1-2 transport_id:3
)
self.assertEqual(result.wifi_serial, "192.168.0.173:5555")
self.assertEqual(result.devices[0].connection_type, "Wi-Fi")
self.assertEqual(
[device.serial for device in result.devices],
["USB-001", "192.168.0.173:5555"],
)
self.assertIn(
["adb", "-s", "USB-001", "tcpip", "5555"],
commands,
)
self.assertEqual(
commands.count(["adb", "connect", "192.168.0.173:5555"]),
2,
1,
)
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,
self.assertEqual(commands.count(["adb", "devices", "-l"]), 3)
self.assertTrue(
any("保存后拔掉 USB" in message for message in progress)
)
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(
@@ -278,12 +274,7 @@ USB-002 unauthorized usb:1-2 transport_id:3
)
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,
)
service = AndroidDeviceService(runner, sleeper=lambda _seconds: None)
with self.assertRaisesRegex(AndroidDeviceSearchError, "状态为 offline"):
service.convert_usb_to_wifi("USB-001")