231 lines
7.9 KiB
Python
231 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from topology_api.app import create_app
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
HTML_PATH = ROOT / "data" / "wiki" / "guanghan_topology_demo.html"
|
|
CONFIG_PATH = ROOT / "topology_api" / "config.py"
|
|
|
|
|
|
class TopologyApiServingTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
app = create_app()
|
|
app.config.update(TESTING=True)
|
|
cls.client = app.test_client()
|
|
|
|
def test_app_serves_demo_page_and_health(self) -> None:
|
|
health = self.client.get("/health")
|
|
self.assertEqual(health.status_code, 200)
|
|
self.assertEqual(health.get_json(), {"status": "ok"})
|
|
|
|
page = self.client.get("/guanghan_topology_demo.html")
|
|
self.assertEqual(page.status_code, 200)
|
|
self.assertIn("广寒基地拓扑结构", page.get_data(as_text=True))
|
|
page.close()
|
|
|
|
def test_page_uses_same_origin_api(self) -> None:
|
|
html = HTML_PATH.read_text(encoding="utf-8")
|
|
self.assertNotIn("http://localhost:5000", html)
|
|
self.assertIn("const API_BASE = ''", html)
|
|
|
|
def test_config_has_no_default_password(self) -> None:
|
|
source = CONFIG_PATH.read_text(encoding="utf-8")
|
|
self.assertNotRegex(
|
|
source,
|
|
r'os\.environ\.get\(\s*["\']PGPASSWORD["\']\s*,',
|
|
)
|
|
|
|
|
|
class TopologyApiQueryTests(unittest.TestCase):
|
|
EXPECTED_LAST_EVENTS = {
|
|
"BL-01": "2024-05-17T23:01:00+00:00",
|
|
"GHC-01": "2024-06-16T07:23:00+00:00",
|
|
"JC-01": "2024-06-17T08:19:00+00:00",
|
|
"BL-02": "2024-06-23T05:23:00+00:00",
|
|
"GHC-02": "2024-07-15T03:16:00+00:00",
|
|
"GHC-03": "2024-08-30T12:45:00+00:00",
|
|
"BL-03": "2024-09-23T19:20:00+00:00",
|
|
"GHC-04": "2024-11-20T13:57:00+00:00",
|
|
"GHC-05": "2025-01-22T10:52:00+00:00",
|
|
}
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
app = create_app()
|
|
app.config.update(TESTING=True)
|
|
cls.client = app.test_client()
|
|
|
|
def test_invalid_time_is_json_400(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?at=not-a-time"
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.get_json()["error"], "invalid_time")
|
|
|
|
def test_time_and_mission_preserve_context(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot"
|
|
"?mission=GHC-01&at=2025-01-02T08:00:00%2B08:00"
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
query = response.get_json()["query"]
|
|
self.assertEqual(query["requestedMission"], "GHC-01")
|
|
self.assertEqual(query["anchorSource"], "explicit_time")
|
|
|
|
def test_mission_list_is_database_backed(self) -> None:
|
|
response = self.client.get("/api/v1/bases/guanghan/missions")
|
|
self.assertEqual(response.status_code, 200)
|
|
missions = response.get_json()["missions"]
|
|
self.assertIn("GHC-05", [mission["code"] for mission in missions])
|
|
for mission in missions:
|
|
self.assertIn("title", mission)
|
|
self.assertIn("occurrenceStatus", mission)
|
|
self.assertIn("lastEventAt", mission)
|
|
|
|
def test_each_mission_uses_its_last_timeline_record(self) -> None:
|
|
response = self.client.get("/api/v1/bases/guanghan/missions")
|
|
actual = {
|
|
mission["code"]: datetime.fromisoformat(
|
|
mission["lastEventAt"]
|
|
).astimezone(timezone.utc)
|
|
for mission in response.get_json()["missions"]
|
|
}
|
|
expected = {
|
|
code: datetime.fromisoformat(value)
|
|
for code, value in self.EXPECTED_LAST_EVENTS.items()
|
|
}
|
|
self.assertEqual(actual, expected)
|
|
|
|
def test_binglun_connections_use_west_south_east_hub_ports(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?mission=GHC-05"
|
|
)
|
|
connections = response.get_json()["connections"]
|
|
hub_ports = {}
|
|
for connection in connections:
|
|
from_component = connection["fromPort"].split(".", 1)[0]
|
|
if (
|
|
from_component in {"binglun_1", "binglun_2", "binglun_3"}
|
|
and connection["type"] == "power"
|
|
):
|
|
hub_ports[from_component] = connection["toPort"]
|
|
self.assertEqual(
|
|
hub_ports,
|
|
{
|
|
"binglun_1": "power_module_hub.west",
|
|
"binglun_2": "power_module_hub.south",
|
|
"binglun_3": "power_module_hub.east",
|
|
},
|
|
)
|
|
|
|
def test_habitat_walkway_terminates_at_power_north_module(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?mission=GHC-05"
|
|
)
|
|
pressurized = {
|
|
frozenset((connection["fromPort"], connection["toPort"]))
|
|
for connection in response.get_json()["connections"]
|
|
if connection["type"] == "pressurized_passage"
|
|
}
|
|
self.assertIn(
|
|
frozenset((
|
|
"habitat_1.distal",
|
|
"power_storage_rack.passage",
|
|
)),
|
|
pressurized,
|
|
)
|
|
|
|
def test_layout_contains_members_labels_and_radiators(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?mission=GHC-05"
|
|
)
|
|
nodes = response.get_json()["layout"]["nodes"]
|
|
by_key = {node["nodeKey"]: node for node in nodes}
|
|
|
|
self.assertEqual(
|
|
by_key["power_west_group"]["memberKeys"],
|
|
["isru_module", "mining_module", "ore_fuel_storage"],
|
|
)
|
|
self.assertEqual(
|
|
by_key["power_east_group"]["memberKeys"],
|
|
["fuel_storage_mod", "centrifuge_mod"],
|
|
)
|
|
self.assertEqual(
|
|
by_key["core_cabin"]["memberKeys"],
|
|
["core_cabin"],
|
|
)
|
|
|
|
for node in nodes:
|
|
self.assertIn("displayMeta", node)
|
|
self.assertIn("displayHint", node)
|
|
self.assertIn("decorations", node)
|
|
|
|
radiator_nodes = {
|
|
node["nodeKey"]
|
|
for node in nodes
|
|
if node["decorations"].get("radiator")
|
|
}
|
|
self.assertEqual(
|
|
radiator_nodes,
|
|
{
|
|
"power_storage_rack",
|
|
"power_west_group",
|
|
"power_east_group",
|
|
"nuclear_reactor",
|
|
},
|
|
)
|
|
|
|
def test_components_include_introduction_mission_and_time(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?mission=GHC-05"
|
|
)
|
|
components = {
|
|
component["key"]: component
|
|
for component in response.get_json()["components"]
|
|
}
|
|
self.assertEqual(
|
|
components["core_cabin"]["introducedMission"],
|
|
"GHC-01",
|
|
)
|
|
self.assertEqual(
|
|
datetime.fromisoformat(
|
|
components["core_cabin"]["introducedAt"]
|
|
).astimezone(timezone.utc),
|
|
datetime.fromisoformat("2024-06-16T07:23:00+00:00"),
|
|
)
|
|
|
|
def test_known_vehicle_dimensions_are_recorded(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/bases/guanghan/snapshot?mission=GHC-05"
|
|
)
|
|
components = {
|
|
component["key"]: component
|
|
for component in response.get_json()["components"]
|
|
}
|
|
self.assertEqual(
|
|
components["binglun_1"]["dimensions"],
|
|
{"heightM": 30.2, "massT": 25.0},
|
|
)
|
|
self.assertEqual(
|
|
components["binglun_2"]["dimensions"],
|
|
{"heightM": 30.2, "massT": 25.0},
|
|
)
|
|
self.assertEqual(
|
|
components["binglun_3"]["dimensions"],
|
|
{"heightM": 36.7, "massT": 28.194},
|
|
)
|
|
self.assertEqual(
|
|
components["wugang_2"]["dimensions"],
|
|
{"diameterM": 11.2, "massT": 9.64},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|