Add test for invalid node expansion request

Author: ericfe-google

bigquery_magics/graph_server.py

@@ -220,20 +220,16 @@ def handle_post_node_expansion(self):
         - params: A JSON string containing connection parameters (project, instance, database, graph)
         - request: A dictionary with node details (uid, node_labels, node_properties, direction, edge_label)
         """
-        try:
-            data = self.parse_post_data()
-
-            # Execute node expansion with:
-            # - params_str: JSON string with connection parameters (project, instance, database, graph)
-            # - request: Dict with node details (uid, node_labels, node_properties, direction, edge_label)
-            self.do_data_response(
-                execute_node_expansion(
-                    params=data.get("params"), request=data.get("request")
-                )
+        data = self.parse_post_data()
+
+        # Execute node expansion with:
+        # - params_str: JSON string with connection parameters (project, instance, database, graph)
+        # - request: Dict with node details (uid, node_labels, node_properties, direction, edge_label)
+        self.do_data_response(
+            execute_node_expansion(
+                params=data.get("params"), request=data.get("request")
             )
-        except BaseException as e:
-            self.do_error_response(e)
-            return
+        )
 
     def do_GET(self):
         assert self.path == GraphServer.endpoints["get_ping"]

tests/unit/test_graph_server.py

@@ -441,5 +441,21 @@ def test_post_node_expansion(self):
         )
 
 
+    @pytest.mark.skipif(
+        graph_visualization is None, reason="Requires `spanner-graph-notebook`"
+    )
+    def test_post_node_expansion_invalid_request(self):
+        self.assertTrue(self.server_thread.is_alive())
+        route = graph_server.graph_server.build_route(
+            graph_server.GraphServer.endpoints["post_node_expansion"]
+        )
+        request = {}
+        response = requests.post(route, json={"params": json.dumps(request)})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.json(), {"error": "Node expansion not yet implemented"}
+        )
+
+
 def test_stop_server_never_started():
     graph_server.graph_server.stop_server()