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"), ) if __name__ == "__main__": unittest.main()