94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""uiautomator2 设备连接边界测试，不连接真机。"""
|
|
|
|
import threading
|
|
import unittest
|
|
|
|
from src.pdd_device_service import PddDeviceError, PddDeviceService
|
|
from src.performance_timing import TaskPerformanceTrace
|
|
|
|
|
|
class FakeDevice:
|
|
def __init__(self):
|
|
self.current_calls = 0
|
|
|
|
def app_current(self):
|
|
self.current_calls += 1
|
|
return {"package": "com.xunmeng.pinduoduo"}
|
|
|
|
|
|
class PddDeviceServiceTest(unittest.TestCase):
|
|
def test_connect_validates_and_releases_exclusive_session(self):
|
|
connected = []
|
|
service = PddDeviceService(lambda serial: connected.append(serial) or FakeDevice())
|
|
|
|
with service.connect("USB-001") as device:
|
|
self.assertIsInstance(device, FakeDevice)
|
|
with self.assertRaisesRegex(PddDeviceError, "正在执行"):
|
|
service.connect("USB-002")
|
|
|
|
with service.connect("USB-002"):
|
|
pass
|
|
self.assertEqual(connected, ["USB-001", "USB-002"])
|
|
|
|
def test_connect_exposes_initial_state_and_records_two_separate_stages(self):
|
|
device = FakeDevice()
|
|
records = []
|
|
trace = TaskPerformanceTrace(sink=records.append)
|
|
trace.bind_task("COL-001")
|
|
|
|
with trace.activate():
|
|
session = PddDeviceService(lambda _serial: device).connect("USB-001")
|
|
self.assertEqual(
|
|
session.initial_app_state["package"],
|
|
"com.xunmeng.pinduoduo",
|
|
)
|
|
self.assertEqual(device.current_calls, 1)
|
|
self.assertEqual(
|
|
[record["operation"] for record in records],
|
|
["uiautomator2_connect", "first_app_current"],
|
|
)
|
|
session.__exit__(None, None, None)
|
|
|
|
def test_invalid_serial_is_rejected_before_connect(self):
|
|
with self.assertRaisesRegex(PddDeviceError, "设备号无效") as raised:
|
|
PddDeviceService(lambda _serial: FakeDevice()).connect("bad serial")
|
|
self.assertEqual(raised.exception.code, "DEVICE_ADDRESS_INVALID")
|
|
|
|
def test_offline_connector_error_has_specific_code(self):
|
|
def fail(_serial):
|
|
raise OSError("offline")
|
|
|
|
with self.assertRaises(PddDeviceError) as raised:
|
|
PddDeviceService(fail).connect("USB-001")
|
|
self.assertEqual(raised.exception.code, "DEVICE_OFFLINE")
|
|
self.assertIn("offline", raised.exception.message)
|
|
|
|
def test_other_connector_error_becomes_connect_failed(self):
|
|
def fail(_serial):
|
|
raise OSError("permission denied")
|
|
|
|
with self.assertRaises(PddDeviceError) as raised:
|
|
PddDeviceService(fail).connect("USB-001")
|
|
self.assertEqual(raised.exception.code, "DEVICE_CONNECT_FAILED")
|
|
|
|
def test_device_object_cannot_be_used_from_another_thread(self):
|
|
session = PddDeviceService(lambda _serial: FakeDevice()).connect("USB-001")
|
|
errors = []
|
|
|
|
def read_device():
|
|
try:
|
|
session.device
|
|
except PddDeviceError as exc:
|
|
errors.append(exc.code)
|
|
|
|
worker = threading.Thread(target=read_device)
|
|
worker.start()
|
|
worker.join()
|
|
session.__exit__(None, None, None)
|
|
|
|
self.assertEqual(errors, ["DEVICE_THREAD_VIOLATION"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|