Refactor Builder interface

.gitignore

@@ -4,4 +4,5 @@ clab-*
 **/__pycache__
 *.pyc
 .venv/*
-*.tar**
+*.tar**
+.idea/

builders/builder2.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+from topology import Topology
+
+
+class TopologyBuilder(ABC):
+    def __init__(self, topology: Topology):
+        self.topology = topology
+
+    @abstractmethod
+    def build_topology(self):
+        pass
+
+    @abstractmethod
+    def destroy_topology(self):
+        pass

builders/clab_builder.py

@@ -0,0 +1,110 @@
+from builders.builder2 import TopologyBuilder
+from topology import Topology
+import yaml
+import subprocess
+import logging
+
+
+class TopologyDumper:
+    def __init__(self, topology: Topology):
+        self.topology = topology
+
+    def __to_dict(self):
+        return {
+            "name": self.topology.name,
+            "topology": {
+                "nodes": {
+                    node.name: {
+                        "kind": node.kind,
+                        "image": f"{node.kind}:latest",
+                    }
+                    for node in self.topology.nodes
+                },
+                "links": [
+                    {
+                        "endpoints": [
+                            f"{link.name_from}:{link.interface_from}",
+                            f"{link.name_to}:{link.interface_to}",
+                        ]
+                    }
+                    for link in self.topology.links
+                ],
+            },
+        }
+
+    def dump(self) -> str:
+        data = self.__to_dict()
+        return yaml.safe_dump(data, default_flow_style=False, sort_keys=False)
+
+
+class ClabBuilder(TopologyBuilder):
+    def __init__(self, topology: Topology):
+        super().__init__(topology)
+        self.logger = logging.getLogger(__name__)
+
+    def build_topology(self):
+        topology_spec = TopologyDumper(self.topology).dump()
+        self.logger.info(
+            f"Building topology {self.topology.name} with Containerlab..."
+        )
+        try:
+            proc = subprocess.Popen(
+                "clab deploy --topo -",
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
+            stdout, stderr = proc.communicate(topology_spec)
+            if proc.returncode != 0:
+                self.logger.info(
+                    f"Error creating topology {self.topology.name}: {stderr}"
+                )
+                raise RuntimeError(
+                    f"Topology deployment failed. "
+                    f"Code: {proc.returncode}: {stderr}"
+                )
+
+            self.logger.info(
+                f"Successfully built topology {self.topology.name}"
+            )
+
+        except FileNotFoundError:
+            self.logger.info(
+                "Containerlab not installed. Aborting topology creation"
+            )
+            raise RuntimeError(
+                "Containerlab is required to use the clab builder."
+            )
+
+    def destroy_topology(self):
+        topology_spec = TopologyDumper(self.topology).dump()
+        self.logger.info(
+            f"Destroying Containerlab topology {self.topology.name}..."
+        )
+        try:
+            proc = subprocess.Popen(
+                "clab destroy --cleanup --topo -",
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
+            stdout, stderr = proc.communicate(topology_spec)
+            if proc.returncode != 0:
+                self.logger.info(
+                    f"Error destroying topology {self.topology.name}: {stderr}"
+                )
+                raise RuntimeError(
+                    f"Topology destruction failed. "
+                    f"Code: {proc.returncode}: {stderr}"
+                )
+
+            self.logger.info(
+                f"Successfully removed topology {self.topology.name}"
+            )
+
+        except FileNotFoundError:
+            self.logger.info(
+                "Containerlab not installed. Aborting topology destruction"
+            )

config/cli.py

@@ -1,7 +1,7 @@
 import argparse
 
 
-__all__ = ['ArgParser']
+__all__ = ["ArgParser"]
 
 
 class ArgParser:

config/settings.py

@@ -13,14 +13,26 @@
 import yaml
 
 
-__all__ = ['TopologyType', 'TopologyAdjustment', 'KafkaSettings',
-           'RabbitSettings', 'TopologyAdjustmentAdd',
-           'TopologyAdjustmentRemove', 'TopologyAdjustmentRemoveLink',
-           'TopologyAdjustmentAddLink', 'InterfaceSettings',
-           'InterfaceCredentials', 'RealnetSettings',
-           'ControllerSettings', 'SiblingSettings',
-           'BuilderSettings', 'read_config', 'validate_config',
-           'AppSettings', 'Settings']
+__all__ = [
+    "TopologyType",
+    "TopologyAdjustment",
+    "KafkaSettings",
+    "RabbitSettings",
+    "TopologyAdjustmentAdd",
+    "TopologyAdjustmentRemove",
+    "TopologyAdjustmentRemoveLink",
+    "TopologyAdjustmentAddLink",
+    "InterfaceSettings",
+    "InterfaceCredentials",
+    "RealnetSettings",
+    "ControllerSettings",
+    "SiblingSettings",
+    "BuilderSettings",
+    "read_config",
+    "validate_config",
+    "AppSettings",
+    "Settings",
+]
 
 
 class TopologyType(BaseModel):

tests/topology/test_topology.py

@@ -0,0 +1,40 @@
+import unittest
+from ...topology import TopologyBuilder
+
+
+class TopologyTests(unittest.TestCase):
+
+    def setUp(self):
+        self.b = TopologyBuilder()
+
+    def test_topology_builder(self):
+        self.b = TopologyBuilder()
+        self.b.add_node("srl1", "nokia_srlinux")
+        self.b.add_node("srl2", "nokia_srlinux")
+        self.b.add_link("srl1", "srl2", "e1-1", "e1-1")
+
+        topo = self.b.build()
+        self.assertEqual(len(topo.nodes), 2)
+        self.assertEqual(len(topo.links), 1)
+
+        self.assertEqual(topo.nodes[0].name, "srl1")
+        self.assertEqual(topo.nodes[1].name, "srl2")
+        self.assertEqual(topo.links[0].name_from, "srl1")
+        self.assertEqual(topo.links[0].name_to, "srl2")
+        self.assertEqual(topo.links[0].interface_to, "e1-1")
+        self.assertEqual(topo.links[0].interface_from, "e1-1")
+
+    def test_topology_faulty_link(self):
+        self.b.add_node("srl1", "nokia_srlinux")
+        with self.assertRaises(AssertionError):
+            self.b.add_link("srl1", "srl2", "e1-1", "e1-1")
+
+    def test_topology_clear(self):
+        self.b.add_node("srl1", "nokia_srlinux")
+        self.b.add_node("srl2", "nokia_srlinux")
+        self.b.add_link("srl1", "srl2", "e1-1", "e1-1")
+        self.b.clear()
+        topo = self.b.build()
+
+        self.assertEqual(len(topo.nodes), 0)
+        self.assertEqual(len(topo.links), 0)

topology/__init__.py

@@ -0,0 +1,3 @@
+from .link import *  # noqa: F403, F401
+from .node import *  # noqa: F403, F401
+from .topology import *  # noqa: F403, F401

topology/link.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+__all__ = ["Link"]
+
+
+@dataclass
+class Link:
+    name_from: str
+    name_to: str
+    interface_from: str
+    interface_to: str

topology/node.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass
+from typing import List
+from .link import Link
+
+__all__ = ["Node"]
+
+
+@dataclass
+class Node:
+    name: str
+    kind: str
+    links: List[Link]

topology/topology.py

@@ -0,0 +1,49 @@
+from .link import Link
+from .node import Node
+from typing import List
+
+__all__ = ["Topology", "TopologyBuilder"]
+
+
+class Topology:
+    def __init__(self):
+        self.name = ""
+        self.nodes: List[Node] = []
+        self.links: List[Link] = []
+
+    @staticmethod
+    def builder():
+        return TopologyBuilder()
+
+
+class TopologyBuilder:
+    def __init__(self) -> None:
+        self._topo = Topology()
+
+    def add_node(self, name: str, kind: str) -> None:
+        self._topo.nodes.append(Node(name, kind, []))
+
+    def add_link(
+        self,
+        node_from: str,
+        node_to: str,
+        interface_from: str,
+        interface_to: str,
+    ) -> None:
+        # assert
+        assert any(node.name == node_from for node in self._topo.nodes)
+        assert any(node.name == node_to for node in self._topo.nodes)
+        self._topo.links.append(
+            Link(node_from, node_to, interface_from, interface_to)
+        )
+
+    def name(self, name: str) -> None:
+        self._topo.name = name
+
+    def clear(self) -> None:
+        self._topo.name = ""
+        self._topo.nodes = []
+        self._topo.links = []
+
+    def build(self) -> Topology:
+        return self._topo