From 2ae3cef70b37a8b0781a225f99f48147af5bc73a Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Mon, 4 Nov 2024 17:25:30 +0100
Subject: [PATCH] Calculate IP space in LAN Switch Interconnect, add IPAM steps
to workflows
---
gso/cli/imports.py | 14 ++---
...e0c35e20_update_lan_switch_interconnect.py | 48 +++++++++++++++
gso/oss-params-example.json | 7 +++
.../product_blocks/lan_switch_interconnect.py | 15 -----
.../product_blocks/service_binding_port.py | 14 ++---
gso/services/infoblox.py | 59 ++++++++++++------
gso/settings.py | 7 ++-
gso/utils/helpers.py | 14 ++++-
gso/utils/types/ip_address.py | 12 +---
gso/workflows/iptrunk/create_iptrunk.py | 2 +-
gso/workflows/iptrunk/migrate_iptrunk.py | 2 +-
.../create_imported_l3_core_service.py | 6 +-
.../l3_core_service/create_l3_core_service.py | 6 +-
.../l3_core_service/modify_l3_core_service.py | 6 +-
...create_imported_lan_switch_interconnect.py | 7 ---
.../create_lan_switch_interconnect.py | 53 ++++++++++++++--
.../terminate_lan_switch_interconnect.py | 24 +++++++-
.../validate_lan_switch_interconnect.py | 38 +++++++++++-
test/cli/test_imports.py | 5 --
.../lan_switch_interconnect_fixtures.py | 11 ----
test/fixtures/site_fixtures.py | 2 +-
...create_imported_lan_switch_interconnect.py | 5 --
.../test_create_lan_switch_interconnect.py | 61 +++++++++----------
.../test_terminate_lan_switch_interconnect.py | 10 ++-
24 files changed, 283 insertions(+), 145 deletions(-)
create mode 100644 gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index 4c973ace..7bfcbd2a 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -43,13 +43,11 @@ from gso.utils.types.base_site import BaseSiteValidatorModel
from gso.utils.types.geant_ids import IMPORTED_GA_ID, IMPORTED_GS_ID
from gso.utils.types.interfaces import BandwidthString, LAGMember, LAGMemberList, PhysicalPortCapacity
from gso.utils.types.ip_address import (
- AddressSpace,
IPAddress,
IPv4AddressType,
- IPV4Netmask,
- IPv4NetworkType,
+ IPv4Netmask,
IPv6AddressType,
- IPV6Netmask,
+ IPv6Netmask,
PortNumber,
)
from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
@@ -276,9 +274,9 @@ class L3CoreServiceImportModel(BaseModel):
vlan_id: VLAN_ID
custom_firewall_filters: bool = False
ipv4_address: IPv4AddressType
- ipv4_mask: IPV4Netmask
+ ipv4_mask: IPv4Netmask
ipv6_address: IPv6AddressType
- ipv6_mask: IPV6Netmask
+ ipv6_mask: IPv6Netmask
is_multi_hop: bool = True
bgp_peers: list["L3CoreServiceImportModel.BaseBGPPeer"]
v4_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
@@ -316,7 +314,6 @@ class LanSwitchInterconnectRouterSideImportModel(BaseModel):
node: UUIDstr
ae_iface: str
ae_members: LAGMemberList[LAGMember]
- ipv4_address: IPv4AddressType
class LanSwitchInterconnectSwitchSideImportModel(BaseModel):
@@ -325,15 +322,12 @@ class LanSwitchInterconnectSwitchSideImportModel(BaseModel):
switch: UUIDstr
ae_iface: str
ae_members: LAGMemberList[LAGMember]
- ipv4_address: IPv4AddressType
class LanSwitchInterconnectImportModel(BaseModel):
"""Import LAN Switch Interconnect model."""
lan_switch_interconnect_description: str
- lan_switch_interconnect_ip_network: IPv4NetworkType | None
- address_space: AddressSpace
minimum_links: int
router_side: LanSwitchInterconnectRouterSideImportModel
switch_side: LanSwitchInterconnectSwitchSideImportModel
diff --git a/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py b/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
new file mode 100644
index 00000000..f041528c
--- /dev/null
+++ b/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
@@ -0,0 +1,48 @@
+"""Update LAN Switch Interconnect.
+
+Revision ID: e854e0c35e20
+Revises: 0e7e7d749617
+Create Date: 2024-11-04 17:21:14.612740
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'e854e0c35e20'
+down_revision = '0e7e7d749617'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+ conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectSwitchSideBlock', 'LanSwitchInterconnectRouterSideBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_address'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectSwitchSideBlock', 'LanSwitchInterconnectRouterSideBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_address'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('lan_switch_interconnect_ip_network'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('lan_switch_interconnect_ip_network'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space', 'lan_switch_interconnect_ip_network'))
+ """))
+ conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('address_space', 'lan_switch_interconnect_ip_network')
+ """))
+
+
+def downgrade() -> None:
+ conn = op.get_bind()
+
diff --git a/gso/oss-params-example.json b/gso/oss-params-example.json
index 63de6f89..f6e85ed8 100644
--- a/gso/oss-params-example.json
+++ b/gso/oss-params-example.json
@@ -51,6 +51,13 @@
"domain_name": ".geantip",
"dns_view": "default",
"network_view": "default"
+ },
+ "LAN_SWITCH_INTERCONNECT": {
+ "V4": {"containers": ["10.2.0.0/16"], "networks": [], "mask": 24},
+ "V6": {"containers": [], "networks": [], "mask": 126},
+ "domain_name": ".geant.net",
+ "dns_view": "default",
+ "network_view": "default"
}
},
"MONITORING": {
diff --git a/gso/products/product_blocks/lan_switch_interconnect.py b/gso/products/product_blocks/lan_switch_interconnect.py
index a3128a72..30c2fb93 100644
--- a/gso/products/product_blocks/lan_switch_interconnect.py
+++ b/gso/products/product_blocks/lan_switch_interconnect.py
@@ -6,7 +6,6 @@ from orchestrator.types import SubscriptionLifecycle
from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
from gso.utils.types.interfaces import LAGMemberList
-from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
class LanSwitchInterconnectInterfaceBlockInactive(
@@ -48,7 +47,6 @@ class LanSwitchInterconnectRouterSideBlockInactive(
node: RouterBlockInactive
ae_iface: str | None = None
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
- ipv4_address: IPv4AddressType | None = None
class LanSwitchInterconnectRouterSideBlockProvisioning(
@@ -59,7 +57,6 @@ class LanSwitchInterconnectRouterSideBlockProvisioning(
node: RouterBlockProvisioning
ae_iface: str | None = None
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning] # type: ignore[assignment]
- ipv4_address: IPv4AddressType | None
class LanSwitchInterconnectRouterSideBlock(
@@ -70,7 +67,6 @@ class LanSwitchInterconnectRouterSideBlock(
node: RouterBlock
ae_iface: str
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock] # type: ignore[assignment]
- ipv4_address: IPv4AddressType | None
class LanSwitchInterconnectSwitchSideBlockInactive(
@@ -83,7 +79,6 @@ class LanSwitchInterconnectSwitchSideBlockInactive(
switch: SwitchBlockInactive
ae_iface: str | None = None
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
- ipv4_address: IPv4AddressType | None = None
class LanSwitchInterconnectSwitchSideBlockProvisioning(
@@ -94,7 +89,6 @@ class LanSwitchInterconnectSwitchSideBlockProvisioning(
switch: SwitchBlockProvisioning
ae_iface: str | None = None
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning] # type: ignore[assignment]
- ipv4_address: IPv4AddressType | None
class LanSwitchInterconnectSwitchSideBlock(
@@ -105,7 +99,6 @@ class LanSwitchInterconnectSwitchSideBlock(
switch: SwitchBlock
ae_iface: str
ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock] # type: ignore[assignment]
- ipv4_address: IPv4AddressType | None
class LanSwitchInterconnectBlockInactive(
@@ -116,8 +109,6 @@ class LanSwitchInterconnectBlockInactive(
"""A LAN Switch Interconnect that's currently inactive, see ``LanSwitchInterconnectBlock``."""
lan_switch_interconnect_description: str | None = None
- lan_switch_interconnect_ip_network: IPv4NetworkType | None = None
- address_space: AddressSpace | None = None
minimum_links: int | None = None
router_side: LanSwitchInterconnectRouterSideBlockInactive
switch_side: LanSwitchInterconnectSwitchSideBlockInactive
@@ -129,8 +120,6 @@ class LanSwitchInterconnectBlockProvisioning(
"""A LAN Switch Interconnect that's currently being provisioned, see ``LanSwitchInterconnectBlock``."""
lan_switch_interconnect_description: str | None = None
- lan_switch_interconnect_ip_network: IPv4NetworkType | None
- address_space: AddressSpace | None = None
minimum_links: int | None = None
router_side: LanSwitchInterconnectRouterSideBlockProvisioning
switch_side: LanSwitchInterconnectSwitchSideBlockProvisioning
@@ -141,10 +130,6 @@ class LanSwitchInterconnectBlock(LanSwitchInterconnectBlockProvisioning, lifecyc
#: A human-readable description of this LAN Switch Interconnect.
lan_switch_interconnect_description: str
- #: The IP resources for this VLAN Switch Interconnect.
- lan_switch_interconnect_ip_network: IPv4NetworkType | None
- #: The address space of the VLAN Switch Interconnect. It can be private or public.
- address_space: AddressSpace
#: The minimum amount of links the LAN Switch Interconnect should consist of.
minimum_links: int
#: The router side of the LAN Switch Interconnect.
diff --git a/gso/products/product_blocks/service_binding_port.py b/gso/products/product_blocks/service_binding_port.py
index 5f10b33c..6d736f56 100644
--- a/gso/products/product_blocks/service_binding_port.py
+++ b/gso/products/product_blocks/service_binding_port.py
@@ -14,7 +14,7 @@ from gso.products.product_blocks.bgp_session import (
)
from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
from gso.utils.shared_enums import SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
from gso.utils.types.virtual_identifiers import VLAN_ID
@@ -60,9 +60,9 @@ class ServiceBindingPortInactive(
vlan_id: VLAN_ID | None = None
sbp_type: SBPType | None = None
ipv4_address: IPv4AddressType | None = None
- ipv4_mask: IPV4Netmask | None = None
+ ipv4_mask: IPv4Netmask | None = None
ipv6_address: IPv6AddressType | None = None
- ipv6_mask: IPV6Netmask | None = None
+ ipv6_mask: IPv6Netmask | None = None
custom_firewall_filters: bool | None = None
gs_id: str | None = None
bgp_session_list: list[BGPSessionInactive] = Field(default_factory=list)
@@ -78,9 +78,9 @@ class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[Subs
vlan_id: VLAN_ID | None = None
sbp_type: SBPType
ipv4_address: IPv4AddressType | None = None
- ipv4_mask: IPV4Netmask | None = None
+ ipv4_mask: IPv4Netmask | None = None
ipv6_address: IPv6AddressType | None = None
- ipv6_mask: IPV6Netmask | None = None
+ ipv6_mask: IPv6Netmask | None = None
custom_firewall_filters: bool
gs_id: str
bgp_session_list: list[BGPSessionProvisioning] # type: ignore[assignment]
@@ -101,11 +101,11 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
#: If layer 3, IPv4 resources.
ipv4_address: IPv4AddressType | None = None
#: IPV4 subnet mask.
- ipv4_mask: IPV4Netmask | None = None
+ ipv4_mask: IPv4Netmask | None = None
#: If layer 3, IPv6 resources.
ipv6_address: IPv6AddressType | None = None
#: IPV6 subnet mask.
- ipv6_mask: IPV6Netmask | None = None
+ ipv6_mask: IPv6Netmask | None = None
#: Any custom firewall filters that the partner may require.
custom_firewall_filters: bool
#: The GÉANT service ID of this binding port.
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index cad3fb43..1068be97 100644
--- a/gso/services/infoblox.py
+++ b/gso/services/infoblox.py
@@ -10,7 +10,7 @@ from infoblox_client.exceptions import (
)
from gso.settings import IPAMParams, load_oss_params
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv4NetworkType, IPv6AddressType, IPv6Netmask
logger = getLogger(__name__)
NULL_MAC = "00:00:00:00:00:00"
@@ -41,7 +41,7 @@ def _allocate_network( # noqa: PLR0917
conn: connector.Connector,
dns_view: str,
network_view: str,
- netmask: int,
+ netmask: IPv4Netmask | IPv6Netmask,
containers: list[str],
comment: str | None = "",
) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
@@ -73,6 +73,21 @@ def _allocate_network( # noqa: PLR0917
raise AllocationError(msg)
+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."""
+ conn, _ = _setup_connection()
+ created_net = objects.NetworkV4.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**.
@@ -229,38 +244,48 @@ def allocate_host(
def create_host_by_ip(
hostname: str,
- ipv4_address: IPv4AddressType,
- ipv6_address: IPv6AddressType,
service_type: str,
comment: str,
+ ipv4_address: IPv4AddressType | None = None,
+ ipv6_address: IPv6AddressType | None = None,
) -> None:
"""Create a new host record with a given IPv4 and IPv6 address.
Args:
hostname: The FQDN of the new host.
- ipv4_address: The IPv4 address of the new host.
- ipv6_address: The IPv6 address of the new host.
service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
Infoblox.
comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in GSO.
+ ipv4_address: The IPv4 address of the new host.
+ ipv6_address: The IPv6 address of the new host.
"""
if not hostname_available(hostname):
msg = f"FQDN '{hostname}' is already in use, allocation aborted."
raise AllocationError(msg)
+ if not (ipv4_address or ipv6_address):
+ msg = "At least one IP address must be given to create a record for. Received None."
+ raise AllocationError(msg)
conn, oss = _setup_connection()
- ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
- ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
dns_view = getattr(oss, service_type).dns_view
network_view = getattr(oss, service_type).network_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, view=dns_view, network_view=network_view
- )
- new_host = find_host_by_fqdn(hostname)
- new_host.ipv4addrs = [ipv4_object]
- new_host.update()
+ if ipv6_address: # We first register the IPv6 record.
+ ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
+ objects.HostRecord.create(
+ conn, ip=ipv6_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
+ )
+ if ipv4_address: # If set, we also register the IPv4 record.
+ new_host = find_host_by_fqdn(hostname)
+ ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
+ new_host.ipv4addrs = [ipv4_object]
+ new_host.update()
+ else: # IPv6 is not set, it must therefore be IPv4 only.
+ ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
+ objects.HostRecord.create(
+ conn, ip=ipv4_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
+ )
def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
@@ -290,11 +315,7 @@ def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
fqdn: The FQDN of a host that is searched for.
"""
conn, _ = _setup_connection()
- return objects.HostRecord.search(
- conn,
- name=fqdn,
- return_fields=["ipv4addrs", "name", "view", "aliases", "comment"],
- )
+ return objects.HostRecord.search(conn, name=fqdn, return_fields=["ipv4addrs", "name", "view", "aliases", "comment"])
def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
diff --git a/gso/settings.py b/gso/settings.py
index 7f71a560..5b3beefd 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -15,7 +15,7 @@ from pydantic import EmailStr
from pydantic_forms.types import strEnum
from pydantic_settings import BaseSettings
-from gso.utils.types.ip_address import IPV4Netmask, IPV6Netmask, PortNumber
+from gso.utils.types.ip_address import IPv4Netmask, IPv6Netmask, PortNumber
logger = logging.getLogger(__name__)
@@ -65,7 +65,7 @@ class V4NetworkParams(BaseSettings):
containers: list[ipaddress.IPv4Network]
networks: list[ipaddress.IPv4Network]
- mask: IPV4Netmask
+ mask: IPv4Netmask
class V6NetworkParams(BaseSettings):
@@ -73,7 +73,7 @@ class V6NetworkParams(BaseSettings):
containers: list[ipaddress.IPv6Network]
networks: list[ipaddress.IPv6Network]
- mask: IPV6Netmask
+ mask: IPv6Netmask
class ServiceNetworkParams(BaseSettings):
@@ -98,6 +98,7 @@ class IPAMParams(BaseSettings):
GEANT_IP: ServiceNetworkParams
SI: ServiceNetworkParams
LT_IAS: ServiceNetworkParams
+ LAN_SWITCH_INTERCONNECT: ServiceNetworkParams
class MonitoringSNMPV2Params(BaseSettings):
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 852e152f..776d4629 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -2,6 +2,7 @@
import random
import re
+from ipaddress import IPv4Network
from typing import TYPE_CHECKING
from uuid import UUID
@@ -18,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
+from gso.utils.types.ip_address import IPv4AddressType, IPv4NetworkType
from gso.utils.types.virtual_identifiers import VC_ID
if TYPE_CHECKING:
@@ -126,6 +127,17 @@ 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."""
+ 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.
+ result.append(str(site_internal_id)) # Append the side ID as the third octet.
+ result.append(f"0/{ipam_oss.V4.mask}") # Append the fourth octet, together with the netmask.
+
+ return IPv4Network(".".join(result))
+
+
def generate_inventory_for_routers(
router_role: RouterRole,
exclude_routers: list[str] | None = None,
diff --git a/gso/utils/types/ip_address.py b/gso/utils/types/ip_address.py
index 3b91167f..83710c3c 100644
--- a/gso/utils/types/ip_address.py
+++ b/gso/utils/types/ip_address.py
@@ -4,7 +4,6 @@ import ipaddress
from typing import Annotated, Any
from pydantic import AfterValidator, Field, PlainSerializer
-from pydantic_forms.types import strEnum
from typing_extensions import Doc
@@ -40,8 +39,8 @@ IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(_str, return_
IPv6NetworkType = Annotated[ipaddress.IPv6Network, PlainSerializer(_str, return_type=str, when_used="always")]
IPAddress = Annotated[str, AfterValidator(validate_ipv4_or_ipv6)]
IPNetwork = Annotated[str, AfterValidator(validate_ipv4_or_ipv6_network)]
-IPV4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
-IPV6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
+IPv4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
+IPv6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
PortNumber = Annotated[
int,
Field(
@@ -53,10 +52,3 @@ PortNumber = Annotated[
"and can therefore not be selected for permanent allocation."
),
]
-
-
-class AddressSpace(strEnum):
- """Types of address space. Can be private or public."""
-
- PRIVATE = "PRIVATE"
- PUBLIC = "PUBLIC"
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 962cc5d0..909da8b2 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -505,7 +505,7 @@ 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, ipv4_addr, ipv6_addr, "TRUNK", str(subscription.subscription_id))
+ infoblox.create_host_by_ip(fqdn, "TRUNK", str(subscription.subscription_id), ipv4_addr, ipv6_addr)
return {"subscription": subscription}
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 7e7f32b4..50dd2823 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/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, v4_addr, v6_addr, "TRUNK", comment)
+ infoblox.create_host_by_ip(new_fqdn, "TRUNK", comment, v4_addr, v6_addr)
return {"subscription": subscription}
diff --git a/gso/workflows/l3_core_service/create_imported_l3_core_service.py b/gso/workflows/l3_core_service/create_imported_l3_core_service.py
index 7280a225..a4948f7d 100644
--- a/gso/workflows/l3_core_service/create_imported_l3_core_service.py
+++ b/gso/workflows/l3_core_service/create_imported_l3_core_service.py
@@ -22,7 +22,7 @@ from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import SBPType
from gso.utils.types.geant_ids import IMPORTED_GS_ID
-from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
from gso.utils.types.virtual_identifiers import VLAN_ID
@@ -57,9 +57,9 @@ def initial_input_form_generator() -> FormGenerator:
vlan_id: VLAN_ID
custom_firewall_filters: bool = False
ipv4_address: IPv4AddressType
- ipv4_mask: IPV4Netmask
+ ipv4_mask: IPv4Netmask
ipv6_address: IPv6AddressType
- ipv6_mask: IPV6Netmask
+ ipv6_mask: IPv6Netmask
rtbh_enabled: bool = True
is_multi_hop: bool = True
bgp_peers: list[BaseBGPPeer]
diff --git a/gso/workflows/l3_core_service/create_l3_core_service.py b/gso/workflows/l3_core_service/create_l3_core_service.py
index cd42c761..3cf358d2 100644
--- a/gso/workflows/l3_core_service/create_l3_core_service.py
+++ b/gso/workflows/l3_core_service/create_l3_core_service.py
@@ -26,7 +26,7 @@ from gso.utils.helpers import (
partner_choice,
)
from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
from gso.utils.types.tt_number import TTNumber
from gso.utils.types.virtual_identifiers import VLAN_ID
from gso.utils.workflow_steps import start_moodi, stop_moodi
@@ -120,12 +120,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
divider: Divider = Field(None, exclude=True)
v4_label: Label = Field("IPV4 SBP interface params", exclude=True)
ipv4_address: IPv4AddressType
- ipv4_mask: IPV4Netmask
+ ipv4_mask: IPv4Netmask
v4_bfd_settings: BFDSettingsForm
divider2: Divider = Field(None, exclude=True)
v6_label: Label = Field("IPV6 SBP interface params", exclude=True)
ipv6_address: IPv6AddressType
- ipv6_mask: IPV6Netmask
+ ipv6_mask: IPv6Netmask
v6_bfd_settings: BFDSettingsForm
divider3: Divider = Field(None, exclude=True)
v4_bgp_peer: IPv4BGPPeer
diff --git a/gso/workflows/l3_core_service/modify_l3_core_service.py b/gso/workflows/l3_core_service/modify_l3_core_service.py
index 24b408ab..af3218c5 100644
--- a/gso/workflows/l3_core_service/modify_l3_core_service.py
+++ b/gso/workflows/l3_core_service/modify_l3_core_service.py
@@ -21,7 +21,7 @@ from gso.products.product_types.edge_port import EdgePort
from gso.products.product_types.l3_core_service import L3CoreService
from gso.utils.helpers import active_edge_port_selector
from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
from gso.utils.types.virtual_identifiers import VLAN_ID
@@ -158,9 +158,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
# occur since it's a layer 3 service. The ignore statements are there to put our type checker at ease.
vlan_id: VLAN_ID = current_sbp.vlan_id # type: ignore[assignment]
ipv4_address: IPv4AddressType = current_sbp.ipv4_address # type: ignore[assignment]
- ipv4_mask: IPV4Netmask = current_sbp.ipv4_mask # type: ignore[assignment]
+ ipv4_mask: IPv4Netmask = current_sbp.ipv4_mask # type: ignore[assignment]
ipv6_address: IPv6AddressType = current_sbp.ipv6_address # type: ignore[assignment]
- ipv6_mask: IPV6Netmask = current_sbp.ipv6_mask # type: ignore[assignment]
+ ipv6_mask: IPv6Netmask = current_sbp.ipv6_mask # type: ignore[assignment]
custom_firewall_filters: bool = current_sbp.custom_firewall_filters
v4_bfd_settings: BFDInputModel = BFDInputModel(
bfd_enabled=current_sbp.v4_bfd_settings.bfd_enabled,
diff --git a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
index ae26125f..6d6632b2 100644
--- a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
@@ -21,14 +21,11 @@ from gso.products.product_types.router import Router
from gso.products.product_types.switch import Switch
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name
-from gso.utils.types.ip_address import AddressSpace, IPv4NetworkType
def _initial_input_form_generator() -> FormGenerator:
class ImportLanSwitchInterconnect(FormPage):
lan_switch_interconnect_description: str
- lan_switch_interconnect_ip_network: IPv4NetworkType | None
- address_space: AddressSpace
minimum_links: int
router_side: LanSwitchInterconnectRouterSideImportModel
switch_side: LanSwitchInterconnectSwitchSideImportModel
@@ -51,16 +48,12 @@ def create_subscription(partner: str) -> State:
def initialize_subscription(
subscription: ImportedLanSwitchInterconnectInactive,
lan_switch_interconnect_description: str,
- lan_switch_interconnect_ip_network: IPv4NetworkType | None,
- address_space: AddressSpace,
minimum_links: int,
router_side: dict,
switch_side: dict,
) -> State:
"""Initialize the subscription using input data."""
subscription.lan_switch_interconnect.lan_switch_interconnect_description = lan_switch_interconnect_description
- subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = lan_switch_interconnect_ip_network
- subscription.lan_switch_interconnect.address_space = address_space
subscription.lan_switch_interconnect.minimum_links = minimum_links
router_block = Router.from_subscription(router_side.pop("node")).router
diff --git a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
index eeec84e1..4f92cc6d 100644
--- a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
@@ -1,5 +1,6 @@
"""A creation workflow for creating a new interconnect between a switch and a router."""
+from ipaddress import IPv4Address
from typing import Annotated
from uuid import uuid4
@@ -7,6 +8,7 @@ from annotated_types import Len
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, begin, done, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
@@ -19,12 +21,15 @@ 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.partners import get_partner_by_name
+from gso.settings import load_oss_params
from gso.utils.helpers import (
active_router_selector,
active_switch_selector,
available_interfaces_choices,
available_lags_choices,
+ generate_lan_switch_interconnect_subnet,
)
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import (
@@ -35,7 +40,6 @@ from gso.utils.types.interfaces import (
PhysicalPortCapacity,
validate_interface_names_are_unique,
)
-from gso.utils.types.ip_address import AddressSpace
from gso.utils.types.tt_number import TTNumber
from gso.workflows.shared import create_summary_form
@@ -48,7 +52,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
router_side: active_router_selector() # type: ignore[valid-type]
switch_side: active_switch_selector() # type: ignore[valid-type]
- address_space: AddressSpace
description: str
minimum_link_count: int
divider: Divider
@@ -107,7 +110,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
"partner",
"router_side",
"switch_side",
- "address_space",
"description",
"minimum_link_count",
"vlan_id",
@@ -133,7 +135,6 @@ def create_subscription(product: UUIDstr, partner: str) -> State:
def initialize_subscription(
subscription: LanSwitchInterconnectInactive,
description: str,
- address_space: AddressSpace,
minimum_link_count: int,
router_side: UUIDstr,
router_side_iface: JuniperPhyInterface,
@@ -144,7 +145,6 @@ def initialize_subscription(
) -> State:
"""Update the product model with all input from the operator."""
subscription.lan_switch_interconnect.lan_switch_interconnect_description = description
- subscription.lan_switch_interconnect.address_space = address_space
subscription.lan_switch_interconnect.minimum_links = minimum_link_count
subscription.lan_switch_interconnect.router_side.node = Router.from_subscription(router_side).router
subscription.lan_switch_interconnect.router_side.ae_iface = router_side_iface
@@ -162,6 +162,47 @@ 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`."""
+ 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)
+ 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}}
+
+
+@step("Register devices in IPAM")
+def register_dns_records_devices(
+ subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, str]
+) -> State:
+ """Register :term:`DNS` records for both switch and router side in :term:`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_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.
+
+ 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)
+
+ return {"ipam_registrations": {switch_hostname: switch_side_ip, router_hostname: router_side_ip}}
+
+
@workflow(
"Create LAN Switch Interconnect",
initial_input_form=wrap_create_initial_input_form(_initial_input_form),
@@ -174,6 +215,8 @@ def create_lan_switch_interconnect() -> StepList:
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
+ >> register_dns_records_network
+ >> register_dns_records_devices
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> done
diff --git a/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
index ac9aef3f..8b9549c0 100644
--- a/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
@@ -4,13 +4,15 @@ from orchestrator import begin, workflow
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, done
+from orchestrator.workflow import StepList, done, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic_forms.types import FormGenerator
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.types.tt_number import TTNumber
@@ -31,6 +33,25 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
return {"subscription": lan_switch_interconnect}
+@step("Release IPAM resources")
+def clean_up_ipam(subscription: LanSwitchInterconnect) -> None:
+ """Clear up :term:`IPAM` resources in :term:`DNS`.
+
+ First, remove :term:`DNS` records for both the router and switch side of the interconnect.
+ Second, remove the registered network for this interconnect.
+ """
+ delete_host_by_fqdn(subscription.lan_switch_interconnect.switch_side.switch.fqdn)
+ delete_host_by_fqdn(
+ f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+ f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+ )
+ delete_network(
+ generate_lan_switch_interconnect_subnet(
+ subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
+ )
+ )
+
+
@workflow(
"Terminate LAN Switch Interconnect",
initial_input_form=wrap_modify_initial_input_form(_input_form_generator),
@@ -42,6 +63,7 @@ def terminate_lan_switch_interconnect() -> StepList:
begin
>> store_process_subscription(Target.TERMINATE)
>> unsync
+ >> clean_up_ipam
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
>> done
diff --git a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
index 9e6cfdd3..fbb6886d 100644
--- a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
@@ -3,11 +3,39 @@
from typing import Any
from orchestrator.targets import Target
+from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, begin, done, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
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
+
+
+@step("Validate IPAM configuration")
+def validate_ipam(subscription: LanSwitchInterconnect) -> None:
+ """Validate that :term:`DNS` records are registered in :term:`IPAM` correctly."""
+ switch_fqdn = subscription.lan_switch_interconnect.switch_side.switch.fqdn
+ router_fqdn = (
+ f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+ f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+ )
+
+ for fqdn in [switch_fqdn, router_fqdn]:
+ host_record = find_host_by_fqdn(fqdn)
+ if not host_record or str(subscription.subscription_id) not in host_record.comment:
+ msg = "DNS record is incorrectly configured in IPAM, please investigate this manually!"
+ raise ProcessFailureError(msg)
+
+ lan_interconnect_network = generate_lan_switch_interconnect_subnet(
+ subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
+ )
+ network_record = find_network_by_cidr(lan_interconnect_network)
+ if not network_record or str(subscription.subscription_id) not in network_record.comment:
+ msg = "LAN Switch Interconnect network is incorrectly configured in IPAM, please investigate this manually!"
+ raise ProcessFailureError(msg)
@step("Check config for drift")
@@ -15,7 +43,14 @@ def validate_config(subscription: dict[str, Any]) -> LSOState:
"""Workflow step for running a playbook that checks whether config has drifted."""
return {
"playbook_name": "lan_switch_interconnect.yaml",
- "inventory": {"all": {"hosts": {None: None}}},
+ "inventory": {
+ "all": {
+ "hosts": {
+ subscription["lan_switch_interconnect"]["switch_side"]["switch"]["fqdn"]: None,
+ subscription["lan_switch_interconnect"]["router_side"]["node"]["router_fqdn"]: None,
+ }
+ }
+ },
"extra_vars": {
"subscription_json": subscription,
"verb": "deploy",
@@ -34,6 +69,7 @@ def validate_lan_switch_interconnect() -> StepList:
begin
>> store_process_subscription(Target.SYSTEM)
>> unsync
+ >> validate_ipam
>> anonymous_lso_interaction(validate_config)
>> resync
>> done
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 571e64b6..3a915979 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -30,7 +30,6 @@ from gso.products.product_types.site import Site
from gso.utils.helpers import generate_unique_vc_id, iso_from_ipv4
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import AddressSpace
##############
@@ -230,8 +229,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
def _lan_switch_interconnect_data(**kwargs):
lan_switch_interconnect_data = {
"lan_switch_interconnect_description": faker.sentence(),
- "lan_switch_interconnect_ip_network": str(faker.ipv4_network()),
- "address_space": AddressSpace.PUBLIC,
"minimum_links": 1,
"router_side": {
"node": router_subscription_factory(),
@@ -240,7 +237,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
{"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
for _ in range(2)
],
- "ipv4_address": faker.ipv4(),
},
"switch_side": {
"switch": switch_subscription_factory(),
@@ -249,7 +245,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
{"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
for _ in range(2)
],
- "ipv4_address": faker.ipv4(),
},
}
lan_switch_interconnect_data.update(**kwargs)
diff --git a/test/fixtures/lan_switch_interconnect_fixtures.py b/test/fixtures/lan_switch_interconnect_fixtures.py
index fe0cb87f..0ea52e20 100644
--- a/test/fixtures/lan_switch_interconnect_fixtures.py
+++ b/test/fixtures/lan_switch_interconnect_fixtures.py
@@ -18,7 +18,6 @@ from gso.products.product_types.lan_switch_interconnect import (
from gso.products.product_types.router import Router
from gso.products.product_types.switch import Switch
from gso.services.subscriptions import get_product_id_by_name
-from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
@pytest.fixture()
@@ -31,17 +30,13 @@ def lan_switch_interconnect_subscription_factory(
status: SubscriptionLifecycle | None = None,
start_date: str | None = "2024-01-01T10:20:30+01:02",
lan_switch_interconnect_description: str | None = None,
- lan_switch_interconnect_ip_network: IPv4NetworkType | None = None,
- address_space: AddressSpace | None = None,
minimum_links: int | None = None,
router_side_node: UUIDstr | None = None,
router_side_ae_iface: str | None = None,
router_side_ae_members: list[dict[str, str]] | None = None,
- router_side_ipv4_address: IPv4AddressType | None = None,
switch_side_switch: UUIDstr | None = None,
switch_side_ae_iface: str | None = None,
switch_side_ae_members: list[dict[str, str]] | None = None,
- switch_side_ipv4_address: IPv4AddressType | None = None,
*,
is_imported: bool | None = True,
) -> UUIDstr:
@@ -70,24 +65,18 @@ def lan_switch_interconnect_subscription_factory(
subscription.lan_switch_interconnect.lan_switch_interconnect_description = (
lan_switch_interconnect_description or faker.sentence()
)
- subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = (
- lan_switch_interconnect_ip_network or faker.ipv4_network()
- )
- subscription.lan_switch_interconnect.address_space = address_space or AddressSpace.PRIVATE
subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
uuid4(),
node=router_side_node or Router.from_subscription(router_subscription_factory()).router,
ae_iface=router_side_ae_iface or faker.network_interface(),
ae_members=router_side_ae_members,
- ipv4_address=router_side_ipv4_address or faker.ipv4(),
)
subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
uuid4(),
switch=switch_side_switch or Switch.from_subscription(switch_subscription_factory()).switch,
ae_iface=switch_side_ae_iface or faker.network_interface(),
ae_members=switch_side_ae_members,
- ipv4_address=switch_side_ipv4_address or faker.ipv4(),
)
subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
diff --git a/test/fixtures/site_fixtures.py b/test/fixtures/site_fixtures.py
index 6fa90418..d14025e9 100644
--- a/test/fixtures/site_fixtures.py
+++ b/test/fixtures/site_fixtures.py
@@ -51,7 +51,7 @@ def site_subscription_factory(faker, geant_partner):
site_subscription.site.site_latitude = site_latitude or str(faker.latitude())
site_subscription.site.site_longitude = site_longitude or str(faker.longitude())
site_subscription.site.site_bgp_community_id = site_bgp_community_id or faker.pyint()
- site_subscription.site.site_internal_id = site_internal_id or faker.pyint()
+ 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()
diff --git a/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
index 69349ae0..74e6e879 100644
--- a/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
@@ -3,7 +3,6 @@ from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnect
-from gso.utils.types.ip_address import AddressSpace
from test.workflows import (
assert_complete,
extract_state,
@@ -15,20 +14,16 @@ from test.workflows import (
def workflow_input_data(faker, router_subscription_factory, switch_subscription_factory):
return {
"lan_switch_interconnect_description": faker.sentence(),
- "lan_switch_interconnect_ip_network": faker.ipv4_network(),
- "address_space": AddressSpace.PUBLIC,
"minimum_links": 1,
"router_side": {
"node": router_subscription_factory(),
"ae_iface": faker.nokia_lag_interface_name(),
"ae_members": faker.link_members_nokia(),
- "ipv4_address": faker.ipv4(),
},
"switch_side": {
"switch": switch_subscription_factory(),
"ae_iface": faker.juniper_ae_interface_name(),
"ae_members": faker.link_members_juniper(),
- "ipv4_address": faker.ipv4(),
},
}
diff --git a/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
index 2c6657d2..9d69017f 100644
--- a/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py
@@ -6,7 +6,6 @@ from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
from gso.services.subscriptions import get_product_id_by_name
-from gso.utils.types.ip_address import AddressSpace
from test.services.conftest import MockedNetboxClient
from test.workflows import assert_complete, extract_state, run_workflow
@@ -28,47 +27,45 @@ def _netbox_client_mock():
@pytest.fixture()
def input_form_data(faker, router_subscription_factory, switch_subscription_factory):
- def _generate_form_data(address_space: AddressSpace):
- return [
- {
- "product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
- },
- {
- "tt_number": faker.tt_number(),
- "router_side": router_subscription_factory(),
- "switch_side": switch_subscription_factory(),
- "address_space": address_space,
- "description": faker.sentence(),
- "minimum_link_count": 2,
- "vlan_id": 111, # VLAN ID for new interconnections is always 111
- },
- {
- "router_side_iface": "lag-1",
- "router_side_ae_members": faker.link_members_nokia()[:2],
- },
- {
- "switch_side_iface": faker.network_interface(),
- "switch_side_ae_members": faker.link_members_juniper()[:2],
- },
- {},
- ]
+ return [
+ {
+ "product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
+ },
+ {
+ "tt_number": faker.tt_number(),
+ "router_side": router_subscription_factory(),
+ "switch_side": switch_subscription_factory(),
+ "description": faker.sentence(),
+ "minimum_link_count": 2,
+ "vlan_id": 111, # VLAN ID for new interconnections is always 111
+ },
+ {
+ "router_side_iface": "lag-1",
+ "router_side_ae_members": faker.link_members_nokia()[:2],
+ },
+ {
+ "switch_side_iface": faker.network_interface(),
+ "switch_side_ae_members": faker.link_members_juniper()[:2],
+ },
+ {},
+ ]
- return _generate_form_data
-
-@pytest.mark.parametrize("address_space", [AddressSpace.PRIVATE, AddressSpace.PUBLIC])
@pytest.mark.workflow()
+@patch("gso.services.infoblox.create_v4_network_by_ip")
+@patch("gso.services.infoblox.create_host_by_ip")
def test_create_lan_switch_interconnect_success(
- address_space,
+ mock_create_host,
+ mock_create_v4_network,
input_form_data,
_netbox_client_mock, # noqa: PT019
):
- initial_data = input_form_data(address_space)
-
- result, _, _ = run_workflow("create_lan_switch_interconnect", initial_data)
+ result, _, _ = run_workflow("create_lan_switch_interconnect", input_form_data)
assert_complete(result)
state = extract_state(result)
subscription_id = state["subscription_id"]
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
diff --git a/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
index 6d5ae538..9b69c628 100644
--- a/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
@@ -1,3 +1,5 @@
+from unittest.mock import patch
+
import pytest
from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
@@ -5,7 +7,11 @@ from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.workflow()
-def test_terminate_lan_switch_interconnect(lan_switch_interconnect_subscription_factory, faker):
+@patch("gso.services.infoblox.find_host_by_fqdn")
+@patch("gso.services.infoblox.delete_network")
+def test_terminate_lan_switch_interconnect(
+ mock_delete_network, mock_find_host, lan_switch_interconnect_subscription_factory, faker
+):
subscription_id = lan_switch_interconnect_subscription_factory()
initial_lan_switch_interconnect_data = [{"subscription_id": subscription_id}, {"tt_number": faker.tt_number()}]
result, _, _ = run_workflow("terminate_lan_switch_interconnect", initial_lan_switch_interconnect_data)
@@ -15,3 +21,5 @@ def test_terminate_lan_switch_interconnect(lan_switch_interconnect_subscription_
subscription_id = state["subscription_id"]
subscription = LanSwitchInterconnect.from_subscription(subscription_id)
assert subscription.status == "terminated"
+ assert mock_find_host.call_count == 2
+ assert mock_delete_network.call_count == 1
--
GitLab