#!/usr/bin/env python3 from __future__ import print_function import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request class GateError(RuntimeError): pass class Client: def __init__(self, base_url, timeout): self.base_url = base_url.rstrip("/") self.timeout = timeout def request(self, method, path, token="", payload=None, expected_statuses=(200,)): body = None headers = {} if token: headers["Authorization"] = "Bearer " + token if payload is not None: body = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = urllib.request.Request(self.base_url + path, data=body, headers=headers, method=method) try: response = urllib.request.urlopen(request, timeout=self.timeout) status = response.getcode() raw = response.read() except urllib.error.HTTPError as error: status = error.code raw = error.read() if status not in expected_statuses: message = "HTTP {0}".format(status) try: envelope = json.loads(raw.decode("utf-8")) message = envelope.get("message") or envelope.get("error") or message except (ValueError, UnicodeDecodeError): pass raise GateError("{0} {1} failed: {2}".format(method, path, message)) if not raw: return status, None try: return status, json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return status, raw.decode("utf-8", "replace") def data(self, method, path, token="", payload=None): _, envelope = self.request(method, path, token=token, payload=payload) if not isinstance(envelope, dict) or "data" not in envelope: raise GateError("{0} {1} returned no data envelope".format(method, path)) return envelope["data"] def query(path, values): return path + "?" + urllib.parse.urlencode(values) def normalized_vins(value): vins = [] for item in value.replace("\n", ",").split(","): vin = item.strip().upper() if vin and vin not in vins: vins.append(vin) if not vins: raise GateError("at least one expected VIN is required") return vins def require(condition, message): if not condition: raise GateError(message) def static_route(client, path): status, body = client.request("GET", path) require(status == 200 and isinstance(body, str) and "羚牛车辆数据中台" in body, "SPA route failed: " + path) def run_gate(args, password): client = Client(args.base_url, args.timeout) expected_vins = normalized_vins(args.expected_vins) expected_menus = set(item.strip() for item in args.expected_menus.split(",") if item.strip()) login = client.data("POST", "/api/v2/auth/login", payload={"username": args.username, "password": password}) token = login.get("accessToken") if isinstance(login, dict) else "" require(bool(token), "login did not return an access token") try: session = client.data("GET", "/api/v2/session", token=token) require(session.get("role") == "customer", "account is not a customer") require(set(session.get("menuKeys") or []) == expected_menus, "customer menu grants do not match the four-menu baseline") require(session.get("vehicleCount") == len(expected_vins), "session vehicle count does not match the expected demo cohort") fleet = client.data("GET", query("/api/realtime/vehicles", {"limit": 100, "offset": 0}), token=token) fleet_by_vin = dict((item.get("vin", "").upper(), item) for item in fleet.get("items") or []) visible_vins = set(fleet_by_vin) require(fleet.get("total") == len(expected_vins), "realtime vehicle total is outside the assigned Scope") require(visible_vins == set(expected_vins), "realtime vehicle set does not match the assigned Scope") sample_checks = [ (args.multi_source_vin, lambda vehicle: (vehicle.get("sourceCount") or 0) > 1 and vehicle.get("locationAvailable") is True, "multi-source"), (args.single_source_vin, lambda vehicle: vehicle.get("sourceCount") == 1 and vehicle.get("locationAvailable") is True, "single-source"), (args.no_location_vin, lambda vehicle: vehicle.get("locationAvailable") is False, "no-location"), (args.offline_vin, lambda vehicle: vehicle.get("online") is False and vehicle.get("locationAvailable") is True, "offline-with-location") ] checked_samples = 0 for sample_vin, predicate, label in sample_checks: if not sample_vin: continue normalized = sample_vin.strip().upper() require(normalized in expected_vins, label + " sample is not in the expected cohort") require(predicate(fleet_by_vin.get(normalized) or {}), label + " sample no longer matches its demo category") checked_samples += 1 for vin in expected_vins: detail = client.data("GET", query("/api/vehicle-service", {"keyword": vin, "limit": 20}), token=token) require(detail.get("lookupResolved") is True and detail.get("vin") == vin, "vehicle detail is unavailable for " + vin) if args.denied_vin: denied_path = query("/api/vehicle-service", {"keyword": args.denied_vin.strip().upper(), "limit": 20}) status, envelope = client.request("GET", denied_path, token=token, expected_statuses=(200, 403, 404)) if status == 200: denied = envelope.get("data") if isinstance(envelope, dict) else None require(not isinstance(denied, dict) or denied.get("lookupResolved") is not True, "out-of-Scope VIN was resolved") if args.track_vin: require(args.track_vin.strip().upper() in expected_vins, "track VIN is not in the expected cohort") require(args.track_from and args.track_to, "track date range is required with --track-vin") track = client.data("GET", query("/api/v2/tracks", { "keyword": args.track_vin.strip().upper(), "dateFrom": args.track_from, "dateTo": args.track_to, "maxPoints": 1600 }), token=token) require(len(track.get("points") or []) > 1, "track gate returned fewer than two points") if args.mileage_from or args.mileage_to: require(args.mileage_from and args.mileage_to, "both mileage dates are required") mileage = client.data("GET", query("/api/v2/statistics/mileage", { "vins": ",".join(expected_vins), "dateFrom": args.mileage_from, "dateTo": args.mileage_to, "protocols": "GB32960,JT808,YUTONG_MQTT" }), token=token) require((mileage.get("vehicleCount") or 0) > 0, "mileage gate returned no vehicles") require((mileage.get("recordCount") or 0) > 0, "mileage gate returned no vehicle-day records") for route in ("/monitor", "/vehicles/" + expected_vins[0], "/tracks", "/statistics"): static_route(client, route) return { "username": args.username, "vehicleCount": len(expected_vins), "menus": sorted(expected_menus), "trackChecked": bool(args.track_vin), "mileageChecked": bool(args.mileage_from), "deniedChecked": bool(args.denied_vin), "sampleChecks": checked_samples } finally: try: client.request("POST", "/api/v2/auth/logout", token=token) except GateError: pass def parser(): value = argparse.ArgumentParser(description="Verify a customer demo account without persisting its password.") value.add_argument("--base-url", default="http://127.0.0.1:20300") value.add_argument("--username", required=True) value.add_argument("--password-env", default="DEMO_PASSWORD") value.add_argument("--expected-vins", required=True, help="Comma-separated VINs assigned to the account") value.add_argument("--expected-menus", default="monitor,vehicles,tracks,statistics") value.add_argument("--denied-vin", default="") value.add_argument("--multi-source-vin", default="") value.add_argument("--single-source-vin", default="") value.add_argument("--no-location-vin", default="") value.add_argument("--offline-vin", default="") value.add_argument("--track-vin", default="") value.add_argument("--track-from", default="") value.add_argument("--track-to", default="") value.add_argument("--mileage-from", default="") value.add_argument("--mileage-to", default="") value.add_argument("--timeout", type=float, default=20.0) return value def main(argv=None): args = parser().parse_args(argv) password = os.environ.get(args.password_env, "") if not password: print("customer_demo_gate=failed reason=password_environment_missing", file=sys.stderr) return 2 try: result = run_gate(args, password) except GateError as error: print("customer_demo_gate=failed reason={0}".format(str(error).replace("\n", " ")), file=sys.stderr) return 1 print("customer_demo_gate=ok username={0} vehicles={1} menus={2} samples={3} track={4} mileage={5} denied={6}".format( result["username"], result["vehicleCount"], ",".join(result["menus"]), result["sampleChecks"], int(result["trackChecked"]), int(result["mileageChecked"]), int(result["deniedChecked"]) )) return 0 if __name__ == "__main__": sys.exit(main())