diff --git a/test/conftest.py b/test/conftest.py
index 5b1e91f17ff182286d05db81dddd1e9f2b73f891..d27d20d4be345b085598189156d82a0cd13659d7 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -36,13 +36,17 @@ from gso.main import init_gso_app
from gso.services.partners import PartnerSchema, create_partner
from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import ( # noqa: F401
+ bgp_session_subscription_factory,
edge_port_subscription_factory,
+ geant_ip_subscription_factory,
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
+ nren_access_port_factory,
office_router_subscription_factory,
opengear_subscription_factory,
router_subscription_factory,
site_subscription_factory,
+ service_binding_port_factory,
super_pop_switch_subscription_factory,
)
diff --git a/test/fixtures/__init__.py b/test/fixtures/__init__.py
index 2269a6b867b32a9adc6179b0f814e89b82cd2352..a5849b0d4e1023676adf6e87d2d34690579fb807 100644
--- a/test/fixtures/__init__.py
+++ b/test/fixtures/__init__.py
@@ -1,4 +1,10 @@
from test.fixtures.edge_port_fixtures import edge_port_subscription_factory
+from test.fixtures.geant_ip_fixtures import (
+ bgp_session_subscription_factory,
+ geant_ip_subscription_factory,
+ nren_access_port_factory,
+ service_binding_port_factory,
+)
from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
from test.fixtures.office_router_fixtures import office_router_subscription_factory
from test.fixtures.opengear_fixtures import opengear_subscription_factory
@@ -16,4 +22,8 @@ __all__ = [
"opengear_subscription_factory",
"site_subscription_factory",
"super_pop_switch_subscription_factory",
+ "geant_ip_subscription_factory",
+ "bgp_session_subscription_factory",
+ "service_binding_port_factory",
+ "nren_access_port_factory",
]
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/geant_ip_fixtures.py
new file mode 100644
index 0000000000000000000000000000000000000000..0a703a121ca6860235619dd3d74fb0a6282b11b4
--- /dev/null
+++ b/test/fixtures/geant_ip_fixtures.py
@@ -0,0 +1,153 @@
+import random
+from uuid import uuid4
+
+import pytest
+from orchestrator.db import db
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+
+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.services import subscriptions
+from gso.utils.shared_enums import APType, SBPType
+from gso.utils.types.ip_address import IPAddress
+
+
+@pytest.fixture()
+def bgp_session_subscription_factory(faker):
+ def create_bgp_session(
+ peer_address: IPAddress | None = None,
+ bfd_interval: int = 2,
+ bfd_multiplier: int = 2,
+ families: list[IPFamily] | None = None,
+ authentication_key: str | None = None,
+ *,
+ is_multi_hop: bool = False,
+ has_custom_policies: bool = False,
+ bfd_enabled: bool = True,
+ multipath_enabled: bool | None = True,
+ send_default_route: bool | None = True,
+ ):
+ return BGPSession.new(
+ subscription_id=uuid4(),
+ peer_address=peer_address or faker.ipv4(),
+ bfd_enabled=bfd_enabled,
+ families=families or [IPFamily.V4UNICAST],
+ has_custom_policies=has_custom_policies,
+ authentication_key=authentication_key or faker.password(),
+ multipath_enabled=multipath_enabled,
+ send_default_route=send_default_route,
+ is_multi_hop=is_multi_hop,
+ bfd_interval=bfd_interval,
+ bfd_multiplier=bfd_multiplier,
+ )
+
+ return create_bgp_session
+
+
+@pytest.fixture()
+def service_binding_port_factory(faker, bgp_session_subscription_factory):
+ def create_service_binding_port(
+ sbp_bgp_session_list: list | None = None,
+ geant_sid: str | None = None,
+ sbp_type: SBPType = SBPType.L3,
+ ipv4_address: str | None = None,
+ ipv6_address: str | None = None,
+ vlan_id: int | None = None,
+ *,
+ custom_firewall_filters: bool = False,
+ is_tagged: bool = False,
+ ):
+ return ServiceBindingPort.new(
+ subscription_id=uuid4(),
+ is_tagged=is_tagged,
+ vlan_id=vlan_id or faker.pyint(min_value=1, max_value=4096),
+ sbp_type=sbp_type,
+ ipv4_address=ipv4_address or faker.ipv4(),
+ ipv6_address=ipv6_address or faker.ipv6(),
+ 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)],
+ )
+
+ 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
+ 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
+
+
+@pytest.fixture()
+def geant_ip_subscription_factory(
+ faker,
+ partner_factory,
+ edge_port_subscription_factory,
+ bgp_session_subscription_factory,
+ service_binding_port_factory,
+ nren_access_port_factory,
+):
+ def create_geant_ip_subscription(
+ description=None,
+ partner: dict | None = None,
+ nren_ap_list: list[NRENAccessPort] | None = None,
+ start_date="2023-05-24T00:00:00+00:00",
+ status: SubscriptionLifecycle | None = None,
+ ) -> UUIDstr:
+ product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
+ partner = partner or partner_factory(name=faker.company(), email=faker.email())
+
+ # 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
+ )
+
+ # Default nren_ap_list creation with primary and backup access ports
+ nren_ap_list = nren_ap_list or [
+ nren_access_port_factory(nren_ap_type=APType.PRIMARY,
+ edge_port=edge_port_subscription_factory(partner=partner)),
+ nren_access_port_factory(nren_ap_type=APType.BACKUP,
+ edge_port=edge_port_subscription_factory(partner=partner)),
+ ]
+
+ # Assign and save edge port and service binding ports
+ for nren_ap in nren_ap_list:
+ edge_port = nren_ap.geant_ip_ep
+ edge_port.edge_port_sbp_list = [
+ service_binding_port_factory(),
+ service_binding_port_factory(),
+ ]
+
+ # Update subscription with description, start date, and status
+ geant_ip_subscription = SubscriptionModel.from_other_lifecycle(
+ geant_ip_subscription,
+ SubscriptionLifecycle.ACTIVE,
+ )
+
+ if status:
+ geant_ip_subscription.status = status
+ geant_ip_subscription.description = description or faker.text(max_nb_chars=30)
+ geant_ip_subscription.start_date = start_date
+ geant_ip_subscription.status = status or SubscriptionLifecycle.ACTIVE
+ geant_ip_subscription.save()
+
+ db.session.commit()
+
+ return str(geant_ip_subscription.subscription_id)
+
+ return create_geant_ip_subscription