Updated test and improved functionality.

gso/cli/netbox.py

 import typer
 from pynetbox import RequestError
 
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 
 app: typer.Typer = typer.Typer()
 
@@ -17,7 +17,7 @@ def netbox_initial_setup() -> None:
     typer.echo("Initial setup of NetBox ...")
     typer.echo("Connecting to NetBox ...")
 
-    nbclient = NetBoxClient()
+    nbclient = NetboxClient()
 
     typer.echo("Creating GÉANT site ...")
     try:

gso/services/netbox_client.py

@@ -45,7 +45,7 @@ class Site(pydantic.BaseModel):
     slug: str
 
 
-class NetBoxClient:
+class NetboxClient:
     """Implement all methods to communicate with the NetBox API."""
 
     def __init__(self) -> None:

gso/workflows/iptrunk/create_iptrunk.py

-from typing import NoReturn
-
 from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Choice, ChoiceList, UniqueConstrainedList
 from orchestrator.targets import Target
@@ -16,13 +14,14 @@ from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router
 from gso.services import infoblox, provisioning_proxy, subscriptions
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
 from gso.workflows.utils import (
     available_interfaces_choices,
     available_lags_choices,
     customer_selector,
     get_router_vendor,
+    validate_router_in_netbox,
 )
 
 
@@ -59,13 +58,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         iptrunk_sideA_node_id: router_enum_a  # type: ignore[valid-type]
 
         @validator("iptrunk_sideA_node_id", allow_reuse=True)
-        def validate_device_exists_in_netbox(cls, iptrunk_sideA_node_id: UUIDstr) -> str | NoReturn:
-            router = Router.from_subscription(iptrunk_sideA_node_id).router
-            if router.router_vendor == RouterVendor.NOKIA:
-                device = NetBoxClient().get_device_by_name(router.router_fqdn)
-                if not device:
-                    raise ValueError("The selected router does not exist in Netbox.")
-            return iptrunk_sideA_node_id
+        def validate_device_exists_in_netbox(cls, iptrunk_sideA_node_id: UUIDstr) -> str | None:
+            return validate_router_in_netbox(iptrunk_sideA_node_id)
 
     user_input_router_side_a = yield SelectRouterSideA
     router_a = user_input_router_side_a.iptrunk_sideA_node_id.name
@@ -106,13 +100,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         iptrunk_sideB_node_id: router_enum_b  # type: ignore[valid-type]
 
         @validator("iptrunk_sideB_node_id", allow_reuse=True)
-        def validate_device_exists_in_netbox(cls, iptrunk_sideB_node_id: UUIDstr) -> str | NoReturn:
-            router = Router.from_subscription(iptrunk_sideB_node_id).router
-            if router.router_vendor == RouterVendor.NOKIA:
-                device = NetBoxClient().get_device_by_name(router.router_fqdn)
-                if not device:
-                    raise ValueError("The selected router does not exist in Netbox.")
-            return iptrunk_sideB_node_id
+        def validate_device_exists_in_netbox(cls, iptrunk_sideB_node_id: UUIDstr) -> str | None:
+            return validate_router_in_netbox(iptrunk_sideB_node_id)
 
     user_input_router_side_b = yield SelectRouterSideB
     router_b = user_input_router_side_b.iptrunk_sideB_node_id.name
@@ -143,10 +132,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     return (
         initial_user_input.dict()
-        | user_input_side_a.dict()
-        | user_input_side_b.dict()
         | user_input_router_side_a.dict()
+        | user_input_side_a.dict()
         | user_input_router_side_b.dict()
+        | user_input_side_b.dict()
     )
 
 
