175 lines
8.0 KiB
Python
175 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
DIRECTORY = Path(__file__).resolve().parent
|
|
PRIVATE_NAME = "测试设备.local.json"
|
|
EXAMPLE_NAME = "测试设备.example.json"
|
|
|
|
|
|
class RegistrationError(RuntimeError):
|
|
"""A sanitized registration error; never include device values."""
|
|
|
|
|
|
def validate_document(document: object) -> list[dict]:
|
|
if not isinstance(document, dict) or document.get("schema_version") != 1:
|
|
raise RegistrationError("schema_version 必须为 1。")
|
|
devices = document.get("devices")
|
|
if not isinstance(devices, list) or not devices:
|
|
raise RegistrationError("devices 必须为非空列表。")
|
|
aliases: set[str] = set()
|
|
serials: set[str] = set()
|
|
for device in devices:
|
|
if not isinstance(device, dict):
|
|
raise RegistrationError("devices 条目格式错误。")
|
|
for field in ("alias", "serial", "model", "android_release"):
|
|
value = device.get(field)
|
|
if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 for c in value):
|
|
raise RegistrationError(f"字段缺失或格式错误：{field}")
|
|
if "<" in value or ">" in value or "待填写" in value:
|
|
raise RegistrationError(f"字段仍为占位内容：{field}")
|
|
if not re.fullmatch(r"android-test-[a-z0-9-]+", device["alias"]):
|
|
raise RegistrationError("alias 必须使用 android-test- 开头的 ASCII 测试代号。")
|
|
if device["serial"] != device["serial"].strip() or any(c.isspace() for c in device["serial"]):
|
|
raise RegistrationError("serial 格式错误。")
|
|
api = device.get("api_level")
|
|
if type(api) is not int or api < 26:
|
|
raise RegistrationError("api_level 必须为不小于 26 的整数。")
|
|
if device.get("connection") != "usb":
|
|
raise RegistrationError("当前自动登记仅支持 usb。")
|
|
for field in ("dedicated_test_device", "automation_allowed", "visual_check_requires_user_instruction"):
|
|
if device.get(field) is not True:
|
|
raise RegistrationError(f"授权或门禁字段未明确启用：{field}")
|
|
if device["alias"] in aliases or device["serial"] in serials:
|
|
raise RegistrationError("登记存在重复 alias 或 serial。")
|
|
aliases.add(device["alias"])
|
|
serials.add(device["serial"])
|
|
return devices
|
|
|
|
|
|
def load_registration(directory: Path = DIRECTORY) -> list[dict]:
|
|
path = directory / PRIVATE_NAME
|
|
if not path.exists():
|
|
try:
|
|
with path.open("xb") as stream:
|
|
stream.write((directory / EXAMPLE_NAME).read_bytes())
|
|
except FileExistsError:
|
|
pass
|
|
raise RegistrationError("已准备空白私有登记；请填写真实登记后重新检查，当前停止真机步骤。")
|
|
try:
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
raise RegistrationError("真实登记无法读取或不是有效 UTF-8 JSON；原文件未改动。") from None
|
|
return validate_document(document)
|
|
|
|
|
|
def run_adb(adb: str, arguments: list[str]) -> str:
|
|
try:
|
|
result = subprocess.run(
|
|
[adb, *arguments], capture_output=True, text=True,
|
|
encoding="utf-8", errors="replace", timeout=20,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
raise RegistrationError("ADB 不可执行或操作超时；未输出命令及设备标识。") from None
|
|
if result.returncode != 0:
|
|
raise RegistrationError("ADB 操作失败；未输出原始设备日志。")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def list_devices(adb: str) -> list[tuple[str, str, bool]]:
|
|
rows = []
|
|
# Windows ADB often omits usb: from devices -l and returns unknown for
|
|
# get-devpath. -d selects a USB transport only; ambiguity must fail closed.
|
|
try:
|
|
single_usb_serial = run_adb(adb, ["-d", "get-serialno"])
|
|
except RegistrationError:
|
|
single_usb_serial = ""
|
|
for line in run_adb(adb, ["devices", "-l"]).splitlines():
|
|
fields = line.split()
|
|
if len(fields) >= 2 and fields[1] in {"device", "offline", "unauthorized", "no"}:
|
|
usb = any(v.startswith("usb:") for v in fields[2:]) or fields[0] == single_usb_serial
|
|
rows.append((fields[0], fields[1], usb))
|
|
return rows
|
|
|
|
|
|
def read_identity(adb: str, serial: str) -> dict:
|
|
def prop(name: str) -> str:
|
|
return run_adb(adb, ["-s", serial, "shell", "getprop", name])
|
|
|
|
try:
|
|
api = int(prop("ro.build.version.sdk"))
|
|
except ValueError:
|
|
raise RegistrationError("无法读取有效的 Android API 版本。") from None
|
|
return {"serial": serial, "model": prop("ro.product.model"),
|
|
"android_release": prop("ro.build.version.release"), "api_level": api}
|
|
|
|
|
|
def register_connected(adb: str, alias: str, directory: Path = DIRECTORY) -> None:
|
|
path = directory / PRIVATE_NAME
|
|
if path.exists():
|
|
raise RegistrationError("真实登记已存在，不覆盖、不重建；请使用 check。")
|
|
rows = list_devices(adb)
|
|
if len(rows) != 1:
|
|
raise RegistrationError("首次自动登记要求仅接入一台手机；当前数量不符合。")
|
|
serial, status, usb = rows[0]
|
|
if status != "device":
|
|
raise RegistrationError("手机未授权或离线；请完成 USB 调试授权。")
|
|
if not usb:
|
|
raise RegistrationError("未确认 USB 传输，拒绝将无线设备自动登记为 USB。")
|
|
device = {"alias": alias, **read_identity(adb, serial), "connection": "usb",
|
|
"dedicated_test_device": True, "automation_allowed": True,
|
|
"visual_check_requires_user_instruction": True}
|
|
document = {"schema_version": 1, "devices": [device]}
|
|
validate_document(document)
|
|
try:
|
|
with path.open("x", encoding="utf-8", newline="\n") as stream:
|
|
json.dump(document, stream, ensure_ascii=False, indent=2)
|
|
stream.write("\n")
|
|
except FileExistsError:
|
|
raise RegistrationError("真实登记已被创建；拒绝覆盖。") from None
|
|
|
|
|
|
def check_connected(adb: str, device: dict) -> None:
|
|
row = next((r for r in list_devices(adb) if r[0] == device["serial"]), None)
|
|
if row is None or row[1] != "device" or not row[2]:
|
|
raise RegistrationError("登记手机未通过已授权 USB 连接检查；不改用其他手机。")
|
|
current = read_identity(adb, device["serial"])
|
|
if any(current[key] != device[key] for key in ("model", "android_release", "api_level")):
|
|
raise RegistrationError("手机系统或型号与登记不一致；请核实登记，不自动更新。")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="测试手机私有登记门禁；不输出真实标识")
|
|
parser.add_argument("--adb", help="已有 adb 可执行文件")
|
|
parser.add_argument("--alias", default="android-test-a")
|
|
parser.add_argument("action", choices=("check", "register-connected"), default="check", nargs="?")
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.action == "register-connected":
|
|
if not args.adb:
|
|
raise RegistrationError("自动登记需要 --adb，且必须已有用户专用测试授权。")
|
|
register_connected(args.adb, args.alias)
|
|
devices = load_registration()
|
|
device = next((item for item in devices if item["alias"] == args.alias), None)
|
|
if device is None:
|
|
raise RegistrationError("未找到指定测试代号；不自动选择其他设备。")
|
|
if args.adb:
|
|
check_connected(args.adb, device)
|
|
print(f"测试登记通过：{device['alias']}；未输出设备标识，视觉检查仍须单独询问。")
|
|
return 0
|
|
except RegistrationError as error:
|
|
print(f"测试登记未通过：{error}")
|
|
return 2
|
|
except OSError:
|
|
print("测试登记读写失败；未输出路径、设备标识或原始异常。")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|