feat(auth): add customer demo acceptance gate
This commit is contained in:
lingniu
@@ -23,6 +23,7 @@ if test -n "$API_BINARY"; then
|
||||
fi
|
||||
test -x "$SCRIPT_DIR/prepare-web-release-tree.sh" || { printf 'release tree helper is missing\n' >&2; exit 1; }
|
||||
test -f "$SCRIPT_DIR/prune-release-history.py" || { printf 'release pruning helper is missing\n' >&2; exit 1; }
|
||||
test -f "$SCRIPT_DIR/verify-customer-demo.py" || { printf 'customer demo gate is missing\n' >&2; exit 1; }
|
||||
|
||||
old=$(readlink -f "$ROOT/current")
|
||||
next="$ROOT/releases/$RELEASE_ID"
|
||||
@@ -86,8 +87,9 @@ fi
|
||||
cp -a "$old/deploy" "$next/deploy"
|
||||
cp "$SCRIPT_DIR/prepare-web-release-tree.sh" "$next/deploy/prepare-web-release-tree.sh"
|
||||
cp "$SCRIPT_DIR/prune-release-history.py" "$next/deploy/prune-release-history.py"
|
||||
cp "$SCRIPT_DIR/verify-customer-demo.py" "$next/deploy/verify-customer-demo.py"
|
||||
cp "$0" "$next/deploy/install-web-release.sh"
|
||||
chmod +x "$next/platform-api" "$next/deploy/"*.sh "$next/deploy/prune-release-history.py"
|
||||
chmod +x "$next/platform-api" "$next/deploy/"*.sh "$next/deploy/prune-release-history.py" "$next/deploy/verify-customer-demo.py"
|
||||
for runtime_file in alert-evaluator alert-stream-evaluator platform-migrate oneos-scope-sync alert-benchmark; do
|
||||
test ! -f "$next/$runtime_file" || chmod +x "$next/$runtime_file"
|
||||
done
|
||||
|
||||
@@ -41,6 +41,7 @@ test "$(cat "$root/current/alert-evaluator")" = 'old evaluator'
|
||||
test -f "$root/current/web/assets/new.js"
|
||||
test -f "$root/current/web/assets/old.js"
|
||||
test -f "$root/current/web/.compatibility-manifests/1.assets"
|
||||
test -x "$root/current/deploy/verify-customer-demo.py"
|
||||
test ! -e "$root/current/lingniu-vehicle-platform.service"
|
||||
test "$(find "$root/releases" -mindepth 1 -maxdepth 1 -type d | wc -l | tr -d ' ')" = 3
|
||||
grep -q 'web_release_install=ok release=new-release previous=old-release' "$fixture/install.out"
|
||||
|
||||
217
vehicle-data-platform/deploy/verify-customer-demo.py
Executable file
217
vehicle-data-platform/deploy/verify-customer-demo.py
Executable file
@@ -0,0 +1,217 @@
|
||||
#!/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())
|
||||
104
vehicle-data-platform/deploy/verify-customer-demo.test.py
Executable file
104
vehicle-data-platform/deploy/verify-customer-demo.test.py
Executable file
@@ -0,0 +1,104 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import print_function
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
|
||||
VIN = "LB9A32A21R0LS1464"
|
||||
DENIED_VIN = "LA9935J21F0YWT578"
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, _format, *_args):
|
||||
pass
|
||||
|
||||
def envelope(self, data, status=200):
|
||||
body = json.dumps({"data": data, "traceId": "test", "timestamp": 1}).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_POST(self):
|
||||
if self.path == "/api/v2/auth/login":
|
||||
self.envelope({"accessToken": "customer-token", "expiresAt": "2026-07-17T00:00:00+08:00", "session": {}})
|
||||
return
|
||||
if self.path == "/api/v2/auth/logout":
|
||||
self.envelope({"loggedOut": True})
|
||||
return
|
||||
self.envelope({}, 404)
|
||||
|
||||
def do_GET(self):
|
||||
parsed = urlparse(self.path)
|
||||
values = parse_qs(parsed.query)
|
||||
if parsed.path == "/api/v2/session":
|
||||
self.envelope({"name": "客户验收", "role": "customer", "authMode": "enforce", "menuKeys": ["monitor", "vehicles", "tracks", "statistics"], "vehicleCount": 1})
|
||||
elif parsed.path == "/api/realtime/vehicles":
|
||||
self.envelope({"items": [{"vin": VIN, "plate": "粤AGR6863", "sourceCount": 2, "online": True, "locationAvailable": True}], "total": 1, "limit": 100, "offset": 0})
|
||||
elif parsed.path == "/api/vehicle-service" and values.get("keyword") == [VIN]:
|
||||
self.envelope({"lookupResolved": True, "vin": VIN})
|
||||
elif parsed.path == "/api/vehicle-service" and values.get("keyword") == [DENIED_VIN]:
|
||||
self.envelope({"lookupResolved": False, "vin": ""})
|
||||
elif parsed.path == "/api/v2/tracks":
|
||||
self.envelope({"vin": VIN, "points": [{"longitude": 113.1, "latitude": 23.1}, {"longitude": 113.2, "latitude": 23.2}]})
|
||||
elif parsed.path == "/api/v2/statistics/mileage":
|
||||
self.envelope({"vehicleCount": 1, "recordCount": 1})
|
||||
elif parsed.path in ("/monitor", "/vehicles/" + VIN, "/tracks", "/statistics"):
|
||||
body = "羚牛车辆数据中台".encode("utf-8")
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
else:
|
||||
self.envelope({}, 404)
|
||||
|
||||
|
||||
class CustomerDemoGateTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.server = HTTPServer(("127.0.0.1", 0), Handler)
|
||||
cls.thread = threading.Thread(target=cls.server.serve_forever)
|
||||
cls.thread.daemon = True
|
||||
cls.thread.start()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.shutdown()
|
||||
cls.thread.join()
|
||||
|
||||
def test_complete_customer_demo_gate(self):
|
||||
script = str(Path(__file__).with_name("verify-customer-demo.py"))
|
||||
environment = dict(os.environ)
|
||||
environment["DEMO_PASSWORD"] = "not-persisted"
|
||||
result = subprocess.run([
|
||||
sys.executable, script,
|
||||
"--base-url", "http://127.0.0.1:{0}".format(self.server.server_port),
|
||||
"--username", "customer-demo",
|
||||
"--expected-vins", VIN,
|
||||
"--denied-vin", DENIED_VIN,
|
||||
"--multi-source-vin", VIN,
|
||||
"--track-vin", VIN,
|
||||
"--track-from", "2026-07-16T17:00:00+08:00",
|
||||
"--track-to", "2026-07-16T18:00:00+08:00",
|
||||
"--mileage-from", "2026-07-17",
|
||||
"--mileage-to", "2026-07-17"
|
||||
], env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertIn("customer_demo_gate=ok", result.stdout)
|
||||
self.assertIn("vehicles=1", result.stdout)
|
||||
self.assertIn("samples=1 track=1 mileage=1 denied=1", result.stdout)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user