From 5a97c6657e8c3e69743c789aa989efdfa01ddce3 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Mon, 21 Oct 2024 17:39:18 +0200
Subject: [PATCH] Update validate_iptrunk WF_USABLE_MAP
* Allow for iptrunk validation workflow to be run on those that are still provisioning
* Update test fixture for router to deduplicate code
---
gso/workflows/__init__.py | 1 +
test/api/test_subscriptions.py | 36 ++---
test/cli/conftest.py | 2 +-
test/cli/test_imports.py | 6 +-
test/conftest.py | 3 +-
test/fixtures.py | 125 ++++--------------
test/utils/test_helpers.py | 20 +--
.../iptrunk/test_create_imported_iptrunk.py | 6 +-
test/workflows/iptrunk/test_create_iptrunk.py | 8 +-
.../workflows/iptrunk/test_migrate_iptrunk.py | 19 ++-
.../iptrunk/test_modify_trunk_interface.py | 11 +-
.../iptrunk/test_terminate_iptrunk.py | 6 +-
.../iptrunk/test_validate_iptrunk.py | 59 +++------
test/workflows/router/test_activate_router.py | 4 +-
test/workflows/router/test_import_router.py | 4 +-
.../router/test_modify_connection_stratey.py | 4 +-
test/workflows/router/test_promote_p_to_pe.py | 35 +++--
.../router/test_redeploy_base_config.py | 4 +-
.../workflows/router/test_terminate_router.py | 16 +--
.../workflows/router/test_update_ibgp_mesh.py | 14 +-
test/workflows/router/test_validate_router.py | 9 +-
21 files changed, 161 insertions(+), 231 deletions(-)
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index f6a36f9b..d94174f2 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -22,6 +22,7 @@ WF_USABLE_MAP.update({
"terminate_iptrunk": ALL_ALIVE_STATES,
"promote_p_to_pe": [SubscriptionLifecycle.ACTIVE],
"validate_router": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+ "validate_iptrunk": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
})
# IP trunk workflows
diff --git a/test/api/test_subscriptions.py b/test/api/test_subscriptions.py
index 3e457a58..53de229d 100644
--- a/test/api/test_subscriptions.py
+++ b/test/api/test_subscriptions.py
@@ -1,15 +1,17 @@
from orchestrator.types import SubscriptionLifecycle
+from gso.utils.shared_enums import Vendor
+
ROUTER_SUBSCRIPTION_ENDPOINT = "/api/v1/subscriptions/routers"
DASHBOARD_DEVICES_ENDPOINT = "/api/v1/subscriptions/dashboard_devices"
-def test_router_subscriptions_endpoint_with_valid_api_key(test_client, nokia_router_subscription_factory):
- nokia_router_subscription_factory()
- nokia_router_subscription_factory()
- nokia_router_subscription_factory()
- nokia_router_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
- nokia_router_subscription_factory(status=SubscriptionLifecycle.INITIAL)
+def test_router_subscriptions_endpoint_with_valid_api_key(test_client, router_subscription_factory):
+ router_subscription_factory()
+ router_subscription_factory()
+ router_subscription_factory()
+ router_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
+ router_subscription_factory(status=SubscriptionLifecycle.INITIAL)
response = test_client.get(
ROUTER_SUBSCRIPTION_ENDPOINT, headers={"Authorization": "Bearer another_REALY_random_AND_3cure_T0keN"}
@@ -19,14 +21,14 @@ def test_router_subscriptions_endpoint_with_valid_api_key(test_client, nokia_rou
assert len(response.json()) == 3
-def test_router_subscriptions_endpoint_with_invalid_api_key(test_client, nokia_router_subscription_factory):
+def test_router_subscriptions_endpoint_with_invalid_api_key(test_client, router_subscription_factory):
response = test_client.get(ROUTER_SUBSCRIPTION_ENDPOINT, headers={"Authorization": "Bearer fake_invalid_api_key"})
assert response.status_code == 403
assert response.json() == {"detail": "Invalid API Key"}
-def test_router_subscriptions_endpoint_without_api_key(test_client, nokia_router_subscription_factory):
+def test_router_subscriptions_endpoint_without_api_key(test_client, router_subscription_factory):
response = test_client.get(ROUTER_SUBSCRIPTION_ENDPOINT)
assert response.status_code == 403
@@ -35,14 +37,18 @@ def test_router_subscriptions_endpoint_without_api_key(test_client, nokia_router
def test_dashboard_devices_endpoint_with_valid_api_key(
test_client,
- nokia_router_subscription_factory,
+ router_subscription_factory,
office_router_subscription_factory,
super_pop_switch_subscription_factory,
):
- nokia_router_subscription_factory(router_fqdn="mx1.ams.nl.geant.net")
- nokia_router_subscription_factory(router_fqdn="mx2.ams.nl.geant.net")
- nokia_router_subscription_factory(status=SubscriptionLifecycle.PROVISIONING, router_fqdn="mx3.ams.nl.geant.net")
- nokia_router_subscription_factory(status=SubscriptionLifecycle.INITIAL, router_fqdn="mx4.ams.nl.geant.net")
+ router_subscription_factory(router_fqdn="mx1.ams.nl.geant.net")
+ router_subscription_factory(router_fqdn="mx2.ams.nl.geant.net")
+ router_subscription_factory(
+ vendor=Vendor.NOKIA, status=SubscriptionLifecycle.PROVISIONING, router_fqdn="mx3.ams.nl.geant.net"
+ )
+ router_subscription_factory(
+ vendor=Vendor.NOKIA, status=SubscriptionLifecycle.INITIAL, router_fqdn="mx4.ams.nl.geant.net"
+ )
office_router_subscription_factory(office_router_fqdn="office1.ams.nl.geant.net")
office_router_subscription_factory(
office_router_fqdn="office2.ams.nl.geant.net", status=SubscriptionLifecycle.TERMINATED
@@ -75,14 +81,14 @@ def test_dashboard_devices_endpoint_with_valid_api_key(
]
-def test_dashboard_devices_endpoint_with_invalid_api_key(test_client, nokia_router_subscription_factory):
+def test_dashboard_devices_endpoint_with_invalid_api_key(test_client, router_subscription_factory):
response = test_client.get(DASHBOARD_DEVICES_ENDPOINT, headers={"Authorization": "Bearer fake_invalid_api_key"})
assert response.status_code == 403
assert response.json() == {"detail": "Invalid API Key"}
-def test_dashboard_devices_endpoint_without_api_key(test_client, nokia_router_subscription_factory):
+def test_dashboard_devices_endpoint_without_api_key(test_client, router_subscription_factory):
response = test_client.get(DASHBOARD_DEVICES_ENDPOINT)
assert response.status_code == 403
diff --git a/test/cli/conftest.py b/test/cli/conftest.py
index 92d726fa..031c5dc6 100644
--- a/test/cli/conftest.py
+++ b/test/cli/conftest.py
@@ -1,9 +1,9 @@
from test.fixtures import ( # noqa: F401
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
- nokia_router_subscription_factory,
office_router_subscription_factory,
opengear_subscription_factory,
+ router_subscription_factory,
site_subscription_factory,
super_pop_switch_subscription_factory,
)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index d794d0f0..98332386 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -30,7 +30,7 @@ def temp_file(tmp_path) -> Path:
@pytest.fixture()
-def iptrunk_data(temp_file, nokia_router_subscription_factory, faker) -> (Path, dict):
+def iptrunk_data(temp_file, router_subscription_factory, faker) -> (Path, dict):
def _iptrunk_data(
*,
ipv4_network=None,
@@ -44,8 +44,8 @@ def iptrunk_data(temp_file, nokia_router_subscription_factory, faker) -> (Path,
side_a_ae_name=None,
side_b_ae_name=None,
):
- router_side_a = nokia_router_subscription_factory()
- router_side_b = nokia_router_subscription_factory()
+ router_side_a = router_subscription_factory()
+ router_side_b = router_subscription_factory()
ipv4_network = ipv4_network or str(faker.ipv4_network(max_subnet=31))
ipv6_network = ipv6_network or str(faker.ipv6_network(max_subnet=126))
diff --git a/test/conftest.py b/test/conftest.py
index ea765207..1cf0d399 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -38,10 +38,9 @@ from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import ( # noqa: F401
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
- juniper_router_subscription_factory,
- nokia_router_subscription_factory,
office_router_subscription_factory,
opengear_subscription_factory,
+ router_subscription_factory,
site_subscription_factory,
super_pop_switch_subscription_factory,
test_workflow,
diff --git a/test/fixtures.py b/test/fixtures.py
index 0bb920c0..6c392855 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -36,8 +36,10 @@ from gso.products.product_types.router import ImportedRouterInactive, Router, Ro
from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInactive
from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
from gso.services import subscriptions
+from gso.utils.helpers import iso_from_ipv4
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
from test.workflows import WorkflowInstanceForTests
@@ -110,35 +112,26 @@ def site_subscription_factory(faker, geant_partner):
@pytest.fixture()
-def nokia_router_subscription_factory(site_subscription_factory, faker, geant_partner):
+def router_subscription_factory(site_subscription_factory, faker, geant_partner):
def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- router_fqdn=None,
- router_ts_port=None,
- router_access_via_ts=None,
- router_lo_ipv4_address=None,
- router_lo_ipv6_address=None,
- router_lo_iso_address=None,
- router_role=RouterRole.PE,
+ description: str | None = None,
+ start_date: str | None = "2023-05-24T00:00:00+00:00",
+ router_fqdn: str | None = None,
+ router_ts_port: int | None = None,
+ router_lo_ipv4_address: IPv4AddressType | None = None,
+ router_lo_ipv6_address: IPv6AddressType | None = None,
+ router_lo_iso_address: str | None = None,
+ router_role: RouterRole | None = RouterRole.PE,
router_site=None,
status: SubscriptionLifecycle | None = None,
partner: dict | None = None,
+ vendor: Vendor | None = Vendor.NOKIA,
*,
+ router_access_via_ts: bool | None = None,
is_imported: bool | None = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
-
- description = description or faker.text(max_nb_chars=30)
- router_fqdn = router_fqdn or faker.domain_name(levels=4)
- router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
- router_access_via_ts = router_access_via_ts or faker.boolean()
- 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_site = router_site or site_subscription_factory()
-
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
router_subscription = RouterInactive.from_product_id(
@@ -150,85 +143,23 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa
product_id, customer_id=partner["partner_id"], insync=True
)
- router_subscription.router.router_fqdn = router_fqdn
- router_subscription.router.router_ts_port = router_ts_port
- router_subscription.router.router_access_via_ts = router_access_via_ts
- 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_fqdn = router_fqdn or faker.domain_name(levels=4)
+ router_subscription.router.router_ts_port = router_ts_port or faker.port_number(is_user=True)
+ router_subscription.router.router_access_via_ts = router_access_via_ts or faker.boolean()
+ router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(
+ faker.ipv4()
+ )
+ router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(
+ faker.ipv6()
+ )
+ router_subscription.router.router_lo_iso_address = router_lo_iso_address or iso_from_ipv4(faker.ipv4())
router_subscription.router.router_role = router_role
- router_subscription.router.router_site = Site.from_subscription(router_site).site
- router_subscription.router.vendor = Vendor.NOKIA
+ router_subscription.router.router_site = Site.from_subscription(router_site or site_subscription_factory()).site
+ router_subscription.router.vendor = vendor
router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
router_subscription.insync = True
- router_subscription.description = description
- router_subscription.start_date = start_date
-
- if status:
- router_subscription.status = status
-
- router_subscription.save()
- db.session.commit()
-
- return str(router_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture()
-def juniper_router_subscription_factory(site_subscription_factory, faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- router_fqdn=None,
- router_ts_port=None,
- router_access_via_ts=None,
- router_lo_ipv4_address=None,
- router_lo_ipv6_address=None,
- router_lo_iso_address=None,
- router_role=RouterRole.PE,
- router_site=None,
- status: SubscriptionLifecycle | None = None,
- partner: dict | None = None,
- *,
- is_imported: bool | None = True,
- ) -> UUIDstr:
- if partner is None:
- partner = geant_partner
-
- description = description or faker.text(max_nb_chars=30)
- router_fqdn = router_fqdn or faker.domain_name(levels=4)
- router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
- router_access_via_ts = router_access_via_ts or faker.boolean()
- 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_site = router_site or site_subscription_factory()
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
- router_subscription = RouterInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
- router_subscription = ImportedRouterInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- router_subscription.router.router_fqdn = router_fqdn
- router_subscription.router.router_ts_port = router_ts_port
- router_subscription.router.router_access_via_ts = router_access_via_ts
- 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_role = router_role
- router_subscription.router.router_site = Site.from_subscription(router_site).site
- router_subscription.router.vendor = Vendor.JUNIPER
-
- router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
- router_subscription.description = description
+ router_subscription.description = description or faker.text(max_nb_chars=30)
router_subscription.start_date = start_date
if status:
@@ -243,7 +174,7 @@ def juniper_router_subscription_factory(site_subscription_factory, faker, geant_
@pytest.fixture()
-def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
+def iptrunk_side_subscription_factory(router_subscription_factory, faker):
def subscription_create(
iptrunk_side_node=None,
iptrunk_side_ae_iface=None,
@@ -251,7 +182,7 @@ def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
iptrunk_side_ae_members=None,
iptrunk_side_ae_members_description=None,
) -> IptrunkSideBlock:
- iptrunk_side_node_id = iptrunk_side_node or nokia_router_subscription_factory()
+ iptrunk_side_node_id = iptrunk_side_node or router_subscription_factory()
iptrunk_side_node = Router.from_subscription(iptrunk_side_node_id).router
iptrunk_side_ae_iface = iptrunk_side_ae_iface or faker.pystr()
iptrunk_side_ae_geant_a_sid = iptrunk_side_ae_geant_a_sid or faker.geant_sid()
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index 28779cd7..0df4ced9 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -99,9 +99,9 @@ def test_tt_number(generate_tt_numbers):
validate_tt_number(tt_number)
-def test_generate_inventory_for_active_routers_with_single_active_router(nokia_router_subscription_factory):
+def test_generate_inventory_for_active_routers_with_single_active_router(router_subscription_factory):
"""Test the generation of inventory for a single active P router."""
- router = Router.from_subscription(nokia_router_subscription_factory(router_role=RouterRole.P))
+ router = Router.from_subscription(router_subscription_factory(router_role=RouterRole.P))
expected_result = {
"all": {
"hosts": {
@@ -116,14 +116,14 @@ def test_generate_inventory_for_active_routers_with_single_active_router(nokia_r
assert generate_inventory_for_active_routers(RouterRole.P) == expected_result
-def test_generate_inventory_for_active_routers_with_multiple_routers(nokia_router_subscription_factory):
+def test_generate_inventory_for_active_routers_with_multiple_routers(router_subscription_factory):
"""Test the generation of inventory for multiple active P and PE routers"""
for _ in range(5):
- nokia_router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.P)
for _ in range(3):
- nokia_router_subscription_factory(router_role=RouterRole.PE)
- nokia_router_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
- nokia_router_subscription_factory(status=SubscriptionLifecycle.INITIAL)
+ router_subscription_factory(router_role=RouterRole.PE)
+ router_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
+ router_subscription_factory(status=SubscriptionLifecycle.INITIAL)
# Test the generation of inventory for multiple active P routers.
inventory = generate_inventory_for_active_routers(RouterRole.P)
assert len(inventory["all"]["hosts"]) == 5
@@ -131,11 +131,11 @@ def test_generate_inventory_for_active_routers_with_multiple_routers(nokia_route
assert len(inventory["all"]["hosts"]) == 3
-def test_generate_inventory_for_active_routers_with_excluded_router(nokia_router_subscription_factory):
+def test_generate_inventory_for_active_routers_with_excluded_router(router_subscription_factory):
"""Test the generation of inventory for active P routers with an excluded router."""
for _ in range(5):
- nokia_router_subscription_factory(router_role=RouterRole.P)
- router = nokia_router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.P)
+ router = router_subscription_factory(router_role=RouterRole.P)
excluded_routers = [Router.from_subscription(router).router.router_fqdn]
inventory = generate_inventory_for_active_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.
diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py
index d172dc8d..25d68d8f 100644
--- a/test/workflows/iptrunk/test_create_imported_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py
@@ -13,7 +13,7 @@ from test.workflows import (
@pytest.fixture()
-def workflow_input_data(faker, nokia_router_subscription_factory):
+def workflow_input_data(faker, router_subscription_factory):
return {
"partner": "GEANT",
"geant_s_sid": faker.geant_sid(),
@@ -22,13 +22,13 @@ def workflow_input_data(faker, nokia_router_subscription_factory):
"iptrunk_speed": PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND,
"iptrunk_minimum_links": 2,
"iptrunk_isis_metric": 10000,
- "side_a_node_id": nokia_router_subscription_factory(),
+ "side_a_node_id": router_subscription_factory(),
"side_a_ae_iface": faker.network_interface(),
"side_a_ae_geant_a_sid": faker.geant_sid(),
"side_a_ae_members": [
{"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(3)
],
- "side_b_node_id": nokia_router_subscription_factory(),
+ "side_b_node_id": router_subscription_factory(),
"side_b_ae_iface": faker.network_interface(),
"side_b_ae_geant_a_sid": faker.geant_sid(),
"side_b_ae_members": [
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 267cd382..6be42fae 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -47,16 +47,16 @@ def _netbox_client_mock():
@pytest.fixture()
-def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_router_subscription_factory, faker):
+def input_form_wizard_data(request, router_subscription_factory, faker):
vendor = getattr(request, "param", Vendor.NOKIA)
- router_side_a = nokia_router_subscription_factory()
+ router_side_a = router_subscription_factory()
# Set side b router to Juniper
if vendor == Vendor.JUNIPER:
- router_side_b = juniper_router_subscription_factory()
+ router_side_b = router_subscription_factory(vendor=Vendor.JUNIPER)
side_b_members = faker.link_members_juniper()[0:2]
else:
- router_side_b = nokia_router_subscription_factory()
+ router_side_b = router_subscription_factory()
side_b_members = [
{"interface_name": f"Interface{interface}", "interface_description": faker.sentence()}
for interface in range(2)
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
index 083d85d9..525ce49b 100644
--- a/test/workflows/iptrunk/test_migrate_iptrunk.py
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -25,8 +25,7 @@ def migrate_form_input(
request,
faker,
iptrunk_subscription_factory,
- juniper_router_subscription_factory,
- nokia_router_subscription_factory,
+ router_subscription_factory,
iptrunk_side_subscription_factory,
):
use_juniper = getattr(request, "param", UseJuniperSide.NONE)
@@ -37,31 +36,31 @@ def migrate_form_input(
# Nokia -> Juniper
product_id = iptrunk_subscription_factory()
old_subscription = Iptrunk.from_subscription(product_id)
- new_router = juniper_router_subscription_factory()
+ new_router = router_subscription_factory(vendor=Vendor.JUNIPER)
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
new_side_ae_members = new_side_ae_members_juniper
lag_name = "ae1"
elif use_juniper == UseJuniperSide.SIDE_B:
# Juniper -> Nokia
- old_side_a_node = juniper_router_subscription_factory()
+ old_side_a_node = router_subscription_factory(vendor=Vendor.JUNIPER)
old_side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=old_side_a_node)
- old_side_b_node = juniper_router_subscription_factory()
+ old_side_b_node = router_subscription_factory(vendor=Vendor.JUNIPER)
old_side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=old_side_b_node)
product_id = iptrunk_subscription_factory(iptrunk_sides=[old_side_a_node, old_side_b_node])
old_subscription = Iptrunk.from_subscription(product_id)
- new_router = nokia_router_subscription_factory()
+ new_router = router_subscription_factory()
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
new_side_ae_members = new_side_ae_members_nokia
lag_name = "lag-1"
elif use_juniper == UseJuniperSide.SIDE_BOTH:
# Juniper -> Juniper
- old_side_a_node = juniper_router_subscription_factory()
+ old_side_a_node = router_subscription_factory(vendor=Vendor.JUNIPER)
old_side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=old_side_a_node)
- old_side_b_node = juniper_router_subscription_factory()
+ old_side_b_node = router_subscription_factory(vendor=Vendor.JUNIPER)
old_side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=old_side_b_node)
product_id = iptrunk_subscription_factory(iptrunk_sides=[old_side_a_node, old_side_b_node])
old_subscription = Iptrunk.from_subscription(product_id)
- new_router = juniper_router_subscription_factory()
+ new_router = router_subscription_factory(vendor=Vendor.JUNIPER)
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
new_side_ae_members = new_side_ae_members_juniper
lag_name = "ae1"
@@ -69,7 +68,7 @@ def migrate_form_input(
# Nokia -> Nokia
product_id = iptrunk_subscription_factory()
old_subscription = Iptrunk.from_subscription(product_id)
- new_router = nokia_router_subscription_factory()
+ new_router = router_subscription_factory()
replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
new_side_ae_members = new_side_ae_members_nokia
lag_name = "lag-1"
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 85c395e8..359308cd 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -21,26 +21,25 @@ def input_form_iptrunk_data(
request,
faker,
iptrunk_subscription_factory,
- juniper_router_subscription_factory,
- nokia_router_subscription_factory,
+ router_subscription_factory,
iptrunk_side_subscription_factory,
):
use_juniper = getattr(request, "param", UseJuniperSide.NONE)
if use_juniper == UseJuniperSide.SIDE_A:
- side_node = juniper_router_subscription_factory()
+ side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node)
side_b_node = iptrunk_side_subscription_factory()
new_side_a_ae_members = faker.link_members_juniper()[0:2]
new_side_b_ae_members = faker.link_members_nokia()[0:2]
elif use_juniper == UseJuniperSide.SIDE_B:
- side_node = juniper_router_subscription_factory()
+ side_node = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory()
side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node)
new_side_a_ae_members = faker.link_members_nokia()[0:2]
new_side_b_ae_members = faker.link_members_juniper()[0:2]
elif use_juniper == UseJuniperSide.SIDE_BOTH:
- side_node_1 = juniper_router_subscription_factory()
- side_node_2 = juniper_router_subscription_factory()
+ side_node_1 = router_subscription_factory(vendor=Vendor.JUNIPER)
+ side_node_2 = router_subscription_factory(vendor=Vendor.JUNIPER)
side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_1)
side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_2)
new_side_a_ae_members = faker.link_members_juniper()[0:2]
diff --git a/test/workflows/iptrunk/test_terminate_iptrunk.py b/test/workflows/iptrunk/test_terminate_iptrunk.py
index 7319596f..c745a615 100644
--- a/test/workflows/iptrunk/test_terminate_iptrunk.py
+++ b/test/workflows/iptrunk/test_terminate_iptrunk.py
@@ -27,7 +27,7 @@ def test_successful_iptrunk_termination(
iptrunk_subscription_factory,
faker,
data_config_filename,
- nokia_router_subscription_factory,
+ router_subscription_factory,
):
# Set up mock return values
product_id = iptrunk_subscription_factory()
@@ -35,8 +35,8 @@ def test_successful_iptrunk_termination(
mocked_delete_interface.return_value = mocked_netbox.delete_interface()
mocked_free_interface.return_value = mocked_netbox.free_interface()
# Add two more routers to our fake network
- nokia_router_subscription_factory(router_role=RouterRole.P)
- nokia_router_subscription_factory(router_role=RouterRole.PE)
+ router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.PE)
# Run workflow
oss_params = load_oss_params()
diff --git a/test/workflows/iptrunk/test_validate_iptrunk.py b/test/workflows/iptrunk/test_validate_iptrunk.py
index 6b132abc..88f998cd 100644
--- a/test/workflows/iptrunk/test_validate_iptrunk.py
+++ b/test/workflows/iptrunk/test_validate_iptrunk.py
@@ -2,9 +2,10 @@ from unittest.mock import patch
import pytest
from infoblox_client import objects
+from orchestrator.types import SubscriptionLifecycle
from gso.products.product_types.iptrunk import Iptrunk
-from test.conftest import UseJuniperSide
+from gso.utils.shared_enums import Vendor
from test.services.conftest import MockedNetboxClient
from test.workflows import (
assert_complete,
@@ -14,31 +15,6 @@ from test.workflows import (
)
-@pytest.fixture()
-def trunk_validation_setup(
- request,
- iptrunk_subscription_factory,
- juniper_router_subscription_factory,
- nokia_router_subscription_factory,
- iptrunk_side_subscription_factory,
-):
- if request.param == UseJuniperSide.SIDE_A:
- # Nokia -> Juniper
- side_a = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory())
- side_b = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory())
- elif request.param == UseJuniperSide.SIDE_B:
- # Juniper -> Nokia
- side_a = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory())
- side_b = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory())
- else:
- # Nokia -> Nokia
- side_a = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory())
- side_b = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory())
-
- subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b])
- return [{"subscription_id": subscription_id}]
-
-
@pytest.fixture()
def _mocked_netbox_client():
with (
@@ -61,9 +37,9 @@ def _mocked_netbox_client():
yield
-@pytest.mark.parametrize(
- "trunk_validation_setup", [UseJuniperSide.NONE, UseJuniperSide.SIDE_A, UseJuniperSide.SIDE_B], indirect=True
-)
+@pytest.mark.parametrize("side_a_vendor", [Vendor.JUNIPER, Vendor.NOKIA])
+@pytest.mark.parametrize("side_b_vendor", [Vendor.JUNIPER, Vendor.NOKIA])
+@pytest.mark.parametrize("subscription_status", [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE])
@pytest.mark.workflow()
@pytest.mark.usefixtures("_mocked_netbox_client")
@patch("gso.services.infoblox.find_network_by_cidr")
@@ -79,10 +55,17 @@ def test_validate_iptrunk_success(
mock_find_network_by_cidr,
faker,
data_config_filename,
- trunk_validation_setup,
+ iptrunk_subscription_factory,
+ iptrunk_side_subscription_factory,
+ router_subscription_factory,
+ side_a_vendor,
+ side_b_vendor,
+ subscription_status,
):
# Mock value setup
- subscription_id = trunk_validation_setup[0]["subscription_id"]
+ side_a = iptrunk_side_subscription_factory(iptrunk_side_node=router_subscription_factory(side_a_vendor))
+ side_b = iptrunk_side_subscription_factory(iptrunk_side_node=router_subscription_factory(side_b_vendor))
+ subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b], status=subscription_status)
trunk = Iptrunk.from_subscription(subscription_id).iptrunk
mock_find_network_by_cidr.side_effects = [
objects.Network(connector=None, ipv4addrs=[trunk.iptrunk_ipv4_network]),
@@ -207,13 +190,14 @@ def test_validate_iptrunk_success(
subscription_id = state["subscription_id"]
subscription = Iptrunk.from_subscription(subscription_id)
- assert subscription.status == "active"
+ assert subscription.status == subscription_status
assert mock_validate_iptrunk.call_count == 3
assert mock_find_host_by_fqdn.call_count == 2
assert mock_find_v6_host_by_fqdn.call_count == 2
assert mock_find_network_by_cidr.call_count == 2
+@pytest.mark.parametrize("subscription_status", [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE])
@pytest.mark.workflow()
@pytest.mark.usefixtures("_mocked_netbox_client")
@patch("gso.services.infoblox.find_network_by_cidr")
@@ -231,12 +215,13 @@ def test_validate_iptrunk_skip_legacy_trunks(
data_config_filename,
iptrunk_subscription_factory,
iptrunk_side_subscription_factory,
- juniper_router_subscription_factory,
+ router_subscription_factory,
+ subscription_status,
):
# Mock value setup
- side_a = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory())
- side_b = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory())
- subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b])
+ side_a = iptrunk_side_subscription_factory(iptrunk_side_node=router_subscription_factory(vendor=Vendor.JUNIPER))
+ side_b = iptrunk_side_subscription_factory(iptrunk_side_node=router_subscription_factory(vendor=Vendor.JUNIPER))
+ subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b], status=subscription_status)
# Run workflow
initial_router_data = [{"subscription_id": subscription_id}]
@@ -247,7 +232,7 @@ def test_validate_iptrunk_skip_legacy_trunks(
subscription_id = state["subscription_id"]
subscription = Iptrunk.from_subscription(subscription_id)
- assert subscription.status == "active"
+ assert subscription.status == subscription_status
assert mock_get_interface_by_name.call_count == 0
assert mock_validate_iptrunk.call_count == 0
assert mock_find_host_by_fqdn.call_count == 0
diff --git a/test/workflows/router/test_activate_router.py b/test/workflows/router/test_activate_router.py
index 2e60fd7c..da0c24a6 100644
--- a/test/workflows/router/test_activate_router.py
+++ b/test/workflows/router/test_activate_router.py
@@ -12,11 +12,11 @@ from test.workflows import (
@pytest.mark.workflow()
def test_activate_router_success(
- nokia_router_subscription_factory,
+ router_subscription_factory,
faker,
):
# Set up mock return values
- product_id = nokia_router_subscription_factory(status="provisioning")
+ product_id = router_subscription_factory(status="provisioning")
# Sanity check
assert Router.from_subscription(product_id).status == "provisioning"
diff --git a/test/workflows/router/test_import_router.py b/test/workflows/router/test_import_router.py
index 20938a7e..3c06b338 100644
--- a/test/workflows/router/test_import_router.py
+++ b/test/workflows/router/test_import_router.py
@@ -7,8 +7,8 @@ from test.workflows import assert_complete, run_workflow
@pytest.mark.workflow()
-def test_import_site_success(nokia_router_subscription_factory):
- imported_router = nokia_router_subscription_factory(is_imported=False)
+def test_import_site_success(router_subscription_factory):
+ imported_router = router_subscription_factory(is_imported=False)
result, _, _ = run_workflow("import_router", [{"subscription_id": imported_router}])
subscription = Router.from_subscription(imported_router)
diff --git a/test/workflows/router/test_modify_connection_stratey.py b/test/workflows/router/test_modify_connection_stratey.py
index 669ad10f..e460f2e9 100644
--- a/test/workflows/router/test_modify_connection_stratey.py
+++ b/test/workflows/router/test_modify_connection_stratey.py
@@ -6,8 +6,8 @@ from test.workflows import assert_complete, run_workflow
@pytest.mark.workflow()
-def test_modify_connection_strategy(responses, nokia_router_subscription_factory):
- subscription_id = nokia_router_subscription_factory(router_access_via_ts=True)
+def test_modify_connection_strategy(responses, router_subscription_factory):
+ subscription_id = router_subscription_factory(router_access_via_ts=True)
subscription = Router.from_subscription(subscription_id)
assert subscription.router.router_access_via_ts is True
form_data = [
diff --git a/test/workflows/router/test_promote_p_to_pe.py b/test/workflows/router/test_promote_p_to_pe.py
index 5e14dcaf..a0245726 100644
--- a/test/workflows/router/test_promote_p_to_pe.py
+++ b/test/workflows/router/test_promote_p_to_pe.py
@@ -5,6 +5,7 @@ from orchestrator.types import SubscriptionLifecycle
from pydantic_forms.exceptions import FormValidationError
from gso.products.product_blocks.router import RouterRole
+from gso.utils.shared_enums import Vendor
from test import USER_CONFIRM_EMPTY_FORM
from test.services.conftest import MockedKentikClient
from test.workflows import (
@@ -22,16 +23,18 @@ from test.workflows import (
def test_promote_p_to_pe_success(
mock_kentik_client,
mock_execute_playbook,
- nokia_router_subscription_factory,
+ router_subscription_factory,
data_config_filename,
faker,
):
"""Test the successful promotion of a Nokia P router to a PE router."""
mock_kentik_client.return_value = MockedKentikClient
- router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
+ router_id = router_subscription_factory(
+ vendor=Vendor.NOKIA, router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE
+ )
# Add two more routers to our fake network
- nokia_router_subscription_factory(router_role=RouterRole.P)
- nokia_router_subscription_factory(router_role=RouterRole.PE)
+ router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.PE)
input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
result, process_stat, step_log = run_workflow("promote_p_to_pe", input_data)
for _ in range(3):
@@ -46,9 +49,11 @@ def test_promote_p_to_pe_success(
@pytest.mark.workflow()
-def test_promote_p_to_pe_juniper_router(juniper_router_subscription_factory, data_config_filename, faker):
+def test_promote_p_to_pe_juniper_router(router_subscription_factory, data_config_filename, faker):
"""Test that the workflow does not run for a Juniper P router since this workflow is only for Nokia routers."""
- router_id = juniper_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
+ router_id = router_subscription_factory(
+ vendor=Vendor.JUNIPER, router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE
+ )
input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
with pytest.raises(FormValidationError) as error:
run_workflow("promote_p_to_pe", input_data)
@@ -60,10 +65,12 @@ def test_promote_p_to_pe_juniper_router(juniper_router_subscription_factory, dat
@pytest.mark.workflow()
@patch("gso.services.lso_client._send_request")
def test_promote_p_to_pe_nokia_pe_router(
- mock_execute_playbook, nokia_router_subscription_factory, data_config_filename, faker
+ mock_execute_playbook, router_subscription_factory, data_config_filename, faker
):
"""Test that the workflow does not run for a Nokia PE router since it is already a PE router."""
- router_id = nokia_router_subscription_factory(router_role=RouterRole.PE, status=SubscriptionLifecycle.ACTIVE)
+ router_id = router_subscription_factory(
+ vendor=Vendor.NOKIA, router_role=RouterRole.PE, status=SubscriptionLifecycle.ACTIVE
+ )
input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
with pytest.raises(FormValidationError) as error:
run_workflow("promote_p_to_pe", input_data)
@@ -72,9 +79,11 @@ def test_promote_p_to_pe_nokia_pe_router(
assert error["loc"][0] == "__root__"
-def test_promote_p_to_pe_missing_tt_number(nokia_router_subscription_factory):
+def test_promote_p_to_pe_missing_tt_number(router_subscription_factory):
"""Test that a missing TT number results in a validation error."""
- router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
+ router_id = router_subscription_factory(
+ vendor=Vendor.NOKIA, router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE
+ )
with pytest.raises(FormValidationError) as error:
run_workflow("promote_p_to_pe", [{"subscription_id": router_id}, {}])
error = error.value.errors[0]
@@ -82,9 +91,11 @@ def test_promote_p_to_pe_missing_tt_number(nokia_router_subscription_factory):
assert error["loc"][0] == "tt_number"
-def test_promote_p_to_pe_with_invalid_router_life_cycle(nokia_router_subscription_factory, faker):
+def test_promote_p_to_pe_with_invalid_router_life_cycle(router_subscription_factory, faker):
"""Test that the router life cycle must be ACTIVE to run this workflow."""
- router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.PROVISIONING)
+ router_id = router_subscription_factory(
+ vendor=Vendor.NOKIA, router_role=RouterRole.P, status=SubscriptionLifecycle.PROVISIONING
+ )
with pytest.raises(FormValidationError) as error:
run_workflow("promote_p_to_pe", [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}])
error = error.value.errors[0]
diff --git a/test/workflows/router/test_redeploy_base_config.py b/test/workflows/router/test_redeploy_base_config.py
index d624d62e..c9697cb7 100644
--- a/test/workflows/router/test_redeploy_base_config.py
+++ b/test/workflows/router/test_redeploy_base_config.py
@@ -15,11 +15,11 @@ from test.workflows import (
@patch("gso.services.lso_client._send_request")
def test_redeploy_base_config_success(
mock_provision_router,
- nokia_router_subscription_factory,
+ router_subscription_factory,
faker,
):
# Set up mock return values
- product_id = nokia_router_subscription_factory()
+ product_id = router_subscription_factory()
# Run workflow
initial_input_data = [{"subscription_id": product_id}, {"tt_number": faker.tt_number()}]
diff --git a/test/workflows/router/test_terminate_router.py b/test/workflows/router/test_terminate_router.py
index a5f4fd3d..a182cc74 100644
--- a/test/workflows/router/test_terminate_router.py
+++ b/test/workflows/router/test_terminate_router.py
@@ -24,15 +24,15 @@ def test_terminate_pe_router_full_success(
mock_execute_playbook,
remove_configuration,
update_ibgp_mesh,
- nokia_router_subscription_factory,
+ router_subscription_factory,
faker,
data_config_filename,
):
# Prepare mock values and expected results
- product_id = nokia_router_subscription_factory()
+ product_id = router_subscription_factory()
# Add two more routers to our fake network
- nokia_router_subscription_factory(router_role=RouterRole.P)
- nokia_router_subscription_factory(router_role=RouterRole.PE)
+ router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.PE)
router_termination_input_form_data = {
"tt_number": faker.tt_number(),
"remove_configuration": remove_configuration,
@@ -78,15 +78,15 @@ def test_terminate_p_router_full_success(
mock_execute_playbook,
remove_configuration,
update_ibgp_mesh,
- nokia_router_subscription_factory,
+ router_subscription_factory,
faker,
data_config_filename,
):
# Prepare mock values and expected results
- product_id = nokia_router_subscription_factory(router_role=RouterRole.P)
+ product_id = router_subscription_factory(router_role=RouterRole.P)
# Add two more routers to our fake network
- nokia_router_subscription_factory(router_role=RouterRole.P)
- nokia_router_subscription_factory(router_role=RouterRole.PE)
+ router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory(router_role=RouterRole.PE)
router_termination_input_form_data = {
"tt_number": faker.tt_number(),
"remove_configuration": remove_configuration,
diff --git a/test/workflows/router/test_update_ibgp_mesh.py b/test/workflows/router/test_update_ibgp_mesh.py
index e1ebb16f..023cd2cb 100644
--- a/test/workflows/router/test_update_ibgp_mesh.py
+++ b/test/workflows/router/test_update_ibgp_mesh.py
@@ -31,18 +31,16 @@ def test_update_ibgp_mesh_success(
trunk_status,
iptrunk_subscription_factory,
iptrunk_side_subscription_factory,
- nokia_router_subscription_factory,
+ router_subscription_factory,
data_config_filename,
faker,
):
mock_librenms_device_exists.return_value = False
- side_a = iptrunk_side_subscription_factory(
- iptrunk_side_node=nokia_router_subscription_factory(router_role=router_role)
- )
+ side_a = iptrunk_side_subscription_factory(iptrunk_side_node=router_subscription_factory(router_role=router_role))
side_b = iptrunk_side_subscription_factory()
# Add some extra devices to the network.
- nokia_router_subscription_factory(router_role=RouterRole.P)
- nokia_router_subscription_factory()
+ router_subscription_factory(router_role=RouterRole.P)
+ router_subscription_factory()
ip_trunk = Iptrunk.from_subscription(
iptrunk_subscription_factory(status=trunk_status, iptrunk_sides=[side_a, side_b])
@@ -85,8 +83,8 @@ def test_update_ibgp_mesh_failure(iptrunk_subscription_factory, data_config_file
@pytest.mark.workflow()
-def test_update_ibgp_mesh_isolated_router(nokia_router_subscription_factory, data_config_filename):
- router_id = nokia_router_subscription_factory(router_role=RouterRole.P)
+def test_update_ibgp_mesh_isolated_router(router_subscription_factory, data_config_filename):
+ router_id = router_subscription_factory(router_role=RouterRole.P)
exception_message = "Selected router does not terminate any available IP trunks."
with pytest.raises(FormValidationError, match=exception_message):
diff --git a/test/workflows/router/test_validate_router.py b/test/workflows/router/test_validate_router.py
index 22e7359a..a72063d9 100644
--- a/test/workflows/router/test_validate_router.py
+++ b/test/workflows/router/test_validate_router.py
@@ -5,6 +5,7 @@ from infoblox_client import objects
from orchestrator.types import SubscriptionLifecycle
from gso.products.product_types.router import Router
+from gso.utils.shared_enums import Vendor
from test.services.conftest import MockedKentikClient
from test.workflows import (
assert_complete,
@@ -27,7 +28,7 @@ def test_validate_nokia_router_success(
mock_get_device_by_name,
mock_execute_playbook,
mock_find_host_by_fqdn,
- nokia_router_subscription_factory,
+ router_subscription_factory,
faker,
data_config_filename,
geant_partner,
@@ -36,7 +37,7 @@ def test_validate_nokia_router_success(
mock_validate_librenms_device.return_value = None
mock_kentik_client.return_value = MockedKentikClient
# Run workflow
- subscription_id = nokia_router_subscription_factory(status=router_state)
+ subscription_id = router_subscription_factory(status=router_state)
mock_fqdn = Router.from_subscription(subscription_id).router.router_fqdn
mock_v4 = faker.ipv4()
mock_find_host_by_fqdn.return_value = objects.HostRecord(
@@ -79,13 +80,13 @@ def test_validate_nokia_router_success(
@pytest.mark.workflow()
def test_validate_juniper_router_success(
- juniper_router_subscription_factory,
+ router_subscription_factory,
faker,
data_config_filename,
geant_partner,
):
# Run workflow
- subscription_id = juniper_router_subscription_factory()
+ subscription_id = router_subscription_factory(vendor=Vendor.JUNIPER)
initial_router_data = [{"subscription_id": subscription_id}]
result, _, _ = run_workflow("validate_router", initial_router_data)
--
GitLab