Feature/trunk migration ipam

Implement the IPAM step in the IP trunk migration workflow.

@neda.moeini if you could also have a look for the sake of code quality, that would be much appreciated (unfortunately I can only assign a single reviewer)

gso/services/infoblox.py

@@ -12,6 +12,7 @@ from infoblox_client.exceptions import (
 from gso.settings import IPAMParams, load_oss_params
 
 logger = getLogger(__name__)
+NULL_MAC = "00:00:00:00:00:00"
 
 
 class AllocationError(Exception):
@@ -40,11 +41,7 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
 
 
 def _allocate_network(
-    conn: connector.Connector,
-    dns_view: str,
-    netmask: int,
-    containers: list[str],
-    comment: str | None = "",
+    conn: connector.Connector, dns_view: str, netmask: int, containers: list[str], comment: str | None = ""
 ) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
     """Allocate a new network in Infoblox.
 
@@ -160,10 +157,7 @@ def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) ->
 
 
 def allocate_host(
-    hostname: str,
-    service_type: str,
-    cname_aliases: list[str],
-    comment: str,
+    hostname: str, service_type: str, cname_aliases: list[str], comment: str
 ) -> tuple[ipaddress.IPv4Address, ipaddress.IPv6Address]:
     """Allocate a new host record in Infoblox.
 
@@ -194,7 +188,7 @@ def allocate_host(
     created_v6 = None
     for ipv6_range in allocation_networks_v6:
         v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv6_range))
-        ipv6_object = objects.IP.create(ip=v6_alloc, mac="00:00:00:00:00:00", configure_for_dhcp=False)
+        ipv6_object = objects.IP.create(ip=v6_alloc, mac=NULL_MAC, configure_for_dhcp=False)
         try:
             new_host = objects.HostRecord.create(
                 conn,
@@ -216,7 +210,7 @@ def allocate_host(
     created_v4 = None
     for ipv4_range in allocation_networks_v4:
         v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv4_range))
-        ipv4_object = objects.IP.create(ip=v4_alloc, mac="00:00:00:00:00:00", configure_for_dhcp=False)
+        ipv4_object = objects.IP.create(ip=v4_alloc, mac=NULL_MAC, configure_for_dhcp=False)
         new_host = objects.HostRecord.search(conn, name=hostname)
         new_host.ipv4addrs = [ipv4_object]
         try:
@@ -234,9 +228,39 @@ def allocate_host(
     return created_v4, created_v6
 
 
-def find_host_by_ip(
-    ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address,
-) -> objects.HostRecord | None:
+def create_host_by_ip(
+    hostname: str,
+    ipv4_address: ipaddress.IPv4Address,
+    ipv6_address: ipaddress.IPv6Address,
+    service_type: str,
+    comment: str,
+) -> None:
+    """Create a new host record with a given IPv4 and IPv6 address.
+
+    :param str hostname: The :term:`FQDN` of the new host.
+    :param IPv4Address ipv4_address: The IPv4 address of the new host.
+    :param IPv6Address ipv6_address: The IPv6 address of the new host.
+    :param str service_type: The relevant service type, used to deduce the correct ``dns_view`` in Infoblox.
+    :param str comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in
+                        :term:`GSO`.
+    """
+    if not hostname_available(hostname):
+        msg = f"Cannot allocate new host, FQDN {hostname} already taken."
+        raise AllocationError(msg)
+
+    conn, oss = _setup_connection()
+    ipv6_object = objects.IP.create(ip=ipv6_address, mac=NULL_MAC, configure_for_dhcp=False)
+    ipv4_object = objects.IP.create(ip=ipv4_address, mac=NULL_MAC, configure_for_dhcp=False)
+    dns_view = getattr(oss, service_type).dns_view
+
+    # This needs to be done in two steps, otherwise only one of the IP addresses is stored.
+    objects.HostRecord.create(conn, ip=ipv6_object, name=hostname, comment=comment, dns_view=dns_view)
+    new_host = find_host_by_fqdn(hostname)
+    new_host.ipv4addrs = [ipv4_object]
+    new_host.update()
+
+
+def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
     """Find a host record in Infoblox by its associated IP address.
 
     :param ip_addr: The IP address of a host that is searched for.
@@ -249,14 +273,14 @@ def find_host_by_ip(
             ipv4addr=ip_addr,
             return_fields=["ipv4addrs", "name", "view", "aliases", "comment"],
         )
-    return objects.HostRecord.search(
+    return objects.HostRecordV6.search(
         conn,
         ipv6addr=ip_addr,
         return_fields=["ipv6addrs", "name", "view", "aliases", "comment"],
     )
 
 
-def find_host_by_fqdn(fqdn: str) -> objects.HostRecord | None:
+def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
     """Find a host record by its associated :term:`FQDN`.
 
     :param fqdn: The :term:`FQDN` of a host that is searched for.
@@ -270,6 +294,18 @@ def find_host_by_fqdn(fqdn: str) -> objects.HostRecord | None:
     )
 
 
+def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
+    """Find a host record by its associated :term:`FQDN`.
+
+    This specific method will return the IPv6 variant of a record, if it exists.
+    :param str fqdn: The :term:`FQDN` of a host that is searched for.
+    """
+    conn, _ = _setup_connection()
+    return objects.HostRecordV6.search(
+        conn, name=fqdn, return_fields=["ipv6addrs", "name", "view", "aliases", "comment"]
+    )
+
+
 def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
     """Delete a host from Infoblox.