From 223e6810f78809fd9c9f6a9041c86f18bf2e194b Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Fri, 6 Dec 2024 15:15:15 +0100
Subject: [PATCH] Reserve IPv6 resources when creating a LAN Switch
Interconnect
---
gso/services/infoblox.py | 35 +++++++-
gso/utils/helpers.py | 19 ++++-
gso/workflows/iptrunk/create_iptrunk.py | 4 +-
gso/workflows/iptrunk/migrate_iptrunk.py | 2 +-
.../create_lan_switch_interconnect.py | 80 ++++++++++++++-----
.../terminate_lan_switch_interconnect.py | 4 +-
.../validate_lan_switch_interconnect.py | 4 +-
test/cli/test_imports.py | 2 +-
test/fixtures/site_fixtures.py | 2 +
test/fixtures/switch_fixtures.py | 2 +-
test/utils/test_helpers.py | 25 ++++++
.../test_create_lan_switch_interconnect.py | 5 +-
.../site/test_create_imported_site.py | 1 +
.../switch/test_create_imported_switch.py | 2 +-
test/workflows/switch/test_create_switch.py | 2 +
15 files changed, 153 insertions(+), 36 deletions(-)
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index 1068be97..a0f5df25 100644
--- a/gso/services/infoblox.py
+++ b/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:
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 776d4629..aca14107 100644
--- a/gso/utils/helpers.py
+++ b/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,
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 909da8b2..cb37490e 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -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}
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 50dd2823..1282f636 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, "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}
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 4f92cc6d..15e14ef6 100644
--- a/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py
@@ -1,6 +1,6 @@
"""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
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 8b9549c0..5403d07f 100644
--- a/gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py
+++ b/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
)
)
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 c54d53c2..b97a601a 100644
--- a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
+++ b/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)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 3a915979..a1b0c757 100644
--- a/test/cli/test_imports.py
+++ b/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)
diff --git a/test/fixtures/site_fixtures.py b/test/fixtures/site_fixtures.py
index d14025e9..c6a884b1 100644
--- a/test/fixtures/site_fixtures.py
+++ b/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"
diff --git a/test/fixtures/switch_fixtures.py b/test/fixtures/switch_fixtures.py
index fa7f34d6..c0167a1f 100644
--- a/test/fixtures/switch_fixtures.py
+++ b/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
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index e9113378..6b62f2bb 100644
--- a/test/utils/test_helpers.py
+++ b/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"
+ )
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 9d69017f..a3511f21 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
@@ -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
diff --git a/test/workflows/site/test_create_imported_site.py b/test/workflows/site/test_create_imported_site.py
index 4e8b8e35..e63f44c6 100644
--- a/test/workflows/site/test_create_imported_site.py
+++ b/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",
}
diff --git a/test/workflows/switch/test_create_imported_switch.py b/test/workflows/switch/test_create_imported_switch.py
index 76ad61c7..b38a0c94 100644
--- a/test/workflows/switch/test_create_imported_switch.py
+++ b/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,
}
diff --git a/test/workflows/switch/test_create_switch.py b/test/workflows/switch/test_create_switch.py
index c11edf80..64006141 100644
--- a/test/workflows/switch/test_create_switch.py
+++ b/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,
},
{},
]
--
GitLab