diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index d3a076adf7ad5aae7044954fae24df3f51d86b05..2f860b7b1bac8ce0b083bbae1d6df0c04a2fbf16 100644
--- a/gso/products/__init__.py
+++ b/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):
diff --git a/test/conftest.py b/test/conftest.py
index d27d20d4be345b085598189156d82a0cd13659d7..ccae3c61668deace3f08dfbb4bdd25daadd03270 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -119,6 +119,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:
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index cfed2090e27de3c7cb0c9221c7fd10709a6db8c2..58d7220376541dc7c8acd83be7df1890fde9f5f9 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/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,
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/geant_ip_fixtures.py
index 1c349a21dccd67659ea5a45ca0e668bcfbd82e61..d024e4050a9c4cfed77c012b22c15b881abbc543 100644
--- a/test/fixtures/geant_ip_fixtures.py
+++ b/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
diff --git a/test/workflows/geant_ip/__init__.py b/test/workflows/geant_ip/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/test/workflows/geant_ip/test_create_geant_ip.py b/test/workflows/geant_ip/test_create_geant_ip.py
new file mode 100644
index 0000000000000000000000000000000000000000..56f63954b6bbacbe8fa0b7ea8881d39f444a3c7f
--- /dev/null
+++ b/test/workflows/geant_ip/test_create_geant_ip.py
@@ -0,0 +1,52 @@
+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)