From 77b6f3f6a48e5646ff987c1af8d105dc81c8b7ad Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Wed, 9 Oct 2024 15:06:57 +0200
Subject: [PATCH] =?UTF-8?q?Add=20unit=20tests=20for=20G=C3=89ANT=20IP=20im?=
=?UTF-8?q?port=20workflow.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.../geant_ip/create_imported_geant_ip.py | 4 +-
test/fixtures/geant_ip_fixtures.py | 20 ++++--
.../geant_ip/test_create_imported_geant_ip.py | 65 +++++++++++++++++++
.../geant_ip/test_import_geant_ip.py | 17 +++++
4 files changed, 96 insertions(+), 10 deletions(-)
create mode 100644 test/workflows/geant_ip/test_create_imported_geant_ip.py
create mode 100644 test/workflows/geant_ip/test_import_geant_ip.py
diff --git a/gso/workflows/geant_ip/create_imported_geant_ip.py b/gso/workflows/geant_ip/create_imported_geant_ip.py
index fd2cb77c..ef620e73 100644
--- a/gso/workflows/geant_ip/create_imported_geant_ip.py
+++ b/gso/workflows/geant_ip/create_imported_geant_ip.py
@@ -78,9 +78,7 @@ def initialize_subscription(subscription: ImportedGeantIPInactive, service_bindi
for service_binding_port in service_binding_ports:
edge_port_subscription = EdgePort.from_subscription(service_binding_port.pop("edge_port"))
bgp_peers = service_binding_port.pop("bgp_peers")
- sbp_bgp_session_list = [
- BGPSession.new(subscription_id=uuid4(), **session) for session in bgp_peers
- ]
+ sbp_bgp_session_list = [BGPSession.new(subscription_id=uuid4(), **session) for session in bgp_peers]
service_binding_port_subscription = ServiceBindingPortInactive.new(
subscription_id=uuid4(),
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/geant_ip_fixtures.py
index 084cc927..5e35a7c8 100644
--- a/test/fixtures/geant_ip_fixtures.py
+++ b/test/fixtures/geant_ip_fixtures.py
@@ -10,7 +10,7 @@ from gso.products import EdgePort, ProductName
from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
from gso.products.product_blocks.geant_ip import NRENAccessPort
from gso.products.product_blocks.service_binding_port import ServiceBindingPort
-from gso.products.product_types.geant_ip import GeantIPInactive
+from gso.products.product_types.geant_ip import GeantIPInactive, ImportedGeantIP
from gso.services import subscriptions
from gso.utils.shared_enums import APType, SBPType
from gso.utils.types.ip_address import IPAddress
@@ -113,14 +113,20 @@ def geant_ip_subscription_factory(
nren_ap_list: list[NRENAccessPort] | None = None,
start_date="2023-05-24T00:00:00+00:00",
status: SubscriptionLifecycle | None = None,
+ *,
+ is_imported=True,
) -> UUIDstr:
- product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
partner = partner or partner_factory()
-
- # Create GEANT IP subscription with product_id and partner details
- geant_ip_subscription = GeantIPInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
+ if is_imported:
+ product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
+ geant_ip_subscription = GeantIPInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ else:
+ product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
+ geant_ip_subscription = ImportedGeantIP.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
# Default nren_ap_list creation with primary and backup access ports
geant_ip_subscription.geant_ip.geant_ip_ap_list = nren_ap_list or [
diff --git a/test/workflows/geant_ip/test_create_imported_geant_ip.py b/test/workflows/geant_ip/test_create_imported_geant_ip.py
new file mode 100644
index 00000000..ffd975f2
--- /dev/null
+++ b/test/workflows/geant_ip/test_create_imported_geant_ip.py
@@ -0,0 +1,65 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ImportedGeantIP
+from gso.products.product_blocks.bgp_session import IPFamily
+from gso.utils.shared_enums import SBPType
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.fixture()
+def imported_geant_ip_creation_input_form_data(edge_port_subscription_factory, partner_factory, faker):
+ return {
+ "partner": partner_factory()["name"],
+ "service_binding_ports": [
+ {
+ "edge_port": edge_port_subscription_factory(),
+ "ap_type": "PRIMARY",
+ "geant_sid": faker.geant_sid(),
+ "sbp_type": SBPType.L3,
+ "is_tagged": faker.boolean(),
+ "vlan_id": faker.vlan_id(),
+ "ipv4_address": faker.ipv4(),
+ "ipv6_address": faker.ipv6(),
+ "custom_firewall_filters": faker.boolean(),
+ "bgp_peers": [
+ {
+ "bfd_enabled": faker.boolean(),
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": faker.boolean(),
+ "authentication_key": faker.password(),
+ "multipath_enabled": faker.boolean(),
+ "send_default_route": faker.boolean(),
+ "is_passive": faker.boolean(),
+ "peer_address": faker.ipv4(),
+ "families": [IPFamily.V4UNICAST, IPFamily.V4MULTICAST],
+ "is_multi_hop": faker.boolean(),
+ "rtbh_enabled": faker.boolean(),
+ },
+ {
+ "bfd_enabled": faker.boolean(),
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": faker.boolean(),
+ "authentication_key": faker.password(),
+ "multipath_enabled": faker.boolean(),
+ "send_default_route": faker.boolean(),
+ "is_passive": faker.boolean(),
+ "peer_address": faker.ipv6(),
+ "families": [IPFamily.V6UNICAST],
+ "is_multi_hop": faker.boolean(),
+ "rtbh_enabled": faker.boolean(),
+ },
+ ],
+ }
+ ],
+ }
+
+
+def test_create_imported_geant_ip_success(faker, imported_geant_ip_creation_input_form_data):
+ result, _, _ = run_workflow("create_imported_geant_ip", [imported_geant_ip_creation_input_form_data])
+ state = extract_state(result)
+ subscription = ImportedGeantIP.from_subscription(state["subscription_id"])
+ assert_complete(result)
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
diff --git a/test/workflows/geant_ip/test_import_geant_ip.py b/test/workflows/geant_ip/test_import_geant_ip.py
new file mode 100644
index 00000000..cac68c2f
--- /dev/null
+++ b/test/workflows/geant_ip/test_import_geant_ip.py
@@ -0,0 +1,17 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import GeantIP, ProductName
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_import_edge_port_success(geant_ip_subscription_factory):
+ imported_geant_ip = geant_ip_subscription_factory(is_imported=False)
+ result, _, _ = run_workflow("import_geant_ip", [{"subscription_id": imported_geant_ip}])
+ subscription = GeantIP.from_subscription(imported_geant_ip)
+
+ assert_complete(result)
+ assert subscription.product.name == ProductName.GEANT_IP
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.insync
--
GitLab