cmautobuy/client/test/test_android_device_service.py

"""ADB Android 设备搜索服务测试。"""

import subprocess
import unittest

from src.android_device_service import (
    AndroidDeviceSearchError,
    AndroidDeviceService,
)


def completed(command, stdout="", stderr="", returncode=0):
    """创建一条假的 ADB 命令结果。"""

    return subprocess.CompletedProcess(command, returncode, stdout, stderr)


class AndroidDeviceServiceTest(unittest.TestCase):
    def test_parse_wifi_ipv4_from_route(self):
        output = (
            "10.20.0.0/16 dev rmnet_data0 scope link src 10.20.1.2\n"
            "default via 192.168.0.1 dev wlan0 proto dhcp\n"
            "192.168.0.0/24 dev wlan0 proto kernel scope link "
            "src 192.168.0.173\n"
        )

        self.assertEqual(
            AndroidDeviceService.parse_wifi_ipv4(output),
            "192.168.0.173",
        )

    def test_missing_wifi_ipv4_has_recovery_message(self):
        with self.assertRaisesRegex(AndroidDeviceSearchError, "连接.*Wi-Fi"):
            AndroidDeviceService.parse_wifi_ipv4(
                "default via 10.0.0.1 dev wlan0\n"
            )

    def test_parse_usb_wifi_and_unavailable_devices(self):
        output = """List of devices attached
USB-001 device product:p model:Pixel_8 device:husky transport_id:1
192.168.0.173:5555 offline transport_id:2
USB-002 unauthorized usb:1-2 transport_id:3
"""

        devices = AndroidDeviceService.parse_devices(output)

        self.assertEqual(len(devices), 3)
        self.assertEqual(devices[0].connection_type, "USB")
        self.assertEqual(devices[0].model, "Pixel 8")
        self.assertEqual(devices[1].connection_type, "Wi-Fi")
        self.assertEqual(devices[1].status, "offline")
        self.assertEqual(devices[2].status, "unauthorized")

    def test_search_reads_version_only_for_connected_device(self):
        commands = []

        def runner(command, _timeout):
            commands.append(list(command))
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\n"
                    "USB-001 device model:Pixel_8 transport_id:1\n"
                    "192.168.0.173:5555 offline transport_id:2\n",
                )
            return completed(command, "14\n")

        devices = AndroidDeviceService(runner).search()

        self.assertEqual(devices[0].android_version, "14")
        self.assertEqual(devices[1].android_version, "—")
        self.assertEqual(len(commands), 2)
        self.assertIn("ro.build.version.release", commands[1])

    def test_require_connected_accepts_exact_usb_and_wifi_serial(self):
        output = (
            "List of devices attached\n"
            "USB-001 device model:Phone\n"
            "192.168.0.173:5555 device model:Phone\n"
        )
        service = AndroidDeviceService(
            lambda command, _timeout: completed(command, output)
        )

        usb = service.require_connected("USB-001")
        wifi = service.require_connected("192.168.0.173:5555")

        self.assertEqual(usb.serial, "USB-001")
        self.assertEqual(wifi.serial, "192.168.0.173:5555")

    def test_require_connected_rejects_missing_offline_and_unauthorized(self):
        cases = (
            ("USB-MISSING", "", "USB.*未连接"),
            ("192.168.0.173:5555", "offline", "offline"),
            ("USB-002", "unauthorized", "未授权"),
        )
        for serial, status, expected in cases:
            with self.subTest(serial=serial, status=status):
                line = f"{serial} {status}\n" if status else ""
                service = AndroidDeviceService(
                    lambda command, _timeout, value=line: completed(
                        command, "List of devices attached\n" + value
                    )
                )
                with self.assertRaisesRegex(AndroidDeviceSearchError, expected):
                    service.require_connected(serial)

    def test_restore_saved_wifi_connects_before_search(self):
        commands = []

        def runner(command, _timeout):
            command = list(command)
            commands.append(command)
            if command[:2] == ["adb", "connect"]:
                return completed(
                    command,
                    "already connected to 192.168.0.173:5555\n",
                )
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\n"
                    "192.168.0.173:5555 device model:Phone\n",
                )
            return completed(command, "14\n")

        devices = AndroidDeviceService(runner).restore_saved_device(
            "192.168.0.173:5555"
        )

        self.assertEqual(devices[0].serial, "192.168.0.173:5555")
        self.assertEqual(
            commands[0],
            ["adb", "connect", "192.168.0.173:5555"],
        )

    def test_restore_saved_wifi_waits_for_authorization_handshake(self):
        device_list_count = 0
        sleeps = []

        def runner(command, _timeout):
            nonlocal device_list_count
            command = list(command)
            if command[:2] == ["adb", "connect"]:
                return completed(command, "connected to 192.168.0.173:5555\n")
            if command == ["adb", "devices", "-l"]:
                device_list_count += 1
                status = "unauthorized" if device_list_count == 1 else "device"
                return completed(
                    command,
                    "List of devices attached\n"
                    f"192.168.0.173:5555 {status} model:Phone\n",
                )
            return completed(command, "14\n")

        devices = AndroidDeviceService(
            runner,
            sleeper=sleeps.append,
        ).restore_saved_device("192.168.0.173:5555")

        self.assertEqual(devices[0].status, "device")
        self.assertEqual(sleeps, [0.25])

    def test_restore_saved_usb_only_searches(self):
        commands = []

        def runner(command, _timeout):
            command = list(command)
            commands.append(command)
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\n"
                    "USB-001 device model:Phone\n",
                )
            return completed(command, "14\n")

        devices = AndroidDeviceService(runner).restore_saved_device(
            "USB-001"
        )

        self.assertEqual(devices[0].serial, "USB-001")
        self.assertFalse(
            any(command[:2] == ["adb", "connect"] for command in commands)
        )

    def test_missing_model_is_read_from_device(self):
        def runner(command, _timeout):
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\nUSB-001 device transport_id:1\n",
                )
            if command[-1] == "ro.product.model":
                return completed(command, "Redmi K70\n")
            return completed(command, "15\n")

        device = AndroidDeviceService(runner).search()[0]

        self.assertEqual(device.model, "Redmi K70")
        self.assertEqual(device.android_version, "15")

    def test_property_failure_keeps_discovered_device(self):
        def runner(command, _timeout):
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\nUSB-001 device model:Phone\n",
                )
            return completed(command, stderr="device disconnected", returncode=1)

        device = AndroidDeviceService(runner).search()[0]

        self.assertEqual(device.serial, "USB-001")
        self.assertEqual(device.android_version, "—")

    def test_missing_adb_has_recovery_message(self):
        def runner(_command, _timeout):
            raise FileNotFoundError("adb")

        with self.assertRaisesRegex(AndroidDeviceSearchError, "未找到 adb"):
            AndroidDeviceService(runner).search()

    def test_timeout_has_recovery_message(self):
        def runner(command, timeout):
            raise subprocess.TimeoutExpired(command, timeout)

        with self.assertRaisesRegex(AndroidDeviceSearchError, "搜索设备超时"):
            AndroidDeviceService(runner).search()

    def test_nonzero_exit_includes_adb_error(self):
        def runner(command, _timeout):
            return completed(command, stderr="ADB server didn't ACK", returncode=1)

        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 = []

        def runner(command, _timeout):
            command = list(command)
            commands.append(command)
            if command[-3:] == ["shell", "ip", "route"]:
                return completed(
                    command,
                    "192.168.0.0/24 dev wlan0 scope link "
                    "src 192.168.0.173\n",
                )
            if command[:2] == ["adb", "connect"]:
                return completed(command, "connected to 192.168.0.173:5555\n")
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\n"
                    "USB-001 device model:Phone\n"
                    "192.168.0.173:5555 device model:Phone\n",
                )
            if command[-1] == "ro.build.version.release":
                return completed(command, "14\n")
            return completed(command, "restarting in TCP mode port: 5555\n")

        progress = []
        service = AndroidDeviceService(runner)

        result = service.convert_usb_to_wifi(
            "USB-001",
            on_progress=progress.append,
        )

        self.assertEqual(result.wifi_serial, "192.168.0.173:5555")
        self.assertEqual(
            [device.serial for device in result.devices],
            ["USB-001", "192.168.0.173:5555"],
        )
        self.assertIn(
            ["adb", "-s", "USB-001", "tcpip", "5555"],
            commands,
        )
        self.assertEqual(
            commands.count(["adb", "connect", "192.168.0.173:5555"]),
            1,
        )
        self.assertEqual(commands.count(["adb", "devices", "-l"]), 3)
        self.assertTrue(
            any("保存后拔掉 USB" in message for message in progress)
        )

    def test_convert_usb_to_wifi_rejects_wifi_serial(self):
        with self.assertRaisesRegex(AndroidDeviceSearchError, "已经是 Wi-Fi"):
            AndroidDeviceService().convert_usb_to_wifi(
                "192.168.0.173:5555"
            )

    def test_convert_usb_to_wifi_requires_connected_wifi_status(self):
        def runner(command, _timeout):
            command = list(command)
            if command[-3:] == ["shell", "ip", "route"]:
                return completed(command, "local src 192.168.0.173\n")
            if command[:2] == ["adb", "connect"]:
                return completed(command, "connected to 192.168.0.173:5555\n")
            if command == ["adb", "devices", "-l"]:
                return completed(
                    command,
                    "List of devices attached\n"
                    "192.168.0.173:5555 offline\n",
                )
            return completed(command, "restarting in TCP mode port: 5555\n")

        service = AndroidDeviceService(runner, sleeper=lambda _seconds: None)

        with self.assertRaisesRegex(AndroidDeviceSearchError, "状态为 offline"):
            service.convert_usb_to_wifi("USB-001")


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