Add unit tests for GÉANT IP import workflow.

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(),

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 [

test/workflows/geant_ip/test_create_imported_geant_ip.py

+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

test/workflows/geant_ip/test_import_geant_ip.py

+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