From c86f3d4b44ab0bc754f83b927e6af9f85ae817a6 Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@ga0479-nmoeini.home>
Date: Tue, 10 Oct 2023 18:31:43 +0200
Subject: [PATCH] Updated test and improved functionality.
---
gso/cli/netbox.py | 4 +-
gso/services/netbox_client.py | 2 +-
gso/workflows/iptrunk/create_iptrunk.py | 31 +++----
gso/workflows/router/create_router.py | 8 +-
gso/workflows/router/terminate_router.py | 4 +-
gso/workflows/utils.py | 26 +++++-
.../iptrunks/iptrunks/test_create_iptrunks.py | 90 ++++++++++++++++---
utils/netboxcli.py | 22 ++---
8 files changed, 133 insertions(+), 54 deletions(-)
diff --git a/gso/cli/netbox.py b/gso/cli/netbox.py
index 08694be5..bfd2318c 100644
--- a/gso/cli/netbox.py
+++ b/gso/cli/netbox.py
@@ -1,7 +1,7 @@
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:
diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py
index 95f1bcfc..a94da723 100644
--- a/gso/services/netbox_client.py
+++ b/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:
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index b96b9d88..b08da129 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -1,5 +1,3 @@
-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,
)
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 74d59ad3..91e7e82a 100644
--- a/gso/workflows/router/create_router.py
+++ b/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")
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 6c036392..4989823f 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/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}
diff --git a/gso/workflows/utils.py b/gso/workflows/utils.py
index 3bfa5163..cd461b81 100644
--- a/gso/workflows/utils.py
+++ b/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
diff --git a/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py b/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py
index 3f59831e..cf50c2cc 100644
--- a/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py
+++ b/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py
@@ -1,3 +1,4 @@
+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()
diff --git a/utils/netboxcli.py b/utils/netboxcli.py
index b934f5e7..1d46bbb3 100644
--- a/utils/netboxcli.py
+++ b/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)
--
GitLab