183 lines
7.8 KiB
Python
183 lines
7.8 KiB
Python
"""ADB 设备边界测试：所有命令执行器均为 mock，不连接真机。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
|
|
|
|
CLIENT_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
|
|
|
from cmbuyer_client.device.adb import (
|
|
AdbClient,
|
|
CommandResult,
|
|
DeviceIdentityUnconfirmedError,
|
|
DeviceNotFoundError,
|
|
DeviceOfflineError,
|
|
DeviceStateError,
|
|
DeviceUnauthorizedError,
|
|
DuplicatePhysicalDeviceError,
|
|
SerialRequiredError,
|
|
)
|
|
|
|
|
|
USB_SERIAL = "3B65BD02H7F00000"
|
|
WIFI_SERIAL = "192.168.0.173:5555"
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self, devices_output: str, properties: dict[tuple[str, str], CommandResult | str]) -> None:
|
|
self.devices_output = devices_output
|
|
self.properties = properties
|
|
self.calls: list[tuple[str, ...]] = []
|
|
|
|
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
|
self.calls.append(tuple(arguments))
|
|
if tuple(arguments) == ("devices", "-l"):
|
|
return CommandResult(stdout=self.devices_output)
|
|
key = (arguments[1], arguments[-1])
|
|
value = self.properties.get(key, "")
|
|
return value if isinstance(value, CommandResult) else CommandResult(stdout=value)
|
|
|
|
|
|
def _properties(serials: tuple[str, ...]) -> dict[tuple[str, str], str]:
|
|
values: dict[tuple[str, str], str] = {}
|
|
for serial in serials:
|
|
values[(serial, "ro.serialno")] = "physical-phone-1"
|
|
values[(serial, "ro.boot.serialno")] = "physical-phone-1"
|
|
values[(serial, "ro.product.model")] = "PKG110"
|
|
values[(serial, "ro.product.name")] = "PKG110"
|
|
values[(serial, "ro.product.device")] = "OP5D2BL1"
|
|
values[(serial, "ro.build.version.release")] = "16"
|
|
return values
|
|
|
|
|
|
class AdbClientTests(unittest.TestCase):
|
|
def test_requires_explicit_serial(self) -> None:
|
|
runner = FakeRunner("List of devices attached\n", {})
|
|
|
|
with self.assertRaises(SerialRequiredError):
|
|
AdbClient(runner).inspect(" ")
|
|
|
|
self.assertEqual(runner.calls, [])
|
|
|
|
def test_missing_offline_and_unauthorized_are_distinct(self) -> None:
|
|
missing = AdbClient(FakeRunner("List of devices attached\n", {}))
|
|
with self.assertRaises(DeviceNotFoundError):
|
|
missing.inspect(USB_SERIAL)
|
|
|
|
offline = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\toffline\n", {}))
|
|
with self.assertRaises(DeviceOfflineError):
|
|
offline.inspect(USB_SERIAL)
|
|
|
|
unauthorized = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\tunauthorized\n", {}))
|
|
with self.assertRaises(DeviceUnauthorizedError):
|
|
unauthorized.inspect(USB_SERIAL)
|
|
|
|
def test_two_channels_with_same_physical_identity_fail_closed(self) -> None:
|
|
output = (
|
|
"List of devices attached\n"
|
|
f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
)
|
|
runner = FakeRunner(output, _properties((WIFI_SERIAL, USB_SERIAL)))
|
|
|
|
with self.assertRaises(DuplicatePhysicalDeviceError):
|
|
AdbClient(runner).inspect(USB_SERIAL)
|
|
|
|
self.assertIn(("-s", WIFI_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls)
|
|
self.assertIn(("-s", USB_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls)
|
|
|
|
def test_multiple_online_devices_with_failed_identity_fail_closed(self) -> None:
|
|
output = (
|
|
"List of devices attached\n"
|
|
f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
)
|
|
properties = _properties((WIFI_SERIAL, USB_SERIAL))
|
|
properties[(WIFI_SERIAL, "ro.serialno")] = CommandResult(stdout="", returncode=1)
|
|
runner = FakeRunner(output, properties)
|
|
|
|
with self.assertRaises(DeviceIdentityUnconfirmedError):
|
|
AdbClient(runner).inspect(USB_SERIAL)
|
|
|
|
def test_online_explicit_serial_reads_non_sensitive_metadata(self) -> None:
|
|
output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
inspection = AdbClient(FakeRunner(output, _properties((USB_SERIAL,)))).inspect(USB_SERIAL)
|
|
|
|
self.assertEqual(inspection.device.serial, USB_SERIAL)
|
|
self.assertEqual(inspection.model, "PKG110")
|
|
self.assertEqual(inspection.android_version, "16")
|
|
|
|
def test_single_online_device_does_not_require_hardware_identity(self) -> None:
|
|
output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
properties = _properties((USB_SERIAL,))
|
|
properties[(USB_SERIAL, "ro.serialno")] = ""
|
|
properties[(USB_SERIAL, "ro.boot.serialno")] = ""
|
|
|
|
inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
|
|
|
|
self.assertEqual(inspection.model, "PKG110")
|
|
|
|
def test_multiple_online_devices_without_hardware_identity_are_unconfirmed(self) -> None:
|
|
output = (
|
|
"List of devices attached\n"
|
|
f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
)
|
|
properties = _properties((WIFI_SERIAL, USB_SERIAL))
|
|
properties[(WIFI_SERIAL, "ro.serialno")] = ""
|
|
properties[(WIFI_SERIAL, "ro.boot.serialno")] = ""
|
|
|
|
with self.assertRaises(DeviceIdentityUnconfirmedError):
|
|
AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
|
|
|
|
def test_multiple_online_devices_with_different_identity_keep_explicit_selection(self) -> None:
|
|
output = (
|
|
"List of devices attached\n"
|
|
f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
)
|
|
properties = _properties((WIFI_SERIAL, USB_SERIAL))
|
|
properties[(WIFI_SERIAL, "ro.serialno")] = "physical-phone-2"
|
|
properties[(WIFI_SERIAL, "ro.boot.serialno")] = "physical-phone-2"
|
|
|
|
inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
|
|
|
|
self.assertEqual(inspection.device.serial, USB_SERIAL)
|
|
|
|
def test_shared_boot_serial_is_duplicate_even_when_ro_serial_differs(self) -> None:
|
|
output = (
|
|
"List of devices attached\n"
|
|
f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
|
|
)
|
|
properties = _properties((WIFI_SERIAL, USB_SERIAL))
|
|
properties[(WIFI_SERIAL, "ro.serialno")] = "wifi-transport-serial"
|
|
properties[(USB_SERIAL, "ro.serialno")] = "usb-transport-serial"
|
|
properties[(WIFI_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial"
|
|
properties[(USB_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial"
|
|
|
|
with self.assertRaises(DuplicatePhysicalDeviceError):
|
|
AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
|
|
|
|
def test_unknown_adb_state_is_rejected(self) -> None:
|
|
client = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\trecovery\n", {}))
|
|
|
|
with self.assertRaises(DeviceStateError):
|
|
client.inspect(USB_SERIAL)
|
|
|
|
def test_runner_timeout_is_a_distinct_connection_error(self) -> None:
|
|
class TimeoutRunner:
|
|
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
|
raise subprocess.TimeoutExpired(arguments, timeout_seconds)
|
|
|
|
from cmbuyer_client.device.adb import DeviceCommandTimeoutError
|
|
|
|
with self.assertRaises(DeviceCommandTimeoutError):
|
|
AdbClient(TimeoutRunner()).inspect(USB_SERIAL)
|