From 73df957c855b6bbe2947f3b73e0115fb10583db2 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Fri, 10 Nov 2023 10:07:49 +0100
Subject: [PATCH] update all workflows and associated tests with the router
remodel accordingly
---
gso/api/v1/imports.py | 4 --
gso/api/v1/subscriptions.py | 2 +-
gso/products/__init__.py | 9 +--
gso/services/subscriptions.py | 4 +-
gso/workflows/iptrunk/create_iptrunk.py | 4 +-
gso/workflows/iptrunk/migrate_iptrunk.py | 4 +-
.../iptrunk/modify_trunk_interface.py | 32 +++++-----
gso/workflows/router/create_router.py | 59 +------------------
gso/workflows/router/terminate_router.py | 22 +------
gso/workflows/tasks/import_router.py | 17 +-----
test/fixtures.py | 11 ----
test/imports/test_imports.py | 1 -
.../workflows/iptrunk/test_migrate_iptrunk.py | 2 +-
test/workflows/router/test_create_router.py | 24 +-------
.../workflows/router/test_terminate_router.py | 5 +-
tox.ini | 2 +-
16 files changed, 38 insertions(+), 164 deletions(-)
diff --git a/gso/api/v1/imports.py b/gso/api/v1/imports.py
index c54ebd6b8..b389025a6 100644
--- a/gso/api/v1/imports.py
+++ b/gso/api/v1/imports.py
@@ -78,13 +78,9 @@ class RouterImportModel(BaseModel):
ts_port: int
router_vendor: RouterVendor
router_role: RouterRole
- is_ias_connected: bool
router_lo_ipv4_address: ipaddress.IPv4Address
router_lo_ipv6_address: ipaddress.IPv6Address
router_lo_iso_address: str
- router_si_ipv4_network: ipaddress.IPv4Network | None = None
- router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
- router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
class IptrunkImportModel(BaseModel):
diff --git a/gso/api/v1/subscriptions.py b/gso/api/v1/subscriptions.py
index 65eae878f..dece88060 100644
--- a/gso/api/v1/subscriptions.py
+++ b/gso/api/v1/subscriptions.py
@@ -14,7 +14,7 @@ router = APIRouter(prefix="/subscriptions", tags=["Subscriptions"], dependencies
@router.get("/routers", status_code=status.HTTP_200_OK, response_model=list[SubscriptionDomainModelSchema])
def subscription_routers() -> list[dict[str, Any]]:
- """Retrieve all active routers subscriptions."""
+ """Retrieve all active router subscriptions."""
subscriptions = []
for r in get_active_router_subscriptions():
subscription = SubscriptionModel.from_subscription(r["subscription_id"])
diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index 1ac51f802..fe77c370e 100644
--- a/gso/products/__init__.py
+++ b/gso/products/__init__.py
@@ -14,11 +14,12 @@ from gso.products.product_types.site import Site
class ProductType(strEnum):
- SITE = "Site"
- NOKIA_ROUTER = "Nokia router"
- JUNIPER_ROUTER = "Juniper router"
- IP_TRUNK = "IP trunk"
IAS_GATEWAY = "IAS gateway"
+ IP_TRUNK = "IP trunk"
+ JUNIPER_ROUTER = "Juniper router"
+ NOKIA_ROUTER = "Nokia router"
+ ROUTER = "Router"
+ SITE = "Site"
SUBSCRIPTION_MODEL_REGISTRY.update(
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index f402ed0bc..0ad9c1b48 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -72,9 +72,7 @@ def get_active_router_subscriptions(includes: list[str] | None = None) -> list[S
:return: A list of Subscription objects for routers.
:rtype: list[Subscription]
"""
- active_nokia_routers = get_active_subscriptions(product_type=ProductType.NOKIA_ROUTER, includes=includes)
- active_juniper_routers = get_active_subscriptions(product_type=ProductType.JUNIPER_ROUTER, includes=includes)
- return active_nokia_routers + active_juniper_routers
+ return get_active_subscriptions(product_type="Router", includes=includes)
def get_product_id_by_name(product_name: ProductType) -> UUID:
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 0df2f4560..82d45edfd 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -279,7 +279,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
nbclient = NetboxClient()
for trunk_side in subscription.iptrunk.iptrunk_sides:
- if trunk_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+ if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
# Create LAG interfaces
lag_interface: Interfaces = nbclient.create_interface(
iface_name=trunk_side.iptrunk_side_ae_iface,
@@ -311,7 +311,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
def allocate_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
"""Allocate the LAG interfaces in NetBox and attach the lag interfaces to the physical interfaces."""
for trunk_side in subscription.iptrunk.iptrunk_sides:
- if trunk_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+ if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
for interface in trunk_side.iptrunk_side_ae_members:
NetboxClient().allocate_interface(
device_name=trunk_side.iptrunk_side_node.router_fqdn,
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 1287aca7f..a54ebec34 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -481,13 +481,13 @@ def update_netbox(
) -> State:
new_side = subscription.iptrunk.iptrunk_sides[replace_index]
nbclient = NetboxClient()
- if new_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+ if get_router_vendor(new_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
for interface in new_side.iptrunk_side_ae_members:
nbclient.allocate_interface(
device_name=new_side.iptrunk_side_node.router_fqdn,
iface_name=interface.interface_name,
)
- if old_side_data["iptrunk_side_node"]["router_vendor"] == RouterVendor.NOKIA:
+ if get_router_vendor(old_side_data["iptrunk_side_node"]["owner_subscription_id"]) == RouterVendor.NOKIA:
# Set interfaces to free
for iface in old_side_data["iptrunk_side_ae_members"]:
nbclient.free_interface(old_side_data["iptrunk_side_node"]["router_fqdn"], iface["interface_name"])
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 042ee163a..dd077e0ed 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -203,30 +203,30 @@ def provision_ip_trunk_iface_real(
@step("Update interfaces in Netbox. Reserving interfaces.")
def update_interfaces_in_netbox(subscription: Iptrunk, removed_ae_members: dict, previous_ae_members: dict) -> State:
nbclient = NetboxClient()
- for side in range(0, 2):
- if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
- lag_interface = subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface
- router_name = subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn
+ for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
+ if get_router_vendor(side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
+ lag_interface = side.iptrunk_side_ae_iface
+ router_name = side.iptrunk_side_node.router_fqdn
# Free removed interfaces
- for member in removed_ae_members[str(side)]:
+ for member in removed_ae_members[str(index)]:
nbclient.free_interface(router_name, member["interface_name"])
# Attach physical interfaces to LAG
# Update interface description to subscription ID
# Reserve interfaces
- for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
+ for interface in side.iptrunk_side_ae_members:
if any(
ae_member.get("interface_name") == interface.interface_name
- for ae_member in previous_ae_members[str(side)]
+ for ae_member in previous_ae_members[str(index)]
):
continue
nbclient.attach_interface_to_lag(
- device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+ device_name=side.iptrunk_side_node.router_fqdn,
lag_name=lag_interface,
iface_name=interface.interface_name,
description=str(subscription.subscription_id),
)
nbclient.reserve_interface(
- device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+ device_name=side.iptrunk_side_node.router_fqdn,
iface_name=interface.interface_name,
)
return {
@@ -241,23 +241,23 @@ def allocate_interfaces_in_netbox(subscription: Iptrunk, previous_ae_members: di
attach the lag interfaces to the physical interfaces detach old ones from the LAG.
"""
- for side in range(0, 2):
+ for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
nbclient = NetboxClient()
- if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
- for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
+ if get_router_vendor(side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
+ for interface in side.iptrunk_side_ae_members:
if any(
ae_member.get("interface_name") == interface.interface_name
- for ae_member in previous_ae_members[str(side)]
+ for ae_member in previous_ae_members[str(index)]
):
continue
nbclient.allocate_interface(
- device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+ device_name=side.iptrunk_side_node.router_fqdn,
iface_name=interface.interface_name,
)
# detach the old interfaces from lag
nbclient.detach_interfaces_from_lag(
- device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
- lag_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface,
+ device_name=side.iptrunk_side_node.router_fqdn,
+ lag_name=side.iptrunk_side_ae_iface,
)
return {"subscription": subscription}
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 5bc93583c..a09036da0 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -1,4 +1,3 @@
-from ipaddress import IPv4Network, IPv6Network
from typing import Any
# noinspection PyProtectedMember
@@ -6,7 +5,7 @@ from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, conditional, done, init, step, workflow
+from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from pydantic import validator
@@ -41,7 +40,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
hostname: str
ts_port: PortNumber
router_role: RouterRole
- is_ias_connected: bool | None = False
@validator("hostname", allow_reuse=True)
def hostname_must_be_available(cls, hostname: str, **kwargs: dict[str, Any]) -> str:
@@ -76,12 +74,10 @@ def initialize_subscription(
subscription: RouterInactive,
hostname: str,
ts_port: PortNumber,
- vendor: RouterVendor,
router_site: str,
router_role: RouterRole,
) -> State:
subscription.router.router_ts_port = ts_port
- subscription.vendor = vendor
subscription.router.router_site = Site.from_subscription(router_site).site
fqdn = generate_fqdn(
hostname, subscription.router.router_site.site_name, subscription.router.router_site.site_country_code
@@ -97,31 +93,13 @@ def initialize_subscription(
@step("Allocate loopback interfaces in IPAM")
-def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: bool) -> State:
+def ipam_allocate_loopback(subscription: RouterProvisioning) -> State:
fqdn = subscription.router.router_fqdn
loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
subscription.router.router_lo_ipv4_address = loopback_v4
subscription.router.router_lo_ipv6_address = loopback_v6
subscription.router.router_lo_iso_address = iso_from_ipv4(subscription.router.router_lo_ipv4_address)
- subscription.router.router_is_ias_connected = is_ias_connected
-
- return {"subscription": subscription}
-
-
-@step("Allocate IAS connection in IPAM")
-def ipam_allocate_ias_networks(subscription: RouterProvisioning) -> State:
- fqdn = subscription.router.router_fqdn
-
- subscription.router.router_si_ipv4_network = infoblox.allocate_v4_network(
- "SI", f"SI for {fqdn} - {subscription.subscription_id}"
- )
- subscription.router.router_ias_lt_ipv4_network = infoblox.allocate_v4_network(
- "LT_IAS", f"LT for {fqdn} - {subscription.subscription_id}"
- )
- subscription.router.router_ias_lt_ipv6_network = infoblox.allocate_v6_network(
- "LT_IAS", f"LT for {fqdn} - {subscription.subscription_id}"
- )
return {"subscription": subscription}
@@ -164,54 +142,21 @@ def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
return {"subscription": subscription}
-@step("Verify IPAM resources for IAS/LT networks")
-def verify_ipam_ias(subscription: RouterProvisioning) -> State:
- si_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_si_ipv4_network))
- ias_lt_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_ias_lt_ipv4_network))
- ias_lt_ipv6_network = infoblox.find_network_by_cidr(IPv6Network(subscription.router.router_ias_lt_ipv6_network))
-
- new_state = {}
-
- if not si_ipv4_network or str(subscription.subscription_id) not in si_ipv4_network.comment:
- new_state = {
- "ipam_si_warning": f"SI IPv4 network expected at {subscription.router.router_si_ipv4_network}, "
- f"but it was not found or misconfigured, please investigate and adjust if necessary."
- }
- if not ias_lt_ipv4_network or str(subscription.subscription_id) not in ias_lt_ipv4_network.comment:
- new_state = new_state | {
- "ipam_ias_lt_ipv4_warning": "IAS/LT IPv4 network expected at "
- f"{subscription.router.router_ias_lt_ipv4_network}, but it was not found or misconfigured, please "
- "investigate and adjust if necessary."
- }
- if not ias_lt_ipv6_network or str(subscription.subscription_id) not in ias_lt_ipv6_network.comment:
- new_state = new_state | {
- "ipam_ias_lt_ipv6_warning": f"IAS/LT IPv6 network expected at "
- f"{subscription.router.router_ias_lt_ipv6_network}, but it was not found or misconfigured, please "
- "investigate and adjust if necessary."
- }
-
- return new_state
-
-
@workflow(
"Create router",
initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
target=Target.CREATE,
)
def create_router() -> StepList:
- should_allocate_ias = conditional(lambda state: state["is_ias_connected"])
-
return (
init
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
>> ipam_allocate_loopback
- >> should_allocate_ias(ipam_allocate_ias_networks)
>> pp_interaction(provision_router_dry)
>> pp_interaction(provision_router_real)
>> verify_ipam_loopback
- >> should_allocate_ias(verify_ipam_ias)
>> create_netbox_device
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index f1a5e3107..a3d6c1a88 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -39,21 +39,6 @@ def deprovision_loopback_ips(subscription: Router) -> dict:
return {"subscription": subscription}
-@step("Deprovision SI interface network from IPAM")
-def deprovision_si_ips(subscription: Router) -> dict:
- infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_si_ipv4_network))
-
- return {"subscription": subscription}
-
-
-@step("Deprovision IAS LT interfaces from IPAM")
-def deprovision_lt_ips(subscription: Router) -> dict:
- infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_ias_lt_ipv4_network))
- infoblox.delete_network(ipaddress.IPv6Network(subscription.router.router_ias_lt_ipv6_network))
-
- return {"subscription": subscription}
-
-
@step("Remove configuration from router")
def remove_config_from_router() -> None:
# FIXME: Add actual content
@@ -76,17 +61,12 @@ def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
def terminate_router() -> StepList:
run_ipam_steps = conditional(lambda state: state["clean_up_ipam"])
run_config_steps = conditional(lambda state: state["remove_configuration"])
- run_ias_removal = conditional(lambda state: state["subscription"]["router"]["router_is_ias_connected"])
-
- ipam_steps = (
- init >> deprovision_loopback_ips >> run_ias_removal(deprovision_si_ips) >> run_ias_removal(deprovision_lt_ips)
- )
return (
init
>> store_process_subscription(Target.TERMINATE)
>> unsync
- >> run_ipam_steps(ipam_steps)
+ >> run_ipam_steps(deprovision_loopback_ips)
>> run_config_steps(remove_config_from_router)
>> remove_device_from_netbox
>> set_status(SubscriptionLifecycle.TERMINATED)
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index 59f0a19f4..a71a8ab17 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -35,10 +35,11 @@ def _get_site_by_name(site_name: str) -> Site:
@step("Create subscription")
def create_subscription(customer: str, router_vendor: RouterVendor) -> State:
customer_id = get_customer_by_name(customer)["id"]
+ product_id: UUID
if router_vendor == RouterVendor.NOKIA:
- product_id: UUID = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
+ product_id = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
elif router_vendor == RouterVendor.JUNIPER:
- product_id: UUID = subscriptions.get_product_id_by_name(ProductType.JUNIPER_ROUTER)
+ product_id = subscriptions.get_product_id_by_name(ProductType.JUNIPER_ROUTER)
else:
raise ValueError(f"Unknown router vendor: {router_vendor}")
@@ -61,13 +62,9 @@ def initial_input_form_generator() -> FormGenerator:
ts_port: int
router_vendor: RouterVendor
router_role: RouterRole
- is_ias_connected: bool
router_lo_ipv4_address: ipaddress.IPv4Address
router_lo_ipv6_address: ipaddress.IPv6Address
router_lo_iso_address: str
- router_si_ipv4_network: ipaddress.IPv4Network | None = None
- router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
- router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
user_input = yield ImportRouter
@@ -81,13 +78,9 @@ def initialize_subscription(
ts_port: PortNumber,
router_site: str,
router_role: router_pb.RouterRole,
- is_ias_connected: bool | None = None,
router_lo_ipv4_address: ipaddress.IPv4Address | None = None,
router_lo_ipv6_address: ipaddress.IPv6Address | None = None,
router_lo_iso_address: str | None = None,
- router_si_ipv4_network: ipaddress.IPv4Network | None = None,
- router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None,
- router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None,
) -> State:
subscription.router.router_ts_port = ts_port
subscription.router.router_site = _get_site_by_name(router_site).site
@@ -100,13 +93,9 @@ def initialize_subscription(
subscription.router.router_role = router_role
subscription.router.router_access_via_ts = True
subscription.description = f"Router {fqdn}"
- subscription.router.router_is_ias_connected = is_ias_connected
subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
subscription.router.router_lo_iso_address = router_lo_iso_address
- subscription.router.router_si_ipv4_network = router_si_ipv4_network
- subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
- subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
subscription = router.RouterProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
diff --git a/test/fixtures.py b/test/fixtures.py
index 061557ed5..72792d0e3 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -79,12 +79,8 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
router_lo_ipv4_address=None,
router_lo_ipv6_address=None,
router_lo_iso_address=None,
- router_si_ipv4_network=None,
- router_ias_lt_ipv4_network=None,
- router_ias_lt_ipv6_network=None,
router_role=RouterRole.PE,
router_site=None,
- router_is_ias_connected=True,
status: SubscriptionLifecycle | None = None,
) -> UUIDstr:
description = description or faker.text(max_nb_chars=30)
@@ -94,9 +90,6 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
router_lo_iso_address = router_lo_iso_address or faker.word()
- router_si_ipv4_network = router_si_ipv4_network or faker.ipv4_network()
- router_ias_lt_ipv4_network = router_ias_lt_ipv4_network or faker.ipv4_network()
- router_ias_lt_ipv6_network = router_ias_lt_ipv6_network or faker.ipv6_network()
router_site = router_site or site_subscription_factory()
product_id = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
@@ -107,12 +100,8 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
router_subscription.router.router_lo_iso_address = router_lo_iso_address
- router_subscription.router.router_si_ipv4_network = router_si_ipv4_network
- router_subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
- router_subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
router_subscription.router.router_role = router_role
router_subscription.router.router_site = Site.from_subscription(router_site).site
- router_subscription.router.router_is_ias_connected = router_is_ias_connected
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.description = description
diff --git a/test/imports/test_imports.py b/test/imports/test_imports.py
index 99015441d..84365a55d 100644
--- a/test/imports/test_imports.py
+++ b/test/imports/test_imports.py
@@ -107,7 +107,6 @@ def router_data(faker, site_data):
"router_site": site_data["site_name"],
"ts_port": 1234,
"customer": "GÉANT",
- "is_ias_connected": True,
"router_lo_ipv4_address": mock_ipv4,
"router_lo_ipv6_address": faker.ipv6(),
"router_lo_iso_address": iso_from_ipv4(mock_ipv4),
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
index 097641b92..12672252e 100644
--- a/test/workflows/iptrunk/test_migrate_iptrunk.py
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -40,7 +40,7 @@ def test_migrate_iptrunk_success(
mock_provision_ip_trunk,
mock_migrate_ip_trunk,
iptrunk_subscription_factory,
- nokia_router_subscription_factory,
+ nokia_router_subscription_factory,
faker,
data_config_filename: PathLike,
):
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
index 23a09ed74..7aeff06d2 100644
--- a/test/workflows/router/test_create_router.py
+++ b/test/workflows/router/test_create_router.py
@@ -5,7 +5,7 @@ from infoblox_client import objects
from gso.products import ProductType, Site
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_types.router import Router, RouterVendor
+from gso.products.product_types.router import Router
from gso.services.crm import customer_selector, get_customer_by_name
from gso.services.subscriptions import get_product_id_by_name
from test.workflows import (
@@ -27,9 +27,7 @@ def router_creation_input_form_data(site_subscription_factory, faker):
"router_site": router_site,
"hostname": faker.pystr(),
"ts_port": faker.pyint(),
- "vendor": RouterVendor.NOKIA,
"router_role": faker.random_choices(elements=(RouterRole.P, RouterRole.PE, RouterRole.AMT), length=1)[0],
- "is_ias_connected": True,
}
@@ -37,17 +35,11 @@ def router_creation_input_form_data(site_subscription_factory, faker):
@patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
@patch("gso.workflows.router.create_router.NetboxClient.create_device")
@patch("gso.workflows.router.create_router.infoblox.hostname_available")
-@patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
@patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
-@patch("gso.workflows.router.create_router.infoblox.allocate_v6_network")
-@patch("gso.workflows.router.create_router.infoblox.allocate_v4_network")
@patch("gso.workflows.router.create_router.infoblox.allocate_host")
def test_create_nokia_router_success(
mock_allocate_host,
- mock_allocate_v4_network,
- mock_allocate_v6_network,
mock_find_host_by_fqdn,
- mock_find_network_by_cidr,
mock_hostname_available,
mock_netbox_create_device,
mock_provision_router,
@@ -59,7 +51,6 @@ def test_create_nokia_router_success(
product_id = get_product_id_by_name(ProductType.NOKIA_ROUTER)
mock_site = Site.from_subscription(router_creation_input_form_data["router_site"]).site
mock_v4 = faker.ipv4()
- mock_v4_net = faker.ipv4_network()
mock_v6 = faker.ipv6()
mock_fqdn = (
f"{router_creation_input_form_data['hostname']}.{mock_site.site_name.lower()}."
@@ -67,8 +58,6 @@ def test_create_nokia_router_success(
)
mock_hostname_available.return_value = True
mock_allocate_host.return_value = str(mock_v4), str(mock_v6)
- mock_allocate_v4_network.return_value = mock_v4_net
- mock_allocate_v6_network.return_value = faker.ipv6_network()
# Run workflow
initial_router_data = [{"product": product_id}, router_creation_input_form_data]
@@ -91,13 +80,6 @@ def test_create_nokia_router_success(
],
name=mock_fqdn,
)
- mock_find_network_by_cidr.return_value = objects.NetworkV4(
- connector=None,
- comment=subscription_id,
- network=str(mock_v4_net),
- network_view="default",
- cidr=str(mock_v4_net),
- )
for _ in range(2):
result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
@@ -113,9 +95,7 @@ def test_create_nokia_router_success(
assert mock_provision_router.call_count == 2
assert mock_netbox_create_device.call_count == 1
assert mock_find_host_by_fqdn.call_count == 1
- assert mock_find_network_by_cidr.call_count == 3
- for error in ["ipam_warning", "ipam_si_warning", "ipam_ias_lt_ipv4_warning", "ipam_ias_lt_ipv6_warning"]:
- assert error not in state
+ assert "ipam_warning" not in state
@pytest.mark.workflow
diff --git a/test/workflows/router/test_terminate_router.py b/test/workflows/router/test_terminate_router.py
index 1b47d79fb..aa1049683 100644
--- a/test/workflows/router/test_terminate_router.py
+++ b/test/workflows/router/test_terminate_router.py
@@ -14,13 +14,11 @@ def router_termination_input_form_data(site_subscription_factory, faker):
@pytest.mark.workflow
@patch("gso.workflows.router.terminate_router.NetboxClient.delete_device")
@patch("gso.workflows.router.terminate_router.infoblox.delete_host_by_ip")
-@patch("gso.workflows.router.terminate_router.infoblox.delete_network")
def test_terminate_router_success(
- mock_delete_network,
mock_delete_host_by_ip,
mock_delete_device,
router_termination_input_form_data,
- nokia_router_subscription_factory,
+ nokia_router_subscription_factory,
faker,
data_config_filename,
):
@@ -37,6 +35,5 @@ def test_terminate_router_success(
subscription = Router.from_subscription(subscription_id)
assert "terminated" == subscription.status
- assert mock_delete_network.call_count == 3
assert mock_delete_device.call_count == 1
assert mock_delete_host_by_ip.call_count == 1
diff --git a/tox.ini b/tox.ini
index 791bcd09d..2b0bccf99 100644
--- a/tox.ini
+++ b/tox.ini
@@ -41,4 +41,4 @@ commands =
sh -c "if [ $SKIP_ALL_TESTS -eq 1 ]; then echo 'Skipping coverage report'; else coverage report --fail-under 80; fi"
allowlist_externals =
- sh
\ No newline at end of file
+ sh
--
GitLab