rpc: add graph valiate subcommand

src/cli/graph.py

@@ -36,3 +36,11 @@ def create(
             },
         )
     )
+
+@graph.command()
+@click.argument("graph", type=Path)
+def validate(graph: Path):
+    """
+    Validate a graph file against the schema.
+    """
+    print(rpc_call("graph_validate", {"graph_path": graph.as_posix()}))

src/warnet/server.py

@@ -16,6 +16,7 @@
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
 
+import jsonschema
 import networkx as nx
 import scenarios
 from flask import Flask, jsonify, request
@@ -24,6 +25,8 @@
 from warnet.utils import (
     create_cycle_graph,
     gen_config_dir,
+    load_schema,
+    validate_graph_schema,
 )
 from warnet.warnet import Warnet
 
@@ -169,6 +172,7 @@ def setup_rpc(self):
         self.jsonrpc.register(self.network_export)
         # Graph
         self.jsonrpc.register(self.graph_generate)
+        self.jsonrpc.register(self.graph_validate)
         # Debug
         self.jsonrpc.register(self.generate_deployment)
         # Server
@@ -457,6 +461,17 @@ def graph_generate(
             self.logger.error(msg)
             raise ServerError(message=msg) from e
 
+    def graph_validate(self, graph_path: str) -> str:
+
+        schema = load_schema()
+        with open(graph_path) as f:
+            graph = nx.parse_graphml(f.read(), node_type=int)
+        try:
+            validate_graph_schema(schema, graph)
+        except (jsonschema.ValidationError, jsonschema.SchemaError) as e:
+            raise ServerError(message=f"Schema of {graph_path} is invalid: {e}") from e
+        return f"Schema of {graph_path} is valid"
+
     def network_down(self, network: str = "warnet") -> str:
         """
         Stop all containers in <network>.

test/graph_test.py

@@ -23,6 +23,10 @@
         with open(tf, "w") as file:
             file.write(xml)
 
+    # Validate the graph schema
+    assert "invalid" not in base.warcli(f"graph validate {Path(tf)}")
+    print(f"Graph at {tf} validated successfully")
+
     # Test that the graph actually works
     print(base.warcli(f"network start {Path(tf)}"))
     base.wait_for_all_tanks_status(target="running")