added apply_kubernetes_yaml_obj

Author: m3dwards

src/warnet/cli/k8s.py

@@ -1,5 +1,8 @@
 import json
+import yaml
+import tempfile
 from importlib.resources import files
+from pathlib import Path
 from typing import Any, Dict
 
 from kubernetes import client, config
@@ -89,6 +92,17 @@ def apply_kubernetes_yaml(yaml_file: str):
     return stream_command(command)
 
 
+def apply_kubernetes_yaml_obj(yaml_obj: str):
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as temp_file:
+        yaml.dump(yaml_obj, temp_file)
+        temp_file_path = temp_file.name
+
+    try:
+        apply_kubernetes_yaml(temp_file_path)
+    finally:
+        Path(temp_file_path).unlink()
+
+
 def delete_namespace(namespace: str):
     command = f"kubectl delete namespace {namespace} --ignore-not-found"
     return stream_command(command)

src/warnet/cli/network.py

@@ -10,6 +10,7 @@
 
 from .bitcoin import _rpc
 from .k8s import (
+    apply_kubernetes_yaml_obj,
     create_kubernetes_object,
     delete_namespace,
     get_edges,
@@ -32,10 +33,29 @@ def network():
     """Network commands"""
 
 
-def create_edges_map(graph):
+class Edge:
+    def __init__(self, src: str, dst: str, data: Dict[str, Any]):
+        self.src = src
+        self.dst = dst
+        self.data = data
+
+    def to_dict(self):
+        return {"src": self.src, "dst": self.dst, "data": self.data}
+
+
+def edges_from_network_file(network_file: Dict[str, Any]) -> List[Edge]:
+    edges = []
+    for node in network_file["nodes"]:
+        if "connect" in node:
+            for connection in node["connect"]:
+                edges.append(Edge(node["name"], connection, ""))
+    return edges
+
+
+def create_edges_map(network_file: Dict[str, Any]):
     edges = []
-    for src, dst, data in graph.edges(data=True):
-        edges.append({"src": src, "dst": dst, "data": data})
+    for edge in edges_from_network_file(network_file):
+        edges.append(edge.to_dict())
     config_map = create_kubernetes_object(
         kind="ConfigMap",
         metadata={
@@ -44,7 +64,7 @@ def create_edges_map(graph):
         },
     )
     config_map["data"] = {"data": json.dumps(edges)}
-    return config_map
+    apply_kubernetes_yaml_obj(config_map)
 
 
 def setup_logging_helm() -> bool:
@@ -107,6 +127,8 @@ def start(network_name: str, logging: bool, network: str):
             if temp_override_file_path:
                 Path(temp_override_file_path).unlink()
 
+    create_edges_map(network_file)
+
 
 @network.command()
 def down():