add iptrunks rest api

gso/api/v1/__init__.py

@@ -3,6 +3,7 @@
 from fastapi import APIRouter
 
 from gso.api.v1.imports import router as imports_router
+from gso.api.v1.network import router as network_router
 from gso.api.v1.processes import router as processes_router
 from gso.api.v1.subscriptions import router as subscriptions_router
 
@@ -11,3 +12,4 @@ router = APIRouter()
 router.include_router(imports_router)
 router.include_router(subscriptions_router)
 router.include_router(processes_router)
+router.include_router(network_router)

gso/api/v1/imports.py

@@ -10,7 +10,7 @@ from orchestrator.services import processes
 from pydantic import BaseModel, root_validator, validator
 
 from gso.auth.security import opa_security_default
-from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
 from gso.services import subscriptions
@@ -65,7 +65,7 @@ class IptrunkImportModel(BaseModel):
     geant_s_sid: str
     iptrunk_type: IptrunkType
     iptrunk_description: str
-    iptrunk_speed: PhyPortCapacity
+    iptrunk_speed: PhysicalPortCapacity
     iptrunk_minimum_links: int
     iptrunk_isis_metric: int
     side_a_node_id: str

gso/api/v1/network.py

+"""API endpoints for network related operations."""
+
+from uuid import UUID
+
+from fastapi import APIRouter
+from orchestrator.domain import SubscriptionModel
+from orchestrator.schemas.base import OrchestratorBaseModel
+from orchestrator.services.subscriptions import build_extended_domain_model
+from starlette import status
+
+from gso.products.product_blocks.iptrunk import PhysicalPortCapacity
+from gso.services.subscriptions import get_active_iptrunk_subscriptions
+
+router = APIRouter(
+    prefix="/networks",
+    tags=["Network"],
+)
+
+
+class RouterBlock(OrchestratorBaseModel):
+    """Router block schema."""
+
+    subscription_instance_id: UUID
+    router_fqdn: str
+
+
+class IptrunkSideBlock(OrchestratorBaseModel):
+    """Iptrunk side block schema."""
+
+    subscription_instance_id: UUID
+    iptrunk_side_node: RouterBlock
+
+
+class IptrunkBlock(OrchestratorBaseModel):
+    """Iptrunk block schema."""
+
+    subscription_instance_id: UUID
+    iptrunk_speed: str
+    iptrunk_capacity: str
+    iptrunk_isis_metric: int
+    iptrunk_sides: list[IptrunkSideBlock]
+
+
+class IptrunkSchema(OrchestratorBaseModel):
+    """Iptrunk schema."""
+
+    subscription_id: UUID
+    insync: bool
+    iptrunk: IptrunkBlock
+
+
+class NetworkTopologyDomainModelSchema(OrchestratorBaseModel):
+    """Network topology domain model schema."""
+
+    iptrunks: list[IptrunkSchema]
+
+
+def _calculate_iptrunk_capacity(iptrunk_sides: list, iptrunk_speed: PhysicalPortCapacity) -> str:
+    """Calculate the total capacity of an IP trunk."""
+    int_iptrunk_speed = int(iptrunk_speed.value.replace("G", ""))
+    capacity = int_iptrunk_speed * len(iptrunk_sides[0]["iptrunk_side_ae_members"])
+    return f"{capacity}G"
+
+
+@router.get(
+    "/topology",
+    status_code=status.HTTP_200_OK,
+    response_model=NetworkTopologyDomainModelSchema,
+)
+def network_topology() -> NetworkTopologyDomainModelSchema:
+    """Retrieve all active or provisioning iptrunk subscriptions."""
+    topology: dict = {"iptrunks": []}
+    active_iptrunks = get_active_iptrunk_subscriptions()
+    for iptrunk in active_iptrunks:
+        subscription = SubscriptionModel.from_subscription(iptrunk["subscription_id"])
+        extended_model = build_extended_domain_model(subscription)
+        formatted_model = {
+            "subscription_id": extended_model["subscription_id"],
+            "insync": extended_model["insync"],
+            "iptrunk": {
+                "subscription_instance_id": extended_model["iptrunk"]["subscription_instance_id"],
+                "iptrunk_speed": extended_model["iptrunk"]["iptrunk_speed"],
+                "iptrunk_isis_metric": extended_model["iptrunk"]["iptrunk_isis_metric"],
+                "iptrunk_capacity": _calculate_iptrunk_capacity(
+                    extended_model["iptrunk"]["iptrunk_sides"], extended_model["iptrunk"]["iptrunk_speed"]
+                ),
+                "iptrunk_sides": [
+                    {
+                        "subscription_instance_id": side["subscription_instance_id"],
+                        "iptrunk_side_node": {
+                            "subscription_instance_id": side["iptrunk_side_node"]["subscription_instance_id"],
+                            "router_fqdn": side["iptrunk_side_node"]["router_fqdn"],
+                        },
+                    }
+                    for side in extended_model["iptrunk"]["iptrunk_sides"]
+                ],
+            },
+        }
+        topology["iptrunks"].append(IptrunkSchema(**formatted_model))
+
+    return NetworkTopologyDomainModelSchema(**topology)

