#!/usr/bin/env python3 from __future__ import print_function import csv import json import os import subprocess import sys import tempfile import threading import unittest from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from urllib.parse import urlparse VIN = "LB9A32A21R0LS1464" SOURCE_REF = "a" * 64 def diagnostic(version=1, provider=""): return { "evidence": { "vin": VIN, "locationSources": [{ "protocol": "JT808", "sourceLabel": provider or "接入方未维护", "providerOverride": provider, "sourceRef": SOURCE_REF, "sourceKind": "PLATFORM", "enabled": True, "priority": 88, "policyRemark": "保留策略备注" }] }, "policy": {"vin": VIN, "version": version, "audit": []} } class Handler(BaseHTTPRequestHandler): puts = [] 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_GET(self): self.assert_auth() if urlparse(self.path).path == "/api/v2/operations/vehicles/{0}/sources".format(VIN): self.envelope(diagnostic()) return self.envelope({}, 404) def do_POST(self): self.assert_auth() if urlparse(self.path).path == "/api/v2/access/vehicles": size = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(size).decode("utf-8")) self.envelope({ "items": [{ "vin": VIN, "plate": "粤AGR6863", "masterDataIssues": ["JT808 接入方未维护"] }], "total": 1, "limit": payload.get("limit"), "offset": payload.get("offset") }) return self.envelope({}, 404) def do_PUT(self): self.assert_auth() expected = "/api/v2/operations/vehicles/{0}/sources/{1}".format(VIN, SOURCE_REF) if urlparse(self.path).path != expected: self.envelope({}, 404) return size = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(size).decode("utf-8")) Handler.puts.append(payload) self.envelope(diagnostic(version=2, provider=payload.get("providerName", ""))) def assert_auth(self): if self.headers.get("Authorization") != "Bearer admin-test-token": self.envelope({}, 401) raise RuntimeError("missing test authorization") class SourceProviderImportTest(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 setUp(self): Handler.puts = [] self.tempdir = tempfile.TemporaryDirectory() def tearDown(self): self.tempdir.cleanup() def write_csv(self, evidence="GPS 运维终端清单 2026-07-16", expected_provider="接入方未维护"): path = Path(self.tempdir.name) / "providers.csv" with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=[ "vin", "source_ref", "provider_name", "evidence", "expected_protocol", "expected_current_provider" ]) writer.writeheader() writer.writerow({ "vin": VIN, "source_ref": SOURCE_REF, "provider_name": "G7s", "evidence": evidence, "expected_protocol": "JT808", "expected_current_provider": expected_provider }) return str(path) def run_script(self, csv_path, apply=False): script = str(Path(__file__).with_name("import-source-providers.py")) environment = dict(os.environ) environment["SOURCE_PROVIDER_ADMIN_TOKEN"] = "admin-test-token" command = [ sys.executable, script, "--csv", csv_path, "--base-url", "http://127.0.0.1:{0}".format(self.server.server_port) ] if apply: command.append("--apply") return subprocess.run( command, env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True ) def test_dry_run_preflights_without_writing(self): result = self.run_script(self.write_csv()) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("source_provider_import=ready", result.stdout) self.assertIn("changes=1", result.stdout) self.assertEqual(Handler.puts, []) def test_apply_preserves_policy_and_sends_evidence(self): result = self.run_script(self.write_csv(), apply=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("source_provider_import=ok", result.stdout) self.assertEqual(len(Handler.puts), 1) self.assertEqual(Handler.puts[0], { "version": 1, "providerName": "G7s", "providerEvidence": "GPS 运维终端清单 2026-07-16", "enabled": True, "priority": 88, "remark": "保留策略备注" }) def test_missing_evidence_fails_before_any_write(self): result = self.run_script(self.write_csv(evidence=""), apply=True) self.assertEqual(result.returncode, 1) self.assertIn("evidence is required", result.stderr) self.assertEqual(Handler.puts, []) def test_stale_expected_provider_fails_preflight(self): result = self.run_script(self.write_csv(expected_provider="旧提供方"), apply=True) self.assertEqual(result.returncode, 1) self.assertIn("current provider changed", result.stderr) self.assertEqual(Handler.puts, []) def test_export_missing_writes_fillable_template(self): script = str(Path(__file__).with_name("import-source-providers.py")) environment = dict(os.environ) environment["SOURCE_PROVIDER_ADMIN_TOKEN"] = "admin-test-token" path = str(Path(self.tempdir.name) / "missing.csv") result = subprocess.run([ sys.executable, script, "--export-missing", path, "--base-url", "http://127.0.0.1:{0}".format(self.server.server_port) ], env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("source_provider_export=ok rows=1 vehicles=1", result.stdout) with open(path, "r", encoding="utf-8-sig", newline="") as handle: rows = list(csv.DictReader(handle)) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["vin"], VIN) self.assertEqual(rows[0]["source_ref"], SOURCE_REF) self.assertEqual(rows[0]["terminal_label"], "") self.assertEqual(rows[0]["provider_name"], "") if __name__ == "__main__": unittest.main()