@@ -280,7 +269,7 @@ def check_ip_trunk_isis(subscription: IptrunkProvisioning, process_id: UUIDstr,
 def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
     """Create the LAG interfaces in NetBox and attach the lag interfaces to the physical interfaces."""
 
-    nbclient = NetBoxClient()
+    nbclient = NetboxClient()
     for side in range(0, 2):
         if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
             # Create LAG interfaces
@@ -317,7 +306,7 @@ def allocate_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
     for side in range(0, 2):
         if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
             for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
-                NetBoxClient().allocate_interface(
+                NetboxClient().allocate_interface(
                     device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
                     iface_name=interface,
                 )

gso/workflows/router/create_router.py

@@ -16,7 +16,7 @@ from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site
 from gso.products.shared import PortNumber
 from gso.services import infoblox, provisioning_proxy, subscriptions
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
 from gso.workflows.utils import customer_selector, iso_from_ipv4
 
@@ -158,11 +158,11 @@ def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr,
 @step("Create NetBox Device")
 def create_netbox_device(subscription: RouterProvisioning) -> State:
     if subscription.router.router_vendor == RouterVendor.NOKIA:
-        NetBoxClient().create_device(
+        NetboxClient().create_device(
             subscription.router.router_fqdn, subscription.router.router_site.site_tier  # type: ignore
         )
-
-    return {"subscription": subscription}
+        return {"subscription": subscription, "label_text": "Creating NetBox device"}
+    return {"subscription": subscription, "label_text": "Skipping NetBox device creation for Juniper router."}
 
 
 @step("Verify IPAM resources for loopback interface")

gso/workflows/router/terminate_router.py

@@ -12,7 +12,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.router import Router
 from gso.services import infoblox
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +63,7 @@ def remove_config_from_router() -> None:
 @step("Remove Device from NetBox")
 def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
     if subscription.router.router_vendor == RouterVendor.NOKIA:
-        NetBoxClient().delete_device(subscription.router.router_fqdn)
+        NetboxClient().delete_device(subscription.router.router_fqdn)
     return {"subscription": subscription}
 
 

gso/workflows/utils.py

@@ -3,11 +3,12 @@ from ipaddress import IPv4Address
 from uuid import UUID
 
 from orchestrator.forms.validators import Choice
+from orchestrator.types import UUIDstr
 
 from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.router import Router
 from gso.services.crm import all_customers
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 
 
 def customer_selector() -> Choice:
@@ -28,7 +29,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
         return None
     interfaces = {
         interface["name"]: f"{interface['name']} - {interface['module']['display']} - {interface['description']}"
-        for interface in NetBoxClient().get_available_interfaces(router_id, speed)
+        for interface in NetboxClient().get_available_interfaces(router_id, speed)
     }
     return Choice("ae member", zip(interfaces.keys(), interfaces.items()))  # type: ignore[arg-type]
 
@@ -42,7 +43,7 @@ def available_lags_choices(router_id: UUID) -> Choice | None:
 
     if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
         return None
-    side_a_ae_iface_list = NetBoxClient().get_available_lags(router_id)
+    side_a_ae_iface_list = NetboxClient().get_available_lags(router_id)
     return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list))  # type: ignore[arg-type]
 
 
@@ -70,3 +71,22 @@ def iso_from_ipv4(ipv4_address: IPv4Address) -> str:
     joined_octets = "".join(padded_octets)
     re_split = ".".join(re.findall("....", joined_octets))
     return ".".join(["49.51e5.0001", re_split, "00"])
+
+
+def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr | None:
+    """Verify if a device exists in Netbox.
+
+    Args:
+    ----
+    subscription_id (UUID): The {term}`UUID` of the router subscription.
+
+    Returns:
+    -------
+    UUID: The {term}`UUID` of the router subscription or raises an error.
+    """
+    router = Router.from_subscription(subscription_id).router
+    if router.router_vendor == RouterVendor.NOKIA:
+        device = NetboxClient().get_device_by_name(router.router_fqdn)
+        if not device:
+            raise ValueError("The selected router does not exist in Netbox.")
+    return subscription_id

test/workflows/iptrunks/iptrunks/test_create_iptrunks.py

+from os import PathLike
 from unittest.mock import patch
 
 import pytest
@@ -19,6 +20,65 @@ from test.workflows import (
 )
 
 
+class MockedNetboxClient:
+    class BaseMockObject:
+        def __init__(self, **kwargs):
+            for key, value in kwargs.items():
+                setattr(self, key, value)
+
+    def get_device_by_name(self):
+        return self.BaseMockObject(id=1, name="test")
+
+    def get_available_lags(self) -> list[str]:
+        return [f"LAG{lag}" for lag in range(1, 5)]
+
+    def get_available_interfaces(self):
+        interfaces = []
+        for interface in range(1, 5):
+            interface_data = {
+                "name": f"Interface{interface}",
+                "module": {"display": f"Module{interface}"},
+                "description": f"Description{interface}",
+            }
+            interfaces.append(interface_data)
+        return interfaces
+
+    def create_interface(self):
+        return self.BaseMockObject(id=1, name="test")
+
+    def attach_interface_to_lag(self):
+        return self.BaseMockObject(id=1, name="test")
+
+    def reserve_interface(self):
+        return self.BaseMockObject(id=1, name="test")
+
+    def allocate_interface(self):
+        return {"id": 1, "name": "test"}
+
+
+@pytest.fixture
+def netbox_client_mock():
+    # Mock NetboxClient methods
+    with (
+        patch("gso.services.netbox_client.NetboxClient.get_device_by_name") as mock_get_device_by_name,
+        patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") as mock_get_available_interfaces,
+        patch("gso.services.netbox_client.NetboxClient.get_available_lags") as mock_get_available_lags,
+        patch("gso.services.netbox_client.NetboxClient.create_interface") as mock_create_interface,
+        patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag") as mock_attach_interface_to_lag,
+        patch("gso.services.netbox_client.NetboxClient.reserve_interface") as mock_reserve_interface,
+        patch("gso.services.netbox_client.NetboxClient.allocate_interface") as mock_allocate_interface,
+    ):
+        mock_get_device_by_name.return_value = MockedNetboxClient().get_device_by_name()
+        mock_get_available_interfaces.return_value = MockedNetboxClient().get_available_interfaces()
+        mock_get_available_lags.return_value = MockedNetboxClient().get_available_lags()
+        mock_create_interface.return_value = MockedNetboxClient().create_interface()
+        mock_attach_interface_to_lag.return_value = MockedNetboxClient().attach_interface_to_lag()
+        mock_reserve_interface.return_value = MockedNetboxClient().reserve_interface()
+        mock_allocate_interface.return_value = MockedNetboxClient().allocate_interface()
+
+        yield
+
+
 @pytest.fixture
 def input_form_wizard_data(router_subscription_factory, faker):
     router_side_a = router_subscription_factory()
@@ -31,25 +91,31 @@ def input_form_wizard_data(router_subscription_factory, faker):
         "iptrunk_type": IptrunkType.DARK_FIBER,
         "iptrunk_description": faker.sentence(),
         "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
-        "iptrunk_minimum_links": 5,
+        "iptrunk_minimum_links": 2,
     }
