feat(topology): render temporal graph from database

tests/test_guanghan_topology_demo.py

@@ -1,114 +1,44 @@
 from __future__ import annotations
 
-import re
 import unittest
 from pathlib import Path
 
 
 ROOT = Path(__file__).resolve().parents[1]
 HTML_PATH = ROOT / "data" / "wiki" / "guanghan_topology_demo.html"
-TIMELINE_PATH = ROOT / "user_input" / "广寒计划" / "Timeline.md"
-
-CLICKABLE_MODULES = {
-    "core",
-    "habitat-1",
-    "habitat-2",
-    "greenhouse",
-    "laboratory",
-    "power-node",
-    "power-north",
-    "power-east",
-    "power-south",
-    "power-west",
-    "wugang-2",
-    "binglun-1",
-    "binglun-2",
-    "binglun-3",
-}
 
 
 class GuanghanTopologySourceTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
         cls.html = HTML_PATH.read_text(encoding="utf-8")
-        timeline = TIMELINE_PATH.read_text(encoding="utf-8")
-        cls.ghc05 = timeline.split("GHC-05任务", 1)[1].split("JC-02任务", 1)[0]
 
-    def test_clickable_module_inventory_is_exact(self) -> None:
-        found = set(re.findall(r'data-module="([^"]+)"', self.html))
-        self.assertEqual(found, CLICKABLE_MODULES)
+    def test_topology_is_rendered_from_snapshot_layout(self) -> None:
+        self.assertNotIn('<g class="topology-node"', self.html)
+        self.assertNotIn("GROUP_MEMBERS", self.html)
+        self.assertNotIn("data-conn=", self.html)
+        self.assertIn("currentData.layout", self.html)
+        self.assertIn("currentData.connections", self.html)
+        self.assertIn("function renderTopology", self.html)
 
-    def test_connection_inventory_matches_approved_topology(self) -> None:
-        self.assertEqual(self.html.count('data-connection="pressurized"'), 2)
-        self.assertIn('data-link="wugang-2--greenhouse"', self.html)
-        self.assertIn('data-link="habitat-1--power-north"', self.html)
-        self.assertNotIn("perimeter-corridor", self.html)
+    def test_radiators_are_dynamic_layout_decorations(self) -> None:
+        self.assertNotIn('<g class="radiator">', self.html)
+        self.assertIn("decorations.radiator", self.html)
+        self.assertIn("function createRadiator", self.html)
 
-    def test_walkways_and_radiators_are_not_clickable(self) -> None:
-        self.assertNotRegex(
-            self.html,
-            r'<(?:line|path|rect)[^>]+data-connection="pressurized"[^>]+data-module=',
-        )
-        self.assertNotRegex(
-            self.html,
-            r'<(?:g|rect)[^>]+class="[^"]*radiator[^"]*"[^>]+data-module=',
-        )
-
-    def test_each_radiator_has_a_visible_text_label(self) -> None:
-        radiator_groups = re.findall(
-            r'<g class="radiator"[^>]*>(.*?)</g>',
-            self.html,
-            flags=re.S,
-        )
-        self.assertEqual(len(radiator_groups), 4)
-        for group in radiator_groups:
-            self.assertIn(">散热器</text>", group)
-            rect = re.search(r'<rect x="(\d+)" y="(\d+)" width="(\d+)"', group)
-            label = re.search(r'<text[^>]+x="(\d+)" y="(\d+)"[^>]*>散热器</text>', group)
-            self.assertIsNotNone(rect)
-            self.assertIsNotNone(label)
-            rect_x, rect_y, rect_width = map(int, rect.groups())
-            label_x, label_y = map(int, label.groups())
-            self.assertGreaterEqual(label_x, rect_x)
-            self.assertLessEqual(label_x, rect_x + rect_width)
-            self.assertLess(label_y, rect_y)
-
-    def test_future_extensions_exist_only_as_source_comments(self) -> None:
-        comments = "\n".join(re.findall(r"<!--(.*?)-->", self.html, flags=re.S))
-        visible_source = re.sub(r"<!--.*?-->", "", self.html, flags=re.S)
+    def test_future_expansion_anchors_remain_source_comments(self) -> None:
         for direction in ("north", "east", "west"):
