From 924c0dd10aaff6a24f43bbe07669b09e006856ec Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Wed, 8 May 2024 19:53:05 +0200
Subject: [PATCH] Add unit tests for importing Iptrunks
---
test/fixtures.py | 52 +++++++++++++------
.../iptrunk/test_create_imported_iptrunk.py | 45 ++++++++++++++++
test/workflows/iptrunk/test_import_iptrunk.py | 18 +++++++
3 files changed, 100 insertions(+), 15 deletions(-)
create mode 100644 test/workflows/iptrunk/test_create_imported_iptrunk.py
create mode 100644 test/workflows/iptrunk/test_import_iptrunk.py
diff --git a/test/fixtures.py b/test/fixtures.py
index 946bc3f8..3ce5bc71 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -23,11 +23,11 @@ from gso.products.product_blocks.iptrunk import (
)
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
-from gso.products.product_types.iptrunk import IptrunkInactive, ImportedIptrunkInactive
-from gso.products.product_types.office_router import OfficeRouterInactive, ImportedOfficeRouterInactive
-from gso.products.product_types.router import Router, RouterInactive, ImportedRouterInactive
+from gso.products.product_types.iptrunk import ImportedIptrunkInactive, IptrunkInactive
+from gso.products.product_types.office_router import ImportedOfficeRouterInactive, OfficeRouterInactive
+from gso.products.product_types.router import ImportedRouterInactive, Router, RouterInactive
from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInactive
-from gso.products.product_types.super_pop_switch import SuperPopSwitchInactive, ImportedSuperPopSwitchInactive
+from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
from test.workflows import WorkflowInstanceForTests
@@ -71,7 +71,9 @@ def site_subscription_factory(faker, geant_partner):
site_subscription = SiteInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SITE)
- site_subscription = ImportedSiteInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ site_subscription = ImportedSiteInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
site_subscription.site.site_city = site_city
site_subscription.site.site_name = site_name
@@ -129,10 +131,14 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa
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)
+ 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 = 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
@@ -190,10 +196,14 @@ def juniper_router_subscription_factory(site_subscription_factory, faker, geant_
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)
+ 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 = 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
@@ -280,10 +290,14 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
- iptrunk_subscription = IptrunkInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ iptrunk_subscription = IptrunkInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IP_TRUNK)
- iptrunk_subscription = ImportedIptrunkInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ iptrunk_subscription = ImportedIptrunkInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
description = description or faker.sentence()
geant_s_sid = geant_s_sid or faker.geant_sid()
@@ -350,10 +364,14 @@ def office_router_subscription_factory(site_subscription_factory, faker, geant_p
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.OFFICE_ROUTER)
- office_router_subscription = OfficeRouterInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ office_router_subscription = OfficeRouterInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OFFICE_ROUTER)
- office_router_subscription = ImportedOfficeRouterInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ office_router_subscription = ImportedOfficeRouterInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
office_router_subscription.office_router.office_router_fqdn = office_router_fqdn
office_router_subscription.office_router.office_router_ts_port = office_router_ts_port
@@ -403,10 +421,14 @@ def super_pop_switch_subscription_factory(site_subscription_factory, faker, gean
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.SUPER_POP_SWITCH)
- super_pop_switch_subscription = SuperPopSwitchInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ super_pop_switch_subscription = SuperPopSwitchInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
else:
product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SUPER_POP_SWITCH)
- super_pop_switch_subscription = ImportedSuperPopSwitchInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
+ super_pop_switch_subscription = ImportedSuperPopSwitchInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
super_pop_switch_subscription.super_pop_switch.super_pop_switch_fqdn = super_pop_switch_fqdn
super_pop_switch_subscription.super_pop_switch.super_pop_switch_ts_port = super_pop_switch_ts_port
diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py
new file mode 100644
index 00000000..4623fe89
--- /dev/null
+++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py
@@ -0,0 +1,45 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+from products import ProductName
+
+from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_types.iptrunk import ImportedIptrunk
+from test.workflows import (
+ assert_complete,
+ extract_state,
+ run_workflow,
+)
+
+
+@pytest.mark.workflow()
+def test_create_imported_iptrunk_success(faker, nokia_router_subscription_factory):
+ input_data = {
+ "partner": "GEANT",
+ "geant_s_sid": faker.geant_sid(),
+ "iptrunk_description": faker.sentence(),
+ "iptrunk_type": IptrunkType.DARK_FIBER,
+ "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_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_ae_iface": faker.network_interface(),
+ "side_b_ae_geant_a_sid": faker.geant_sid(),
+ "side_b_ae_members": [
+ {"interface_name": faker.network_interface(), "interface_description": faker.sentence()} for _ in range(3)
+ ],
+ "iptrunk_ipv4_network": faker.ipv4_network(max_subnet=31),
+ "iptrunk_ipv6_network": faker.ipv6_network(max_subnet=126),
+ }
+ result, _, _ = run_workflow("create_imported_iptrunk", [input_data])
+ state = extract_state(result)
+ imported_router = ImportedIptrunk.from_subscription(state["subscription_id"])
+
+ assert_complete(result)
+ assert imported_router.product.name == ProductName.IMPORTED_IP_TRUNK
+ assert imported_router.status == SubscriptionLifecycle.ACTIVE
diff --git a/test/workflows/iptrunk/test_import_iptrunk.py b/test/workflows/iptrunk/test_import_iptrunk.py
new file mode 100644
index 00000000..99cdbfd9
--- /dev/null
+++ b/test/workflows/iptrunk/test_import_iptrunk.py
@@ -0,0 +1,18 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.iptrunk import Iptrunk
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_import_iptrunk_success(iptrunk_subscription_factory):
+ imported_iptrunk = iptrunk_subscription_factory(is_imported=False)
+ result, _, _ = run_workflow("import_iptrunk", [{"subscription_id": imported_iptrunk}])
+ subscription = Iptrunk.from_subscription(imported_iptrunk)
+
+ assert_complete(result)
+ assert subscription.product.name == ProductName.IP_TRUNK
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.insync
--
GitLab