+    create_ip_trunk_side_a_router_name = {"iptrunk_sideA_node_id": router_side_a}
     create_ip_trunk_side_a_step = {
-        "iptrunk_sideA_node_id": router_side_a,
-        "iptrunk_sideA_ae_iface": faker.pystr(),
+        "iptrunk_sideA_ae_iface": "LAG1",
         "iptrunk_sideA_ae_geant_a_sid": faker.pystr(),
-        "iptrunk_sideA_ae_members": [faker.pystr() for _ in range(5)],
-        "iptrunk_sideA_ae_members_descriptions": [faker.sentence() for _ in range(5)],
+        "iptrunk_sideA_ae_members": ["Interface1", "Interface2"],
+        "iptrunk_sideA_ae_members_descriptions": ["Interface1 Description", "Interface2 Description"],
     }
 
+    create_ip_trunk_side_b_router_name = {"iptrunk_sideB_node_id": router_side_b}
     create_ip_trunk_side_b_step = {
-        "iptrunk_sideB_node_id": router_side_b,
-        "iptrunk_sideB_ae_iface": faker.pystr(),
+        "iptrunk_sideB_ae_iface": "LAG1",
         "iptrunk_sideB_ae_geant_a_sid": faker.pystr(),
-        "iptrunk_sideB_ae_members": [faker.pystr() for _ in range(5)],
-        "iptrunk_sideB_ae_members_descriptions": [faker.sentence() for _ in range(5)],
+        "iptrunk_sideB_ae_members": ["Interface1", "Interface2"],
+        "iptrunk_sideB_ae_members_descriptions": ["Interface1 Description", "Interface2 Description"],
     }
 
-    return [create_ip_trunk_step, create_ip_trunk_side_a_step, create_ip_trunk_side_b_step]
+    return [
+        create_ip_trunk_step,
+        create_ip_trunk_side_a_router_name,
+        create_ip_trunk_side_a_step,
+        create_ip_trunk_side_b_router_name,
+        create_ip_trunk_side_b_step,
+    ]
 
 
 def _user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
@@ -73,6 +139,8 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     responses,
     input_form_wizard_data,
     faker,
+    data_config_filename: PathLike,
+    netbox_client_mock,
 ):
     mock_allocate_v4_network.return_value = faker.ipv4_network()
     mock_allocate_v6_network.return_value = faker.ipv6_network()
@@ -122,6 +190,8 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
     responses,
     input_form_wizard_data,
     faker,
+    netbox_client_mock,
+    data_config_filename: PathLike,
 ):
     mock_allocate_v4_network.return_value = faker.ipv4_network()
     mock_allocate_v6_network.return_value = faker.ipv6_network()

utils/netboxcli.py

@@ -4,7 +4,7 @@ from typing import Any, Dict, List
 import click
 import pandas as pd
 
-from gso.services.netbox_client import NetBoxClient
+from gso.services.netbox_client import NetboxClient
 
 
 def convert_to_table(data: List[Dict[str, Any]], fields: List[str]) -> pd.DataFrame:
