-
Karel van Klink authored
* Allow for iptrunk validation workflow to be run on those that are still provisioning * Update test fixture for router to deduplicate code
Karel van Klink authored* Allow for iptrunk validation workflow to be run on those that are still provisioning * Update test fixture for router to deduplicate code
fixtures.py 23.59 KiB
import ipaddress
from collections.abc import Generator
from typing import Any
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.workflow import done, init, inputstep
from pydantic_forms.core import FormPage
from pydantic_forms.types import FormGenerator, SubscriptionMapping
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
@step("Insert UUID in state")
def insert_object():
return {"subscription_id": str(uuid4()), "model": generic_product_one.from_subscription(generic_subscription_1)}
@step("Test that it is a string now")
def check_object(subscription_id: Any, model: dict) -> None:
# This is actually a test. It would be nicer to have this in a proper test but that takes to much setup that
# already happens here. So we hijack this fixture and run this test for all tests that use this fixture
# (which should not be an issue)
assert isinstance(subscription_id, str)
assert isinstance(model, dict)
@inputstep("Modify", assignee=Assignee.CHANGES)
def modify(subscription_id: UUIDstr) -> FormGenerator:
class TestChoice(Choice):
A = "A"
B = "B"
C = "C"
class TestForm(FormPage):
generic_select: TestChoice
user_input = yield TestForm
return user_input.model_dump()
@workflow("Workflow")
def workflow_for_testing_processes_py():
return init >> insert_object >> check_object >> modify >> done
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