212 lines
7.5 KiB
Python
212 lines
7.5 KiB
Python
#!/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()
|