From 69de1d55fa9c18012c19b8f275bbcf9161b3ef37 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Wed, 23 Oct 2024 14:43:14 +0200
Subject: [PATCH] Deduplicate fixtures
---
test/cli/test_imports.py | 8 +-
test/conftest.py | 2 +-
test/fixtures.py | 522 +-----------------
test/fixtures/__init__.py | 5 +-
test/fixtures/edge_port_fixtures.py | 9 +-
test/fixtures/iptrunk_fixtures.py | 9 +-
test/fixtures/office_router_fixtures.py | 4 +-
test/fixtures/opengear_fixtures.py | 4 +-
test/fixtures/router_fixtures.py | 125 +----
test/fixtures/site_fixtures.py | 4 +-
test/fixtures/super_pop_switch_fixtures.py | 4 +-
.../edge_port/test_create_edge_port.py | 5 +-
.../test_create_imported_edge_port.py | 5 +-
13 files changed, 54 insertions(+), 652 deletions(-)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index b0ec0adb..33ab32dd 100644
--- a/test/cli/test_imports.py
+++ b/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"])
diff --git a/test/conftest.py b/test/conftest.py
index 88e43569..98176865 100644
--- a/test/conftest.py
+++ b/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,
)
diff --git a/test/fixtures.py b/test/fixtures.py
index 6c392855..e2b905c1 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -1,4 +1,3 @@
-import ipaddress
from collections.abc import Generator
from typing import Any
from uuid import uuid4
@@ -6,462 +5,15 @@ from uuid import uuid4
import pytest
from orchestrator import step, workflow
from orchestrator.config.assignee import Assignee
-from orchestrator.db import (
- ProductTable,
- SubscriptionInstanceTable,
- SubscriptionInstanceValueTable,
- SubscriptionTable,
- db,
-)
-from orchestrator.domain import SubscriptionModel
-from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from orchestrator.utils.datetime import nowtz
+from orchestrator.types import UUIDstr
from orchestrator.workflow import done, init, inputstep
from pydantic_forms.core import FormPage
-from pydantic_forms.types import FormGenerator, SubscriptionMapping
+from pydantic_forms.types import FormGenerator
from pydantic_forms.validators import Choice
-from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import (
- IptrunkInterfaceBlock,
- IptrunkSideBlock,
- IptrunkType,
-)
-from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import SiteTier
-from gso.products.product_types.iptrunk import ImportedIptrunkInactive, IptrunkInactive
-from gso.products.product_types.office_router import ImportedOfficeRouterInactive, OfficeRouterInactive
-from gso.products.product_types.opengear import ImportedOpengearInactive, OpengearInactive
-from gso.products.product_types.router import ImportedRouterInactive, Router, RouterInactive
-from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInactive
-from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
-from gso.services import subscriptions
-from gso.utils.helpers import iso_from_ipv4
-from gso.utils.shared_enums import Vendor
-from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
from test.workflows import WorkflowInstanceForTests
-@pytest.fixture()
-def site_subscription_factory(faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- site_name=None,
- site_city=None,
- site_country=None,
- site_country_code=None,
- site_latitude=None,
- site_longitude=None,
- site_bgp_community_id=None,
- site_internal_id=None,
- site_tier=SiteTier.TIER1,
- site_ts_address=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 "Site Subscription"
- site_name = site_name or faker.site_name()
- site_city = site_city or faker.city()
- site_country = site_country or faker.country()
- site_country_code = site_country_code or faker.country_code()
- site_latitude = site_latitude or str(faker.latitude())
- site_longitude = site_longitude or str(faker.longitude())
- site_bgp_community_id = site_bgp_community_id or faker.pyint()
- site_internal_id = site_internal_id or faker.pyint()
- site_ts_address = site_ts_address or faker.ipv4()
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.SITE)
- site_subscription = SiteInactive.from_product_id(product_id, customer_id=partner["partner_id"], insync=True)
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SITE)
- site_subscription = ImportedSiteInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- site_subscription.site.site_city = site_city
- site_subscription.site.site_name = site_name
- site_subscription.site.site_country = site_country
- site_subscription.site.site_country_code = site_country_code
- site_subscription.site.site_latitude = site_latitude
- site_subscription.site.site_longitude = site_longitude
- site_subscription.site.site_bgp_community_id = site_bgp_community_id
- site_subscription.site.site_internal_id = site_internal_id
- site_subscription.site.site_tier = site_tier
- site_subscription.site.site_ts_address = site_ts_address
-
- site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
- site_subscription.description = description
- site_subscription.start_date = start_date
- if status:
- site_subscription.status = status
-
- site_subscription.save()
- db.session.commit()
-
- return str(site_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture()
-def router_subscription_factory(site_subscription_factory, faker, geant_partner):
- def subscription_create(
- 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
- 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 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 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 or faker.text(max_nb_chars=30)
- 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 iptrunk_side_subscription_factory(router_subscription_factory, faker):
- def subscription_create(
- iptrunk_side_node=None,
- iptrunk_side_ae_iface=None,
- iptrunk_side_ae_geant_a_sid=None,
- iptrunk_side_ae_members=None,
- iptrunk_side_ae_members_description=None,
- ) -> IptrunkSideBlock:
- iptrunk_side_node_id = iptrunk_side_node or router_subscription_factory()
- 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()
- iptrunk_side_ae_members = iptrunk_side_ae_members or [
- IptrunkInterfaceBlock.new(
- faker.uuid4(),
- interface_name=faker.network_interface(),
- interface_description=faker.sentence(),
- ),
- IptrunkInterfaceBlock.new(
- faker.uuid4(),
- interface_name=faker.network_interface(),
- interface_description=faker.sentence(),
- ),
- ]
-
- return IptrunkSideBlock.new(
- faker.uuid4(),
- iptrunk_side_node=iptrunk_side_node,
- iptrunk_side_ae_iface=iptrunk_side_ae_iface,
- iptrunk_side_ae_geant_a_sid=iptrunk_side_ae_geant_a_sid,
- iptrunk_side_ae_members=iptrunk_side_ae_members,
- iptrunk_side_ae_members_description=iptrunk_side_ae_members_description,
- )
-
- return subscription_create
-
-
-@pytest.fixture()
-def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- geant_s_sid=None,
- iptrunk_description=None,
- iptrunk_type=IptrunkType.LEASED,
- iptrunk_speed=PhysicalPortCapacity.ONE_GIGABIT_PER_SECOND,
- iptrunk_isis_metric=None,
- iptrunk_ipv4_network=None,
- iptrunk_ipv6_network=None,
- iptrunk_sides=None,
- status: SubscriptionLifecycle | None = None,
- partner: dict | None = None,
- *,
- is_imported: bool | None = True,
- ) -> UUIDstr:
- if partner is None:
- partner = geant_partner
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
- iptrunk_subscription = IptrunkInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IP_TRUNK)
- iptrunk_subscription = ImportedIptrunkInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- description = description or faker.sentence()
- geant_s_sid = geant_s_sid or faker.geant_sid()
- iptrunk_description = iptrunk_description or faker.sentence()
- iptrunk_isis_metric = iptrunk_isis_metric or faker.pyint()
- iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4_network(max_subnet=31)
- iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6_network(max_subnet=126)
- iptrunk_minimum_links = 1
- iptrunk_side_a = iptrunk_side_subscription_factory()
- iptrunk_side_b = iptrunk_side_subscription_factory()
- iptrunk_sides = iptrunk_sides or [iptrunk_side_a, iptrunk_side_b]
-
- iptrunk_subscription.iptrunk.geant_s_sid = geant_s_sid
- iptrunk_subscription.iptrunk.iptrunk_description = iptrunk_description
- iptrunk_subscription.iptrunk.iptrunk_type = iptrunk_type
- iptrunk_subscription.iptrunk.iptrunk_speed = iptrunk_speed
- iptrunk_subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
- iptrunk_subscription.iptrunk.iptrunk_isis_metric = iptrunk_isis_metric
- iptrunk_subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
- iptrunk_subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
- iptrunk_subscription.iptrunk.iptrunk_sides = iptrunk_sides
-
- iptrunk_subscription = SubscriptionModel.from_other_lifecycle(
- iptrunk_subscription,
- SubscriptionLifecycle.ACTIVE,
- )
-
- if status:
- iptrunk_subscription.status = status
-
- iptrunk_subscription.description = description
- iptrunk_subscription.start_date = start_date
- iptrunk_subscription.save()
- db.session.commit()
-
- return str(iptrunk_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture()
-def office_router_subscription_factory(site_subscription_factory, faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- office_router_fqdn=None,
- office_router_ts_port=None,
- office_router_lo_ipv4_address=None,
- office_router_lo_ipv6_address=None,
- office_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)
- office_router_fqdn = office_router_fqdn or faker.domain_name(levels=4)
- office_router_ts_port = office_router_ts_port or faker.random_int(min=1, max=49151)
- office_router_lo_ipv4_address = office_router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
- office_router_lo_ipv6_address = office_router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
- office_router_site = office_router_site or site_subscription_factory()
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.OFFICE_ROUTER)
- office_router_subscription = OfficeRouterInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OFFICE_ROUTER)
- office_router_subscription = ImportedOfficeRouterInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- office_router_subscription.office_router.office_router_fqdn = office_router_fqdn
- office_router_subscription.office_router.office_router_ts_port = office_router_ts_port
- office_router_subscription.office_router.office_router_lo_ipv4_address = office_router_lo_ipv4_address
- office_router_subscription.office_router.office_router_lo_ipv6_address = office_router_lo_ipv6_address
- office_router_subscription.office_router.office_router_site = Site.from_subscription(office_router_site).site
- office_router_subscription.office_router.vendor = Vendor.NOKIA
-
- office_router_subscription = SubscriptionModel.from_other_lifecycle(
- office_router_subscription, SubscriptionLifecycle.ACTIVE
- )
- office_router_subscription.description = description
- office_router_subscription.start_date = start_date
-
- if status:
- office_router_subscription.status = status
-
- office_router_subscription.save()
- db.session.commit()
-
- return str(office_router_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture()
-def super_pop_switch_subscription_factory(site_subscription_factory, faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- super_pop_switch_fqdn=None,
- super_pop_switch_ts_port=None,
- super_pop_switch_mgmt_ipv4_address=None,
- super_pop_switch_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)
- super_pop_switch_fqdn = super_pop_switch_fqdn or faker.domain_name(levels=4)
- super_pop_switch_ts_port = super_pop_switch_ts_port or faker.random_int(min=1, max=49151)
- super_pop_switch_mgmt_ipv4_address = super_pop_switch_mgmt_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
- super_pop_switch_site = super_pop_switch_site or site_subscription_factory()
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.SUPER_POP_SWITCH)
- super_pop_switch_subscription = SuperPopSwitchInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SUPER_POP_SWITCH)
- super_pop_switch_subscription = ImportedSuperPopSwitchInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- super_pop_switch_subscription.super_pop_switch.super_pop_switch_fqdn = super_pop_switch_fqdn
- super_pop_switch_subscription.super_pop_switch.super_pop_switch_ts_port = super_pop_switch_ts_port
- super_pop_switch_subscription.super_pop_switch.super_pop_switch_mgmt_ipv4_address = (
- super_pop_switch_mgmt_ipv4_address
- )
- super_pop_switch_subscription.super_pop_switch.super_pop_switch_site = Site.from_subscription(
- super_pop_switch_site
- ).site
- super_pop_switch_subscription.super_pop_switch.vendor = Vendor.NOKIA
-
- super_pop_switch_subscription = SubscriptionModel.from_other_lifecycle(
- super_pop_switch_subscription, SubscriptionLifecycle.ACTIVE
- )
- super_pop_switch_subscription.description = description
- super_pop_switch_subscription.start_date = start_date
-
- if status:
- super_pop_switch_subscription.status = status
-
- super_pop_switch_subscription.save()
- db.session.commit()
-
- return str(super_pop_switch_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture()
-def opengear_subscription_factory(site_subscription_factory, faker, geant_partner):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- opengear_site=None,
- opengear_hostname=None,
- opengear_wan_address=None,
- opengear_wan_netmask=None,
- opengear_wan_gateway=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)
- opengear_site = opengear_site or site_subscription_factory()
- opengear_hostname = opengear_hostname or faker.domain_name(levels=4)
- opengear_wan_address = opengear_wan_address or faker.ipv4()
- opengear_wan_netmask = opengear_wan_netmask or faker.ipv4()
- opengear_wan_gateway = opengear_wan_gateway or faker.ipv4()
-
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.OPENGEAR)
- opengear_subscription = OpengearInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_OPENGEAR)
- opengear_subscription = ImportedOpengearInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
-
- opengear_subscription.opengear.opengear_site = Site.from_subscription(opengear_site).site
- opengear_subscription.opengear.opengear_hostname = opengear_hostname
- opengear_subscription.opengear.opengear_wan_address = opengear_wan_address
- opengear_subscription.opengear.opengear_wan_netmask = opengear_wan_netmask
- opengear_subscription.opengear.opengear_wan_gateway = opengear_wan_gateway
-
- opengear_subscription = SubscriptionModel.from_other_lifecycle(
- opengear_subscription, SubscriptionLifecycle.ACTIVE
- )
- opengear_subscription.description = description
- opengear_subscription.start_date = start_date
-
- if status:
- opengear_subscription.status = status
-
- opengear_subscription.save()
- db.session.commit()
-
- return str(opengear_subscription.subscription_id)
-
- return subscription_create
-
-
@pytest.fixture()
def test_workflow(generic_subscription_1: UUIDstr, generic_product_type_1) -> Generator:
_, generic_product_one = generic_product_type_1
@@ -497,73 +49,3 @@ def test_workflow(generic_subscription_1: UUIDstr, generic_product_type_1) -> Ge
with WorkflowInstanceForTests(workflow_for_testing_processes_py, "workflow_for_testing_processes_py") as wf:
yield wf
-
-
-def create_subscription_for_mapping(
- product: ProductTable, mapping: SubscriptionMapping, values: dict[str, Any], **kwargs: Any
-) -> SubscriptionTable:
- """Create a subscription in the test coredb for the given subscription_mapping and values.
-
- This function handles optional resource types starting with a ? in the mapping not supplied in the values array.
-
- Args:
- product: the ProductTable to create a sub for
- mapping: the subscription_mapping belonging to that product
- values: a dictionary of keys from the sub_map and their corresponding test values
- kwargs: The rest of the arguments
-
- Returns: The conforming subscription.
- """
-
- def build_instance(name, value_mapping):
- block = product.find_block_by_name(name)
-
- def build_value(rt, value):
- resource_type = block.find_resource_type_by_name(rt)
- return SubscriptionInstanceValueTable(resource_type_id=resource_type.resource_type_id, value=value)
-
- return SubscriptionInstanceTable(
- product_block_id=block.product_block_id,
- values=[
- build_value(resource_type, values[value_key]) for (resource_type, value_key) in value_mapping.items()
- ],
- )
-
- # recreate the mapping: leave out the ?keys if no value supplied for them
- mapping = {
- name: [
- {
- **{k: value_map[k] for k in value_map if not value_map[k].startswith("?")},
- **{
- k: value_map[k][1:]
- for k in value_map
- if value_map[k].startswith("?") and value_map[k][1:] in values
- },
- }
- for value_map in mapping[name]
- ]
- for name in mapping
- }
-
- instances = [
- build_instance(name, value_mapping)
- for (name, value_mappings) in mapping.items()
- for value_mapping in value_mappings
- ]
-
- return create_subscription(instances=instances, product=product, **kwargs)
-
-
-def create_subscription(**kwargs):
- attrs = {
- "description": "A subscription.",
- "customer_id": kwargs.get("customer_id", "85938c4c-0a11-e511-80d0-005056956c1a"),
- "start_date": nowtz(),
- "status": "active",
- "insync": True,
- **kwargs,
- }
- o = SubscriptionTable(**attrs)
- db.session.add(o)
- db.session.commit()
- return o
diff --git a/test/fixtures/__init__.py b/test/fixtures/__init__.py
index 645e6d20..0b69c5fb 100644
--- a/test/fixtures/__init__.py
+++ b/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",
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index f689c191..669d42ff 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/test/fixtures/edge_port_fixtures.py
@@ -1,7 +1,5 @@
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(
diff --git a/test/fixtures/iptrunk_fixtures.py b/test/fixtures/iptrunk_fixtures.py
index f2251fc7..70ccdeb5 100644
--- a/test/fixtures/iptrunk_fixtures.py
+++ b/test/fixtures/iptrunk_fixtures.py
@@ -1,7 +1,5 @@
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()
diff --git a/test/fixtures/office_router_fixtures.py b/test/fixtures/office_router_fixtures.py
index 2f43fa6b..0c55ada0 100644
--- a/test/fixtures/office_router_fixtures.py
+++ b/test/fixtures/office_router_fixtures.py
@@ -1,9 +1,7 @@
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
diff --git a/test/fixtures/opengear_fixtures.py b/test/fixtures/opengear_fixtures.py
index d3cd83da..b6dbff50 100644
--- a/test/fixtures/opengear_fixtures.py
+++ b/test/fixtures/opengear_fixtures.py
@@ -1,7 +1,5 @@
import pytest
-from orchestrator.db import (
- db,
-)
+from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
diff --git a/test/fixtures/router_fixtures.py b/test/fixtures/router_fixtures.py
index d08dde35..04c77ced 100644
--- a/test/fixtures/router_fixtures.py
+++ b/test/fixtures/router_fixtures.py
@@ -1,9 +1,7 @@
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:
diff --git a/test/fixtures/site_fixtures.py b/test/fixtures/site_fixtures.py
index b9155a29..e0a3a0bd 100644
--- a/test/fixtures/site_fixtures.py
+++ b/test/fixtures/site_fixtures.py
@@ -1,7 +1,5 @@
import pytest
-from orchestrator.db import (
- db,
-)
+from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
diff --git a/test/fixtures/super_pop_switch_fixtures.py b/test/fixtures/super_pop_switch_fixtures.py
index 33459570..5350e2a7 100644
--- a/test/fixtures/super_pop_switch_fixtures.py
+++ b/test/fixtures/super_pop_switch_fixtures.py
@@ -1,9 +1,7 @@
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
diff --git a/test/workflows/edge_port/test_create_edge_port.py b/test/workflows/edge_port/test_create_edge_port.py
index d087134f..a9a47b8d 100644
--- a/test/workflows/edge_port/test_create_edge_port.py
+++ b/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(),
diff --git a/test/workflows/edge_port/test_create_imported_edge_port.py b/test/workflows/edge_port/test_create_imported_edge_port.py
index 8ebd12a1..c5590fe4 100644
--- a/test/workflows/edge_port/test_create_imported_edge_port.py
+++ b/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,
--
GitLab