1177 lines
44 KiB
Python
1177 lines
44 KiB
Python
"""设置页保存当前 Client 并后台登记的离屏测试。"""
|
|
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PyQt5.QtCore import Qt, QTimer
|
|
from PyQt5.QtTest import QTest
|
|
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
|
|
from src.settings_repository import SettingsRepository
|
|
from src.settings_ui import AndroidDeviceRow, SettingsPage
|
|
from src.task_models import TaskType
|
|
|
|
|
|
class SlowMockAdminGateway(MockAdminGateway):
|
|
"""让登记停留一小段时间，用来证明主线程仍能处理事件。"""
|
|
|
|
def register_client(self, client, capabilities):
|
|
time.sleep(0.08)
|
|
return super().register_client(client, capabilities)
|
|
|
|
|
|
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)]
|
|
self.call_count += 1
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return result
|
|
|
|
def restore_saved_device(self, serial, is_cancelled=None):
|
|
self.restore_serials.append(serial)
|
|
return self.search(is_cancelled)
|
|
|
|
def is_package_installed(self, _serial, _package_name):
|
|
return True
|
|
|
|
|
|
class SlowAndroidDeviceService:
|
|
"""模拟较慢的 ADB 搜索。"""
|
|
|
|
def __init__(self, delay=0.08):
|
|
self.delay = delay
|
|
self.call_count = 0
|
|
|
|
def search(self, _is_cancelled=None):
|
|
self.call_count += 1
|
|
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)
|
|
|
|
def is_package_installed(self, _serial, _package_name):
|
|
return True
|
|
|
|
|
|
class ControlledWifiConversionService:
|
|
"""模拟 USB 转 Wi-Fi 的进度、成功、失败和延迟。"""
|
|
|
|
def __init__(self, delay=0.0, error=None):
|
|
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,
|
|
is_cancelled=None,
|
|
on_progress=None,
|
|
):
|
|
self.call_count += 1
|
|
if on_progress is not None:
|
|
on_progress("正在连接 Wi-Fi 设备 192.168.0.173:5555…")
|
|
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(usb_serial, "USB", "Phone", "14"),
|
|
AndroidDevice(wifi_serial, "Wi-Fi", "Phone", "14"),
|
|
],
|
|
)
|
|
|
|
|
|
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 设备设置写入速度和失败，其他设置保持正常。"""
|
|
|
|
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):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self):
|
|
self.temp_directory = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.temp_directory.name) / "client.db"
|
|
self.repository = SettingsRepository(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.temp_directory.cleanup()
|
|
|
|
def _wait_until(self, predicate, timeout_ms=2000):
|
|
elapsed = 0
|
|
while not predicate() and elapsed < timeout_ms:
|
|
QTest.qWait(10)
|
|
elapsed += 10
|
|
self.assertTrue(predicate(), "等待异步操作超时")
|
|
|
|
def test_existing_client_info_is_restored_when_page_opens(self):
|
|
self.repository.set_many(
|
|
{
|
|
"admin.client_id": "CLIENT-EXISTING",
|
|
"admin.client_name": "办公室电脑",
|
|
}
|
|
)
|
|
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
)
|
|
|
|
self.assertEqual(page.deviceIdInput.text(), "CLIENT-EXISTING")
|
|
self.assertEqual(page.deviceNameInput.text(), "办公室电脑")
|
|
self.assertEqual(page.currentDeviceStatusLabel.text(), "本地设备信息已加载")
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_settings_page_has_no_manual_live_purchase_authorization(self):
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
live_purchase_adapter_ready=True,
|
|
)
|
|
|
|
self.assertFalse(hasattr(page, "livePurchaseCard"))
|
|
self.assertFalse(hasattr(page, "livePurchaseEnableButton"))
|
|
self.assertFalse(hasattr(page, "livePurchaseConfirmationInput"))
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_ready_client_automatically_registers_live_capability(self):
|
|
self.repository.set_many(
|
|
{
|
|
"admin.client_id": "CLIENT-LIVE",
|
|
"admin.client_name": "采购电脑",
|
|
SELECTED_ANDROID_SERIAL_KEY: "USB-LIVE",
|
|
}
|
|
)
|
|
gateway = MockAdminGateway()
|
|
service = SequenceAndroidDeviceService([[]])
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=gateway,
|
|
android_device_service=service,
|
|
live_purchase_adapter_ready=True,
|
|
)
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(lambda: gateway.registration_count == 1)
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
registered = gateway.registered_client("CLIENT-LIVE")
|
|
self.assertEqual(registered[1].purchase_mode, "live")
|
|
self.assertIn(TaskType.PURCHASE, registered[1].supported_types)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_save_persists_locally_and_registers_with_mock(self):
|
|
gateway = MockAdminGateway()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=gateway,
|
|
)
|
|
page.deviceNameInput.setText(" 办公室电脑 ")
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self.assertFalse(page.currentDeviceSaveButton.isEnabled())
|
|
self._wait_until(lambda: gateway.registration_count == 1)
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
client_id = self.repository.get("admin.client_id")
|
|
self.assertTrue(client_id.startswith("CLIENT-"))
|
|
self.assertEqual(
|
|
self.repository.get("admin.client_name"), "办公室电脑"
|
|
)
|
|
self.assertEqual(page.deviceIdInput.text(), client_id)
|
|
self.assertEqual(
|
|
page.currentDeviceStatusLabel.text(),
|
|
"本地已保存，已登记到 Admin",
|
|
)
|
|
self.assertTrue(page.currentDeviceSaveButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_admin_failure_keeps_local_values_and_allows_retry(self):
|
|
gateway = MockAdminGateway()
|
|
gateway.fail_next_call_temporarily()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=gateway,
|
|
)
|
|
page.deviceNameInput.setText("仓库电脑")
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
self.assertTrue(self.repository.get("admin.client_id"))
|
|
self.assertEqual(self.repository.get("admin.client_name"), "仓库电脑")
|
|
self.assertIn("本地已保存，Admin 登记失败", page.currentDeviceStatusLabel.text())
|
|
self.assertIn("可再次点击保存重试", page.currentDeviceStatusLabel.text())
|
|
self.assertTrue(page.currentDeviceSaveButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_slow_registration_does_not_block_main_event_loop(self):
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=SlowMockAdminGateway(),
|
|
)
|
|
timer_fired = []
|
|
QTimer.singleShot(10, lambda: timer_fired.append(True))
|
|
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
|
|
self._wait_until(lambda: page.eventBinder._thread is None)
|
|
|
|
self.assertTrue(timer_fired)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_shutdown_ignores_late_registration_result(self):
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=SlowMockAdminGateway(),
|
|
)
|
|
page.currentDeviceSaveButton.click()
|
|
self._wait_until(
|
|
lambda: "正在登记" in page.currentDeviceStatusLabel.text()
|
|
)
|
|
status_before_close = page.currentDeviceStatusLabel.text()
|
|
|
|
page.eventBinder.shutdown()
|
|
QTest.qWait(120)
|
|
|
|
self.assertEqual(page.currentDeviceStatusLabel.text(), status_before_close)
|
|
page.deleteLater()
|
|
|
|
def test_search_updates_table_and_device_status(self):
|
|
service = SequenceAndroidDeviceService(
|
|
[[
|
|
AndroidDevice("USB-001", "USB", "Pixel 8", "14"),
|
|
AndroidDevice(
|
|
"192.168.0.173:5555",
|
|
"Wi-Fi",
|
|
status="offline",
|
|
),
|
|
]]
|
|
)
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self.assertFalse(page.searchButton.isEnabled())
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
model = page.deviceTableModel
|
|
self.assertEqual(model.rowCount(), 2)
|
|
self.assertEqual(model.device_for_serial("USB-001").model, "Pixel 8")
|
|
self.assertEqual(
|
|
model.device_for_serial("192.168.0.173:5555").status,
|
|
"offline",
|
|
)
|
|
offline_index = model.index(
|
|
model.row_for_serial("192.168.0.173:5555"), 0
|
|
)
|
|
self.assertFalse(
|
|
model.setData(offline_index, Qt.Checked, Qt.CheckStateRole)
|
|
)
|
|
self.assertEqual(
|
|
page.deviceStatusLabel.text(),
|
|
"搜索完成：找到 2 台 Android 设备",
|
|
)
|
|
self.assertTrue(page.searchButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_empty_search_keeps_placeholder_rows(self):
|
|
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.assertGreaterEqual(page.deviceTableModel.rowCount(), 1)
|
|
self.assertIn("未找到 Android 设备", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_search_failure_preserves_previous_rows(self):
|
|
service = SequenceAndroidDeviceService(
|
|
[AndroidDeviceSearchError("未找到 adb，请检查 PATH 后重试")]
|
|
)
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
page.set_android_devices(
|
|
[AndroidDeviceRow("USB-OLD", "USB", "Old Phone", "12")]
|
|
)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial("USB-OLD")
|
|
)
|
|
self.assertIn("搜索失败：未找到 adb", page.deviceStatusLabel.text())
|
|
self.assertIn("已保存配置未更改", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_search_preserves_or_clears_checked_device_by_serial(self):
|
|
connected = AndroidDevice("USB-001", "USB", "Pixel 8", "14")
|
|
service = SequenceAndroidDeviceService([[connected], [connected], []])
|
|
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)
|
|
page.deviceTableModel.set_checked_serial("USB-001")
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, "USB-001")
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, "")
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_slow_search_does_not_block_or_start_twice(self):
|
|
service = SlowAndroidDeviceService()
|
|
page = SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
android_device_service=service,
|
|
)
|
|
timer_fired = []
|
|
QTimer.singleShot(10, lambda: timer_fired.append(True))
|
|
|
|
page.searchButton.click()
|
|
page.eventBinder._request_search()
|
|
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.call_count, 1)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_shutdown_ignores_late_search_result(self):
|
|
service = SlowAndroidDeviceService()
|
|
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 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_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)
|
|
changed_serials = []
|
|
page.eventBinder.androidDeviceConfigurationChanged.connect(
|
|
changed_serials.append
|
|
)
|
|
|
|
page.saveButton.click()
|
|
self.assertFalse(page.saveButton.isEnabled())
|
|
self._wait_until(
|
|
lambda: self.repository.get(SELECTED_ANDROID_SERIAL_KEY)
|
|
== "USB-001"
|
|
and 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())
|
|
self.assertEqual(changed_serials, ["USB-001"])
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
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(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_manual_search_reconnects_saved_wifi_before_refresh(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")],
|
|
[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)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: service.call_count == 2)
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertEqual(service.restore_serials, [wifi_serial, wifi_serial])
|
|
self.assertEqual(page.deviceTableModel.checked_serial, wifi_serial)
|
|
self.assertIn("搜索完成", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_manual_wifi_reconnect_failure_still_lists_usb_device(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")],
|
|
AndroidDeviceSearchError("目标设备拒绝连接"),
|
|
[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)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: service.call_count == 3)
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertEqual(
|
|
self.repository.get(SELECTED_ANDROID_SERIAL_KEY),
|
|
wifi_serial,
|
|
)
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial("USB-001")
|
|
)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, "")
|
|
self.assertIn("找到 1 台 Android 设备", page.deviceStatusLabel.text())
|
|
self.assertIn("恢复失败", page.deviceStatusLabel.text())
|
|
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 == 2)
|
|
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("未找到其他 Android 设备", page.deviceStatusLabel.text())
|
|
self.assertIn("配置已保留", page.deviceStatusLabel.text())
|
|
self.assertTrue(page.searchButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_search_failure_after_restore_failure_preserves_previous_rows(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")],
|
|
AndroidDeviceSearchError("目标设备拒绝连接"),
|
|
AndroidDeviceSearchError("adb devices 执行失败"),
|
|
]
|
|
)
|
|
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)
|
|
|
|
page.searchButton.click()
|
|
self._wait_until(lambda: service.call_count == 3)
|
|
self._wait_until(lambda: page.eventBinder._search_thread is None)
|
|
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial(wifi_serial)
|
|
)
|
|
self.assertEqual(page.deviceTableModel.checked_serial, wifi_serial)
|
|
self.assertIn("搜索失败", page.deviceStatusLabel.text())
|
|
self.assertIn("adb devices 执行失败", page.deviceStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_saved_wifi_waiting_for_authorization_has_clear_message(self):
|
|
wifi_serial = "192.168.0.173:5555"
|
|
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, wifi_serial)
|
|
service = SequenceAndroidDeviceService(
|
|
[[AndroidDevice(wifi_serial, "Wi-Fi", status="unauthorized")]]
|
|
)
|
|
|
|
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.deviceTableModel.checked_serial, "")
|
|
self.assertIn("尚未授权", page.deviceStatusLabel.text())
|
|
self.assertIn("请在手机上允许调试", page.deviceStatusLabel.text())
|
|
self.assertEqual(
|
|
self.repository.get(SELECTED_ANDROID_SERIAL_KEY),
|
|
wifi_serial,
|
|
)
|
|
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(
|
|
[[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_device_shows_installed_pdd_package(self):
|
|
service = ControlledPackageCheckService(installed=True)
|
|
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)
|
|
|
|
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(
|
|
[
|
|
AndroidDeviceRow("USB-OLD", "USB"),
|
|
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: "原设备配置未更改" in page.deviceStatusLabel.text()
|
|
and 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)
|
|
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))
|
|
|
|
page.saveButton.click()
|
|
page.eventBinder._request_save()
|
|
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
|
|
self._wait_until(
|
|
lambda: repository.selected_write_count == 1
|
|
and 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)
|
|
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(
|
|
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()
|
|
|
|
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(
|
|
[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,
|
|
)
|
|
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(
|
|
page.deviceTableModel.checked_serial,
|
|
"192.168.0.173:5555",
|
|
)
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial("USB-001")
|
|
)
|
|
self.assertIsNotNone(
|
|
page.deviceTableModel.device_for_serial(
|
|
"192.168.0.173:5555"
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.repository.get(SELECTED_ANDROID_SERIAL_KEY),
|
|
"USB-OLD",
|
|
)
|
|
self.assertIn("已连接并勾选", page.deviceStatusLabel.text())
|
|
self.assertIn("保存，再拔掉 USB", 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("Wi-Fi 连接验证失败")
|
|
)
|
|
self.repository.set(SELECTED_ANDROID_SERIAL_KEY, "USB-OLD")
|
|
page = SettingsPage(
|
|
settings_repository=self.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-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.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")
|
|
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
|
|
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")
|
|
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
|
|
|
|
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()
|