94 lines
5.0 KiB
Python
94 lines
5.0 KiB
Python
import importlib.util
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from pathlib import Path
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
SPEC = importlib.util.spec_from_file_location("vehicle_map_server", Path(__file__).parents[1] / "server.py")
|
|
server = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(server)
|
|
|
|
|
|
class DashboardTest(unittest.TestCase):
|
|
def test_dashboard_aggregates_authorized_vehicle_and_station_data(self):
|
|
def fake_post(path, body):
|
|
if path.endswith("realtime/query"):
|
|
return [
|
|
{"vin": "VIN1", "motionStatus": "driving", "speedKmh": 18, "activeToday": True, "online": True},
|
|
{"vin": "VIN2", "motionStatus": "idle", "speedKmh": 0, "activeToday": True, "online": True},
|
|
{"vin": "VIN3", "motionStatus": "offline", "speedKmh": 0, "activeToday": False, "online": False},
|
|
]
|
|
if path.endswith("mileage/query"):
|
|
return [
|
|
{"vin": "VIN1", "status": "NORMAL", "dailyMileageKm": 12.345},
|
|
{"vin": "VIN2", "status": "NORMAL", "dailyMileageKm": 7.655},
|
|
{"vin": "VIN3", "status": "NO_DATA", "dailyMileageKm": None},
|
|
]
|
|
if path.endswith("hydrogen-stations/query"):
|
|
return [{"id": "1", "cooperative": True}, {"id": "2", "cooperative": False}]
|
|
raise AssertionError(path)
|
|
|
|
with patch.object(server, "_post_open_platform", side_effect=fake_post):
|
|
with patch.object(server, "_cache", {}):
|
|
result = server._load_dashboard()
|
|
|
|
self.assertEqual(result["summary"]["totalVehicles"], 3)
|
|
self.assertEqual(result["summary"]["onlineVehicles"], 2)
|
|
self.assertEqual(result["summary"]["drivingVehicles"], 1)
|
|
self.assertEqual(result["summary"]["todayMileageKm"], 20.0)
|
|
self.assertEqual(result["summary"]["totalStations"], 2)
|
|
self.assertEqual(result["summary"]["cooperativeStations"], 1)
|
|
self.assertEqual(result["vehicles"][0]["dailyMileageKm"], 12.345)
|
|
self.assertEqual(result["vehicles"][2]["dailyMileageKm"], 0)
|
|
|
|
def test_public_station_directory_uses_an_allowlist(self):
|
|
stations = [{
|
|
"id": "GD-1", "name": "广州合作站", "shortName": "合作站", "province": "广东省",
|
|
"city": "广州市", "district": "黄埔区", "address": "开源大道1号", "longitude": 113.2,
|
|
"latitude": 23.1, "cooperative": True, "monthlyHydrogenKg": 88.8,
|
|
"totalHydrogenKg": 999.9, "internalOwner": "must-not-leak",
|
|
}]
|
|
with patch.object(server, "_post_open_platform", return_value=stations):
|
|
with patch.object(server, "_cache", {}):
|
|
result = server._load_public_stations()
|
|
|
|
station = result["stations"][0]
|
|
self.assertEqual(result["access"]["level"], "public")
|
|
self.assertEqual(station["name"], "广州合作站")
|
|
self.assertTrue(station["cooperative"])
|
|
self.assertNotIn("monthlyHydrogenKg", station)
|
|
self.assertNotIn("totalHydrogenKg", station)
|
|
self.assertNotIn("internalOwner", station)
|
|
|
|
def test_signed_operations_session_requires_the_operations_scope(self):
|
|
with patch.object(server, "LOCAL_ACCESS_CODE", "LN-map-test-code"), patch.object(server, "SESSION_SECRET", b"test-session-secret"), patch.object(server, "SESSION_TTL_SECONDS", 600):
|
|
principal = server._authorizer().authorize_access_code("LN-map-test-code")
|
|
token = server._create_session(principal)
|
|
session = server._read_session(f"{server.SESSION_COOKIE_NAME}={token}")
|
|
|
|
self.assertEqual(session["subject"], "local-access-code")
|
|
self.assertIn(server.OPERATIONS_READ_SCOPE, session["scopes"])
|
|
self.assertIsNone(server._read_session(f"{server.SESSION_COOKIE_NAME}={token}tampered"))
|
|
|
|
legacy_payload = {"v": 1, "sub": "legacy", "scopes": [server.OPERATIONS_READ_SCOPE], "exp": 4_102_444_800}
|
|
legacy_encoded = server._b64encode(server._json_bytes(legacy_payload))
|
|
legacy_signature = server._b64encode(hmac.new(server.SESSION_SECRET, legacy_encoded.encode("ascii"), hashlib.sha256).digest())
|
|
self.assertIsNone(server._read_session(f"{server.SESSION_COOKIE_NAME}={legacy_encoded}.{legacy_signature}"))
|
|
|
|
def test_default_operations_session_expires_after_thirty_minutes(self):
|
|
self.assertEqual(server.SESSION_TTL_SECONDS, 1800)
|
|
with patch.object(server, "SESSION_SECRET", b"test-session-secret"), patch.object(server, "SESSION_TTL_SECONDS", 1800):
|
|
token = server._create_session({"subject": "test", "scopes": [server.OPERATIONS_READ_SCOPE]})
|
|
encoded, _ = token.split(".", 1)
|
|
payload = json.loads(server._b64decode(encoded))
|
|
self.assertGreaterEqual(payload["exp"], int(time.time()) + 1798)
|
|
self.assertLessEqual(payload["exp"], int(time.time()) + 1801)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|