Deduplicate fixtures

test/cli/test_imports.py

@@ -201,10 +201,10 @@ def opengear_data(temp_file, faker, site_subscription_factory):
 
 
 @pytest.fixture()
-def edge_port_data(temp_file, faker, nokia_router_subscription_factory, partner_factory):
+def edge_port_data(temp_file, faker, router_subscription_factory, partner_factory):
     def _edge_port_data(**kwargs):
         edge_port_data = {
-            "node": Router.from_subscription(nokia_router_subscription_factory()).router.router_fqdn,
+            "node": Router.from_subscription(router_subscription_factory(vendor=Vendor.NOKIA)).router.router_fqdn,
             "service_type": EdgePortType.CUSTOMER,
             "speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
             "encapsulation": EncapsulationType.DOT1Q,
@@ -526,9 +526,9 @@ def test_import_edge_port_successful(mock_start_process, mock_sleep, edge_port_d
 @patch("gso.cli.imports.time.sleep")
 @patch("gso.cli.imports.start_process")
 def test_import_edge_port_with_invalid_router(
-    mock_start_process, mock_sleep, edge_port_data, capfd, nokia_router_subscription_factory
+    mock_start_process, mock_sleep, edge_port_data, capfd, router_subscription_factory
 ):
-    p_router = nokia_router_subscription_factory(router_role=RouterRole.P)
+    p_router = router_subscription_factory(vendor=Vendor.NOKIA, router_role=RouterRole.P)
     broken_data = edge_port_data(node=Router.from_subscription(p_router).router.router_fqdn)
     import_edge_port(broken_data["path"])
 

test/conftest.py

@@ -45,8 +45,8 @@ from test.fixtures import (  # noqa: F401
     office_router_subscription_factory,
     opengear_subscription_factory,
     router_subscription_factory,
-    site_subscription_factory,
     service_binding_port_factory,
+    site_subscription_factory,
     super_pop_switch_subscription_factory,
 )
 

test/fixtures/__init__.py

@@ -8,7 +8,7 @@ from test.fixtures.nren_l3_core_service_fixtures import (
 )
 from test.fixtures.office_router_fixtures import office_router_subscription_factory
 from test.fixtures.opengear_fixtures import opengear_subscription_factory
-from test.fixtures.router_fixtures import juniper_router_subscription_factory, nokia_router_subscription_factory
+from test.fixtures.router_fixtures import router_subscription_factory
 from test.fixtures.site_fixtures import site_subscription_factory
 from test.fixtures.super_pop_switch_fixtures import super_pop_switch_subscription_factory
 
@@ -17,12 +17,11 @@ __all__ = [
     "edge_port_subscription_factory",
     "iptrunk_side_subscription_factory",
     "iptrunk_subscription_factory",
-    "juniper_router_subscription_factory",
-    "nokia_router_subscription_factory",
     "nren_access_port_factory",
     "nren_l3_core_service_subscription_factory",
     "office_router_subscription_factory",
     "opengear_subscription_factory",
+    "router_subscription_factory",
     "service_binding_port_factory",
     "site_subscription_factory",
     "super_pop_switch_subscription_factory",

test/fixtures/edge_port_fixtures.py

 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
@@ -14,11 +12,12 @@ from gso.products.product_blocks.edge_port import (
 from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
 from gso.products.product_types.router import Router
 from gso.services import subscriptions
+from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 
 
 @pytest.fixture()
-def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscription_factory):
+def edge_port_subscription_factory(faker, partner_factory, router_subscription_factory):
     def subscription_create(
         description=None,
         partner: dict | None = None,
@@ -40,7 +39,7 @@ def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscrip
         is_imported=True,
     ) -> UUIDstr:
         partner = partner or partner_factory()
-        node = Router.from_subscription(nokia_router_subscription_factory()).router
+        node = Router.from_subscription(router_subscription_factory(vendor=Vendor.NOKIA)).router
         if is_imported:
             product_id = subscriptions.get_product_id_by_name(ProductName.EDGE_PORT)
             edge_port_subscription = EdgePortInactive.from_product_id(

test/fixtures/iptrunk_fixtures.py

 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
@@ -14,11 +12,12 @@ from gso.products.product_blocks.iptrunk import (
 from gso.products.product_types.iptrunk import ImportedIptrunkInactive, IptrunkInactive
 from gso.products.product_types.router import Router
 from gso.services import subscriptions
+from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 
 
 @pytest.fixture()
-def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
+def iptrunk_side_subscription_factory(router_subscription_factory, faker):
     def subscription_create(
         iptrunk_side_node=None,
         iptrunk_side_ae_iface=None,
@@ -26,7 +25,7 @@ def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
         iptrunk_side_ae_members=None,
         iptrunk_side_ae_members_description=None,
     ) -> IptrunkSideBlock:
-        iptrunk_side_node_id = iptrunk_side_node or nokia_router_subscription_factory()
+        iptrunk_side_node_id = iptrunk_side_node or router_subscription_factory(vendor=Vendor.NOKIA)
         iptrunk_side_node = Router.from_subscription(iptrunk_side_node_id).router
         iptrunk_side_ae_iface = iptrunk_side_ae_iface or faker.pystr()
         iptrunk_side_ae_geant_a_sid = iptrunk_side_ae_geant_a_sid or faker.geant_sid()

test/fixtures/office_router_fixtures.py

 import ipaddress
 
 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 

test/fixtures/opengear_fixtures.py

 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 

test/fixtures/router_fixtures.py

 import ipaddress
 
 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
@@ -12,39 +10,32 @@ from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import ImportedRouterInactive, RouterInactive
 from gso.products.product_types.site import Site
 from gso.services import subscriptions
+from gso.utils.helpers import iso_from_ipv4
 from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 
 
 @pytest.fixture()
-def nokia_router_subscription_factory(site_subscription_factory, faker, geant_partner):
+def router_subscription_factory(site_subscription_factory, faker, geant_partner):
     def subscription_create(
-        description=None,
-        start_date="2023-05-24T00:00:00+00:00",
-        router_fqdn=None,
-        router_ts_port=None,
-        router_access_via_ts=None,
-        router_lo_ipv4_address=None,
-        router_lo_ipv6_address=None,
-        router_lo_iso_address=None,
-        router_role=RouterRole.PE,
+        description: str | None = None,
+        start_date: str | None = "2023-05-24T00:00:00+00:00",
+        router_fqdn: str | None = None,
+        router_ts_port: int | None = None,
+        router_lo_ipv4_address: IPv4AddressType | None = None,
+        router_lo_ipv6_address: IPv6AddressType | None = None,
+        router_lo_iso_address: str | None = None,
+        router_role: RouterRole | None = RouterRole.PE,
         router_site=None,
         status: SubscriptionLifecycle | None = None,
         partner: dict | None = None,
+        vendor: Vendor | None = Vendor.NOKIA,
         *,
+        router_access_via_ts: bool | None = None,
         is_imported: bool | None = True,
     ) -> UUIDstr:
         if partner is None:
             partner = geant_partner
-
-        description = description or faker.text(max_nb_chars=30)
-        router_fqdn = router_fqdn or faker.domain_name(levels=4)
-        router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
-        router_access_via_ts = router_access_via_ts or faker.boolean()
-        router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
-        router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
-        router_lo_iso_address = router_lo_iso_address or faker.word()
-        router_site = router_site or site_subscription_factory()
-
         if is_imported:
             product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
             router_subscription = RouterInactive.from_product_id(
@@ -56,85 +47,23 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa
                 product_id, customer_id=partner["partner_id"], insync=True
             )
 
-        router_subscription.router.router_fqdn = router_fqdn
-        router_subscription.router.router_ts_port = router_ts_port
-        router_subscription.router.router_access_via_ts = router_access_via_ts
-        router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
-        router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
-        router_subscription.router.router_lo_iso_address = router_lo_iso_address
+        router_subscription.router.router_fqdn = router_fqdn or faker.domain_name(levels=4)
+        router_subscription.router.router_ts_port = router_ts_port or faker.port_number(is_user=True)
+        router_subscription.router.router_access_via_ts = router_access_via_ts or faker.boolean()
+        router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(
+            faker.ipv4()
+        )
+        router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(
+            faker.ipv6()
+        )
+        router_subscription.router.router_lo_iso_address = router_lo_iso_address or iso_from_ipv4(faker.ipv4())
         router_subscription.router.router_role = router_role
-        router_subscription.router.router_site = Site.from_subscription(router_site).site
-        router_subscription.router.vendor = Vendor.NOKIA
+        router_subscription.router.router_site = Site.from_subscription(router_site or site_subscription_factory()).site
+        router_subscription.router.vendor = vendor
 
         router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
         router_subscription.insync = True
-        router_subscription.description = description
-        router_subscription.start_date = start_date
-
-        if status:
-            router_subscription.status = status
-
-        router_subscription.save()
-        db.session.commit()
-
-        return str(router_subscription.subscription_id)
-
-    return subscription_create
-
-
-@pytest.fixture()
-def juniper_router_subscription_factory(site_subscription_factory, faker, geant_partner):
-    def subscription_create(
-        description=None,
-        start_date="2023-05-24T00:00:00+00:00",
-        router_fqdn=None,
-        router_ts_port=None,
-        router_access_via_ts=None,
-        router_lo_ipv4_address=None,
-        router_lo_ipv6_address=None,
-        router_lo_iso_address=None,
-        router_role=RouterRole.PE,
-        router_site=None,
-        status: SubscriptionLifecycle | None = None,
-        partner: dict | None = None,
-        *,
-        is_imported: bool | None = True,
-    ) -> UUIDstr:
-        if partner is None:
-            partner = geant_partner
-
-        description = description or faker.text(max_nb_chars=30)
-        router_fqdn = router_fqdn or faker.domain_name(levels=4)
-        router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
-        router_access_via_ts = router_access_via_ts or faker.boolean()
-        router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
-        router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
-        router_lo_iso_address = router_lo_iso_address or faker.word()
-        router_site = router_site or site_subscription_factory()
-
-        if is_imported:
-            product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
-            router_subscription = RouterInactive.from_product_id(
-                product_id, customer_id=partner["partner_id"], insync=True
-            )
-        else:
-            product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_ROUTER)
-            router_subscription = ImportedRouterInactive.from_product_id(
-                product_id, customer_id=partner["partner_id"], insync=True
-            )
-
-        router_subscription.router.router_fqdn = router_fqdn
-        router_subscription.router.router_ts_port = router_ts_port
-        router_subscription.router.router_access_via_ts = router_access_via_ts
-        router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
-        router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
-        router_subscription.router.router_lo_iso_address = router_lo_iso_address
-        router_subscription.router.router_role = router_role
-        router_subscription.router.router_site = Site.from_subscription(router_site).site
-        router_subscription.router.vendor = Vendor.JUNIPER
-
-        router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
-        router_subscription.description = description
+        router_subscription.description = description or faker.text(max_nb_chars=30)
         router_subscription.start_date = start_date
 
         if status:

test/fixtures/site_fixtures.py

 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 

test/fixtures/super_pop_switch_fixtures.py

 import ipaddress
 
 import pytest
-from orchestrator.db import (
-    db,
-)
+from orchestrator.db import db
 from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 

test/workflows/edge_port/test_create_edge_port.py

@@ -9,6 +9,7 @@ from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationTyp
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.router import Router
 from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 from test.services.conftest import MockedNetboxClient
 from test.workflows import (
@@ -43,10 +44,10 @@ def _netbox_client_mock():
 
 
 @pytest.fixture()
-def input_form_wizard_data(request, nokia_router_subscription_factory, partner_factory, faker):
+def input_form_wizard_data(request, router_subscription_factory, partner_factory, faker):
     create_edge_port_step = {
         "tt_number": faker.tt_number(),
-        "node": nokia_router_subscription_factory(),
+        "node": router_subscription_factory(vendor=Vendor.NOKIA),
         "partner": partner_factory(name="GAAR", email=faker.email())["partner_id"],
         "service_type": EdgePortType.PUBLIC,
         "geant_ga_id": faker.geant_gid(),

test/workflows/edge_port/test_create_imported_edge_port.py

@@ -3,14 +3,15 @@ from orchestrator.types import SubscriptionLifecycle
 
 from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_types.edge_port import ImportedEdgePort
+from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 from test.workflows import assert_complete, extract_state, run_workflow
 
 
 @pytest.fixture()
-def imported_edge_port_creation_input_form_data(nokia_router_subscription_factory, partner_factory, faker):
+def imported_edge_port_creation_input_form_data(router_subscription_factory, partner_factory, faker):
     return {
-        "node": nokia_router_subscription_factory(),
+        "node": router_subscription_factory(vendor=Vendor.NOKIA),
         "service_type": EdgePortType.CUSTOMER,
         "speed": PhysicalPortCapacity.TEN_GIGABIT_PER_SECOND,
         "encapsulation": EncapsulationType.DOT1Q,