Update GÉANT IP and Edge Port fixtures, add GÉANT IP creation unit test

gso/products/__init__.py

@@ -41,8 +41,8 @@ class ProductName(strEnum):
     IMPORTED_OPENGEAR = "Imported Opengear"
     EDGE_PORT = "Edge port"
     IMPORTED_EDGE_PORT = "Imported Edge port"
-    GEANT_IP = "GEANT IP"
-    IMPORTED_GEANT_IP = "Imported GEANT IP"
+    GEANT_IP = "GeantIP"
+    IMPORTED_GEANT_IP = "Imported GeantIP"
 
 
 class ProductType(strEnum):

test/conftest.py

@@ -126,6 +126,9 @@ class FakerProvider(BaseProvider):
             for i in range(iface_amount)
         ]
 
+    def vlan_id(self) -> int:
+        return self.generator.random_int(min=1, max=4096)
+
 
 @pytest.fixture(scope="session")
 def faker() -> Faker:

test/fixtures/edge_port_fixtures.py

@@ -32,7 +32,6 @@ def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscrip
         edge_port_type=EdgePortType.PUBLIC,
         geant_ga_id=None,
         edge_port_ae_members=None,
-        edge_port_sbp_list=None,
         status: SubscriptionLifecycle | None = None,
         *,
         enable_lacp=True,
@@ -77,7 +76,6 @@ def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscrip
                 interface_description=faker.sentence(),
             ),
         ]
-        edge_port_subscription.edge_port.edge_port_sbp_list = edge_port_sbp_list or []
         edge_port_subscription = SubscriptionModel.from_other_lifecycle(
             edge_port_subscription,
             SubscriptionLifecycle.ACTIVE,

test/fixtures/geant_ip_fixtures.py

@@ -49,7 +49,7 @@ def bgp_session_subscription_factory(faker):
 
 
 @pytest.fixture()
-def service_binding_port_factory(faker, bgp_session_subscription_factory):
+def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_port_subscription_factory):
     def create_service_binding_port(
         sbp_bgp_session_list: list | None = None,
         geant_sid: str | None = None,
@@ -57,6 +57,7 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory):
         ipv4_address: str | None = None,
         ipv6_address: str | None = None,
         vlan_id: int | None = None,
+        edge_port: EdgePort | None = None,
         *,
         custom_firewall_filters: bool = False,
         is_tagged: bool = False,
@@ -71,23 +72,18 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory):
             custom_firewall_filters=custom_firewall_filters,
             geant_sid=geant_sid or faker.geant_sid(),
             sbp_bgp_session_list=sbp_bgp_session_list or [bgp_session_subscription_factory() for _ in range(2)],
+            edge_port=edge_port or edge_port_subscription_factory(),
         )
 
     return create_service_binding_port
 
 
 @pytest.fixture()
-def nren_access_port_factory(faker, edge_port_subscription_factory):
-    def create_nren_access_port(
-        nren_ap_type: APType | None = None,
-        edge_port: UUIDstr | None = None,
-    ):
-        edge_port = edge_port or edge_port_subscription_factory()
-        geant_ip_ep = EdgePort.from_subscription(edge_port).edge_port
+def nren_access_port_factory(faker):
+    def create_nren_access_port(nren_ap_type: APType | None = None):
         return NRENAccessPort.new(
             subscription_id=uuid4(),
             nren_ap_type=nren_ap_type or random.choice(list(APType)),  # noqa: S311
-            geant_ip_ep=geant_ip_ep,
         )
 
     return create_nren_access_port

test/workflows/geant_ip/test_create_geant_ip.py

+import pytest
+
+from gso.products import ProductName
+from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.shared_enums import APType
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.fixture()
+def base_bgp_peer_input(faker):
+    def _base_bgp_peer_input():
+        bfd_enabled = faker.boolean()
+        return {
+            "bfd_enabled": bfd_enabled,
+            "bfd_interval": faker.pyint() if bfd_enabled else None,
+            "bfd_multiplier": faker.pyint() if bfd_enabled else None,
+            "has_custom_policies": faker.boolean(),
+            "authentication_key": faker.password(),
+            "multipath_enabled": faker.boolean(),
+            "send_default_route": faker.boolean(),
+            "is_passive": faker.boolean(),
+        }
+
+    return _base_bgp_peer_input
+
+
+@pytest.mark.workflow()
+def test_create_geant_ip_success(
+    responses, faker, partner_factory, edge_port_subscription_factory, base_bgp_peer_input
+):
+    partner = partner_factory(name=faker.company(), email=faker.email())
+    product_id = get_product_id_by_name(ProductName.GEANT_IP)
+    edge_port_a = edge_port_subscription_factory(partner=partner)
+
+    form_input_data = [
+        {"product": product_id},
+        {"tt_number": faker.tt_number(), "partner": partner["partner_id"]},
+        {"edge_ports": [{"edge_port": edge_port_a, "ap_type": APType.PRIMARY}]},
+        {
+            "geant_sid": faker.geant_sid(),
+            "is_tagged": faker.boolean(),
+            "vlan_id": faker.vlan_id(),
+            "ipv4_address": faker.ipv4(),
+            "ipv6_address": faker.ipv6(),
+            "custom_firewall_filters": faker.boolean(),
+            "v4_bgp_peer": base_bgp_peer_input() | {"add_v4_multicast": faker.boolean(), "peer_address": faker.ipv4()},
+            "v6_bgp_peer": base_bgp_peer_input() | {"add_v6_multicast": faker.boolean(), "peer_address": faker.ipv6()},
+        },
+    ]
+
+    result, process_stat, step_log = run_workflow("create_geant_ip", form_input_data)
+    assert_complete(result)