Reserve IPv6 resources when creating a LAN Switch Interconnect

gso/services/infoblox.py

@@ -10,7 +10,14 @@ from infoblox_client.exceptions import (
 )
 
 from gso.settings import IPAMParams, load_oss_params
-from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv4NetworkType, IPv6AddressType, IPv6Netmask
+from gso.utils.types.ip_address import (
+    IPv4AddressType,
+    IPv4Netmask,
+    IPv4NetworkType,
+    IPv6AddressType,
+    IPv6Netmask,
+    IPv6NetworkType,
+)
 
 logger = getLogger(__name__)
 NULL_MAC = "00:00:00:00:00:00"
@@ -76,7 +83,11 @@ def _allocate_network(  # noqa: PLR0917
 def create_v4_network_by_ip(
     dns_view: str, network_view: str, network: IPv4NetworkType, comment: str | None = ""
 ) -> None:
-    """Register an IPv4 network at the given location. Raises an :class:`AllocationError` on failure."""
+    """Register an IPv4 network at the given location.
+
+    Raises:
+        AllocationError on failure.
+    """
     conn, _ = _setup_connection()
     created_net = objects.NetworkV4.create(
         conn, network=network, view=dns_view, network_view=network_view, comment=comment
@@ -88,6 +99,25 @@ def create_v4_network_by_ip(
     logger.debug(msg)
 
 
+def create_v6_network_by_ip(
+    dns_view: str, network_view: str, network: IPv6NetworkType, comment: str | None = ""
+) -> None:
+    """Register an IPv6 network at the given location.
+
+    Raises:
+        AllocationError on failure.
+    """
+    conn, _ = _setup_connection()
+    created_net = objects.NetworkV6.create(
+        conn, network=network, view=dns_view, network_view=network_view, comment=comment
+    )
+    if created_net.response != "Infoblox Object created":
+        msg = f"Failed to allocate network at {network}. Response from Netbox: {created_net.response}"
+        raise AllocationError(msg)
+    msg = f"Successfully registered new network at {network}"
+    logger.debug(msg)
+
+
 def hostname_available(hostname: str) -> bool:
     """Check whether a hostname is still available **in Infoblox**.
 
@@ -246,6 +276,7 @@ def create_host_by_ip(
     hostname: str,
     service_type: str,
     comment: str,
+    *,
     ipv4_address: IPv4AddressType | None = None,
     ipv6_address: IPv6AddressType | None = None,
 ) -> None:

gso/utils/helpers.py

@@ -2,7 +2,7 @@
 
 import random
 import re
-from ipaddress import IPv4Network
+from ipaddress import IPv4Network, IPv6Network
 from typing import TYPE_CHECKING
 from uuid import UUID
 
@@ -19,7 +19,7 @@ from gso.services.partners import get_all_partners
 from gso.services.subscriptions import is_virtual_circuit_id_available
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import IPv4AddressType, IPv4NetworkType
+from gso.utils.types.ip_address import IPv4AddressType, IPv4NetworkType, IPv6NetworkType
 from gso.utils.types.virtual_identifiers import VC_ID
 
 if TYPE_CHECKING:
@@ -127,8 +127,8 @@ def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
     return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"
 
 
-def generate_lan_switch_interconnect_subnet(site_internal_id: int) -> IPv4NetworkType:
-    """Generate an IPv4 network in which a :term:`LAN` Switch Interconnect resides, given a Site internal ID."""
+def generate_lan_switch_interconnect_subnet_v4(site_internal_id: int) -> IPv4NetworkType:
+    """Generate an IPv4 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
     ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
 
     result = str(ipam_oss.V4.containers[0]).split(".")[:2]  # Take the first two octets from the IPv4 network.
@@ -138,6 +138,17 @@ def generate_lan_switch_interconnect_subnet(site_internal_id: int) -> IPv4Networ
     return IPv4Network(".".join(result))
 
 
+def generate_lan_switch_interconnect_subnet_v6(site_internal_id: int) -> IPv6NetworkType:
+    """Generate an IPv6 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
+    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+
+    result = IPv6Network(ipam_oss.V6.containers[0]).exploded[:17]  # Take the first 56 bits of the network
+    result += str(hex(site_internal_id)[2:])  # Append the site internal id for bytes 57 to 64 as hexadecimal number
+    result += f"::/{ipam_oss.V6.mask}"  # And fill the rest of the network with empty bits
+
+    return IPv6Network(result)
+
+
 def generate_inventory_for_routers(
     router_role: RouterRole,
     exclude_routers: list[str] | None = None,

gso/workflows/iptrunk/create_iptrunk.py

@@ -507,7 +507,9 @@ def register_dns_records(subscription: IptrunkInactive) -> State:
         ipv4_addr = subscription.iptrunk.iptrunk_ipv4_network[index]
         ipv6_addr = subscription.iptrunk.iptrunk_ipv6_network[index + 1]
 
-        infoblox.create_host_by_ip(fqdn, "TRUNK", str(subscription.subscription_id), ipv4_addr, ipv6_addr)
+        infoblox.create_host_by_ip(
+            fqdn, "TRUNK", str(subscription.subscription_id), ipv4_address=ipv4_addr, ipv6_address=ipv6_addr
+        )
 
     return {"subscription": subscription}
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -740,7 +740,7 @@ def update_ipam(subscription: Iptrunk, replace_index: int, new_node: Router, new
     #  And in with the new
     new_fqdn = f"{new_lag_interface}-0.{new_node.router.router_fqdn}"
     comment = str(subscription.subscription_id)
-    infoblox.create_host_by_ip(new_fqdn, "TRUNK", comment, v4_addr, v6_addr)
+    infoblox.create_host_by_ip(new_fqdn, "TRUNK", comment, ipv4_address=v4_addr, ipv6_address=v6_addr)
 
     return {"subscription": subscription}
 

gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

 """A creation workflow for creating a new interconnect between a switch and a router."""
 
-from ipaddress import IPv4Address
+from ipaddress import IPv4Network, IPv6Network
 from typing import Annotated
 from uuid import uuid4
 
@@ -21,7 +21,7 @@ from gso.products.product_blocks.lan_switch_interconnect import (
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnectInactive
 from gso.products.product_types.router import Router
 from gso.products.product_types.switch import Switch
-from gso.services.infoblox import create_host_by_ip, create_v4_network_by_ip
+from gso.services.infoblox import create_host_by_ip, create_v4_network_by_ip, create_v6_network_by_ip
 from gso.services.partners import get_partner_by_name
 from gso.settings import load_oss_params
 from gso.utils.helpers import (
@@ -29,7 +29,8 @@ from gso.utils.helpers import (
     active_switch_selector,
     available_interfaces_choices,
     available_lags_choices,
-    generate_lan_switch_interconnect_subnet,
+    generate_lan_switch_interconnect_subnet_v4,
+    generate_lan_switch_interconnect_subnet_v6,
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import (
@@ -162,28 +163,28 @@ def initialize_subscription(
     return {"subscription": subscription}
 
 
-@step("Register network in IPAM")
-def register_dns_records_network(subscription: LanSwitchInterconnectInactive) -> State:
-    """Add :term:`DNS` records in :term:`IPAM`."""
+@step("Register IPv4 network in IPAM")
+def register_dns_records_v4_network(subscription: LanSwitchInterconnectInactive) -> State:
+    """Add DNS records in IPAM."""
     router_site = subscription.lan_switch_interconnect.router_side.node.router_site
     if not router_site or not router_site.site_internal_id:
         msg = "Site internal ID not set. Cannot continue."
         raise ProcessFailureError(msg, details=router_site)
 
-    new_network = generate_lan_switch_interconnect_subnet(router_site.site_internal_id)
+    new_network = generate_lan_switch_interconnect_subnet_v4(router_site.site_internal_id)
     ipam_oss_params = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
     create_v4_network_by_ip(
         ipam_oss_params.dns_view, ipam_oss_params.network_view, new_network, str(subscription.subscription_id)
     )
 
-    return {"ipam_registrations": {"network": new_network}}
+    return {"ipam_registrations": {"v4": {"network": new_network}}}
 
 
-@step("Register devices in IPAM")
-def register_dns_records_devices(
-    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, str]
+@step("Register IPv4 devices in IPAM")
+def register_dns_records_v4_devices(
+    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, dict[str, str]]
 ) -> State:
-    """Register :term:`DNS` records for both switch and router side in :term:`IPAM`."""
+    """Register DNS records for both switch and router side in IPAM."""
     switch_hostname = subscription.lan_switch_interconnect.switch_side.switch.fqdn
     router_hostname = (
         f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
@@ -193,14 +194,51 @@ def register_dns_records_devices(
         msg = "Missing switch or router hostname, cannot continue."
         raise ProcessFailureError(msg, details=subscription.lan_switch_interconnect)
 
-    ip_network_prefix = ipam_registrations["network"].split(".")[:3]  # Take the first three octets of the network.
-    switch_side_ip = IPv4Address(".".join([*ip_network_prefix, "10"]))  # Add .10 as the fourth octet of the switch.
-    router_side_ip = IPv4Address(".".join([*ip_network_prefix, "1"]))  # Add .1 as the fourth octet of the router.
+    ip_network = IPv4Network(ipam_registrations["v4"]["network"])
 
-    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, switch_side_ip)
-    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, router_side_ip)
+    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv4_address=ip_network[10])
+    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv4_address=ip_network[1])
 
-    return {"ipam_registrations": {switch_hostname: switch_side_ip, router_hostname: router_side_ip}}
+    return {"ipam_registrations": {"v4": {switch_hostname: ip_network[10], router_hostname: ip_network[1]}}}
+
+
+@step("Register IPv6 network in IPAM")
+def register_dns_records_v6_network(subscription: LanSwitchInterconnectInactive) -> State:
+    """Add DNS records in IPAM."""
+    router_site = subscription.lan_switch_interconnect.router_side.node.router_site
+    if not router_site or not router_site.site_internal_id:
+        msg = "Site internal ID not set. Cannot continue."
+        raise ProcessFailureError(msg, details=router_site)
+
+    new_network = generate_lan_switch_interconnect_subnet_v6(router_site.site_internal_id)
+    ipam_oss_params = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+    create_v6_network_by_ip(
+        ipam_oss_params.dns_view, ipam_oss_params.network_view, new_network, str(subscription.subscription_id)
+    )
+
+    return {"ipam_registrations": {"v6": {"network": new_network}}}
+
+
+@step("Register IPv6 devices in IPAM")
+def register_dns_records_v6_devices(
+    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, dict[str, str]]
+) -> State:
+    """Register DNS records for both switch and router side in IPAM."""
+    switch_hostname = subscription.lan_switch_interconnect.switch_side.switch.fqdn
+    router_hostname = (
+        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+    )
+    if not (switch_hostname and router_hostname):
+        msg = "Missing switch or router hostname, cannot continue."
+        raise ProcessFailureError(msg, details=subscription.lan_switch_interconnect)
+
+    ip_network = IPv6Network(ipam_registrations["v6"]["network"])
+
+    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv6_address=ip_network[10])
+    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv6_address=ip_network[1])
+
+    return {"ipam_registrations": {"v6": {switch_hostname: ip_network[10], router_hostname: ip_network[1]}}}
 
 
 @workflow(
@@ -215,8 +253,10 @@ def create_lan_switch_interconnect() -> StepList:
         >> create_subscription
         >> store_process_subscription(Target.CREATE)
         >> initialize_subscription
-        >> register_dns_records_network
-        >> register_dns_records_devices
+        >> register_dns_records_v4_network
+        >> register_dns_records_v4_devices
+        >> register_dns_records_v6_network
+        >> register_dns_records_v6_devices
         >> set_status(SubscriptionLifecycle.ACTIVE)
         >> resync
         >> done

gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py

@@ -12,7 +12,7 @@ from pydantic_forms.validators import Label
 
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.services.infoblox import delete_host_by_fqdn, delete_network
-from gso.utils.helpers import generate_lan_switch_interconnect_subnet
+from gso.utils.helpers import generate_lan_switch_interconnect_subnet_v4
 from gso.utils.types.tt_number import TTNumber
 
 
@@ -46,7 +46,7 @@ def clean_up_ipam(subscription: LanSwitchInterconnect) -> None:
         f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
     )
     delete_network(
-        generate_lan_switch_interconnect_subnet(
+        generate_lan_switch_interconnect_subnet_v4(
             subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
         )
     )

gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py

@@ -11,7 +11,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.services.infoblox import find_host_by_fqdn, find_network_by_cidr
 from gso.services.lso_client import LSOState, anonymous_lso_interaction
-from gso.utils.helpers import generate_lan_switch_interconnect_subnet
+from gso.utils.helpers import generate_lan_switch_interconnect_subnet_v4
 
 
 @step("Validate IPAM configuration")
@@ -29,7 +29,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None:
             msg = "DNS record is incorrectly configured in IPAM, please investigate this manually!"
             raise ProcessFailureError(msg, details=host_record)
 
-    lan_interconnect_network = generate_lan_switch_interconnect_subnet(
+    lan_interconnect_network = generate_lan_switch_interconnect_subnet_v4(
         subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
     )
     network_record = find_network_by_cidr(lan_interconnect_network)

test/cli/test_imports.py

@@ -214,7 +214,7 @@ def switch_data(temp_file, faker, site_subscription_factory):
             "ts_port": faker.port_number(is_user=True),
             "site": site_subscription_factory(),
             "switch_vendor": Vendor.JUNIPER,
-            "switch_model": SwitchModel.EX3400,
+            "switch_model": SwitchModel.EX3400_48T,
         }
         switch_data.update(**kwargs)
 

test/fixtures/site_fixtures.py

@@ -31,6 +31,7 @@ def site_subscription_factory(faker, geant_partner):
         start_date="2023-05-24T00:00:00+00:00",
         *,
         is_imported: bool | None = True,
+        site_contains_optical_equipment: bool | None = True,
     ) -> UUIDstr:
         if partner is None:
             partner = geant_partner
@@ -54,6 +55,7 @@ def site_subscription_factory(faker, geant_partner):
         site_subscription.site.site_internal_id = site_internal_id or faker.pyint(max_value=254)
         site_subscription.site.site_tier = site_tier or SiteTier.TIER1
         site_subscription.site.site_ts_address = site_ts_address or faker.ipv4()
+        site_subscription.site.site_contains_optical_equipment = site_contains_optical_equipment
 
         site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
         site_subscription.description = description or "Site Subscription"

test/fixtures/switch_fixtures.py

@@ -41,7 +41,7 @@ def switch_subscription_factory(faker, geant_partner, site_subscription_factory)
         switch_subscription.switch.ts_port = ts_port or faker.port_number(is_user=True)
         switch_subscription.switch.site = site or Site.from_subscription(site_subscription_factory()).site
         switch_subscription.switch.switch_vendor = switch_vendor or Vendor.JUNIPER
-        switch_subscription.switch.switch_model = switch_model or SwitchModel.EX3400
+        switch_subscription.switch.switch_model = switch_model or SwitchModel.EX3400_24T
 
         switch_subscription = SubscriptionModel.from_other_lifecycle(switch_subscription, SubscriptionLifecycle.ACTIVE)
         switch_subscription.insync = True

test/utils/test_helpers.py

@@ -6,9 +6,12 @@ from orchestrator.types import SubscriptionLifecycle
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
+from gso.products.product_types.site import Site
 from gso.utils.helpers import (
     available_interfaces_choices_including_current_members,
     generate_inventory_for_routers,
+    generate_lan_switch_interconnect_subnet_v4,
+    generate_lan_switch_interconnect_subnet_v6,
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.tt_number import validate_tt_number
@@ -139,3 +142,25 @@ def test_generate_inventory_for_active_routers_with_excluded_router(router_subsc
     excluded_routers = [Router.from_subscription(router).router.router_fqdn]
     inventory = generate_inventory_for_routers(RouterRole.P, exclude_routers=excluded_routers)
     assert len(inventory["all"]["hosts"]) == 5  # 6 P routers, the last one is excluded, so 5 P routers are left.
+
+
+@pytest.mark.parametrize("execution_count", range(10))
+def test_generate_lan_switch_interconnect_subnet_v4(execution_count, site_subscription_factory):
+    """Test generating a new subnet for a LAN Switch Interconnect.
+
+    We need to ensure that the third octet of the new subnet is set correctly from the Site internal ID.
+    """
+    site = Site.from_subscription(site_subscription_factory())
+    assert (
+        str(generate_lan_switch_interconnect_subnet_v4(site.site.site_internal_id))
+        == f"10.2.{site.site.site_internal_id}.0/24"
+    )
+
+
+@pytest.mark.parametrize("execution_count", range(10))
+def test_generate_lan_switch_interconnect_subnet_v6(execution_count, site_subscription_factory):
+    site = Site.from_subscription(site_subscription_factory())
+    assert (
+        str(generate_lan_switch_interconnect_subnet_v6(site.site.site_internal_id))
+        == f"beef:cafe:0:{hex(site.site.site_internal_id).split("x")[-1]}::/64"
+    )

test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py

@@ -52,11 +52,13 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact
 
 
 @pytest.mark.workflow()
+@patch("gso.services.infoblox.create_v6_network_by_ip")
 @patch("gso.services.infoblox.create_v4_network_by_ip")
 @patch("gso.services.infoblox.create_host_by_ip")
 def test_create_lan_switch_interconnect_success(
     mock_create_host,
     mock_create_v4_network,
+    mock_create_v6_network,
     input_form_data,
     _netbox_client_mock,  # noqa: PT019
 ):
@@ -68,4 +70,5 @@ def test_create_lan_switch_interconnect_success(
     subscription = LanSwitchInterconnect.from_subscription(subscription_id)
     assert subscription.status == SubscriptionLifecycle.ACTIVE
     assert mock_create_v4_network.call_count == 1
-    assert mock_create_host.call_count == 2
+    assert mock_create_v6_network.call_count == 1
+    assert mock_create_host.call_count == 4

test/workflows/site/test_create_imported_site.py

@@ -19,6 +19,7 @@ def workflow_input_data(faker):
         "site_internal_id": faker.pyint(),
         "site_tier": SiteTier.TIER1,
         "site_ts_address": faker.ipv4(),
+        "site_contains_optical_equipment": True,
         "partner": "GEANT",
     }
 

test/workflows/switch/test_create_imported_switch.py

@@ -19,7 +19,7 @@ def workflow_input_data(faker, site_subscription_factory):
         "ts_port": faker.port_number(is_user=True),
         "site": site_subscription_factory(),
         "switch_vendor": Vendor.JUNIPER,
-        "switch_model": SwitchModel.EX3400,
+        "switch_model": SwitchModel.EX3400_24T,
     }
 
 

test/workflows/switch/test_create_switch.py

@@ -3,6 +3,7 @@ from unittest.mock import patch
 import pytest
 
 from gso.products import ProductName
+from gso.products.product_blocks.switch import SwitchModel
 from gso.products.product_types.switch import Switch
 from gso.services.subscriptions import get_product_id_by_name
 from test import USER_CONFIRM_EMPTY_FORM
@@ -32,6 +33,7 @@ def test_create_switch_success(
             "switch_site": site_subscription_factory(),
             "hostname": faker.domain_word(),
             "ts_port": faker.port_number(is_user=True),
+            "model": SwitchModel.EX3400_24T,
         },
         {},
     ]