gso/products/product_blocks/iptrunk.py

@@ -14,7 +14,7 @@ from gso.products.product_blocks.router import (
 )
 
 
-class PhyPortCapacity(strEnum):
+class PhysicalPortCapacity(strEnum):
     """Physical port capacity enumerator.
 
     An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
@@ -113,7 +113,7 @@ class IptrunkBlockInactive(
     geant_s_sid: str | None = None
     iptrunk_description: str | None = None
     iptrunk_type: IptrunkType | None = None
-    iptrunk_speed: PhyPortCapacity | None = None
+    iptrunk_speed: PhysicalPortCapacity | None = None
     iptrunk_minimum_links: int | None = None
     iptrunk_isis_metric: int | None = None
     iptrunk_ipv4_network: ipaddress.IPv4Network | None = None
@@ -127,7 +127,7 @@ class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLife
     geant_s_sid: str | None = None
     iptrunk_description: str | None = None
     iptrunk_type: IptrunkType | None = None
-    iptrunk_speed: PhyPortCapacity | None = None
+    iptrunk_speed: PhysicalPortCapacity | None = None
     iptrunk_minimum_links: int | None = None
     iptrunk_isis_metric: int | None = None
     iptrunk_ipv4_network: ipaddress.IPv4Network | None = None
@@ -145,7 +145,7 @@ class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
     #:  The type of trunk, can be either dark fibre or leased capacity.
     iptrunk_type: IptrunkType
     #:  The speed of the trunk, measured per interface associated with it.
-    iptrunk_speed: PhyPortCapacity
+    iptrunk_speed: PhysicalPortCapacity
     #:  The minimum amount of links the trunk should consist of.
     iptrunk_minimum_links: int
     #:  The :term:`ISIS` metric of this link

gso/workflows/iptrunk/create_iptrunk.py

@@ -20,7 +20,7 @@ from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlockInactive,
     IptrunkSideBlockInactive,
     IptrunkType,
-    PhyPortCapacity,
+    PhysicalPortCapacity,
 )
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router
@@ -60,7 +60,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         geant_s_sid: str
         iptrunk_description: str
         iptrunk_type: IptrunkType
-        iptrunk_speed: PhyPortCapacity
+        iptrunk_speed: PhysicalPortCapacity
         iptrunk_minimum_links: int
 
         @validator("tt_number", allow_reuse=True)
@@ -210,7 +210,7 @@ def initialize_subscription(
     geant_s_sid: str,
     iptrunk_type: IptrunkType,
     iptrunk_description: str,
-    iptrunk_speed: PhyPortCapacity,
+    iptrunk_speed: PhysicalPortCapacity,
     iptrunk_minimum_links: int,
     side_a_node_id: str,
     side_a_ae_iface: str,

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -19,7 +19,7 @@ from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlock,
     IptrunkSideBlock,
     IptrunkType,
-    PhyPortCapacity,
+    PhysicalPortCapacity,
 )
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services.lso_client import execute_playbook, lso_interaction
@@ -86,7 +86,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             "Changing the PhyPortCapacity will result in the deletion of all AE members. "
             "You will need to add the new AE members in the next steps."  # type: ignore[assignment]
         )
-        iptrunk_speed: PhyPortCapacity = subscription.iptrunk.iptrunk_speed
+        iptrunk_speed: PhysicalPortCapacity = subscription.iptrunk.iptrunk_speed
         iptrunk_minimum_links: int = subscription.iptrunk.iptrunk_minimum_links
         iptrunk_isis_metric: int = ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric)
         iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv4_network)
@@ -157,7 +157,7 @@ def modify_iptrunk_subscription(
     geant_s_sid: str,
     iptrunk_type: IptrunkType,
     iptrunk_description: str,
-    iptrunk_speed: PhyPortCapacity,
+    iptrunk_speed: PhysicalPortCapacity,
     iptrunk_minimum_links: int,
     side_a_ae_geant_a_sid: str,
     side_a_ae_members: list[dict],

gso/workflows/tasks/import_iptrunk.py

@@ -12,7 +12,7 @@ from orchestrator.workflow import StepList, done, init, step
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 
 from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhysicalPortCapacity
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router
 from gso.services import subscriptions
@@ -42,7 +42,7 @@ def initial_input_form_generator() -> FormGenerator:
         geant_s_sid: str
         iptrunk_description: str
         iptrunk_type: IptrunkType
-        iptrunk_speed: PhyPortCapacity
+        iptrunk_speed: PhysicalPortCapacity
         iptrunk_minimum_links: int
         iptrunk_isis_metric: int
 
@@ -83,7 +83,7 @@ def initialize_subscription(
     geant_s_sid: str,
     iptrunk_type: IptrunkType,
     iptrunk_description: str,
-    iptrunk_speed: PhyPortCapacity,
+    iptrunk_speed: PhysicalPortCapacity,
     iptrunk_minimum_links: int,
     iptrunk_isis_metric: int,
     side_a_node_id: str,

test/api/test_imports.py

@@ -5,7 +5,7 @@ import pytest
 from orchestrator.db import SubscriptionTable
 from orchestrator.services import subscriptions
 
-from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
 from gso.utils.helpers import iso_from_ipv4
@@ -27,7 +27,7 @@ def iptrunk_data(nokia_router_subscription_factory, faker):
         "geant_s_sid": faker.geant_sid(),
         "iptrunk_type": IptrunkType.DARK_FIBER,
         "iptrunk_description": faker.sentence(),
-        "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
+        "iptrunk_speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
         "iptrunk_minimum_links": 5,
         "iptrunk_isis_metric": 500,
         "side_a_node_id": router_side_a,

test/api/test_networks.py

+from orchestrator.types import SubscriptionLifecycle
+
+TOPOLOGY_ENDPOINT = "/api/v1/networks/topology"
+
+
+def test_iptrunk_subscriptions_endpoint_with_valid_api_key(test_client, iptrunk_subscription_factory):
+    iptrunk_subscription_factory()
+    iptrunk_subscription_factory()
+    iptrunk_subscription_factory()
+    iptrunk_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
+    iptrunk_subscription_factory(status=SubscriptionLifecycle.INITIAL)
+
+    response = test_client.get(TOPOLOGY_ENDPOINT)
+
+    assert response.status_code == 200
+    assert len(response.json()["iptrunks"]) == 3

test/fixtures.py

@@ -10,7 +10,7 @@ from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlock,
     IptrunkSideBlock,
     IptrunkType,
-    PhyPortCapacity,
+    PhysicalPortCapacity,
 )
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
@@ -239,7 +239,7 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant
         geant_s_sid=None,
         iptrunk_description=None,
         iptrunk_type=IptrunkType.DARK_FIBER,
-        iptrunk_speed=PhyPortCapacity.ONE_GIGABIT_PER_SECOND,
+        iptrunk_speed=PhysicalPortCapacity.ONE_GIGABIT_PER_SECOND,
         iptrunk_isis_metric=None,
         iptrunk_ipv4_network=None,
         iptrunk_ipv6_network=None,

test/workflows/iptrunk/test_create_iptrunk.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 import pytest
 
 from gso.products import Iptrunk, ProductName
-from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.helpers import LAGMember
 from gso.utils.shared_enums import Vendor
@@ -65,7 +65,7 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
         "geant_s_sid": faker.geant_sid(),
         "iptrunk_type": IptrunkType.DARK_FIBER,
         "iptrunk_description": faker.sentence(),
-        "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
+        "iptrunk_speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
         "iptrunk_minimum_links": 2,
     }
     create_ip_trunk_side_a_router_name = {"side_a_node_id": router_side_a}

test/workflows/iptrunk/test_modify_trunk_interface.py

@@ -3,7 +3,7 @@ from unittest.mock import patch
 import pytest
 
 from gso.products import Iptrunk
-from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
 from gso.utils.shared_enums import Vendor
 from test.conftest import UseJuniperSide
 from test.workflows import (
@@ -55,7 +55,7 @@ def input_form_iptrunk_data(
     new_sid = faker.geant_sid()
     new_description = faker.sentence()
     new_type = IptrunkType.LEASED
-    new_speed = PhyPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND
+    new_speed = PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND
     new_link_count = 2
 
     new_side_a_sid = faker.geant_sid()