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

fix: 启动时恢复已保存的 Android 设备 (#26)

Browse Source
This commit is contained in:
chengma
2026-08-07 14:53:28 +08:00
parent e40ce5a13c
commit dc9b000a36
4 changed files with 253 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/test/test_android_device_service.py
+52
View File
@@ -72,6 +72,58 @@ USB-002 unauthorized usb:1-2 transport_id:3
self.assertEqual(len(commands), 2)
self.assertIn("ro.build.version.release", commands[1])
def test_restore_saved_wifi_connects_before_search(self):
commands = []
def runner(command, _timeout):
command = list(command)
commands.append(command)
if command[:2] == ["adb", "connect"]:
return completed(
command,
"already 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 device model:Phone\n",
)
return completed(command, "14\n")
devices = AndroidDeviceService(runner).restore_saved_device(
"192.168.0.173:5555"
)
self.assertEqual(devices[0].serial, "192.168.0.173:5555")
self.assertEqual(
commands[0],
["adb", "connect", "192.168.0.173:5555"],
)
def test_restore_saved_usb_only_searches(self):
commands = []
def runner(command, _timeout):
command = list(command)
commands.append(command)
if command == ["adb", "devices", "-l"]:
return completed(
command,
"List of devices attached\n"
"USB-001 device model:Phone\n",
)
return completed(command, "14\n")
devices = AndroidDeviceService(runner).restore_saved_device(
"USB-001"
)
self.assertEqual(devices[0].serial, "USB-001")
self.assertFalse(
any(command[:2] == ["adb", "connect"] for command in commands)
)
def test_missing_model_is_read_from_device(self):
def runner(command, _timeout):
if command == ["adb", "devices", "-l"]:
client/test/test_settings_ui_event.py
+120 -4
View File
@@ -38,6 +38,7 @@ 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)]
@@ -46,6 +47,10 @@ class SequenceAndroidDeviceService:
raise result
return result
def restore_saved_device(self, serial, is_cancelled=None):
self.restore_serials.append(serial)
return self.search(is_cancelled)
class SlowAndroidDeviceService:
"""模拟较慢的 ADB 搜索。"""
@@ -59,6 +64,10 @@ class SlowAndroidDeviceService:
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)
class ControlledWifiConversionService:
"""模拟 USB 转 Wi-Fi 的进度、成功、失败和延迟。"""
@@ -413,21 +422,128 @@ class SettingsPageEventTest(unittest.TestCase):
page.eventBinder.shutdown()
page.deleteLater()
def test_saved_android_device_is_restored_before_search(self):
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001")
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(page.deviceTableModel.checked_serial, "")
self.assertIn("等待搜索确认连接", page.deviceStatusLabel.text())
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_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 == 1)
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("可点击搜索重试", page.deviceStatusLabel.text())
self.assertTrue(page.searchButton.isEnabled())
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(