Adding a single tier rack fabric

.vscode/extensions.json

@@ -1,5 +1,7 @@
 {
   "recommendations": [
+    "ms-python.debugpy",
+    "ms-python.pylint",
     "ms-python.python",
     "esbenp.prettier-vscode",
     "tamasfe.even-better-toml",

.vscode/settings.json

@@ -1,4 +1,10 @@
 {
     "python.terminal.activateEnvironment": true,
-    "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python3"
+    "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python3",
+    "python.testing.pytestArgs": [
+        "-s",
+        "src/tests/"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
 }

src/infragraph/blueprints/single_tier/single_tier_fabric.py

@@ -0,0 +1,50 @@
+from infragraph import *
+from infragraph.switch import Switch
+from infragraph.infragraph_service import InfraGraphService
+
+
+class SingleTierFabric(Infrastructure):
+    """Return a single tier rack fabric with the following characteristics:
+    Inputs:
+        - server (device instance)
+        - server count - 'x'
+        - bandwidth in Gbps
+    Output:
+        - servers with 'x' count
+        - a generic rack switch with n ports where:
+            n = server count 'x' * nics in each server
+        - connectivity between switch and hosts is the provided bandwidth
+    """
+
+    def __init__(self, server: Device, server_count: int, bandwidth: int = 100):
+        super().__init__(name="single-tier-fabric", description="Single Tier Fabric")
+
+        server_nic_component = InfraGraphService.get_component(server, Component.NIC)
+        total_ports = server_nic_component.count * server_count
+
+        switch_device = Switch(total_ports)
+        self.devices.append(server).append(switch_device)
+
+        hosts = self.instances.add(name=server.name, device=server.name, count=server_count)
+        switch_instance = self.instances.add(name="switch", device=switch_device.name, count=1)
+
+        switch_link = self.links.add(
+            name="switch-link",
+            description="Link characteristics for connectivity between servers and rack switches",
+        )
+        switch_link.physical.bandwidth.gigabits_per_second = bandwidth
+
+        switch_component = InfraGraphService.get_component(switch_device, Component.PORT)
+
+        # link each host to one leaf switch
+        for idx in range(total_ports):
+            host_index = idx // server_nic_component.count
+            host_component_index = idx % server_nic_component.count
+            edge = self.edges.add(
+                scheme=InfrastructureEdge.ONE2ONE, link=switch_link.name
+            )
+            edge.ep1.instance = f"{hosts.name}[{host_index}]"
+            edge.ep1.component = f"{server_nic_component.name}[{host_component_index}]"
+            edge.ep2.instance = f"{switch_instance.name}[0]"
+            edge.ep2.component = f"{switch_component.name}[{idx}]"
+

src/tests/test_blueprints/test_single_tier_fabric.py

@@ -0,0 +1,64 @@
+import pytest
+import networkx
+from infragraph.infragraph_service import InfraGraphService
+from infragraph.blueprints.single_tier.single_tier_fabric import SingleTierFabric
+from infragraph.dgx import Dgx
+from infragraph.server import Server
+
+def print_graph(graph):
+    for node, attrs in graph.nodes(data=True):
+        print(f"Node: {node}, Attributes: {attrs}")
+
+    for u, v, attrs in graph.edges(data=True):
+        print(f"Edge: ({u}, {v}), Attributes: {attrs}")
+
+@pytest.mark.asyncio
+async def test_single_tier_fabric_one_dgx():
+    """
+    Generate a single tier fabric with 1 dgx host and validate the infragraph
+    """
+    dgx = Dgx()
+    single_tier_fabric = SingleTierFabric(dgx, 1)
+    # create the graph
+    service = InfraGraphService()
+    service.set_graph(single_tier_fabric)
+
+    # validations
+    g = service.get_networkx_graph()
+    print(networkx.write_network_text(g, vertical_chains=True))
+    print_graph(g)
+
+@pytest.mark.asyncio
+async def test_single_tier_fabric_multi_dgx():
+    """
+    Generate a single tier fabric with multi dgx host and validate the infragraph
+    """
+    dgx = Dgx()
+    single_tier_fabric = SingleTierFabric(dgx, 3)
+    # create the graph
+    service = InfraGraphService()
+    service.set_graph(single_tier_fabric)
+
+    # validations
+    g = service.get_networkx_graph()
+    print(networkx.write_network_text(g, vertical_chains=True))
+    print_graph(g)
+
+@pytest.mark.asyncio
+async def test_single_tier_fabric_multi_server():
+    """
+    Generate a single tier fabric with multi server host and validate the infragraph
+    """
+    server = Server()
+    single_tier_fabric = SingleTierFabric(server, 2)
+    # create the graph
+    service = InfraGraphService()
+    service.set_graph(single_tier_fabric)
+
+    # validations
+    g = service.get_networkx_graph()
+    print(networkx.write_network_text(g, vertical_chains=True))
+    print_graph(g)
+
+if __name__ == "__main__":
+    pytest.main(["-s", __file__])

src/tests/test_infrastructure.py

@@ -20,4 +20,4 @@ async def test_infrastructure():
 
 
 if __name__ == "__main__":
-    pytest.main(["-s", __file__])
+    pytest.main(["-s", __file__])