80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
SPEC = importlib.util.spec_from_file_location("station_navigation_server", Path(__file__).parents[1] / "server.py")
|
|
server = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(server)
|
|
|
|
|
|
class StationNavigationTest(unittest.TestCase):
|
|
def test_directory_is_a_positive_allowlist_without_hydrogen_data(self):
|
|
source = {
|
|
"status": "ok",
|
|
"asOf": "2026-08-12 12:00:00",
|
|
"summary": {"totalStations": 1, "cooperativeStations": 1},
|
|
"stations": [{
|
|
"id": "GD-1", "name": "广州合作站", "province": "广东省", "city": "广州市",
|
|
"district": "黄埔区", "address": "开源大道1号", "longitude": 113.2, "latitude": 23.1,
|
|
"cooperative": True,
|
|
}],
|
|
}
|
|
with patch.object(server, "urlopen", return_value=io.BytesIO(json.dumps(source).encode("utf-8"))):
|
|
result = server._load_station_directory()
|
|
|
|
self.assertEqual(result["summary"], {"totalStations": 1, "cooperativeStations": 1})
|
|
station = result["stations"][0]
|
|
self.assertEqual(station["name"], "广州合作站")
|
|
self.assertEqual(station["province"], "广东省")
|
|
self.assertEqual(server.VEHICLE_MAP_INTERNAL_BASE_URL, "http://127.0.0.1:20800")
|
|
|
|
def test_static_assets_are_fingerprinted(self):
|
|
self.assertRegex(server.STATIC_ASSET_VERSION, r"^[0-9a-f]{16}$")
|
|
|
|
def test_static_assets_resolve_from_the_release_directory(self):
|
|
self.assertIn('os.chdir(str(ROOT))', Path(__file__).parents[1].joinpath('server.py').read_text(encoding='utf-8'))
|
|
|
|
def test_default_cache_window_is_two_minutes(self):
|
|
self.assertEqual(server.STATION_CACHE_SECONDS, 120)
|
|
|
|
def test_cache_reuses_directory_snapshot_within_window(self):
|
|
calls = []
|
|
with patch.object(server, "_cache", {}):
|
|
first = server._cached("station-directory", 120, lambda: calls.append("load") or {"version": 1})
|
|
second = server._cached("station-directory", 120, lambda: calls.append("load") or {"version": 2})
|
|
self.assertEqual(calls, ["load"])
|
|
self.assertEqual(first, second)
|
|
|
|
def test_expired_directory_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 = {"station-directory": (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("station-directory", 120, loader)
|
|
self.assertEqual(result, {"version": 1})
|
|
self.assertTrue(started.wait(1))
|
|
release.set()
|
|
for _ in range(100):
|
|
if cache["station-directory"][1] == {"version": 2}:
|
|
break
|
|
time.sleep(0.01)
|
|
self.assertEqual(cache["station-directory"][1], {"version": 2})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|