Feature/nat 314 integrate iptrunk modification with netbox

gso/services/netbox_client.py

@@ -53,6 +53,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 +92,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 +239,43 @@ 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 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."""