Skip to content
Snippets Groups Projects
Verified Commit 223e6810 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Reserve IPv6 resources when creating a LAN Switch Interconnect

parent ecfdbbca
No related branches found
No related tags found
1 merge request!302Feature/update lan interconnect
Showing
with 153 additions and 36 deletions
......@@ -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:
......
......@@ -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,
......
......@@ -505,7 +505,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}
......
......@@ -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}
......
"""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
......
......@@ -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
)
)
......
......@@ -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)
......
......@@ -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)
......
......@@ -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"
......
......@@ -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
......
......@@ -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"
)
......@@ -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
......@@ -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",
}
......
......@@ -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,
}
......
......@@ -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,
},
{},
]
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment