70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""uiautomator2 设备连接边界测试，不连接真机。"""
|
|
|
|
import threading
|
|
import unittest
|
|
|
|
from src.pdd_device_service import PddDeviceError, PddDeviceService
|
|
|
|
|
|
class FakeDevice:
|
|
def app_current(self):
|
|
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_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()
|