diff --git a/gso/cli/netbox.py b/gso/cli/netbox.py
index d3016443c3870f01cebf521fad0d6a830326c20e..ddaa94ff7f70ab045f266629bc2e4370061cb3b7 100644
--- a/gso/cli/netbox.py
+++ b/gso/cli/netbox.py
@@ -8,7 +8,7 @@ app: typer.Typer = typer.Typer()
@app.command()
def netbox_initial_setup() -> None:
- """Initial setup of NetBox.
+ """Set up NetBox for the first time.
It includes:
- Creating a default site (GEANT)
diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py
index 67cfe77e767346d1dae215fc28d589960f7578db..4d6dbf03630c316151adfa126732a5ca7f55a711 100644
--- a/gso/services/netbox_client.py
+++ b/gso/services/netbox_client.py
@@ -1,35 +1,27 @@
-"""
-This module contains all methods
-to communicate with the
-NetBox API endpoint: Data Center Infrastructure Main (dcim)
-"""
+"""Contain all methods to communicate with the NetBox API endpoint.Data Center Infrastructure Main (DCIM)."""
from uuid import UUID
import pydantic
import pynetbox
+from pynetbox.models.dcim import Devices, DeviceTypes, Interfaces
from gso.products import Router
-from gso.settings import NetBoxParams, load_oss_params
-from gso.utils.device_info import TierInfo, FEASIBLE_LAG_RANGE
-from gso.utils.exceptions import NotFoundError, WorkflowStateException
+from gso.settings import load_oss_params
+from gso.utils.device_info import FEASIBLE_LAG_RANGE, TierInfo
+from gso.utils.exceptions import NotFoundError, WorkflowStateError
# Define device models
class Manufacturer(pydantic.BaseModel):
- """
- The device manufacturer.
- Is mandatory to create a device in netbox.
- Is needed to define the device type.
- """
+ """Defines the manufacturer of a device."""
name: str
slug: str
class DeviceType(pydantic.BaseModel):
- """
- The device type for a device.
- Is mandatory to create a device.
+ """Defines the device type.
+
The manufacturer should be created first to get the manufacturer id,
which is defined here as int.
"""
@@ -40,38 +32,34 @@ class DeviceType(pydantic.BaseModel):
class DeviceRole(pydantic.BaseModel):
- """
- The role of a device in netbox.
- Is mandatory to create a device.
- """
+ """Defines the role of a device."""
name: str
slug: str
class Site(pydantic.BaseModel):
- """
- The site where to place the device.
- Is mandatory to create a device.
- """
+ """Defines the site of a device."""
name: str
slug: str
class NetBoxClient:
- def __init__(self):
+ """Implement all methods to communicate with the NetBox API."""
+
+ def __init__(self) -> None:
netbox_params = load_oss_params().NETBOX
self.netbox = pynetbox.api(netbox_params.api, netbox_params.token)
- def get_all_devices(self):
+ def get_all_devices(self) -> list[Devices]:
return list(self.netbox.dcim.devices.all())
- def get_device_by_name(self, device_name: str):
- """Returns the device object by name from netbox, or None if not found."""
+ def get_device_by_name(self, device_name: str) -> Devices:
+ """Return the device object by name from netbox, or None if not found."""
return self.netbox.dcim.devices.get(name=device_name)
- def get_interfaces_by_device(self, device_name: str, speed: str):
+ def get_interfaces_by_device(self, device_name: str, speed: str) -> list[Interfaces]:
"""Get all interfaces of a device by name and speed."""
device = self.get_device_by_name(device_name)
@@ -79,56 +67,41 @@ class NetBoxClient:
self.netbox.dcim.interfaces.filter(device_id=device.id, enabled=False, mark_connected=False, speed=speed)
)
- def create_interface(self, iface_name: str, type: str, speed: str, device_name: str) -> dict:
- """
- Creates a new interface on
- a device, where device is defined by name.
+ def create_interface(self, iface_name: str, type: str, speed: str, device_name: str) -> Interfaces:
+ """Create new interface on a device, where device is defined by name.
+
The type parameter can be 1000base-t, 10gbase-t, lag...
- For more details on type definidtion have a look in
- choises.py in the netbox API implementation in module dcim.
+ For more details on type definition have a look in
+ choices.py in the netbox API implementation in module DCIM.
Returns the new interface object as dict.
"""
device = self.get_device_by_name(device_name)
-
- new_iface = self.netbox.dcim.interfaces.create(
+ return self.netbox.dcim.interfaces.create(
name=iface_name, type=type, speed=speed, enabled=False, mark_connected=False, device=device.id
)
- return dict(new_iface)
-
- def create_device_type(self, manufacturer: str, model: str, slug: str) -> dict:
- """Creates a new device type in netbox."""
+ def create_device_type(self, manufacturer: str, model: str, slug: str) -> DeviceTypes:
+ """Create a new device type in netbox."""
# First get manufacturer id
manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id)
- device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug})
-
- # Convert the format from DeviceType to dict
- device_type = dict(device_type)
-
- # Create device type
- device_type = self.netbox.dcim.device_types.create(device_type)
- return dict(device_type)
+ device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug}) # type: ignore
+ return self.netbox.dcim.device_types.create(dict(device_type))
- def create_device_role(self, name: str, slug: str) -> dict:
+ def create_device_role(self, name: str, slug: str) -> DeviceRole:
device_role = DeviceRole(**{"name": name, "slug": slug})
- device_role = dict(device_role)
- device_role = self.netbox.dcim.device_roles.create(device_role)
- return dict(device_role)
+ return self.netbox.dcim.device_roles.create(dict(device_role))
- def create_device_site(self, name, slug) -> dict:
+ def create_device_site(self, name: str, slug: str) -> Site:
device_site = Site(**{"name": name, "slug": slug})
- device_site = self.netbox.dcim.sites.create(dict(device_site))
- return dict(device_site)
+ return self.netbox.dcim.sites.create(dict(device_site))
- def create_device_manufacturer(self, name: str, slug: str) -> dict:
+ def create_device_manufacturer(self, name: str, slug: str) -> Manufacturer:
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
- device_manufacturer = dict(device_manufacturer)
- device_manufacturer = self.netbox.dcim.manufacturers.create(device_manufacturer)
- return dict(device_manufacturer)
+ return self.netbox.dcim.manufacturers.create(dict(device_manufacturer))
- def create_device(self, router_name: str, site_tier: str) -> dict:
- """Creates a new device in netbox."""
+ def create_device(self, router_name: str, site_tier: str) -> Devices:
+ """Create a new device in netbox."""
# Get device type id
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
@@ -155,15 +128,15 @@ class NetBoxClient:
comments="Installed via pynetbox",
)
- return dict(device)
+ return device
- def delete_device(self, router_name: str):
+ def delete_device(self, router_name: str) -> None:
self.netbox.dcim.devices.get(name=router_name).delete()
return
- def attach_interface_to_lag(self, device_name: str, lag_name: str, iface_name: str) -> dict:
- """
- Assign a given interface to a lag.
+ def attach_interface_to_lag(self, device_name: str, lag_name: str, iface_name: str) -> Interfaces:
+ """Assign a given interface to a lag.
+
Returns the lag object with the assignend interfaces
"""
# Get device id
@@ -179,11 +152,9 @@ class NetBoxClient:
iface.lag = lag.id
# Update interface
- updated_iface = self.netbox.dcim.interfaces.update(iface)
-
- return dict(updated_iface)
+ return self.netbox.dcim.interfaces.update(iface)
- def reserve_interface(self, device_name: str, iface_name: str) -> dict:
+ def reserve_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Reserve an interface by enabling it."""
# First get interface from device
@@ -196,15 +167,15 @@ class NetBoxClient:
# Check if interface is reserved
if interface.enabled:
- raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
+ raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is already reserved.")
# Reserve interface by enabling it
interface.enabled = True
interface.save()
- return dict(interface)
+ return interface
- def allocate_interface(self, device_name: str, iface_name: str) -> dict:
+ def allocate_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Allocate an interface by marking it as connected."""
device = self.get_device_by_name(device_name)
@@ -216,16 +187,16 @@ class NetBoxClient:
# Check if interface is reserved
if interface.mark_connected:
- raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
+ raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is already allocated.")
# allocate interface by mark as connected
interface.mark_connected = True
interface.save()
- return dict(interface)
+ return interface
def get_available_lags(self, router_id: UUID) -> list[str]:
- """Returns all available lags not assigned to a device."""
+ """Return all available lags not assigned to a device."""
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
@@ -238,10 +209,5 @@ class NetBoxClient:
# Generate all feasible lags
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
- # Find available lags not assigned to the device
- available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
- return available_lags
-
-
-# if __name__ == "__main__":
-# print(dict(create_device_manufacturer("Juniper", "juniper")))
+ # Return available lags not assigned to the device
+ return [lag for lag in all_feasible_lags if lag not in lag_interface_names]
diff --git a/gso/utils/device_info.py b/gso/utils/device_info.py
index 47b2a8bed2d08e93debff83e55c28f6399dc2e5f..c4d09ccb823a62de127093d1856f45ecb544594c 100644
--- a/gso/utils/device_info.py
+++ b/gso/utils/device_info.py
@@ -1,5 +1,3 @@
-from typing import Optional
-
from pydantic import BaseModel
@@ -12,7 +10,7 @@ class ModuleInfo(BaseModel):
class TierInfo:
- def __init__(self):
+ def __init__(self) -> None:
self.Tier1 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
@@ -28,13 +26,9 @@ class TierInfo:
total_10g_interfaces=60,
)
- def get_module_by_name(self, name: str) -> Optional[ModuleInfo]:
- if name == "Tier1":
- return self.Tier1
- elif name == "Tier2":
- return self.Tier2
- else:
- return None
+ def get_module_by_name(self, name: str) -> ModuleInfo:
+ return getattr(self, name)
+# Ranges of LAGs that are feasible for a given device type.
FEASIBLE_LAG_RANGE = range(1, 11)
diff --git a/gso/utils/exceptions.py b/gso/utils/exceptions.py
index a11a741fbd09c272a606d7927987ccf5baf7f058..21c127e88b3144d43eb6474e8a890881f36135b0 100644
--- a/gso/utils/exceptions.py
+++ b/gso/utils/exceptions.py
@@ -4,7 +4,7 @@ class NotFoundError(Exception):
pass
-class WorkflowStateException(Exception):
+class WorkflowStateError(Exception):
"""Exception raised on problems during workflow."""
pass
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 403148d26995c813b2fa35790ea7d374219ffb9f..2d28c06e64879a28f3a2d031a42e7a3771705675 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -146,22 +146,22 @@ def provision_router_dry(subscription: RouterProvisioning, process_id: UUIDstr,
def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr, tt_number: str) -> State:
provisioning_proxy.provision_router(subscription, process_id, tt_number, False)
- return ({
+ return {
"subscription": subscription,
"label_text": (
"Deployment of base config for a new router. Deployment is being taken care of by the"
" provisioning proxy, please wait for the results to come back before continuing."
),
- })
+ }
@step("Create NetBox Device")
def create_netbox_device(subscription: RouterProvisioning) -> State:
- NetBoxClient().create_device(subscription.router.router_fqdn, subscription.router.router_site.site_tier)
+ NetBoxClient().create_device(
+ subscription.router.router_fqdn, subscription.router.router_site.site_tier # type: ignore
+ )
- return {
- "subscription": subscription
- }
+ return {"subscription": subscription}
@step("Verify IPAM resources for loopback interface")
diff --git a/requirements.txt b/requirements.txt
index 0bc2c774ad1855686c36de8de63ff16bb55e6547..b103da38a717e01e8c339a38c8e5534d778183e6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,7 +2,6 @@ orchestrator-core==1.3.0
requests==2.31.0
infoblox-client~=0.6.0
pycountry==22.3.5
-
pytest==7.4.2
faker==19.6.2
responses==0.23.3
@@ -13,4 +12,5 @@ mypy==1.5.1
ruff==0.0.290
sphinx==7.2.6
sphinx-rtd-theme==1.3.0
-urllib3_mock==0.3.3
\ No newline at end of file
+urllib3_mock==0.3.3
+pynetbox==7.2.0
diff --git a/test/conftest.py b/test/conftest.py
index 39dff29c18ca71c53b50d29f8dee86624bace6b9..2bb17b7a4baf57e34ca19bfadcb70f0790cc5786 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -62,7 +62,7 @@ def configuration_data() -> dict:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
yield {
"GENERAL": {"public_hostname": "https://gap.geant.org"},
- "RESOURCE_MANAGEMENT": {"todo": "todo"},
+ "NETBOX": {"api": "https://127.0.0.1:8000", "token": "TOKEN"},
"IPAM": {
"INFOBLOX": {
"scheme": "https",
diff --git a/utils/netboxcli.py b/utils/netboxcli.py
index 2e5b680bd1b44fdf4ff8ec7b4a416aa8e0f40381..b934f5e7c94b1f43d3692865041bdd02d2af95bf 100644
--- a/utils/netboxcli.py
+++ b/utils/netboxcli.py
@@ -1,91 +1,88 @@
-"""
-Command line tool to communicate with
-the NetBox API
-"""
+"""Command line tool to communicate withthe NetBox API."""
+from typing import Any, Dict, List
+
import click
import pandas as pd
-from gso.services import netbox_client as nc
from gso.services.netbox_client import NetBoxClient
-def convert_to_table(data, fields):
- if len(data) == 0:
- print("No data is available for your request")
- exit()
+def convert_to_table(data: List[Dict[str, Any]], fields: List[str]) -> pd.DataFrame:
+ if not data:
+ raise ValueError("No data is available for your request")
df = pd.DataFrame(data)
- df = df[fields] # Selecting specific fields to display
+ if fields:
+ df = df[fields]
+
return df
@click.group()
-def cli():
+def cli() -> None:
pass
@cli.group()
-def create():
+def create() -> None:
pass
@create.command()
-@click.option('--fqdn', prompt='Enter device name', help='Device name')
-@click.option('--model', default="vmx", help='Device model')
-@click.option('--role', default="router", help='Device role')
-@click.option('--site', default="Amsterdam", help='Device role')
-def device(fqdn: str, model: str, role: str, site: str):
- click.echo(f"Creating device: fqdn={fqdn}, model={model}, role={role}")
+@click.option("--fqdn", prompt="Enter device name", help="Device name")
+@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)
- print(new_device)
+ click.echo(new_device)
@create.command()
-@click.option('--name', help='Interfacename')
-@click.option('--type', default='10gbase-t', help='Interface type, default is 10GBASE-T')
-@click.option('--speed', default='1000', help='Interface speed , default is 1000')
-@click.option('--fqdn', help='Device where to create interface')
-def interface(name: str, type: str, speed: str, fqdn: str):
+@click.option("--name", help="Interfacename")
+@click.option("--type", default="10gbase-t", help="Interface type, default is 10GBASE-T")
+@click.option("--speed", default="1000", help="Interface speed , default is 1000")
+@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)
- print(new_interface)
+ click.echo(new_interface)
@create.command()
-@click.option('--name', help='Manufacturer name')
-@click.option('--slug', help='Short name for manufacturer')
-def manufacturer(name: str, slug: str):
+@click.option("--name", help="Manufacturer name")
+@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)
- print(manufacturer)
+ click.echo(manufacturer)
@create.command()
-@click.option('--manufacturer', help='Manufacturer for device')
-@click.option('--model', help='Model for device')
-@click.option('--slug', help='Short name for manufacturer')
-def device_type(manufacturer: str, model: str, slug: str):
+@click.option("--manufacturer", help="Manufacturer for device")
+@click.option("--model", help="Model for device")
+@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)
- print(device_type)
+ click.echo(device_type)
@create.command()
-@click.option('--name', help='Name for device role')
-@click.option('--slug', help='Short name for device role')
-def device_role(name: str, slug: str):
+@click.option("--name", help="Name for device role")
+@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)
- print(device_role)
+ click.echo(device_role)
@create.command()
-@click.option('--name', help='Name for device site')
-@click.option('--slug', help='Short name for device site')
-def device_site(name: str, slug: str):
+@click.option("--name", help="Name for device site")
+@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)
- print(device_site)
+ click.echo(device_site)
create.add_command(device)
@@ -98,14 +95,14 @@ create.add_command(device_site)
# Define list commands here
@cli.group()
-def list():
+def list() -> None:
pass
@list.command()
-@click.option('--fqdn', help='Device name to list interfaces')
-@click.option('--speed', default="1000", help='Interface speed to list interfaces (default 1000=1G)')
-def interfaces(fqdn: str, speed: str):
+@click.option("--fqdn", help="Device name to list interfaces")
+@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)
display_fields = ["name", "enabled", "mark_connected", "custom_fields", "lag", "speed"]
@@ -114,11 +111,11 @@ def interfaces(fqdn: str, speed: str):
iface_list.append(dict(iface))
table = convert_to_table(iface_list, display_fields)
- print(table)
+ click.echo(table)
@list.command()
-def devices():
+def devices() -> None:
click.echo("Listing all devices:")
device_list = NetBoxClient().get_all_devices()
display_fields = ["name", "device_type"]
@@ -127,7 +124,7 @@ def devices():
devices.append(dict(device))
table = convert_to_table(devices, display_fields)
- print(table)
+ click.echo(table)
list.add_command(interfaces)
@@ -136,40 +133,40 @@ list.add_command(devices)
# Define here attach command
@cli.group()
-def attach():
+def attach() -> None:
pass
@attach.command()
-@click.option('--fqdn', help='Device name where to attach interface to lag')
-@click.option('--iface', help='Interface name to attach to lag')
-@click.option('--lag', help='LAG name to attach interface')
-def interface(fqdn: str, iface: str, lag: str):
+@click.option("--fqdn", help="Device name where to attach interface to lag")
+@click.option("--iface", help="Interface name to attach to lag")
+@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)
- print(new_iface)
+ click.echo(new_iface)
-attach.add_command(interface)
+attach.add_command(interface_to_lag)
# The reserve command
@cli.group()
-def reserve():
+def reserve() -> None:
pass
@reserve.command()
-@click.option('--fqdn', help='Device name where to get interface to reserve')
-@click.option('--iface', help='Interface name to reserve')
-def interface(fqdn: str, iface: str):
+@click.option("--fqdn", help="Device name where to get interface to reserve")
+@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)
- print(reserved_iface )
+ click.echo(reserved_iface)
-reserve.add_command(interface)
+reserve.add_command(reserve_interface)
-if __name__ == '__main__':
+if __name__ == "__main__":
cli()