Feature/nat 314 integrate iptrunk modification with netbox

gso/services/netbox_client.py

@@ -3,6 +3,7 @@ from uuid import UUID
 
 import pydantic
 import pynetbox
+from infoblox_client.objects import Interface
 from pynetbox.models.dcim import Devices, DeviceTypes, Interfaces
 
 from gso.products.product_types.router import Router
@@ -53,6 +54,14 @@ class NetboxClient:
     def get_all_devices(self) -> list[Devices]:
         return list(self.netbox.dcim.devices.all())
 
+    def get_allocated_interfaces_by_gso_subscription(self, device_name: str, subscription_id: UUID) -> list[Interfaces]:
+        """Return all allocated interfaces of a device by name."""
+
+        device = self.get_device_by_name(device_name)
+        return self.netbox.dcim.interfaces.filter(
+            device_id=device.id, enabled=True, mark_connected=True, description=subscription_id
+        )
+
     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)
@@ -84,6 +93,13 @@ class NetboxClient:
             description=description,
         )
 
+    def delete_interface(self, device_name: str, iface_name: str) -> None:
+        """Delete an interface from a device by name."""
+
+        device = self.get_device_by_name(device_name)
+        interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
+        return interface.delete()
+
     def create_device_type(self, manufacturer: str, model: str, slug: str) -> DeviceTypes:
         """Create a new device type in Netbox."""
 
@@ -224,6 +240,54 @@ class NetboxClient:
 
         return interface
 
+    def free_interface(self, device_name: str, iface_name: str) -> Interfaces:
+        """Free interface by marking disconnect and disable it."""
+
+        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
+        if interface is None:
+            raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
+
+        interface.mark_connected = False
+        interface.enabled = False
+        interface.description = ""
+        interface.save()
+
+        return interface
+
+    def detach_interfaces_from_lag(self, device_name: str, lag_name: str) -> None:
+        """Detach all interfaces from a LAG."""
+        device = self.get_device_by_name(device_name)
+        lag = self.netbox.dcim.interfaces.get(device_id=device.id, name=lag_name)
+        for interface in self.netbox.dcim.interfaces.filter(
+            device_id=device.id, lag_id=lag.id, enabled=False, mark_connected=False
+        ):
+            interface.lag = None
+            interface.save()
+        return
+
+    def deallocate_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
+        if interface is None:
+            raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
+
+        # Check if interface is reserved
+        if interface.mark_connected:
+            raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is already allocated.")
+
+        # allocate interface by mark as connected
+        interface.mark_connected = False
+        interface.save()
+
+        return interface
+
     def get_available_lags(self, router_id: UUID) -> list[str]:
         """Return all available :term:`LAG`s not assigned to a device."""
 
@@ -258,3 +322,13 @@ class NetboxClient:
         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.")