cmautobuy/client/test/test_pdd_device_service.py

"""uiautomator2 设备连接边界测试，不连接真机。"""

import threading
import unittest

from src.pdd_device_service import (
    PddDeviceError,
    PddDeviceService,
    PersistentPddDeviceService,
    bind_thread_device_service,
    current_thread_device_service,
)
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"])


class PersistentPddDeviceServiceTest(unittest.TestCase):
    def test_same_serial_within_ttl_connects_once_but_checks_each_task(self):
        connected = []
        device = FakeDevice()
        service = PersistentPddDeviceService(
            lambda serial: connected.append(serial) or device,
            ttl_seconds=90,
        )

        with service.connect("USB-001"):
            pass
        with service.connect("USB-001"):
            pass

        self.assertEqual(connected, ["USB-001"])
        self.assertEqual(device.current_calls, 2)

    def test_expired_or_changed_serial_discards_cached_connection(self):
        now = [10.0]
        connected = []
        service = PersistentPddDeviceService(
            lambda serial: connected.append(serial) or FakeDevice(),
            ttl_seconds=5,
            monotonic=lambda: now[0],
        )

        with service.connect("USB-001"):
            pass
        now[0] = 16.0
        with service.connect("USB-001"):
            pass
        with service.connect("USB-002"):
            pass

        self.assertEqual(connected, ["USB-001", "USB-001", "USB-002"])

    def test_failed_cached_health_check_reconnects_only_once(self):
        class UnhealthyDevice(FakeDevice):
            def app_current(self):
                self.current_calls += 1
                if self.current_calls >= 2:
                    raise OSError("offline")
                return {"package": "com.xunmeng.pinduoduo"}

        devices = [UnhealthyDevice(), FakeDevice()]
        connected = []
        service = PersistentPddDeviceService(
            lambda serial: connected.append(serial) or devices.pop(0)
        )

        with service.connect("USB-001"):
            pass
        with service.connect("USB-001") as current:
            self.assertIsInstance(current, FakeDevice)

        self.assertEqual(connected, ["USB-001", "USB-001"])

    def test_release_and_context_binding_stay_on_owner_thread(self):
        service = PersistentPddDeviceService(lambda _serial: FakeDevice())
        with service.connect("USB-001"):
            pass
        with bind_thread_device_service(service):
            self.assertIs(current_thread_device_service(), service)
        self.assertIsNone(current_thread_device_service())

        errors = []

        def release_from_other_thread():
            try:
                service.release_cached()
            except PddDeviceError as exc:
                errors.append(exc.code)

        worker = threading.Thread(target=release_from_other_thread)
        worker.start()
        worker.join()
        self.assertEqual(errors, ["DEVICE_THREAD_VIOLATION"])
        service.release_cached()
        self.assertFalse(service.has_cached_device)


if __name__ == "__main__":
    unittest.main()