Search for all IPs in Infoblox in allocated networks when creating a trunk

gso/services/infoblox.py

@@ -268,11 +268,11 @@ def create_host_by_ip(
     new_host.update()
 
 
-def find_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> objects.HostRecord | None:
+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.
-    :type ip_addr: IPv4AddressType | ipaddress.IPv6Address
+    :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address
     """
     conn, _ = _setup_connection()
     if ip_addr.version == 4:  # noqa: PLR2004, the 4 in IPv4 is well-known and not a "magic value."

gso/workflows/iptrunk/create_iptrunk.py

 """A creation workflow that deploys a new IP trunk service."""
 
 import json
+from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network
 from typing import Annotated
 from uuid import uuid4
 
@@ -214,22 +215,48 @@ def create_subscription(product: UUIDstr, partner: str) -> State:
 @step("Get information from IPAM")
 def get_info_from_ipam(subscription: IptrunkInactive) -> State:
     """Allocate IP resources in :term:`IPAM`."""
-    subscription.iptrunk.iptrunk_ipv4_network = infoblox.allocate_v4_network(
+    new_ipv4_network = infoblox.allocate_v4_network(
         "TRUNK",
         subscription.iptrunk.iptrunk_description,
     )
-    subscription.iptrunk.iptrunk_ipv6_network = infoblox.allocate_v6_network(
+    new_ipv6_network = infoblox.allocate_v6_network(
         "TRUNK",
         subscription.iptrunk.iptrunk_description,
     )
+    subscription.iptrunk.iptrunk_ipv4_network = new_ipv4_network
+    subscription.iptrunk.iptrunk_ipv6_network = new_ipv6_network
 
-    return {"subscription": subscription}
+    return {
+        "subscription": subscription,
+        "new_ipv4_network": str(new_ipv4_network),
+        "new_ipv6_network": str(new_ipv6_network),
+    }
+
+
+@step("Check for existing DNS records in the assigned IPv4 network")
+def dig_all_hosts_v4(new_ipv4_network: str) -> None:
+    """Check if any hosts have already been assigned inside the IPv4 network in Netbox."""
+    registered_hosts = [host for host in IPv4Network(new_ipv4_network) if infoblox.find_host_by_ip(IPv4Address(host))]
+
+    if registered_hosts:
+        msg = "One or more hosts in the assigned IPv4 network are already registered, please investigate."
+        raise ProcessFailureError(msg, details=registered_hosts)
+
+
+@step("Check for existing DNS records in the assigned IPv6 network")
+def dig_all_hosts_v6(new_ipv6_network: str) -> None:
+    """Check if any hosts have already been assigned inside the IPv6 network in Netbox."""
+    registered_hosts = [host for host in IPv6Network(new_ipv6_network) if infoblox.find_host_by_ip(IPv6Address(host))]
+
+    if registered_hosts:
+        msg = "One or more hosts in the assigned IPv6 network are already registered, please investigate."
+        raise ProcessFailureError(msg, details=registered_hosts)
 
 
 @step("Ping all hosts in the assigned IPv4 network")
-def ping_all_hosts_v4(subscription: IptrunkInactive) -> None:
+def ping_all_hosts_v4(new_ipv4_network: str) -> None:
     """Ping all hosts in the IPv4 network to verify they're not in use."""
-    unavailable_hosts = [host for host in subscription.iptrunk.iptrunk_ipv4_network if ping(host, timeout=1)]
+    unavailable_hosts = [host for host in IPv4Network(new_ipv4_network) if ping(str(host), timeout=1)]
 
     if unavailable_hosts:
         msg = "One or more hosts in the assigned IPv4 network are responding to ping, please investigate."
@@ -237,14 +264,16 @@ def ping_all_hosts_v4(subscription: IptrunkInactive) -> None:
 
 
 @step("Ping all hosts in the assigned IPv6 network")
-def ping_all_hosts_v6(subscription: IptrunkInactive) -> None:
+def ping_all_hosts_v6(new_ipv6_network: str) -> State:
     """Ping all hosts in the IPv6 network to verify they're not in use."""
-    unavailable_hosts = [host for host in subscription.iptrunk.iptrunk_ipv6_network if ping(host, timeout=1)]
+    unavailable_hosts = [host for host in IPv6Network(new_ipv6_network) if ping(str(host), timeout=1)]
 
     if unavailable_hosts:
         msg = "One or more hosts in the assigned IPv6 network are responding to ping, please investigate."
         raise ProcessFailureError(msg, details=unavailable_hosts)
 
+    return {"__remove_keys": ["new_ipv4_network", "new_ipv6_network"]}
+
 
 @step("Initialize subscription")
 def initialize_subscription(
@@ -552,7 +581,15 @@ def create_iptrunk() -> StepList:
     side_b_is_nokia = conditional(lambda state: get_router_vendor(state["side_b_node_id"]) == Vendor.NOKIA)
 
     assign_ip_networks = step_group(
-        name="Assign IP networks", steps=begin >> get_info_from_ipam >> ping_all_hosts_v4 >> ping_all_hosts_v6
+        name="Assign IP networks",
+        steps=(
+            begin
+            >> get_info_from_ipam
+            >> dig_all_hosts_v4
+            >> dig_all_hosts_v6
+            >> ping_all_hosts_v4
+            >> ping_all_hosts_v6
+        ),
     )
 
     return (

test/workflows/iptrunk/test_create_iptrunk.py

@@ -2,6 +2,7 @@ from os import PathLike
 from unittest.mock import patch
 
 import pytest
+from infoblox_client.objects import HostRecord
 
 from gso.products import Iptrunk, ProductName
 from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
@@ -11,6 +12,7 @@ from test import USER_CONFIRM_EMPTY_FORM
 from test.services.conftest import MockedNetboxClient, MockedSharePointClient
 from test.workflows import (
     assert_complete,
+    assert_failed,
     assert_lso_interaction_failure,
     assert_lso_interaction_success,
     assert_suspended,
@@ -102,9 +104,13 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.create_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.find_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.ping")
 @patch("gso.workflows.iptrunk.create_iptrunk.SharePointClient")
 def test_successful_iptrunk_creation_with_standard_lso_result(
     mock_sharepoint_client,
+    mock_ping,
+    mock_find_host_by_ip,
     mock_create_host,
     mock_allocate_v4_network,
     mock_allocate_v6_network,
@@ -117,8 +123,10 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     test_client,
 ):
     mock_create_host.return_value = None
-    mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
-    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
+    mock_allocate_v4_network.return_value = faker.ipv4_network(min_subnet=31, max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(min_subnet=126, max_subnet=126)
+    mock_find_host_by_ip.return_value = None
+    mock_ping.return_value = False
     mock_sharepoint_client.return_value = MockedSharePointClient
 
     product_id = get_product_id_by_name(ProductName.IP_TRUNK)
@@ -147,13 +155,20 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     )
 
     assert mock_execute_playbook.call_count == 6
+    #  We search for 6 hosts in total, 2 in a /31 and 4 in a /126
+    assert mock_find_host_by_ip.call_count == 6
+    assert mock_ping.call_count == 6
 
 
 @pytest.mark.workflow()
 @patch("gso.workflows.iptrunk.create_iptrunk.execute_playbook")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.find_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.ping")
 def test_iptrunk_creation_fails_when_lso_return_code_is_one(
+    mock_ping,
+    mock_find_host_by_ip,
     mock_allocate_v4_network,
     mock_allocate_v6_network,
     mock_execute_playbook,
@@ -163,8 +178,10 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
     _netbox_client_mock,  # noqa: PT019
     data_config_filename: PathLike,
 ):
-    mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
-    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
+    mock_allocate_v4_network.return_value = faker.ipv4_network(min_subnet=31, max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(min_subnet=126, max_subnet=126)
+    mock_find_host_by_ip.return_value = None
+    mock_ping.return_value = False
     product_id = get_product_id_by_name(ProductName.IP_TRUNK)
 
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
@@ -175,6 +192,8 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
     assert_lso_interaction_failure(result, process_stat, step_log)
 
     assert mock_execute_playbook.call_count == 2
+    assert mock_find_host_by_ip.call_count == 6
+    assert mock_ping.call_count == 6
 
 
 @pytest.mark.parametrize("input_form_wizard_data", [Vendor.JUNIPER], indirect=True)
@@ -183,9 +202,13 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
 @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.create_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.find_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.ping")
 @patch("gso.workflows.iptrunk.create_iptrunk.SharePointClient")
 def test_successful_iptrunk_creation_with_juniper_interface_names(
     mock_sharepoint_client,
+    mock_ping,
+    mock_find_host_by_ip,
     mock_create_host,
     mock_allocate_v4_network,
     mock_allocate_v6_network,
@@ -198,8 +221,11 @@ def test_successful_iptrunk_creation_with_juniper_interface_names(
     test_client,
 ):
     mock_create_host.return_value = None
-    mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
-    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
+    mock_allocate_v4_network.return_value = faker.ipv4_network(min_subnet=31, max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(min_subnet=126, max_subnet=126)
+    mock_find_host_by_ip.return_value = None
+    mock_ping.return_value = False
+
     mock_sharepoint_client.return_value = MockedSharePointClient
     product_id = get_product_id_by_name(ProductName.IP_TRUNK)
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
@@ -213,3 +239,71 @@ def test_successful_iptrunk_creation_with_juniper_interface_names(
 
     assert_complete(result)
     assert mock_execute_playbook.call_count == 6
+    assert mock_find_host_by_ip.call_count == 6
+    assert mock_ping.call_count == 6
+
+
+@pytest.mark.workflow()
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.find_host_by_ip")
+def test_iptrunk_creation_with_taken_dns_record(
+    mock_find_host_by_ip,
+    mock_allocate_v4_network,
+    mock_allocate_v6_network,
+    input_form_wizard_data,
+    faker,
+    _netbox_client_mock,  # noqa: PT019
+):
+    mock_allocate_v4_network.return_value = faker.ipv4_network(min_subnet=31, max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(min_subnet=126, max_subnet=126)
+    mock_find_host_by_ip.return_value = HostRecord(connector=None, hostname="fake.internal")
+
+    product_id = get_product_id_by_name(ProductName.IP_TRUNK)
+    initial_site_data = [{"product": product_id}, *input_form_wizard_data]
+    result, _, _ = run_workflow("create_iptrunk", initial_site_data)
+
+    assert_failed(result)
+
+    state = extract_state(result)
+    assert (
+        state["error"] == "One or more hosts in the assigned IPv4 network are already registered, please investigate."
+    )
+
+    #  We search for 2 hosts in a /31 and then fail the workflow
+    assert mock_find_host_by_ip.call_count == 2
+
+
+@pytest.mark.workflow()
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.find_host_by_ip")
+@patch("gso.workflows.iptrunk.create_iptrunk.ping")
+def test_iptrunk_creation_with_taken_ip_address(
+    mock_ping,
+    mock_find_host_by_ip,
+    mock_allocate_v4_network,
+    mock_allocate_v6_network,
+    input_form_wizard_data,
+    faker,
+    _netbox_client_mock,  # noqa: PT019
+):
+    mock_allocate_v4_network.return_value = faker.ipv4_network(min_subnet=31, max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(min_subnet=126, max_subnet=126)
+    mock_find_host_by_ip.return_value = None
+    mock_ping.return_value = True
+
+    product_id = get_product_id_by_name(ProductName.IP_TRUNK)
+    initial_site_data = [{"product": product_id}, *input_form_wizard_data]
+    result, _, _ = run_workflow("create_iptrunk", initial_site_data)
+
+    assert_failed(result)
+
+    state = extract_state(result)
+    assert (
+        state["error"] == "One or more hosts in the assigned IPv4 network are responding to ping, please investigate."
+    )
+
+    assert mock_find_host_by_ip.call_count == 6
+    #  We ping 2 hosts in a /31 and then fail the workflow
+    assert mock_ping.call_count == 2