test(topology): require database presentation payload

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-09 23:43:20 +08:00
parent 14ea13d6a4
commit 00026fbbe7
1 changed files with 95 additions and 0 deletions
@@ -200,6 +200,101 @@ class TopologyApiQueryTests(unittest.TestCase):
            datetime.fromisoformat("2024-06-16T07:23:00+00:00"),
        )

+    def test_layout_includes_database_presentation_payload(self) -> None:
+        response = self.client.get(
+            "/api/v1/bases/guanghan/snapshot?mission=GHC-05"
+        )
+        self.assertEqual(response.status_code, 200)
+        layout = response.get_json()["layout"]
+        presentation = layout["presentation"]
+
+        self.assertEqual(
+            presentation["theme"]["key"],
+            "guanghan_dark_engineering_v1",
+        )
+        self.assertIn("nodeStyles", presentation)
+        self.assertIn("edgeStyles", presentation)
+        self.assertIn("decorationStyles", presentation)
+        self.assertIn("legendItems", presentation)
+
+        node_style_keys = {
+            node["styleKey"] for node in layout["nodes"]
+        }
+        self.assertLessEqual(
+            node_style_keys,
+            set(presentation["nodeStyles"].keys()),
+        )
+
+        connection_types = {
+            connection["type"]
+            for connection in response.get_json()["connections"]
+        }
+        self.assertLessEqual(
+            connection_types,
+            set(presentation["edgeStyles"].keys()),
+        )
+
+        self.assertEqual(
+            presentation["edgeStyles"]["pressurized_passage"],
+            {
+                "displayName": "加压走道",
+                "strokeColor": "#51e89c",
+                "strokeWidth": 9.0,
+                "dashArray": None,
+                "lineCap": "square",
+                "legendGroup": "pressurized",
+            },
+        )
+        self.assertEqual(
+            presentation["decorationStyles"]["radiator"]["label"],
+            "散热器",
+        )
+        self.assertEqual(
+            [
+                item["key"]
+                for item in presentation["legendItems"]
+            ],
+            ["structural", "pressurized", "utility", "radiator"],
+        )
+
+    def test_presentation_styles_are_database_backed(self) -> None:
+        from topology_api.db import get_writer_connection
+
+        with get_writer_connection() as conn:
+            conn.execute("SET search_path TO topology, public")
+            with conn.cursor() as cur:
+                cur.execute("""
+                    SELECT stroke_color
+                    FROM diagram_edge_styles
+                    WHERE connection_type = 'pressurized_passage'
+                """)
+                original_color = cur.fetchone()["stroke_color"]
+                cur.execute("""
+                    UPDATE diagram_edge_styles
+                    SET stroke_color = '#00ffaa'
+                    WHERE connection_type = 'pressurized_passage'
+                """)
+                conn.commit()
+            try:
+                response = self.client.get(
+                    "/api/v1/bases/guanghan/snapshot?mission=GHC-05"
+                )
+                self.assertEqual(response.status_code, 200)
+                presentation = response.get_json()["layout"]["presentation"]
+                self.assertEqual(
+                    presentation["edgeStyles"]["pressurized_passage"]
+                    ["strokeColor"],
+                    "#00ffaa",
+                )
+            finally:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE diagram_edge_styles
+                        SET stroke_color = %(original)s
+                        WHERE connection_type = 'pressurized_passage'
+                    """, {"original": original_color})
+                    conn.commit()
+
    def test_known_vehicle_dimensions_are_recorded(self) -> None:
        response = self.client.get(
            "/api/v1/bases/guanghan/snapshot?mission=GHC-05"