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

feat: 保存当前使用的 Android 设备 (#22)

Browse Source
This commit is contained in:
chengma
2026-08-07 14:13:40 +08:00
parent d2de7331bb
commit 27a09e524c
4 changed files with 558 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/test/test_selected_android_device_service.py
+60
View File
@@ -0,0 +1,60 @@
"""当前使用 Android 设备设置服务测试。"""
import tempfile
import unittest
from pathlib import Path
from src.selected_android_device_service import (
SELECTED_ANDROID_SERIAL_KEY,
SelectedAndroidDeviceService,
)
from src.settings_repository import SettingsRepository
class SelectedAndroidDeviceServiceTest(unittest.TestCase):
def setUp(self):
self.temp_directory = tempfile.TemporaryDirectory()
db_path = Path(self.temp_directory.name) / "client.db"
self.repository = SettingsRepository(db_path)
self.service = SelectedAndroidDeviceService(self.repository)
def tearDown(self):
self.temp_directory.cleanup()
def test_missing_setting_loads_as_empty(self):
self.assertEqual(self.service.load(), "")
def test_save_normalizes_and_persists_serial(self):
serial = self.service.save(" 192.168.0.173:5555 ")
self.assertEqual(serial, "192.168.0.173:5555")
self.assertEqual(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY), serial
)
def test_invalid_serial_is_rejected(self):
for serial in ("", "USB 001", None):
with self.subTest(serial=serial):
with self.assertRaises(ValueError):
self.service.save(serial)
def test_corrupted_setting_loads_as_empty(self):
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, {"serial": "bad"})
self.assertEqual(self.service.load(), "")
def test_delete_removes_only_selected_serial(self):
self.repository.set("android.pdd_package", "com.xunmeng.pinduoduo")
self.service.save("USB-001")
self.assertTrue(self.service.delete())
self.assertIsNone(self.repository.get(SELECTED_ANDROID_SERIAL_KEY))
self.assertEqual(
self.repository.get("android.pdd_package"),
"com.xunmeng.pinduoduo",
)
if __name__ == "__main__":
unittest.main()
client/test/test_settings_ui_event.py
+210 -1
View File
@@ -17,6 +17,7 @@ from src.android_device_service import (
AndroidDeviceSearchError,
)
from src.mock_admin_gateway import MockAdminGateway
from src.selected_android_device_service import SELECTED_ANDROID_SERIAL_KEY
from src.settings_repository import SettingsRepository
from src.settings_ui import AndroidDeviceRow, SettingsPage
@@ -57,6 +58,34 @@ class SlowAndroidDeviceService:
return [AndroidDevice("USB-001", "USB", "Pixel 8", "14")]
class ControlledSettingsRepository(SettingsRepository):
"""控制 Android 设备设置写入速度和失败，其他设置保持正常。"""
def __init__(self, db_path, delay=0.0):
super().__init__(db_path)
self.delay = delay
self.fail_selected_write = False
self.selected_write_count = 0
def set(self, setting_key, value, updated_at=None):
if setting_key == SELECTED_ANDROID_SERIAL_KEY:
self.selected_write_count += 1
if self.delay:
time.sleep(self.delay)
if self.fail_selected_write:
raise OSError("模拟数据库写入失败")
return super().set(setting_key, value, updated_at)
def delete(self, setting_key):
if setting_key == SELECTED_ANDROID_SERIAL_KEY:
self.selected_write_count += 1
if self.delay:
time.sleep(self.delay)
if self.fail_selected_write:
raise OSError("模拟数据库删除失败")
return super().delete(setting_key)
class SettingsPageEventTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
@@ -254,7 +283,7 @@ class SettingsPageEventTest(unittest.TestCase):
page.deviceTableModel.device_for_serial("USB-OLD")
)
self.assertIn("搜索失败：未找到 adb", page.deviceStatusLabel.text())
self.assertIn("设备列表未更改", page.deviceStatusLabel.text())
self.assertIn("已保存配置未更改", page.deviceStatusLabel.text())
page.eventBinder.shutdown()
page.deleteLater()
@@ -321,6 +350,186 @@ class SettingsPageEventTest(unittest.TestCase):
)
page.deleteLater()
def test_selected_android_device_is_saved_to_sqlite(self):
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
)
page.set_android_devices(
[AndroidDeviceRow("USB-001", "USB", "Pixel 8", "14")]
)
page.deviceTableModel.set_checked_serial("USB-001")
page.saveButton.click()
self.assertFalse(page.saveButton.isEnabled())
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
)
self.assertEqual(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-001"
)
self.assertEqual(
page.deviceStatusLabel.text(),
"当前使用设备：USB-001（已保存）",
)
self.assertTrue(page.deleteButton.isEnabled())
page.eventBinder.shutdown()
page.deleteLater()
def test_saved_android_device_is_restored_before_search(self):
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001")
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
)
self.assertEqual(page.eventBinder._saved_android_serial, "USB-001")
self.assertEqual(page.deviceTableModel.checked_serial, "")
self.assertIn("等待搜索确认连接", page.deviceStatusLabel.text())
self.assertTrue(page.deleteButton.isEnabled())
page.eventBinder.shutdown()
page.deleteLater()
def test_search_auto_selects_saved_online_device(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,
)
page.searchButton.click()
self._wait_until(lambda: page.eventBinder._search_thread is None)
self.assertEqual(page.deviceTableModel.checked_serial, "USB-001")
self.assertIn("已自动选择", page.deviceStatusLabel.text())
page.eventBinder.shutdown()
page.deleteLater()
def test_search_keeps_saved_device_when_it_is_offline(self):
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-001")
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.assertEqual(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-001"
)
self.assertEqual(page.deviceTableModel.checked_serial, "")
self.assertIn("已保存设备当前未连接", page.deviceStatusLabel.text())
self.assertIn("配置已保留", page.deviceStatusLabel.text())
page.eventBinder.shutdown()
page.deleteLater()
def test_selected_android_device_save_failure_keeps_old_configuration(self):
repository = ControlledSettingsRepository(self.db_path)
repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD")
repository.fail_selected_write = True
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
)
page.set_android_devices(
[
AndroidDeviceRow("USB-OLD", "USB"),
AndroidDeviceRow("USB-NEW", "USB"),
]
)
page.deviceTableModel.set_checked_serial("USB-NEW")
page.saveButton.click()
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
)
self.assertEqual(
repository.get(SELECTED_ANDROID_SERIAL_KEY), "USB-OLD"
)
self.assertEqual(page.deviceTableModel.checked_serial, "USB-NEW")
self.assertIn("原设备配置未更改", page.deviceStatusLabel.text())
self.assertTrue(page.saveButton.isEnabled())
page.eventBinder.shutdown()
page.deleteLater()
def test_delete_clears_saved_offline_device_only(self):
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OFFLINE")
self.repository.set("android.pdd_package", "com.xunmeng.pinduoduo")
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
)
page.deleteButton.click()
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
)
self.assertIsNone(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY)
)
self.assertEqual(
self.repository.get("android.pdd_package"),
"com.xunmeng.pinduoduo",
)
self.assertIn("本地配置已删除", page.deviceStatusLabel.text())
self.assertFalse(page.deleteButton.isEnabled())
page.eventBinder.shutdown()
page.deleteLater()
def test_slow_android_device_save_does_not_block_or_start_twice(self):
repository = ControlledSettingsRepository(self.db_path, delay=0.08)
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
)
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.saveButton.click()
page.eventBinder._request_save()
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
)
self.assertTrue(timer_fired)
self.assertEqual(repository.selected_write_count, 1)
page.eventBinder.shutdown()
page.deleteLater()
def test_shutdown_ignores_late_android_device_save_result(self):
repository = ControlledSettingsRepository(self.db_path, delay=0.08)
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
page.saveButton.click()
self._wait_until(
lambda: page.eventBinder._android_setting_thread is not None
)
status_before_close = page.deviceStatusLabel.text()
page.eventBinder.shutdown()
QTest.qWait(120)
self.assertEqual(page.deviceStatusLabel.text(), status_before_close)
page.deleteLater()
if __name__ == "__main__":
unittest.main()