#!/usr/bin/env python3 """Run the self-expand vehicle frontend safety and integration matrix. The default suite is deliberately non-transmitting: it expects the frontend in ``dry_run`` mode and the native gateway in ``monitor_only`` mode. It verifies every visible vehicle preset from natural-language request through CAN_DEV and the native whitelist, exercises rejection paths, checks live bus reception, and proves that the adapter transmit counter did not change. """ from __future__ import annotations import argparse import json import sys import time from dataclasses import dataclass from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @dataclass(frozen=True) class FrontendCase: case_id: str utterance: str commands: tuple[str, ...] FRONTEND_CASES = ( FrontendCase("seat-vent-on", "打开主驾座椅通风", ("seat driver vent low",)), FrontendCase("seat-heat-on", "打开主驾座椅加热", ("seat driver heat low",)), FrontendCase("seat-vent-off", "关闭主驾座椅通风", ("seat driver vent off",)), FrontendCase("seat-heat-off", "关闭主驾座椅加热", ("seat driver heat off",)), FrontendCase("climate-on", "打开空调", ("ac on",)), FrontendCase("climate-24", "把空调温度调到24度", ("ac temp driver 24", "ac temp passenger 24")), FrontendCase("climate-fan3", "把空调风量调到3档", ("ac fan 3",)), FrontendCase("climate-auto", "把空调设置为自动模式", ("ac control auto",)), FrontendCase("outlet-driver-left", "把主驾左风口调到中间位置", ("outlet driver-left percent 50 50",)), FrontendCase("outlet-driver-right", "把主驾右风口调到中间位置", ("outlet driver-right percent 50 50",)), FrontendCase("outlet-passenger-left", "把副驾左风口调到中间位置", ("outlet passenger-left percent 50 50",)), FrontendCase("outlet-passenger-right", "把副驾右风口调到中间位置", ("outlet passenger-right percent 50 50",)), FrontendCase("shade-open", "打开遮阳帘", ("shade open",)), FrontendCase("shade-close", "关闭遮阳帘", ("shade close",)), FrontendCase("shade-half", "把遮阳帘打开一半", ("shade half",)), FrontendCase("shade-stop", "停止遮阳帘", ("shade stop",)), FrontendCase("light-indoor", "打开车内室内灯", ("light indoor on",)), FrontendCase("light-reading", "打开全部阅读灯", ("light reading all on",)), FrontendCase("light-ambient", "打开车内氛围灯", ("light ambient on",)), FrontendCase("light-brightness", "把氛围灯亮度调到3档", ("light ambient brightness 3",)), FrontendCase("window-open", "打开全部车窗", ("window all open",)), FrontendCase("window-close", "关闭全部车窗", ("window all close",)), FrontendCase("window-front-left-50", "把左前车窗打开到50%", ("window front-left position 50",)), FrontendCase("door-rear-lock", "锁上后排门锁", ("door rear lock",)), ) REJECTED_COMMANDS = ( "raw 123#0102", "status", "help", "quit", "window front-left position 101", "ac fan 8", "ac temp driver 17", "seat driver horizontal forward 99", "light ambient color 128", "outlet driver-left percent 101 50", "shade position 101", ) EXPECTED_FEEDBACK_IDS = {"0x388", "0x404", "0x407", "0x41A", "0x42E", "0x4FA", "0x494"} def request_json( url: str, *, payload: dict[str, Any] | None = None, text: str | None = None, headers: dict[str, str] | None = None, timeout: float = 30, ) -> tuple[int, dict[str, Any]]: request_headers = {"Accept": "application/json", **(headers or {})} data = None if payload is not None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") request_headers["Content-Type"] = "application/json; charset=utf-8" elif text is not None: data = text.encode("utf-8") request_headers["Content-Type"] = "text/plain; charset=utf-8" request = Request(url, data=data, headers=request_headers, method="POST" if data is not None else "GET") try: with urlopen(request, timeout=timeout) as response: # noqa: S310 - operator-selected local endpoints raw, status = response.read(), int(response.status) except HTTPError as exc: raw, status = exc.read(), int(exc.code) except (URLError, TimeoutError, OSError) as exc: raise RuntimeError(f"cannot reach {url}: {exc}") from exc try: decoded = json.loads(raw.decode("utf-8")) if raw else {} except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RuntimeError(f"non-JSON response from {url}") from exc if not isinstance(decoded, dict): raise RuntimeError(f"non-object JSON response from {url}") return status, decoded def record(results: list[dict[str, Any]], case_id: str, passed: bool, detail: str, **extra: Any) -> None: results.append({"id": case_id, "passed": passed, "detail": detail, **extra}) def run(args: argparse.Namespace) -> dict[str, Any]: frontend = args.frontend_url.rstrip("/") native = args.native_url.rstrip("/") results: list[dict[str, Any]] = [] started = time.time() status, health = request_json(f"{frontend}/api/health", timeout=5) vehicle = health.get("vehicle") or {} native_health = vehicle.get("native_can") or {} health_ok = ( 200 <= status < 300 and health.get("success") is True and vehicle.get("connected") is True and native_health.get("connected") is True and vehicle.get("execution_mode") == "dry_run" ) record(results, "frontend-health", health_ok, "frontend, CAN_DEV and native gateway are connected", status=status) status, before = request_json(f"{native}/api/can/runtime", timeout=5) before_tx = int(before.get("tx_count", -1)) live_ids = set((before.get("states") or {}).keys()) runtime_ok = ( 200 <= status < 300 and before.get("mode") == "monitor_only" and before_tx >= 0 and EXPECTED_FEEDBACK_IDS <= live_ids and int(before.get("last_receive_age_ms", 999999)) < args.max_receive_age_ms ) record( results, "native-monitor-runtime", runtime_ok, "adapter is receiving all required feedback families in monitor-only mode", tx_count=before_tx, feedback_ids=sorted(live_ids), ) app_source = (Path(__file__).resolve().parents[1] / "apps" / "ir-toy-console" / "app.js").read_text(encoding="utf-8") source_missing = [case.case_id for case in FRONTEND_CASES if case.utterance not in app_source] record(results, "frontend-source-matrix", not source_missing, "all 24 tested actions are present in the visible frontend", missing=source_missing) for case in FRONTEND_CASES: status, response = request_json( f"{frontend}/api/vehicle/execute", payload={"command": case.utterance, "vehicle_id": args.vehicle_id}, timeout=args.action_timeout, ) report = response.get("native_can_execution") or {} actual_commands = tuple( str(item.get("command")) for item in report.get("commands", []) if isinstance(item, dict) ) passed = ( 200 <= status < 300 and response.get("success") is True and report.get("status") == "validated" and actual_commands == case.commands and report.get("physical_state_confirmed") is False ) record( results, f"preset:{case.case_id}", passed, case.utterance, expected_commands=list(case.commands), actual_commands=list(actual_commands), http_status=status, report_status=report.get("status"), ) for index, command in enumerate(REJECTED_COMMANDS, start=1): status, response = request_json(f"{native}/api/can/validate", text=command, timeout=5) passed = status >= 400 and response.get("valid") is not True and response.get("transmitted") is not True record(results, f"reject:{index:02d}", passed, command, http_status=status) status, response = request_json( f"{native}/api/can/command", text="ac on", headers={"X-CAN-Approval-Token": "intentionally-invalid"}, timeout=5, ) record( results, "monitor-only-command-gate", status in {401, 403} and response.get("accepted") is not True, "monitor-only gateway rejects command execution", http_status=status, ) _, after = request_json(f"{native}/api/can/runtime", timeout=5) after_tx = int(after.get("tx_count", -1)) record( results, "zero-transmit-invariant", before_tx == after_tx, "dry-run suite did not transmit CAN frames", before_tx=before_tx, after_tx=after_tx, ) passed_count = sum(item["passed"] for item in results) return { "suite": "vehicle_frontend_dry_run", "passed": passed_count == len(results), "summary": {"passed": passed_count, "failed": len(results) - passed_count, "total": len(results)}, "duration_ms": round((time.time() - started) * 1000), "frontend_url": frontend, "native_url": native, "results": results, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--frontend-url", default="http://127.0.0.1:8010") parser.add_argument("--native-url", default="http://127.0.0.1:8765") parser.add_argument("--vehicle-id", default="denza_n7") parser.add_argument("--action-timeout", type=float, default=30) parser.add_argument("--max-receive-age-ms", type=int, default=2000) parser.add_argument("--output", type=Path) args = parser.parse_args() try: report = run(args) except Exception as exc: # noqa: BLE001 - command-line diagnostic boundary report = {"suite": "vehicle_frontend_dry_run", "passed": False, "fatal_error": str(exc)} rendered = json.dumps(report, ensure_ascii=False, indent=2) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(rendered + "\n", encoding="utf-8") print(rendered) return 0 if report.get("passed") else 1 if __name__ == "__main__": sys.exit(main())