@@ -33,7 +33,7 @@ def create() -> None:
 @click.option("--model", default="vmx", help="Device model")
 def device(fqdn: str, model: str) -> None:
     click.echo(f"Creating device: fqdn={fqdn}, model={model}")
-    new_device = NetBoxClient().create_device(fqdn, model)
+    new_device = NetboxClient().create_device(fqdn, model)
     click.echo(new_device)
 
 
@@ -44,7 +44,7 @@ def device(fqdn: str, model: str) -> None:
 @click.option("--fqdn", help="Device where to create interface")
 def interface(name: str, type: str, speed: str, fqdn: str) -> None:
     click.echo(f"Creating interface: name={name}, speed={speed}, fqdn={fqdn}")
-    new_interface = NetBoxClient().create_interface(name, type, speed, fqdn)
+    new_interface = NetboxClient().create_interface(name, type, speed, fqdn)
     click.echo(new_interface)
 
 
@@ -53,7 +53,7 @@ def interface(name: str, type: str, speed: str, fqdn: str) -> None:
 @click.option("--slug", help="Short name for manufacturer")
 def manufacturer(name: str, slug: str) -> None:
     click.echo(f"Creating manufacturer: name={name}")
-    manufacturer = NetBoxClient().create_device_manufacturer(name, slug)
+    manufacturer = NetboxClient().create_device_manufacturer(name, slug)
     click.echo(manufacturer)
 
 
@@ -63,7 +63,7 @@ def manufacturer(name: str, slug: str) -> None:
 @click.option("--slug", help="Short name for manufacturer")
 def device_type(manufacturer: str, model: str, slug: str) -> None:
     click.echo(f"Creating device type: manufacturer={manufacturer} model = {model}")
-    device_type = NetBoxClient().create_device_type(manufacturer, model, slug)
+    device_type = NetboxClient().create_device_type(manufacturer, model, slug)
     click.echo(device_type)
 
 
@@ -72,7 +72,7 @@ def device_type(manufacturer: str, model: str, slug: str) -> None:
 @click.option("--slug", help="Short name for device role")
 def device_role(name: str, slug: str) -> None:
     click.echo(f"Creating device role: name={name}")
-    device_role = NetBoxClient().create_device_role(name, slug)
+    device_role = NetboxClient().create_device_role(name, slug)
     click.echo(device_role)
 
 
@@ -81,7 +81,7 @@ def device_role(name: str, slug: str) -> None:
 @click.option("--slug", help="Short name for device site")
 def device_site(name: str, slug: str) -> None:
     click.echo(f"Creating device site: name={name}")
-    device_site = NetBoxClient().create_device_site(name, slug)
+    device_site = NetboxClient().create_device_site(name, slug)
     click.echo(device_site)
 
 
@@ -104,7 +104,7 @@ def list() -> None:
 @click.option("--speed", default="1000", help="Interface speed to list interfaces (default 1000=1G)")
 def interfaces(fqdn: str, speed: str) -> None:
     click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}")
-    interface_list = NetBoxClient().get_interfaces_by_device(fqdn, speed)
+    interface_list = NetboxClient().get_interfaces_by_device(fqdn, speed)
     display_fields = ["name", "enabled", "mark_connected", "custom_fields", "lag", "speed"]
     iface_list = []
     for iface in interface_list:
@@ -117,7 +117,7 @@ def interfaces(fqdn: str, speed: str) -> None:
 @list.command()
 def devices() -> None:
     click.echo("Listing all devices:")
-    device_list = NetBoxClient().get_all_devices()
+    device_list = NetboxClient().get_all_devices()
     display_fields = ["name", "device_type"]
     devices = []
     for device in device_list:
@@ -143,7 +143,7 @@ def attach() -> None:
 @click.option("--lag", help="LAG name to attach interface")
 def interface_to_lag(fqdn: str, iface: str, lag: str) -> None:
     click.echo(f"Attaching interface to lag: device ={fqdn}, interface name={iface} to lag={lag}")
-    new_iface = NetBoxClient().attach_interface_to_lag(fqdn, lag, iface)
+    new_iface = NetboxClient().attach_interface_to_lag(fqdn, lag, iface)
     click.echo(new_iface)
 
 
@@ -161,7 +161,7 @@ def reserve() -> None:
 @click.option("--iface", help="Interface name to reserve")
 def reserve_interface(fqdn: str, iface: str) -> None:
     click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}")
-    reserved_iface = NetBoxClient().reserve_interface(fqdn, iface)
+    reserved_iface = NetboxClient().reserve_interface(fqdn, iface)
     click.echo(reserved_iface)