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.types.interfaces import PhysicalPortCapacity
from gso.utils.shared_enums import Vendor
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 nokia_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.NOKIA
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.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(nokia_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 nokia_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