-            self.assertIn(f"FUTURE_EXPANSION_ANCHOR:{direction}", comments)
-        self.assertNotIn("未来北部十字结构", visible_source)
-        self.assertNotIn("未来东部十字结构", visible_source)
-        self.assertNotIn("未来西部十字结构", visible_source)
-
-    def test_every_clickable_module_has_dimensions_field(self) -> None:
-        for module_id in CLICKABLE_MODULES:
-            match = re.search(
-                rf'"{re.escape(module_id)}"\s*:\s*\{{(.*?)\n    \}}(?:,|\n)',
+            self.assertIn(
+                f"FUTURE_EXPANSION_ANCHOR:{direction}",
                 self.html,
-                flags=re.S,
             )
-            self.assertIsNotNone(match, msg=f"{module_id} detail object is missing")
-            self.assertIn("dimensions:", match.group(1), msg=f"{module_id} is missing dimensions")
+        self.assertNotIn("未来北部十字结构", self.html)
+        self.assertNotIn("未来东部十字结构", self.html)
+        self.assertNotIn("未来西部十字结构", self.html)
 
-    def test_timeline_contains_only_the_two_real_corridors(self) -> None:
-        for incorrect in (
-            "4段加压走道",
-            "居住舱#1↔居住舱#2",
-            "居住舱#2↔实验舱",
-            "实验舱↔温室",
-            "温室↔居住舱#1",
-            "环形加压通路",
-            "8个对接面",
-            "4段加压走道48小时",
-        ):
-            self.assertNotIn(incorrect, self.ghc05)
-        self.assertIn("居住舱#1远端对接口", self.ghc05)
-        self.assertIn("电力模块北部", self.ghc05)
-        self.assertIn("温室远端对接口", self.ghc05)
-        self.assertIn("吴刚#2", self.ghc05)
-        self.assertIn("从温室模块远端接口断开", self.ghc05)
+    def test_page_uses_same_origin_api(self) -> None:
+        self.assertIn("const API_BASE = ''", self.html)
+        self.assertNotIn("http://localhost:5000", self.html)
 
 
 if __name__ == "__main__":

tests/test_guanghan_topology_e2e.py

@@ -1,27 +1,26 @@
 from __future__ import annotations
 
 import asyncio
-import functools
 import threading
 import unittest
 from contextlib import contextmanager
-from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
 
 from playwright.async_api import async_playwright
+from werkzeug.serving import make_server
 
-
-ROOT = Path(__file__).resolve().parents[1]
+from topology_api.app import create_app
 
 
 @contextmanager
-def local_server():
-    handler = functools.partial(SimpleHTTPRequestHandler, directory=str(ROOT))
-    server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+def topology_server():
+    app = create_app()
+    app.config.update(TESTING=True)
+    server = make_server("127.0.0.1", 0, app, threaded=True)
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     try:
-        yield f"http://127.0.0.1:{server.server_port}"
+        port = server.socket.getsockname()[1]
+        yield f"http://127.0.0.1:{port}"
     finally:
         server.shutdown()
         server.server_close()
@@ -29,44 +28,91 @@ def local_server():
 
 
 class GuanghanTopologyE2ETests(unittest.TestCase):
-    def test_rendered_topology_and_modal_interactions(self) -> None:
+    def test_historical_graph_and_ghc05_rendering(self) -> None:
         async def scenario(base_url: str) -> None:
             async with async_playwright() as pw:
                 browser = await pw.chromium.launch(headless=True)
-                page = await browser.new_page(viewport={"width": 1440, "height": 1100})
-                errors: list[str] = []
-                page.on("console", lambda msg: errors.append(msg.text) if msg.type == "error" else None)
-                await page.goto(
-                    f"{base_url}/data/wiki/guanghan_topology_demo.html",
-                    wait_until="networkidle",
+                page = await browser.new_page(
+                    viewport={"width": 1440, "height": 1000}
+                )
+                console_errors: list[str] = []
+                page.on(
+                    "console",
+                    lambda msg: (
+                        console_errors.append(msg.text)
+                        if msg.type == "error"
+                        else None
+                    ),
                 )
 
-                self.assertEqual(await page.locator("[data-module]").count(), 14)
-                self.assertEqual(await page.locator('[data-connection="pressurized"]').count(), 2)
-                strokes = await page.locator('[data-connection="pressurized"]').evaluate_all(
+                async def open_mission(code: str) -> None:
+                    await page.goto(
+                        f"{base_url}/guanghan_topology_demo.html"
+                        f"?mission={code}",
+                        wait_until="networkidle",
+                    )
+                    await page.wait_for_function(
+                        """code => (
+                            document.querySelector('#page-subtitle')
+                              ?.textContent.startsWith(code + ' ·')
+                        )""",
+                        arg=code,
+                    )
+
+                await open_mission("BL-01")
+                self.assertEqual(
+                    await page.locator(".topology-node").count(), 1
+                )
+                self.assertEqual(
+                    await page.locator(".connection").count(), 0
+                )
+                self.assertEqual(await page.locator(".radiator").count(), 0)
+
+                await open_mission("GHC-01")
+                self.assertEqual(
+                    await page.locator(".topology-node").count(), 2
+                )
+                self.assertEqual(
+                    await page.locator(".connection").count(), 0
+                )
+
+                await open_mission("GHC-05")
+                self.assertEqual(
+                    await page.locator(".topology-node").count(), 14
+                )
+                self.assertEqual(
+                    await page.locator(".connection.pressurized").count(), 2
+                )
+                self.assertEqual(await page.locator(".radiator").count(), 4)
+                strokes = await page.locator(
+                    ".connection.pressurized"
+                ).evaluate_all(
                     "(els) => els.map((el) => getComputedStyle(el).stroke)"
                 )
-                self.assertEqual(strokes, ["rgb(81, 232, 156)", "rgb(81, 232, 156)"])
+                self.assertEqual(
+                    strokes,
+                    ["rgb(81, 232, 156)", "rgb(81, 232, 156)"],
+                )
 
-                for forbidden in ("未来北部十字结构", "未来东部十字结构", "未来西部十字结构"):
-                    self.assertEqual(await page.get_by_text(forbidden, exact=False).count(), 0)
-
-                await page.locator('[data-module="core"]').click()
-                self.assertEqual(await page.locator("#detail-modal").count(), 1)
-                self.assertEqual(await page.locator("#detail-modal").get_attribute("aria-hidden"), "false")
-                self.assertEqual(await page.locator("#detail-title").inner_text(), "广寒核心舱")
-                await page.keyboard.press("Escape")
-                self.assertEqual(await page.locator("#detail-modal").get_attribute("aria-hidden"), "true")
-
-                await page.locator('[data-module="habitat-1"]').click()
-                self.assertIn("运输：高2.7 m", await page.locator("#detail-dimensions").inner_text())
-                await page.locator("#detail-modal").click(position={"x": 5, "y": 5})
-                self.assertEqual(await page.locator("#detail-modal").get_attribute("aria-hidden"), "true")
-
-                self.assertEqual(errors, [])
+                overflows = await page.locator(
+                    ".topology-node"
+                ).evaluate_all(
+                    """nodes => nodes.flatMap(node => {
+                        const rect = node.querySelector('rect').getBBox();
+                        return [...node.querySelectorAll('text')]
+                          .map(text => ({text: text.textContent,
+                                        box: text.getBBox()}))
+                          .filter(item =>
+                            item.box.x < rect.x ||
+                            item.box.x + item.box.width > rect.x + rect.width
+                          );
+                    })"""
+                )
+                self.assertEqual(overflows, [])
+                self.assertEqual(console_errors, [])
                 await browser.close()
 
-        with local_server() as base_url:
+        with topology_server() as base_url:
             asyncio.run(scenario(base_url))
 
 

tests/test_topology_api.py

@@ -124,6 +124,23 @@ class TopologyApiQueryTests(unittest.TestCase):
             },
         )
 
+    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"