import importlib.util from pathlib import Path import tempfile import threading 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_public_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_static_asset_version_is_a_content_fingerprint(self): self.assertEqual(len(server.STATIC_ASSET_VERSION), 16) self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$") index_template = (Path(__file__).parents[1] / "index.html").read_text(encoding="utf-8") self.assertEqual(index_template.count("__ASSET_VERSION__"), 2) def test_default_cache_window_is_two_minutes(self): self.assertEqual(server.DASHBOARD_CACHE_SECONDS, 120) self.assertEqual(server.STATION_CACHE_SECONDS, 120) def test_public_station_directory_limits_partner_contact_and_price_to_cooperative_stations(self): stations = [ {"id": "partner", "cooperative": True, "contactPerson": "张工", "contactPhone": "13800000000", "unitPrice": 12.5}, {"id": "external", "cooperative": False, "contactPerson": "李工", "contactPhone": "13900000000", "unitPrice": 10.0}, ] with patch.object(server, "_load_stations", return_value=stations): directory = server._public_station_directory()["stations"] self.assertEqual(directory[0]["contactPerson"], "张工") self.assertEqual(directory[0]["contactPhone"], "13800000000") self.assertEqual(directory[0]["unitPrice"], 12.5) self.assertNotIn("contactPerson", directory[1]) self.assertNotIn("contactPhone", directory[1]) self.assertNotIn("unitPrice", directory[1]) def test_cache_reuses_dashboard_snapshot_within_window(self): calls = [] with patch.object(server, "_cache", {}): first = server._cached("dashboard", 120, lambda: calls.append("load") or {"version": 1}) second = server._cached("dashboard", 120, lambda: calls.append("load") or {"version": 2}) self.assertEqual(calls, ["load"]) self.assertEqual(first, second) def test_expired_cache_returns_stale_value_while_refreshing_in_background(self): started = threading.Event() release = threading.Event() def loader(): started.set() release.wait(1) return {"version": 2} cache = {"dashboard": (time.time() - 121, {"version": 1})} with patch.object(server, "_cache", cache), \ patch.object(server, "_cache_refreshing", set()), \ patch.object(server, "_cache_load_locks", {}): result = server._cached("dashboard", 120, loader) self.assertEqual(result, {"version": 1}) self.assertTrue(started.wait(1)) release.set() for _ in range(100): if cache["dashboard"][1] == {"version": 2}: break time.sleep(0.01) self.assertEqual(cache["dashboard"][1], {"version": 2}) def test_expired_station_region_is_served_while_refresh_is_scheduled(self): station = {"longitude": 113.2, "latitude": 23.1, "province": "错误省份"} coordinate = server._station_coordinate_key(station) stale_region = {"province": "广东省", "city": "广州市", "district": "黄埔区"} cache = {coordinate: {"updatedAt": time.time() - server.STATION_GEOCODE_CACHE_SECONDS - 1, "region": stale_region}} with patch.object(server, "_station_geocode_cache", cache), \ patch.object(server, "_station_geocode_cache_dirty", False), \ patch.object(server, "_schedule_station_region_refresh") as schedule, \ patch.object(server, "_reverse_geocode_station") as reverse: result = server._stations_with_gps_regions([station]) self.assertEqual(result[0]["province"], "广东省") self.assertEqual(result[0]["city"], "广州市") reverse.assert_not_called() schedule.assert_called_once() self.assertEqual(len(schedule.call_args[0][0]), 1) def test_station_region_comes_from_gps_reverse_geocode_and_persists_by_coordinate(self): station = { "longitude": 113.200001, "latitude": 23.100001, "province": "错误省份", "city": "错误城市", "district": "错误区县", } with tempfile.TemporaryDirectory() as directory: cache_path = Path(directory) / "station-geocode.json" with patch.object(server, "STATION_GEOCODE_CACHE_PATH", cache_path), \ patch.object(server, "_station_geocode_cache", None), \ patch.object(server, "_station_geocode_cache_dirty", False), \ patch.object(server, "_reverse_geocode_station", return_value={ "province": "广东省", "city": "广州市", "district": "黄埔区", "adcode": "440112" }) as reverse: first = server._stations_with_gps_regions([station])[0] second = server._stations_with_gps_regions([dict(station, province="仍然错误")])[0] self.assertEqual(first["province"], "广东省") self.assertEqual(first["city"], "广州市") self.assertEqual(first["district"], "黄埔区") self.assertEqual(second["province"], "广东省") self.assertEqual(reverse.call_count, 1) self.assertTrue(cache_path.exists()) if __name__ == "__main__": unittest.main()