Skip to content
Snippets Groups Projects
Commit 5ec342b3 authored by JORGE SASIAIN's avatar JORGE SASIAIN Committed by Neda Moeini
Browse files

NAT-244: create methods to free and deallocate an interface in Netbox

parent 58c576ce
No related branches found
No related tags found
1 merge request!94Feature/Netbox integration terminate ip trunk
...@@ -206,7 +206,7 @@ class NetboxClient: ...@@ -206,7 +206,7 @@ class NetboxClient:
device = self.get_device_by_name(device_name) device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name) interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Reserve interface by enabling it # Check if interface exists
if interface is None: if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.") raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
...@@ -220,13 +220,34 @@ class NetboxClient: ...@@ -220,13 +220,34 @@ class NetboxClient:
return interface return interface
def free_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Reserve an interface by enabling it."""
# First get interface from device
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface exists
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Ensure that interface is reserved
if not interface.enabled:
raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is not reserved.")
# Free interface by disabling it
interface.enabled = False
interface.save()
return interface
def allocate_interface(self, device_name: str, iface_name: str) -> Interfaces: def allocate_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Allocate an interface by marking it as connected.""" """Allocate an interface by marking it as connected."""
device = self.get_device_by_name(device_name) device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name) interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface is available # Check if interface exists
if interface is None: if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.") raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
...@@ -234,7 +255,7 @@ class NetboxClient: ...@@ -234,7 +255,7 @@ class NetboxClient:
if interface.mark_connected: if interface.mark_connected:
raise WorkflowStateError(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 # Allocate interface by marking it as connected
interface.mark_connected = True interface.mark_connected = True
interface.save() interface.save()
...@@ -301,14 +322,4 @@ class NetboxClient: ...@@ -301,14 +322,4 @@ class NetboxClient:
speed_bps = self.calculate_speed_bits_per_sec(speed) speed_bps = self.calculate_speed_bits_per_sec(speed)
return self.netbox.dcim.interfaces.filter( return self.netbox.dcim.interfaces.filter(
device=device.name, enabled=False, mark_connected=False, speed=speed_bps device=device.name, enabled=False, mark_connected=False, speed=speed_bps
) )
\ No newline at end of file
def get_interface_by_name_and_device(self, router_id: UUID, interface_name: str) -> Interface:
"""Return the interface object by name and device from netbox, or ``None`` if not found."""
router = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router)
try:
return self.netbox.dcim.interfaces.get(device=device.name, name=interface_name)
except pynetbox.RequestError:
raise NotFoundError(f"Interface: {interface_name} on device: {device.name} not found.")
"""Command line tool to communicate withthe NetBox API.""" """Command line tool to communicate with the NetBox API."""
from typing import Any, Dict, List from typing import Any, Dict, List
import click import click
...@@ -38,7 +38,7 @@ def device(fqdn: str, model: str) -> None: ...@@ -38,7 +38,7 @@ def device(fqdn: str, model: str) -> None:
@create.command() @create.command()
@click.option("--name", help="Interfacename") @click.option("--name", help="Interface name")
@click.option("--type", default="10gbase-t", help="Interface type, default is 10GBASE-T") @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("--speed", default="1000", help="Interface speed , default is 1000")
@click.option("--fqdn", help="Device where to create interface") @click.option("--fqdn", help="Device where to create interface")
...@@ -150,22 +150,52 @@ def interface_to_lag(fqdn: str, iface: str, lag: str) -> None: ...@@ -150,22 +150,52 @@ def interface_to_lag(fqdn: str, iface: str, lag: str) -> None:
attach.add_command(interface_to_lag) attach.add_command(interface_to_lag)
# The reserve command # The action command
@cli.group() @cli.group()
def reserve() -> None: def action() -> None:
pass pass
@reserve.command() @action.command()
@click.option("--fqdn", help="Device name where to get interface to reserve") @click.option("--fqdn", help="Device name where to get interface to edit")
@click.option("--iface", help="Interface name to reserve") @click.option("--iface", help="Interface name to edit")
def reserve_interface(fqdn: str, iface: str) -> None: def reserve_interface(fqdn: str, iface: str) -> None:
click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}") 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) click.echo(reserved_iface)
reserve.add_command(reserve_interface) @action.command()
@click.option("--fqdn", help="Device name where to get interface to edit")
@click.option("--iface", help="Interface name to edit")
def free_interface(fqdn: str, iface: str) -> None:
click.echo(f"Freeing interface: device={fqdn}, interface name={iface}")
freed_iface = NetboxClient().free_interface(fqdn, iface)
click.echo(freed_iface)
@action.command()
@click.option("--fqdn", help="Device name where to get interface to edit")
@click.option("--iface", help="Interface name to edit")
def allocate_interface(fqdn: str, iface: str) -> None:
click.echo(f"Allocating interface: device={fqdn}, interface name={iface}")
allocated_iface = NetboxClient().allocate_interface(fqdn, iface)
click.echo(allocated_iface)
@action.command()
@click.option("--fqdn", help="Device name where to get interface to edit")
@click.option("--iface", help="Interface name to edit")
def deallocate_interface(fqdn: str, iface: str) -> None:
click.echo(f"Deallocating interface: device={fqdn}, interface name={iface}")
deallocated_iface = NetboxClient().deallocate_interface(fqdn, iface)
click.echo(deallocated_iface)
action.add_command(reserve_interface)
action.add_command(free_interface)
action.add_command(allocate_interface)
action.add_command(deallocate_interface)
if __name__ == "__main__": if __name__ == "__main__":
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment