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

feat: 检测 Android 设备 PDD 安装状态 (#29)

Browse Source
This commit is contained in:
chengma
2026-08-07 15:54:12 +08:00
parent a6e3fe0417
commit 6d863a1a8a
5 changed files with 569 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/android_device_service.py
+21
View File
@@ -10,6 +10,7 @@ from typing import Callable, List, Optional, Sequence
DEFAULT_ADB_TIMEOUT_SECONDS = 5.0
DEFAULT_DEVICE_READY_CHECKS = 8
DEFAULT_DEVICE_READY_INTERVAL_SECONDS = 0.25
PDD_PACKAGE_NAME = "com.xunmeng.pinduoduo"
class AndroidDeviceSearchError(RuntimeError):
@@ -111,6 +112,26 @@ class AndroidDeviceService:
)
return enriched
def is_package_installed(
self,
serial: str,
package_name: str = PDD_PACKAGE_NAME,
) -> bool:
"""查询指定设备是否安装目标包；ADB 失败时抛出可读异常。"""
self._validate_saved_serial(serial)
if not package_name or any(character.isspace() for character in package_name):
raise AndroidDeviceSearchError("应用包名格式不正确")
result = self._run(
["adb", "-s", serial, "shell", "pm", "path", package_name],
"检测 PDD 应用",
)
return any(
line.strip().startswith("package:")
for line in (result.stdout or "").splitlines()
)
def restore_saved_device(
self,
serial: str,
client/src/settings_ui.py
+29 -18
View File
@@ -240,16 +240,11 @@ class SettingsPage(QWidget):
"设备信息尚未保存", self
)
self.currentDeviceStatusLabel.setAccessibleName("当前设备保存状态")
self.currentDeviceStatusLabel.setWordWrap(True)
self.currentDeviceCard = self._build_current_device_card()
self.addressInput = LineEdit(self)
self.addressInput.setPlaceholderText("例如 192.168.0.173:5555")
self.addressInput.setClearButtonEnabled(True)
self.addressInput.setAccessibleName("Android 无线调试地址")
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 连接"
@@ -288,6 +283,10 @@ class SettingsPage(QWidget):
self.deviceStatusLabel = CaptionLabel("当前使用设备：未选择", self)
self.deviceStatusLabel.setAccessibleName("Android 设备搜索和选择状态")
self.deviceStatusLabel.setWordWrap(True)
self.pddAppStatusLabel = CaptionLabel("未检测", self)
self.pddAppStatusLabel.setAccessibleName("PDD 应用安装状态")
self.pddAppStatusLabel.setWordWrap(True)
self.androidDeviceCard = self._build_android_device_card()
content = QWidget(self)
@@ -329,7 +328,13 @@ class SettingsPage(QWidget):
layout = QVBoxLayout(card)
layout.setContentsMargins(24, 20, 24, 22)
layout.setSpacing(12)
layout.addWidget(SubtitleLabel("当前设备", card))
titleLayout = QHBoxLayout()
titleLayout.setSpacing(12)
titleLayout.addWidget(SubtitleLabel("当前设备", card))
titleLayout.addStretch(1)
titleLayout.addWidget(self.currentDeviceStatusLabel, 1)
layout.addLayout(titleLayout)
form = QFormLayout()
form.setHorizontalSpacing(16)
@@ -346,7 +351,6 @@ class SettingsPage(QWidget):
commandLayout.addStretch(1)
commandLayout.addWidget(self.currentDeviceSaveButton)
layout.addLayout(commandLayout)
layout.addWidget(self.currentDeviceStatusLabel)
return card
def _build_android_device_card(self) -> CardWidget:
@@ -354,16 +358,19 @@ class SettingsPage(QWidget):
layout = QVBoxLayout(card)
layout.setContentsMargins(24, 20, 24, 22)
layout.setSpacing(12)
layout.addWidget(SubtitleLabel("Android 设备", card))
addressLayout = QHBoxLayout()
addressLayout.setSpacing(12)
addressLabel = CaptionLabel("无线地址", card)
addressLabel.setBuddy(self.addressInput)
addressLayout.addWidget(addressLabel)
addressLayout.addWidget(self.addressInput, 1)
addressLayout.addWidget(self.connectButton)
layout.addLayout(addressLayout)
titleLayout = QHBoxLayout()
titleLayout.setSpacing(12)
titleLayout.addWidget(SubtitleLabel("Android 设备", card))
titleLayout.addStretch(1)
titleLayout.addWidget(self.deviceStatusLabel, 1)
layout.addLayout(titleLayout)
pddStatusLayout = QHBoxLayout()
pddStatusLayout.setSpacing(12)
pddStatusLayout.addWidget(CaptionLabel("PDD 应用", card))
pddStatusLayout.addWidget(self.pddAppStatusLabel, 1)
layout.addLayout(pddStatusLayout)
commandLayout = QHBoxLayout()
commandLayout.setSpacing(10)
@@ -374,7 +381,6 @@ class SettingsPage(QWidget):
commandLayout.addWidget(self.deleteButton)
layout.addLayout(commandLayout)
layout.addWidget(self.deviceTable, 1)
layout.addWidget(self.deviceStatusLabel)
return card
def set_client_info(self, device_id: str, device_name: str) -> None:
@@ -394,6 +400,11 @@ class SettingsPage(QWidget):
self.deviceTableModel.set_devices(devices)
self._update_device_placeholder_rows()
def set_pdd_app_status(self, message: str) -> None:
"""显示当前勾选设备上的 PDD 安装检测结果。"""
self.pddAppStatusLabel.setText(message)
def set_saved_android_device(self, serial: str) -> None:
"""显示已经保存并实际用于自动化的 Android 设备。"""
client/src/settings_ui_event.py
+285 -53
View File
@@ -19,6 +19,7 @@ from .android_device_service import (
AndroidDeviceConversionCancelled,
AndroidDeviceSearchError,
AndroidDeviceService,
PDD_PACKAGE_NAME,
)
from .admin_gateway import (
AdminGatewayError,
@@ -149,6 +150,48 @@ class AndroidDeviceSearchWorker(QObject):
self.completed.emit()
class PddPackageCheckWorker(QObject):
"""在后台检查一台 Android 设备是否安装 PDD。"""
succeeded = pyqtSignal(str, bool)
failed = pyqtSignal(str, str)
completed = pyqtSignal(str)
def __init__(self, service: AndroidDeviceService, serial: str):
super().__init__()
self._service = service
self._serial = serial
self._cancelled = False
def cancel(self) -> None:
"""ADB 命令无法中断；完成后不再发送业务结果。"""
self._cancelled = True
@pyqtSlot()
def run(self) -> None:
try:
try:
installed = self._service.is_package_installed(
self._serial,
PDD_PACKAGE_NAME,
)
except AndroidDeviceSearchError as exc:
if not self._cancelled:
self.failed.emit(self._serial, str(exc))
except Exception as exc:
if not self._cancelled:
self.failed.emit(
self._serial,
str(exc) or "无法检测 PDD 应用",
)
else:
if not self._cancelled:
self.succeeded.emit(self._serial, installed)
finally:
self.completed.emit(self._serial)
class AndroidWifiConversionWorker(QObject):
"""在后台引导一台 USB 设备切换到 Wi-Fi ADB。"""
@@ -244,7 +287,6 @@ class SettingsPageEventBinder(QObject):
"""绑定设备管理控件，并向应用层发出稳定事件。"""
searchRequested = pyqtSignal()
connectRequested = pyqtSignal(str)
convertWifiRequested = pyqtSignal(str)
currentDeviceSaveRequested = pyqtSignal(str, str)
saveRequested = pyqtSignal(str, str)
@@ -269,6 +311,14 @@ class SettingsPageEventBinder(QObject):
self._search_is_startup_restore = False
self._search_thread: Optional[QThread] = None
self._search_worker: Optional[AndroidDeviceSearchWorker] = None
self._pdd_check_busy = False
self._pdd_check_serial = ""
self._pending_pdd_check_serial = ""
self._pending_android_save_serial = ""
self._pdd_check_result_serial = ""
self._pdd_is_installed = False
self._pdd_check_thread: Optional[QThread] = None
self._pdd_check_worker: Optional[PddPackageCheckWorker] = None
self._wifi_conversion_busy = False
self._wifi_conversion_thread: Optional[QThread] = None
self._wifi_conversion_worker: Optional[
@@ -309,15 +359,15 @@ 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)
self.deleteRequested.connect(self._start_delete_android_device)
page.addressInput.textChanged.connect(self._sync_button_state)
page.deviceTableModel.checkedDeviceChanged.connect(self._sync_button_state)
page.deviceTableModel.checkedDeviceChanged.connect(
self._on_checked_device_changed
)
page.destroyed.connect(self.shutdown)
application = QCoreApplication.instance()
if application is not None:
@@ -469,8 +519,35 @@ class SettingsPageEventBinder(QObject):
self._search_worker = worker
thread.start()
@pyqtSlot()
def _request_connect(self) -> None:
@pyqtSlot(str)
def _on_checked_device_changed(self, serial: str) -> None:
"""勾选变化后刷新按钮，并为最新设备排队检测 PDD。"""
self._sync_button_state()
if self._closing:
return
if self._pending_android_save_serial != serial:
self._pending_android_save_serial = ""
self._pdd_check_result_serial = ""
self._pdd_is_installed = False
if not serial:
self._pending_pdd_check_serial = ""
self._page.set_pdd_app_status("未检测")
return
self._queue_pdd_package_check(serial)
def _queue_pdd_package_check(self, serial: str) -> None:
"""立即检测或只保留最后一次待检测的设备号。"""
if self._closing or not serial:
return
if self._pdd_check_busy:
if serial != self._pdd_check_serial:
self._pending_pdd_check_serial = serial
self._page.set_pdd_app_status("等待检测…")
return
if (
self._busy
or self._current_device_busy
@@ -478,16 +555,115 @@ class SettingsPageEventBinder(QObject):
or self._wifi_conversion_busy
or self._android_setting_busy
):
self._pending_pdd_check_serial = serial
self._page.set_pdd_app_status("等待检测…")
return
self._start_pdd_package_check(serial)
def _start_pdd_package_check(self, serial: str) -> None:
"""启动单次后台包检测；同一时间最多运行一个。"""
if self._closing or self._pdd_check_busy:
return
address = self._page.addressInput.text().strip()
if not address:
self._page.deviceStatusLabel.setText(
"连接失败：请先填写无线调试地址，例如 192.168.0.173:5555"
)
self._page.addressInput.setFocus()
self._pending_pdd_check_serial = ""
self._pdd_check_busy = True
self._pdd_check_serial = serial
self._page.set_pdd_app_status("正在检测…")
self._sync_button_state()
thread = QThread(self)
worker = PddPackageCheckWorker(
self._android_device_service,
serial,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.succeeded.connect(self._on_pdd_package_check_succeeded)
worker.failed.connect(self._on_pdd_package_check_failed)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
thread.finished.connect(self._on_pdd_package_check_finished)
thread.finished.connect(thread.deleteLater)
self._pdd_check_thread = thread
self._pdd_check_worker = worker
thread.start()
@pyqtSlot(str, bool)
def _on_pdd_package_check_succeeded(
self,
serial: str,
installed: bool,
) -> None:
if self._closing:
return
self.connectRequested.emit(address)
if serial != self._page.deviceTableModel.checked_serial:
return
self._pdd_check_result_serial = serial
self._pdd_is_installed = installed
if installed:
self._page.set_pdd_app_status(
f"{PDD_PACKAGE_NAME}（已安装）"
)
return
self._page.set_pdd_app_status("未安装，请先在手机安装拼多多")
if self._pending_android_save_serial == serial:
self._pending_android_save_serial = ""
self._page.deviceStatusLabel.setText(
"保存失败：所选设备未安装 PDD，请安装后重试"
)
@pyqtSlot(str, str)
def _on_pdd_package_check_failed(
self,
serial: str,
message: str,
) -> None:
if self._closing:
return
if serial != self._page.deviceTableModel.checked_serial:
return
self._pdd_check_result_serial = serial
self._pdd_is_installed = False
self._page.set_pdd_app_status(f"检测失败：{message}")
if self._pending_android_save_serial == serial:
self._pending_android_save_serial = ""
self._page.deviceStatusLabel.setText(
f"保存失败：无法确认 PDD 安装状态；{message}；原配置未更改"
)
@pyqtSlot()
def _on_pdd_package_check_finished(self) -> None:
finished_serial = self._pdd_check_serial
self._pdd_check_worker = None
self._pdd_check_thread = None
self._pdd_check_busy = False
self._pdd_check_serial = ""
if self._closing:
return
current_serial = self._page.deviceTableModel.checked_serial
pending_serial = self._pending_pdd_check_serial
if pending_serial and pending_serial == current_serial:
self._queue_pdd_package_check(pending_serial)
return
self._pending_pdd_check_serial = ""
if (
self._pending_android_save_serial == finished_serial
and self._pdd_check_result_serial == finished_serial
and self._pdd_is_installed
and current_serial == finished_serial
):
self._pending_android_save_serial = ""
self.saveRequested.emit("", finished_serial)
return
self._sync_button_state()
@pyqtSlot()
def _request_convert_wifi(self) -> None:
@@ -567,7 +743,11 @@ class SettingsPageEventBinder(QObject):
self._page.deviceTable.setFocus()
return
self.saveRequested.emit("", serial)
self._pending_android_save_serial = serial
self._page.deviceStatusLabel.setText(
"正在确认所选设备是否安装 PDD…"
)
self._queue_pdd_package_check(serial)
@pyqtSlot(str, str)
def _start_save_android_device(
@@ -640,7 +820,6 @@ class SettingsPageEventBinder(QObject):
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
@@ -657,11 +836,10 @@ class SettingsPageEventBinder(QObject):
self._page.searchButton.setEnabled(
device_commands_enabled and not self._current_device_busy
)
self._page.connectButton.setEnabled(
device_commands_enabled and has_address
)
self._page.saveButton.setEnabled(
device_commands_enabled and bool(serial)
device_commands_enabled
and not self._pdd_check_busy
and bool(serial)
)
self._page.convertWifiButton.setEnabled(
device_commands_enabled
@@ -775,6 +953,12 @@ class SettingsPageEventBinder(QObject):
"未找到 Android 设备，请检查 USB 调试或无线连接后重试"
)
selected_serial = self._page.deviceTableModel.checked_serial
if selected_serial:
self._queue_pdd_package_check(selected_serial)
else:
self._page.set_pdd_app_status("未检测")
@pyqtSlot(str)
def _on_search_failed(self, message: str) -> None:
if self._closing:
@@ -803,6 +987,9 @@ class SettingsPageEventBinder(QObject):
self._search_is_startup_restore = False
if not self._closing:
self._sync_button_state()
pending_serial = self._pending_pdd_check_serial
if pending_serial:
self._queue_pdd_package_check(pending_serial)
@pyqtSlot(str)
def _on_wifi_conversion_progress(self, message: str) -> None:
@@ -846,6 +1033,9 @@ class SettingsPageEventBinder(QObject):
self._wifi_conversion_busy = False
if not self._closing:
self._sync_button_state()
pending_serial = self._pending_pdd_check_serial
if pending_serial:
self._queue_pdd_package_check(pending_serial)
@pyqtSlot(str, str)
def _on_android_device_setting_succeeded(
@@ -952,41 +1142,79 @@ class SettingsPageEventBinder(QObject):
search_worker = self._search_worker
search_thread = self._search_thread
if search_worker is not None:
search_worker.cancel()
for signal, slot in (
(search_worker.succeeded, self._on_search_succeeded),
(search_worker.failed, self._on_search_failed),
):
try:
search_worker.cancel()
signal_slots = (
(search_worker.succeeded, self._on_search_succeeded),
(search_worker.failed, self._on_search_failed),
)
except RuntimeError:
# Worker 已调用 deleteLater，但 finished 清理槽还在消息队列中。
signal_slots = ()
for signal, slot in signal_slots:
try:
signal.disconnect(slot)
except TypeError:
except (TypeError, RuntimeError):
pass
if search_thread is not None and search_thread.isRunning():
search_thread.quit()
search_thread.wait(6000)
pdd_worker = self._pdd_check_worker
pdd_thread = self._pdd_check_thread
if pdd_worker is not None:
try:
pdd_worker.cancel()
signal_slots = (
(
pdd_worker.succeeded,
self._on_pdd_package_check_succeeded,
),
(
pdd_worker.failed,
self._on_pdd_package_check_failed,
),
)
except RuntimeError:
# Worker 已调用 deleteLater，但 thread.finished 的清理槽
# 还在主线程队列中时，Python 引用可能暂时仍存在。
signal_slots = ()
for signal, slot in signal_slots:
try:
signal.disconnect(slot)
except (TypeError, RuntimeError):
pass
if pdd_thread is not None and pdd_thread.isRunning():
pdd_thread.quit()
pdd_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:
conversion_worker.cancel()
signal_slots = (
(
conversion_worker.progressChanged,
self._on_wifi_conversion_progress,
),
(
conversion_worker.succeeded,
self._on_wifi_conversion_succeeded,
),
(
conversion_worker.failed,
self._on_wifi_conversion_failed,
),
)
except RuntimeError:
signal_slots = ()
for signal, slot in signal_slots:
try:
signal.disconnect(slot)
except TypeError:
except (TypeError, RuntimeError):
pass
if conversion_thread is not None and conversion_thread.isRunning():
@@ -996,20 +1224,24 @@ class SettingsPageEventBinder(QObject):
setting_worker = self._android_setting_worker
setting_thread = self._android_setting_thread
if setting_worker is not None:
setting_worker.cancel()
for signal, slot in (
(
setting_worker.succeeded,
self._on_android_device_setting_succeeded,
),
(
setting_worker.failed,
self._on_android_device_setting_failed,
),
):
try:
setting_worker.cancel()
signal_slots = (
(
setting_worker.succeeded,
self._on_android_device_setting_succeeded,
),
(
setting_worker.failed,
self._on_android_device_setting_failed,
),
)
except RuntimeError:
signal_slots = ()
for signal, slot in signal_slots:
try:
signal.disconnect(slot)
except TypeError:
except (TypeError, RuntimeError):
pass
if setting_thread is not None and setting_thread.isRunning():
client/test/test_android_device_service.py
+49
View File
@@ -202,6 +202,55 @@ USB-002 unauthorized usb:1-2 transport_id:3
with self.assertRaisesRegex(AndroidDeviceSearchError, "didn't ACK"):
AndroidDeviceService(runner).search()
def test_package_is_installed_when_pm_returns_apk_path(self):
commands = []
def runner(command, _timeout):
commands.append(list(command))
return completed(
command,
"package:/data/app/com.xunmeng.pinduoduo/base.apk\n",
)
installed = AndroidDeviceService(runner).is_package_installed(
"USB-001"
)
self.assertTrue(installed)
self.assertEqual(
commands,
[[
"adb",
"-s",
"USB-001",
"shell",
"pm",
"path",
"com.xunmeng.pinduoduo",
]],
)
def test_package_is_not_installed_when_pm_returns_no_path(self):
def runner(command, _timeout):
return completed(command, "")
installed = AndroidDeviceService(runner).is_package_installed(
"USB-001"
)
self.assertFalse(installed)
def test_package_check_reports_adb_failure(self):
def runner(command, _timeout):
return completed(
command,
stderr="device offline",
returncode=1,
)
with self.assertRaisesRegex(AndroidDeviceSearchError, "device offline"):
AndroidDeviceService(runner).is_package_installed("USB-001")
def test_convert_usb_to_wifi_connects_while_usb_is_attached(self):
commands = []
client/test/test_settings_ui_event.py
+185 -8
View File
@@ -51,6 +51,9 @@ class SequenceAndroidDeviceService:
self.restore_serials.append(serial)
return self.search(is_cancelled)
def is_package_installed(self, _serial, _package_name):
return True
class SlowAndroidDeviceService:
"""模拟较慢的 ADB 搜索。"""
@@ -68,6 +71,9 @@ class SlowAndroidDeviceService:
self.restore_serial = serial
return self.search(is_cancelled)
def is_package_installed(self, _serial, _package_name):
return True
class ControlledWifiConversionService:
"""模拟 USB 转 Wi-Fi 的进度、成功、失败和延迟。"""
@@ -76,10 +82,18 @@ class ControlledWifiConversionService:
self.delay = delay
self.error = error
self.call_count = 0
self.search_count = 0
def search(self, _is_cancelled=None):
self.search_count += 1
return []
def restore_saved_device(self, _serial, is_cancelled=None):
return self.search(is_cancelled)
def is_package_installed(self, _serial, _package_name):
return True
def convert_usb_to_wifi(
self,
usb_serial,
@@ -106,6 +120,34 @@ class ControlledWifiConversionService:
)
class ControlledPackageCheckService:
"""控制 PDD 包检测结果，其他 ADB 行为返回空列表。"""
def __init__(self, installed=True, delay=0.0, error=None):
self.installed = installed
self.delay = delay
self.error = error
self.checked_serials = []
self.search_count = 0
def search(self, _is_cancelled=None):
self.search_count += 1
return []
def restore_saved_device(self, _serial, _is_cancelled=None):
return self.search(_is_cancelled)
def is_package_installed(self, serial, _package_name):
self.checked_serials.append(serial)
if self.delay:
time.sleep(self.delay)
if self.error is not None:
raise self.error
if isinstance(self.installed, dict):
return self.installed.get(serial, False)
return self.installed
class ControlledSettingsRepository(SettingsRepository):
"""控制 Android 设备设置写入速度和失败，其他设置保持正常。"""
@@ -399,19 +441,24 @@ class SettingsPageEventTest(unittest.TestCase):
page.deleteLater()
def test_selected_android_device_is_saved_to_sqlite(self):
service = ControlledPackageCheckService()
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
page.set_android_devices(
[AndroidDeviceRow("USB-001", "USB", "Pixel 8", "14")]
)
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.saveButton.click()
self.assertFalse(page.saveButton.isEnabled())
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
lambda: self.repository.get(SELECTED_ANDROID_SERIAL_KEY)
== "USB-001"
and page.eventBinder._android_setting_thread is None
)
self.assertEqual(
@@ -670,13 +717,89 @@ class SettingsPageEventTest(unittest.TestCase):
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
def test_selected_device_shows_installed_pdd_package(self):
service = ControlledPackageCheckService(installed=True)
page = SettingsPage(
settings_repository=repository,
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")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
self.assertEqual(service.checked_serials, ["USB-001"])
self.assertEqual(
page.pddAppStatusLabel.text(),
"com.xunmeng.pinduoduo（已安装）",
)
self.assertFalse(hasattr(page, "addressInput"))
self.assertFalse(hasattr(page, "connectButton"))
page.eventBinder.shutdown()
page.deleteLater()
def test_save_rechecks_pdd_and_blocks_device_without_app(self):
service = ControlledPackageCheckService(installed=False)
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")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.saveButton.click()
self._wait_until(
lambda: len(service.checked_serials) == 2
and page.eventBinder._pdd_check_thread is None
)
self.assertIsNone(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY)
)
self.assertIn("未安装", page.pddAppStatusLabel.text())
self.assertIn("保存失败", page.deviceStatusLabel.text())
page.eventBinder.shutdown()
page.deleteLater()
def test_save_blocks_when_pdd_check_fails(self):
service = ControlledPackageCheckService(
error=AndroidDeviceSearchError("device offline")
)
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")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.saveButton.click()
self._wait_until(
lambda: len(service.checked_serials) == 2
and page.eventBinder._pdd_check_thread is None
)
self.assertIsNone(
self.repository.get(SELECTED_ANDROID_SERIAL_KEY)
)
self.assertIn("检测失败", page.pddAppStatusLabel.text())
self.assertIn("原配置未更改", page.deviceStatusLabel.text())
page.eventBinder.shutdown()
page.deleteLater()
def test_device_change_ignores_stale_pdd_check_result(self):
service = ControlledPackageCheckService(
installed={"USB-OLD": True, "USB-NEW": False},
delay=0.05,
)
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
page.set_android_devices(
[
@@ -684,11 +807,45 @@ class SettingsPageEventTest(unittest.TestCase):
AndroidDeviceRow("USB-NEW", "USB"),
]
)
page.deviceTableModel.set_checked_serial("USB-OLD")
self._wait_until(lambda: page.eventBinder._pdd_check_busy)
page.deviceTableModel.set_checked_serial("USB-NEW")
self._wait_until(
lambda: service.checked_serials == ["USB-OLD", "USB-NEW"]
and page.eventBinder._pdd_check_thread is None
)
self.assertEqual(page.deviceTableModel.checked_serial, "USB-NEW")
self.assertIn("未安装", page.pddAppStatusLabel.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
service = ControlledPackageCheckService()
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
self._wait_until(lambda: service.search_count == 1)
self._wait_until(lambda: page.eventBinder._search_thread is None)
page.set_android_devices(
[
AndroidDeviceRow("USB-OLD", "USB"),
AndroidDeviceRow("USB-NEW", "USB"),
]
)
page.deviceTableModel.set_checked_serial("USB-NEW")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.saveButton.click()
self._wait_until(
lambda: page.eventBinder._android_setting_thread is None
lambda: "原设备配置未更改" in page.deviceStatusLabel.text()
and page.eventBinder._android_setting_thread is None
)
self.assertEqual(
@@ -727,12 +884,15 @@ class SettingsPageEventTest(unittest.TestCase):
def test_slow_android_device_save_does_not_block_or_start_twice(self):
repository = ControlledSettingsRepository(self.db_path, delay=0.08)
service = ControlledPackageCheckService()
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
timer_fired = []
QTimer.singleShot(10, lambda: timer_fired.append(True))
@@ -740,7 +900,8 @@ class SettingsPageEventTest(unittest.TestCase):
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
lambda: repository.selected_write_count == 1
and page.eventBinder._android_setting_thread is None
)
self.assertTrue(timer_fired)
@@ -750,12 +911,15 @@ class SettingsPageEventTest(unittest.TestCase):
def test_shutdown_ignores_late_android_device_save_result(self):
repository = ControlledSettingsRepository(self.db_path, delay=0.08)
service = ControlledPackageCheckService()
page = SettingsPage(
settings_repository=repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.saveButton.click()
self._wait_until(
@@ -769,21 +933,25 @@ class SettingsPageEventTest(unittest.TestCase):
page.deleteLater()
def test_convert_wifi_button_requires_connected_usb_device(self):
service = ControlledPackageCheckService()
page = SettingsPage(
settings_repository=self.repository,
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
self.assertFalse(page.convertWifiButton.isEnabled())
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
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._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
self.assertFalse(page.convertWifiButton.isEnabled())
page.set_android_devices(
@@ -802,13 +970,17 @@ class SettingsPageEventTest(unittest.TestCase):
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
self._wait_until(lambda: service.search_count == 1)
self._wait_until(lambda: page.eventBinder._search_thread is None)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.convertWifiButton.click()
self._wait_until(
lambda: page.eventBinder._wifi_conversion_thread is None
)
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
self.assertEqual(service.last_usb_serial, "USB-001")
self.assertEqual(
@@ -844,8 +1016,11 @@ class SettingsPageEventTest(unittest.TestCase):
admin_gateway=MockAdminGateway(),
android_device_service=service,
)
self._wait_until(lambda: service.search_count == 1)
self._wait_until(lambda: page.eventBinder._search_thread is None)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.convertWifiButton.click()
self._wait_until(
@@ -875,6 +1050,7 @@ class SettingsPageEventTest(unittest.TestCase):
)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
timer_fired = []
QTimer.singleShot(10, lambda: timer_fired.append(True))
@@ -899,6 +1075,7 @@ class SettingsPageEventTest(unittest.TestCase):
)
page.set_android_devices([AndroidDeviceRow("USB-001", "USB")])
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
page.convertWifiButton.click()
self._wait_until(