diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py index 562c5d26a9a3048af4e165fc952047bc35cb565f..1ed7f82f1ac813e6946851077e93adb9dfceab14 100644 --- a/gso/services/netbox_client.py +++ b/gso/services/netbox_client.py @@ -206,7 +206,7 @@ class NetboxClient: device = self.get_device_by_name(device_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: raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.") @@ -220,13 +220,34 @@ class NetboxClient: 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: """Allocate an interface by marking it as connected.""" device = self.get_device_by_name(device_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: raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.") @@ -234,7 +255,7 @@ class NetboxClient: if interface.mark_connected: 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.save() @@ -301,14 +322,4 @@ class NetboxClient: speed_bps = self.calculate_speed_bits_per_sec(speed) return self.netbox.dcim.interfaces.filter( device=device.name, enabled=False, mark_connected=False, speed=speed_bps - ) - - 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.") + ) \ No newline at end of file diff --git a/utils/netboxcli.py b/utils/netboxcli.py index 1d46bbb3db64520017c1bcedd07545b48d099166..6b2a1c9500d52de970c3e338c016280a5965d201 100644 --- a/utils/netboxcli.py +++ b/utils/netboxcli.py @@ -1,4 +1,4 @@ -"""Command line tool to communicate withthe NetBox API.""" +"""Command line tool to communicate with the NetBox API.""" from typing import Any, Dict, List import click @@ -38,7 +38,7 @@ def device(fqdn: str, model: str) -> None: @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("--speed", default="1000", help="Interface speed , default is 1000") @click.option("--fqdn", help="Device where to create interface") @@ -150,22 +150,52 @@ def interface_to_lag(fqdn: str, iface: str, lag: str) -> None: attach.add_command(interface_to_lag) -# The reserve command +# The action command @cli.group() -def reserve() -> None: +def action() -> None: pass -@reserve.command() -@click.option("--fqdn", help="Device name where to get interface to reserve") -@click.option("--iface", help="Interface name to reserve") +@action.command() +@click.option("--fqdn", help="Device name where to get interface to edit") +@click.option("--iface", help="Interface name to edit") 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) 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__":