From 78f861459a8f28d443df0216e0c776459de6c4dc Mon Sep 17 00:00:00 2001
From: Mohammad Torkashvand <mohammad.torkashvand@geant.org>
Date: Fri, 22 Mar 2024 15:20:35 +0100
Subject: [PATCH] Implement API endpoint for retrieving FQDNs of monitored
dashboard devices
---
gso/api/v1/subscriptions.py | 43 ++++++-
gso/products/__init__.py | 24 ++--
gso/services/subscriptions.py | 55 ++++++---
gso/workflows/tasks/import_iptrunk.py | 4 +-
gso/workflows/tasks/import_office_router.py | 4 +-
gso/workflows/tasks/import_router.py | 4 +-
gso/workflows/tasks/import_site.py | 4 +-
.../tasks/import_super_pop_switch.py | 4 +-
test/api/conftest.py | 2 +
test/api/test_subscriptions.py | 57 +++++++++
test/fixtures.py | 115 +++++++++++++++++-
test/workflows/iptrunk/test_create_iptrunk.py | 8 +-
test/workflows/router/test_create_router.py | 6 +-
test/workflows/site/test_create_site.py | 6 +-
14 files changed, 285 insertions(+), 51 deletions(-)
diff --git a/gso/api/v1/subscriptions.py b/gso/api/v1/subscriptions.py
index 9cc6075a..bf4e96bf 100644
--- a/gso/api/v1/subscriptions.py
+++ b/gso/api/v1/subscriptions.py
@@ -2,14 +2,16 @@
from typing import Any
-from fastapi import Depends, status
+from fastapi import Depends, Response, status
from fastapi.routing import APIRouter
from orchestrator.domain import SubscriptionModel
from orchestrator.schemas import SubscriptionDomainModelSchema
from orchestrator.services.subscriptions import build_extended_domain_model
+from orchestrator.types import SubscriptionLifecycle
from gso.auth.api_key_auth import get_api_key
-from gso.services.subscriptions import get_active_router_subscriptions
+from gso.products import ProductType
+from gso.services.subscriptions import get_router_subscriptions, get_subscriptions
router = APIRouter(
prefix="/subscriptions",
@@ -24,11 +26,44 @@ router = APIRouter(
response_model=list[SubscriptionDomainModelSchema],
)
def subscription_routers() -> list[dict[str, Any]]:
- """Retrieve all active router subscriptions."""
+ """Retrieve all active or provisioning router subscriptions."""
subscriptions = []
- for r in get_active_router_subscriptions():
+ routers = get_router_subscriptions(lifecycles=[SubscriptionLifecycle.ACTIVE, SubscriptionLifecycle.PROVISIONING])
+ for r in routers:
subscription = SubscriptionModel.from_subscription(r["subscription_id"])
extended_model = build_extended_domain_model(subscription)
subscriptions.append(extended_model)
return subscriptions
+
+
+@router.get(
+ "/dashboard_devices",
+ status_code=status.HTTP_200_OK,
+ response_class=Response,
+ responses={
+ 200: {
+ "content": {"text/plain": {}},
+ "description": "Return a flat file of FQDNs.",
+ }
+ },
+)
+def subscription_dashboard_devices() -> Response:
+ """Retrieve FQDN for all dashboard devices that are monitored."""
+ fqdns = []
+ dashboard_devices = get_subscriptions(
+ product_types=[ProductType.ROUTER, ProductType.SUPER_POP_SWITCH, ProductType.OFFICE_ROUTER],
+ lifecycles=[SubscriptionLifecycle.ACTIVE, SubscriptionLifecycle.PROVISIONING],
+ )
+ for device in dashboard_devices:
+ subscription = SubscriptionModel.from_subscription(device["subscription_id"])
+ extended_model = build_extended_domain_model(subscription)
+ if extended_model["product"]["product_type"] == ProductType.ROUTER:
+ fqdns.append(extended_model["router"]["router_fqdn"])
+ elif extended_model["product"]["product_type"] == ProductType.SUPER_POP_SWITCH:
+ fqdns.append(extended_model["super_pop_switch"]["super_pop_switch_fqdn"])
+ elif extended_model["product"]["product_type"] == ProductType.OFFICE_ROUTER:
+ fqdns.append(extended_model["office_router"]["office_router_fqdn"])
+
+ fqdn_flat_file = "\n".join(fqdns)
+ return Response(content=fqdn_flat_file, media_type="text/plain")
diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index 2fd25d9f..32350d65 100644
--- a/gso/products/__init__.py
+++ b/gso/products/__init__.py
@@ -15,8 +15,8 @@ from gso.products.product_types.site import Site
from gso.products.product_types.super_pop_switch import SuperPopSwitch
-class ProductType(strEnum):
- """An enumerator of available products in :term:`GSO`."""
+class ProductName(strEnum):
+ """An enumerator of available product names in :term:`GSO`."""
IP_TRUNK = "IP trunk"
ROUTER = "Router"
@@ -25,12 +25,22 @@ class ProductType(strEnum):
OFFICE_ROUTER = "Office router"
+class ProductType(strEnum):
+ """An enumerator of available product types in :term:`GSO`."""
+
+ IP_TRUNK = Iptrunk.__name__
+ ROUTER = Router.__name__
+ SITE = Site.__name__
+ SUPER_POP_SWITCH = SuperPopSwitch.__name__
+ OFFICE_ROUTER = OfficeRouter.__name__
+
+
SUBSCRIPTION_MODEL_REGISTRY.update(
{
- "IP trunk": Iptrunk,
- "Router": Router,
- "Site": Site,
- "Super PoP switch": SuperPopSwitch,
- "Office router": OfficeRouter,
+ ProductName.IP_TRUNK.value: Iptrunk,
+ ProductName.ROUTER.value: Router,
+ ProductName.SITE.value: Site,
+ ProductName.SUPER_POP_SWITCH.value: SuperPopSwitch,
+ ProductName.OFFICE_ROUTER.value: OfficeRouter,
},
)
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index e89d9c25..d1e789aa 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -19,28 +19,31 @@ from orchestrator.services.subscriptions import query_in_use_by_subscriptions
from orchestrator.types import SubscriptionLifecycle
from pydantic_forms.types import UUIDstr
-from gso.products import ProductType
+from gso.products import ProductName, ProductType
from gso.products.product_types.site import Site
SubscriptionType = dict[str, Any]
def get_subscriptions(
- product_type: str,
- lifecycle: SubscriptionLifecycle,
+ product_types: list[ProductType],
+ lifecycles: list[SubscriptionLifecycle] | None = None,
includes: list[str] | None = None,
excludes: list[str] | None = None,
) -> list[SubscriptionType]:
"""Retrieve active subscriptions for a specific product type.
- :param str product_type: The type of the product for which to retrieve subscriptions.
- :param SubscriptionLifecycle lifecycle: The lifecycle that the products must be in.
+ :param list[ProductName] product_types: The types of the product for which to retrieve subscriptions.
+ :param SubscriptionLifecycle lifecycles: The lifecycles that the products must be in.
:param list[str] includes: List of fields to be included in the returned Subscription objects.
:param list[str] excludes: List of fields to be excluded from the returned Subscription objects.
:return: A list of Subscription objects that match the query.
:rtype: list[Subscription]
"""
+ if not lifecycles:
+ lifecycles = list(SubscriptionLifecycle)
+
if not includes:
includes = [col.name for col in SubscriptionTable.__table__.columns]
@@ -50,8 +53,8 @@ def get_subscriptions(
dynamic_fields = [getattr(SubscriptionTable, field) for field in includes]
query = SubscriptionTable.query.join(ProductTable).filter(
- ProductTable.product_type == product_type,
- SubscriptionTable.status == lifecycle,
+ ProductTable.product_type.in_([str(product_type) for product_type in product_types]),
+ SubscriptionTable.status.in_([str(lifecycle) for lifecycle in lifecycles]),
)
results = query.with_entities(*dynamic_fields).all()
@@ -68,7 +71,23 @@ def get_active_site_subscriptions(includes: list[str] | None = None) -> list[Sub
:return: A list of Subscription objects for sites.
:rtype: list[Subscription]
"""
- return get_subscriptions(product_type=ProductType.SITE, lifecycle=SubscriptionLifecycle.ACTIVE, includes=includes)
+ return get_subscriptions(
+ product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
+
+
+def get_router_subscriptions(
+ includes: list[str] | None = None, lifecycles: list[SubscriptionLifecycle] | None = None
+) -> list[SubscriptionType]:
+ """Retrieve subscriptions specifically for routers.
+
+ :param includes: The fields to be included in the returned Subscription objects.
+ :type includes: list[str]
+
+ :return: A list of Subscription objects for routers.
+ :rtype: list[Subscription]
+ """
+ return get_subscriptions(product_types=[ProductType.ROUTER], lifecycles=lifecycles, includes=includes)
def get_active_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
@@ -80,7 +99,9 @@ def get_active_router_subscriptions(includes: list[str] | None = None) -> list[S
:return: A list of Subscription objects for routers.
:rtype: list[Subscription]
"""
- return get_subscriptions(product_type="Router", lifecycle=SubscriptionLifecycle.ACTIVE, includes=includes)
+ return get_subscriptions(
+ product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
def get_provisioning_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
@@ -89,7 +110,9 @@ def get_provisioning_router_subscriptions(includes: list[str] | None = None) ->
:param list[str] includes: The fields to be included in the returned Subscription objects.
:return list[Subscription]: A list of router Subscription objects.
"""
- return get_subscriptions(product_type="Router", lifecycle=SubscriptionLifecycle.PROVISIONING, includes=includes)
+ return get_subscriptions(
+ product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.PROVISIONING], includes=includes
+ )
def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
@@ -101,7 +124,9 @@ def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[
:return: A list of Subscription objects for IP trunks.
:rtype: list[Subscription]
"""
- return get_subscriptions(product_type="Iptrunk", lifecycle=SubscriptionLifecycle.ACTIVE, includes=includes)
+ return get_subscriptions(
+ product_types=[ProductType.IP_TRUNK], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
def get_active_trunks_that_terminate_on_router(subscription_id: UUIDstr) -> list[SubscriptionTable]:
@@ -120,18 +145,18 @@ def get_active_trunks_that_terminate_on_router(subscription_id: UUIDstr) -> list
query_in_use_by_subscriptions(UUID(subscription_id))
.join(ProductTable)
.filter(
- ProductTable.product_type == "Iptrunk",
- SubscriptionTable.status == "active",
+ ProductTable.product_type == ProductType.IP_TRUNK,
+ SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
)
.all()
)
-def get_product_id_by_name(product_name: ProductType) -> UUID:
+def get_product_id_by_name(product_name: ProductName) -> UUID:
"""Retrieve the :term:`UUID` of a product by its name.
:param product_name: The name of the product.
- :type product_name: ProductType
+ :type product_name: ProductName
:return UUID: The :term:`UUID` of the product.
:rtype: UUID
diff --git a/gso/workflows/tasks/import_iptrunk.py b/gso/workflows/tasks/import_iptrunk.py
index 877c375e..c34be8ed 100644
--- a/gso/workflows/tasks/import_iptrunk.py
+++ b/gso/workflows/tasks/import_iptrunk.py
@@ -11,7 +11,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhyPortCapacity
from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
from gso.products.product_types.router import Router
@@ -68,7 +68,7 @@ def initial_input_form_generator() -> FormGenerator:
def create_subscription(partner: str) -> State:
"""Create a new subscription in the service database."""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id = subscriptions.get_product_id_by_name(ProductType.IP_TRUNK)
+ product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
subscription = IptrunkInactive.from_product_id(product_id, partner_id)
return {
diff --git a/gso/workflows/tasks/import_office_router.py b/gso/workflows/tasks/import_office_router.py
index 255c7f31..9168cdae 100644
--- a/gso/workflows/tasks/import_office_router.py
+++ b/gso/workflows/tasks/import_office_router.py
@@ -9,7 +9,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_types import office_router
from gso.products.product_types.office_router import OfficeRouterInactive
from gso.services import subscriptions
@@ -22,7 +22,7 @@ from gso.utils.shared_enums import PortNumber, Vendor
def create_subscription(partner: str) -> State:
"""Create a new subscription object."""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id = subscriptions.get_product_id_by_name(ProductType.OFFICE_ROUTER)
+ product_id = subscriptions.get_product_id_by_name(ProductName.OFFICE_ROUTER)
subscription = OfficeRouterInactive.from_product_id(product_id, partner_id)
return {
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index d284dbc5..c71ce26e 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -9,7 +9,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_blocks import router as router_pb
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types import router
@@ -25,7 +25,7 @@ from gso.utils.shared_enums import PortNumber, Vendor
def create_subscription(partner: str) -> State:
"""Create a new subscription object."""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
+ product_id = subscriptions.get_product_id_by_name(ProductName.ROUTER)
subscription = RouterInactive.from_product_id(product_id, partner_id)
return {
diff --git a/gso/workflows/tasks/import_site.py b/gso/workflows/tasks/import_site.py
index 026ffb1b..ff49808a 100644
--- a/gso/workflows/tasks/import_site.py
+++ b/gso/workflows/tasks/import_site.py
@@ -8,7 +8,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import SiteInactive
from gso.services import subscriptions
@@ -23,7 +23,7 @@ def create_subscription(partner: str) -> State:
FIXME: all attributes passed by the input form appear to be unused
"""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id: UUID = subscriptions.get_product_id_by_name(ProductType.SITE)
+ product_id: UUID = subscriptions.get_product_id_by_name(ProductName.SITE)
subscription = SiteInactive.from_product_id(product_id, partner_id)
return {
diff --git a/gso/workflows/tasks/import_super_pop_switch.py b/gso/workflows/tasks/import_super_pop_switch.py
index aa6c832f..5f2796c2 100644
--- a/gso/workflows/tasks/import_super_pop_switch.py
+++ b/gso/workflows/tasks/import_super_pop_switch.py
@@ -9,7 +9,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_types import super_pop_switch
from gso.products.product_types.super_pop_switch import SuperPopSwitchInactive
from gso.services import subscriptions
@@ -23,7 +23,7 @@ from gso.utils.shared_enums import PortNumber, Vendor
def create_subscription(partner: str) -> State:
"""Create a new subscription object."""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id = subscriptions.get_product_id_by_name(ProductType.SUPER_POP_SWITCH)
+ product_id = subscriptions.get_product_id_by_name(ProductName.SUPER_POP_SWITCH)
subscription = SuperPopSwitchInactive.from_product_id(product_id, partner_id)
return {
diff --git a/test/api/conftest.py b/test/api/conftest.py
index e015f147..e002fa13 100644
--- a/test/api/conftest.py
+++ b/test/api/conftest.py
@@ -2,5 +2,7 @@ from test.fixtures import ( # noqa: F401
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
nokia_router_subscription_factory,
+ office_router_subscription_factory,
site_subscription_factory,
+ super_pop_switch_subscription_factory,
)
diff --git a/test/api/test_subscriptions.py b/test/api/test_subscriptions.py
index 37c74cd8..3e457a58 100644
--- a/test/api/test_subscriptions.py
+++ b/test/api/test_subscriptions.py
@@ -1,6 +1,7 @@
from orchestrator.types import SubscriptionLifecycle
ROUTER_SUBSCRIPTION_ENDPOINT = "/api/v1/subscriptions/routers"
+DASHBOARD_DEVICES_ENDPOINT = "/api/v1/subscriptions/dashboard_devices"
def test_router_subscriptions_endpoint_with_valid_api_key(test_client, nokia_router_subscription_factory):
@@ -30,3 +31,59 @@ def test_router_subscriptions_endpoint_without_api_key(test_client, nokia_router
assert response.status_code == 403
assert response.json() == {"detail": "Not authenticated"}
+
+
+def test_dashboard_devices_endpoint_with_valid_api_key(
+ test_client,
+ nokia_router_subscription_factory,
+ office_router_subscription_factory,
+ super_pop_switch_subscription_factory,
+):
+ nokia_router_subscription_factory(router_fqdn="mx1.ams.nl.geant.net")
+ nokia_router_subscription_factory(router_fqdn="mx2.ams.nl.geant.net")
+ nokia_router_subscription_factory(status=SubscriptionLifecycle.PROVISIONING, router_fqdn="mx3.ams.nl.geant.net")
+ nokia_router_subscription_factory(status=SubscriptionLifecycle.INITIAL, router_fqdn="mx4.ams.nl.geant.net")
+ office_router_subscription_factory(office_router_fqdn="office1.ams.nl.geant.net")
+ office_router_subscription_factory(
+ office_router_fqdn="office2.ams.nl.geant.net", status=SubscriptionLifecycle.TERMINATED
+ )
+ office_router_subscription_factory(
+ office_router_fqdn="office3.ams.nl.geant.net", status=SubscriptionLifecycle.PROVISIONING
+ )
+ super_pop_switch_subscription_factory(super_pop_switch_fqdn="superpop1.ams.nl.geant.net")
+ super_pop_switch_subscription_factory(
+ super_pop_switch_fqdn="superpop2.ams.nl.geant.net", status=SubscriptionLifecycle.TERMINATED
+ )
+ super_pop_switch_subscription_factory(
+ super_pop_switch_fqdn="superpop3.ams.nl.geant.net", status=SubscriptionLifecycle.PROVISIONING
+ )
+
+ response = test_client.get(
+ DASHBOARD_DEVICES_ENDPOINT, headers={"Authorization": "Bearer another_REALY_random_AND_3cure_T0keN"}
+ )
+
+ assert response.status_code == 200
+ fqdns = response.text.strip().split("\n")
+ assert sorted(fqdns) == [
+ "mx1.ams.nl.geant.net",
+ "mx2.ams.nl.geant.net",
+ "mx3.ams.nl.geant.net",
+ "office1.ams.nl.geant.net",
+ "office3.ams.nl.geant.net",
+ "superpop1.ams.nl.geant.net",
+ "superpop3.ams.nl.geant.net",
+ ]
+
+
+def test_dashboard_devices_endpoint_with_invalid_api_key(test_client, nokia_router_subscription_factory):
+ response = test_client.get(DASHBOARD_DEVICES_ENDPOINT, headers={"Authorization": "Bearer fake_invalid_api_key"})
+
+ assert response.status_code == 403
+ assert response.json() == {"detail": "Invalid API Key"}
+
+
+def test_dashboard_devices_endpoint_without_api_key(test_client, nokia_router_subscription_factory):
+ response = test_client.get(DASHBOARD_DEVICES_ENDPOINT)
+
+ assert response.status_code == 403
+ assert response.json() == {"detail": "Not authenticated"}
diff --git a/test/fixtures.py b/test/fixtures.py
index 0800edec..d1775d8e 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -5,7 +5,7 @@ from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
@@ -15,8 +15,10 @@ from gso.products.product_blocks.iptrunk import (
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.iptrunk import IptrunkInactive
+from gso.products.product_types.office_router import OfficeRouterInactive
from gso.products.product_types.router import Router, RouterInactive
from gso.products.product_types.site import Site, SiteInactive
+from gso.products.product_types.super_pop_switch import SuperPopSwitchInactive
from gso.services import subscriptions
from gso.utils.shared_enums import Vendor
@@ -52,7 +54,7 @@ def site_subscription_factory(faker, geant_partner):
site_internal_id = site_internal_id or faker.pyint()
site_ts_address = site_ts_address or faker.ipv4()
- product_id = subscriptions.get_product_id_by_name(ProductType.SITE)
+ 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)
site_subscription.site.site_city = site_city
site_subscription.site.site_name = site_name
@@ -104,7 +106,7 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
- product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
+ 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)
router_subscription.router.router_fqdn = router_fqdn
router_subscription.router.router_ts_port = router_ts_port
@@ -159,7 +161,7 @@ def juniper_router_subscription_factory(site_subscription_factory, faker, geant_
router_lo_iso_address = router_lo_iso_address or faker.word()
router_site = router_site or site_subscription_factory()
- product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
+ 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)
router_subscription.router.router_fqdn = router_fqdn
@@ -244,7 +246,7 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant
if partner is None:
partner = geant_partner
- product_id = subscriptions.get_product_id_by_name(ProductType.IP_TRUNK)
+ product_id = subscriptions.get_product_id_by_name(ProductName.IP_TRUNK)
description = description or faker.sentence()
geant_s_sid = geant_s_sid or faker.geant_sid()
@@ -286,3 +288,106 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant
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,
+ ) -> 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()
+
+ 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
+ )
+ 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,
+ ) -> 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()
+
+ 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
+ )
+ 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
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 38611a75..589fbabb 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -3,7 +3,7 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk, ProductType
+from gso.products import Iptrunk, ProductName
from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.helpers import LAGMember
@@ -116,7 +116,7 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
mock_create_host.return_value = None
mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
- product_id = get_product_id_by_name(ProductType.IP_TRUNK)
+ product_id = get_product_id_by_name(ProductName.IP_TRUNK)
initial_site_data = [{"product": product_id}, *input_form_wizard_data]
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
@@ -162,7 +162,7 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
):
mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
- product_id = get_product_id_by_name(ProductType.IP_TRUNK)
+ product_id = get_product_id_by_name(ProductName.IP_TRUNK)
initial_site_data = [{"product": product_id}, *input_form_wizard_data]
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
@@ -195,7 +195,7 @@ def test_successful_iptrunk_creation_with_juniper_interface_names(
mock_create_host.return_value = None
mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
- product_id = get_product_id_by_name(ProductType.IP_TRUNK)
+ product_id = get_product_id_by_name(ProductName.IP_TRUNK)
initial_site_data = [{"product": product_id}, *input_form_wizard_data]
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
index a6d61729..33244ae5 100644
--- a/test/workflows/router/test_create_router.py
+++ b/test/workflows/router/test_create_router.py
@@ -3,7 +3,7 @@ from unittest.mock import patch
import pytest
from infoblox_client import objects
-from gso.products import ProductType, Site
+from gso.products import ProductName, Site
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import Router
from gso.services.subscriptions import get_product_id_by_name
@@ -51,7 +51,7 @@ def test_create_nokia_router_success(
data_config_filename,
):
# Set up mock return values
- product_id = get_product_id_by_name(ProductType.ROUTER)
+ product_id = get_product_id_by_name(ProductName.ROUTER)
mock_site = Site.from_subscription(router_creation_input_form_data["router_site"]).site
mock_v4 = faker.ipv4()
mock_v6 = faker.ipv6()
@@ -169,7 +169,7 @@ def test_create_nokia_router_lso_failure(
)
# Run workflow
- product_id = get_product_id_by_name(ProductType.ROUTER)
+ product_id = get_product_id_by_name(ProductName.ROUTER)
initial_router_data = [{"product": product_id}, router_creation_input_form_data]
result, process_stat, step_log = run_workflow("create_router", initial_router_data)
diff --git a/test/workflows/site/test_create_site.py b/test/workflows/site/test_create_site.py
index 5909b5db..e3157615 100644
--- a/test/workflows/site/test_create_site.py
+++ b/test/workflows/site/test_create_site.py
@@ -1,7 +1,7 @@
import pytest
from pydantic_forms.exceptions import FormValidationError
-from gso.products import ProductType
+from gso.products import ProductName
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import Site
from gso.services.partners import get_partner_by_name
@@ -11,7 +11,7 @@ from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.workflow()
def test_create_site(responses, faker):
- product_id = get_product_id_by_name(ProductType.SITE)
+ product_id = get_product_id_by_name(ProductName.SITE)
initial_site_data = [
{"product": product_id},
{
@@ -51,7 +51,7 @@ def test_site_name_is_incorrect(responses, faker):
"""
invalid_site_name = "AMST10"
expected_exception_msg = rf".*Enter a valid site name.+Received: {invalid_site_name}.*"
- product_id = get_product_id_by_name(ProductType.SITE)
+ product_id = get_product_id_by_name(ProductName.SITE)
initial_site_data = [
{"product": product_id},
{
--
GitLab