加固版本怎样使用云真机建立发布前回归矩阵

#!/usr/bin/env python3
import json
import re
import sys
from pathlib import Path

SHA256 = re.compile(r"^[a-f0-9]{64}$")
ALLOWED_TYPES = {"physical", "virtual"}
ALLOWED_FORMS = {"phone", "tablet", "foldable"}
ALLOWED_ORIENTATIONS = {"portrait", "landscape"}

def fail(message, code):
    print(message, file=sys.stderr)
    raise SystemExit(code)

def load_object(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail("cannot read JSON: " + str(exc), 3)
    if not isinstance(value, dict):
        fail("JSON root must be an object", 4)
    return value

def text(record, field):
    value = record.get(field)
    if not isinstance(value, str) or not value.strip():
        fail("missing text field: " + field, 5)
    return value.strip()

def strings(record, field, allow_empty=False):
    value = record.get(field)
    if not isinstance(value, list):
        fail("invalid list field: " + field, 6)
    items = []
    for item in value:
        if not isinstance(item, str) or not item.strip():
            fail("invalid item in: " + field, 7)
        items.append(item.strip())
    if not allow_empty and not items:
        fail("empty required list: " + field, 8)
    if len(items) != len(set(items)):
        fail("duplicate item in: " + field, 9)
    return sorted(items)

def integers(record, field):
    value = record.get(field)
    if not isinstance(value, list) or not value:
        fail("invalid integer list: " + field, 10)
    if not all(isinstance(item, int) and not isinstance(item, bool) for item in value):
        fail("invalid integer item: " + field, 11)
    return sorted(set(value))

def normalize_devices(document):
    devices = document.get("devices")
    if not isinstance(devices, list) or not devices:
        fail("device catalog is empty", 12)
    result = []
    for device in devices:
        if not isinstance(device, dict):
            fail("device must be an object", 13)
        device_type = text(device, "type")
        form = text(device, "form")
        if device_type not in ALLOWED_TYPES or form not in ALLOWED_FORMS:
            fail("unsupported device type or form", 14)
        if device.get("stability") != "stable":
            continue
        api = device.get("api")
        if not isinstance(api, int) or isinstance(api, bool):
            fail("invalid device api", 15)
        orientations = strings(device, "orientations")
        if not set(orientations).issubset(ALLOWED_ORIENTATIONS):
            fail("unsupported orientation", 16)
        result.append({
            "model": text(device, "model"),
            "version": text(device, "version"),
            "api": api,
            "abis": strings(device, "abis"),
            "type": device_type,
            "form": form,
            "locales": strings(device, "locales"),
            "orientations": orientations,
            "features": strings(device, "features", True),
        })
    if not result:
        fail("no stable devices remain", 17)
    return result

def validate_plan(plan):
    artifact = text(plan, "artifactSha256")
    if not SHA256.fullmatch(artifact):
        fail("invalid artifactSha256", 18)
    required_types = strings(plan, "requiredTypes")
    required_forms = strings(plan, "requiredForms")
    if not set(required_types).issubset(ALLOWED_TYPES):
        fail("unsupported requiredTypes", 19)
    if not set(required_forms).issubset(ALLOWED_FORMS):
        fail("unsupported requiredForms", 20)
    profiles = plan.get("testProfiles")
    if not isinstance(profiles, list) or not profiles:
        fail("testProfiles are missing", 21)
    return {
        "artifactSha256": artifact,
        "requiredApis": integers(plan, "requiredApis"),
        "requiredAbis": strings(plan, "requiredAbis"),
        "requiredTypes": required_types,
        "requiredForms": required_forms,
        "requiredRiskTags": strings(plan, "requiredRiskTags"),
        "testProfiles": profiles,
    }

def device_matches(device, profile):
    apis = integers(profile, "apis")
    abis = strings(profile, "abis")
    features = set(strings(profile, "features", True))
    forms = set(strings(profile, "forms"))
    if profile.get("needsPhysical") not in {True, False}:
        fail("needsPhysical must be boolean", 22)
    return (
        device["api"] in apis
        and bool(set(device["abis"]) & set(abis))
        and device["form"] in forms
        and features.issubset(set(device["features"]))
        and (not profile["needsPhysical"] or device["type"] == "physical")
    )

def build_matrix(devices, plan):
    cells = []
    covered = {"apis": set(), "abis": set(), "types": set(), "forms": set(), "riskTags": set()}
    for profile in plan["testProfiles"]:
        if not isinstance(profile, dict):
            fail("test profile must be an object", 23)
        profile_id = text(profile, "id")
        suite = text(profile, "suite")
        risk_tags = strings(profile, "riskTags")
        requested_locales = strings(profile, "locales")
        requested_orientations = strings(profile, "orientations")
        matches = [device for device in devices if device_matches(device, profile)]
        if not matches:
            continue
        for device in matches:
            locales = sorted(set(requested_locales) & set(device["locales"]))
            orientations = sorted(set(requested_orientations) & set(device["orientations"]))
            if not locales or not orientations:
                continue
            abi = sorted(set(device["abis"]) & set(strings(profile, "abis")))[0]
            cell = {
                "profileId": profile_id,
                "suite": suite,
                "model": device["model"],
                "version": device["version"],
                "api": device["api"],
                "abi": abi,
                "type": device["type"],
                "form": device["form"],
                "locale": locales[0],
                "orientation": orientations[0],
                "riskTags": risk_tags,
            }
            cells.append(cell)
            covered["apis"].add(device["api"])
            covered["abis"].add(abi)
            covered["types"].add(device["type"])
            covered["forms"].add(device["form"])
            covered["riskTags"].update(risk_tags)
    requirements = {
        "apis": set(plan["requiredApis"]),
        "abis": set(plan["requiredAbis"]),
        "types": set(plan["requiredTypes"]),
        "forms": set(plan["requiredForms"]),
        "riskTags": set(plan["requiredRiskTags"]),
    }
    uncovered = {key: sorted(requirements[key] - covered[key]) for key in requirements}
    uncovered = {key: value for key, value in uncovered.items() if value}
    return cells, uncovered

def main():
    if len(sys.argv) != 3:
        print("Usage: matrix_plan.py DEVICE_CATALOG_JSON TEST_PLAN_JSON", file=sys.stderr)
        raise SystemExit(2)
    catalog_path = Path(sys.argv[1])
    plan_path = Path(sys.argv[2])
    if not catalog_path.is_file() or not plan_path.is_file():
        print("required input file missing", file=sys.stderr)
        raise SystemExit(2)
    devices = normalize_devices(load_object(catalog_path))
    plan = validate_plan(load_object(plan_path))
    cells, uncovered = build_matrix(devices, plan)
    result = {
        "status": "blocked" if uncovered or not cells else "execution-required",
        "artifactSha256": plan["artifactSha256"],
        "matrix": cells,
        "uncovered": uncovered,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if result["status"] == "blocked":
        raise SystemExit(24)

if __name__ == "__main__":
    main()