Made pipeline happy.

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)

gso/services/netbox_client.py

-"""
-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]

gso/utils/device_info.py

-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)

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

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")

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
+urllib3_mock==0.3.3
\ No newline at end of file
+pynetbox==7.2.0

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",

utils/netboxcli.py

-"""
-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()