docs/source/module/workflows/router/modify_connection_strategy.rst

+``gso.workflows.router.modify_connection_strategy``
+=========================================
+
+.. automodule:: gso.workflows.router.modify_connection_strategy
+   :members:
+   :show-inheritance:

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")

gso/migrations/versions/2023-11-21_e8378fbcfbf3_add_initial_products.py

@@ -11,7 +11,7 @@ from alembic import op
 # revision identifiers, used by Alembic.
 revision = 'e8378fbcfbf3'
 down_revision = 'da5c9f4cce1c'
-branch_labels = ('data',)
+branch_labels = None
 depends_on = None
 
 

gso/migrations/versions/2024-03-21_a2cd3f2e6d7a_modify_connection_streategy_workflow.py

+"""Modify connection streategy workflow..
+
+Revision ID: a2cd3f2e6d7a
+Revises:
+Create Date: 2024-03-21 16:05:59.043106
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'a2cd3f2e6d7a'
+down_revision = None
+branch_labels = None
+depends_on = 'd61c0f92da1e'
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "modify_connection_strategy",
+        "target": "MODIFY",
+        "description": "Modify connection strategy",
+        "product_type": "Router"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

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,
     },
 )

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

gso/utils/shared_enums.py

@@ -19,3 +19,10 @@ class PortNumber(ConstrainedInt):
 
     gt = 0
     le = 49151
+
+
+class ConnectionStrategy(strEnum):
+    """An enumerator for the connection Strategies."""
+
+    IN_BAND = "IN BAND"
+    OUT_OF_BAND = "OUT OF BAND"

gso/workflows/__init__.py

@@ -26,6 +26,7 @@ LazyWorkflowInstance("gso.workflows.router.create_router", "create_router")
 LazyWorkflowInstance("gso.workflows.router.redeploy_base_config", "redeploy_base_config")
 LazyWorkflowInstance("gso.workflows.router.terminate_router", "terminate_router")
 LazyWorkflowInstance("gso.workflows.router.update_ibgp_mesh", "update_ibgp_mesh")
+LazyWorkflowInstance("gso.workflows.router.modify_connection_strategy", "modify_connection_strategy")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")
 LazyWorkflowInstance("gso.workflows.site.terminate_site", "terminate_site")

gso/workflows/router/modify_connection_strategy.py

+"""Modify connection strategy workflow. Flipping the connection between in-band to out-of-band."""
+
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, UUIDstr
+from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products.product_types.router import Router
+from gso.utils.shared_enums import ConnectionStrategy
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    """Modify the connection strategy initial formruff format."""
+    subscription = Router.from_subscription(subscription_id)
+
+    current_connection_strategy = (
+        ConnectionStrategy.OUT_OF_BAND if subscription.router.router_access_via_ts else ConnectionStrategy.IN_BAND
+    )
+
+    class ModifyConnectionStrategyForm(FormPage):
+        class Config:
+            title = f"Modify the connection strategy of {subscription.router.router_fqdn}."
+
+        connection_strategy: ConnectionStrategy = current_connection_strategy
+
+    user_input = yield ModifyConnectionStrategyForm
+
+    return user_input.dict()
+
+
+@step("Update subscription model")
+def update_subscription_model(subscription: Router, connection_strategy: str) -> State:
+    """Update the database model to reflect the new connection strategy.
+
+    If the connection strategy is set to IN-BAND, then access_via_ts should be set to False.
+    Conversely, if the connection strategy is set to OUT-OF-BAND, access_via_ts should be set to True.
+    """
+    subscription.router.router_access_via_ts = connection_strategy == ConnectionStrategy.OUT_OF_BAND
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Modify connection strategy",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def modify_connection_strategy() -> StepList:
+    """Modify the connection strategy."""
+    return init >> store_process_subscription(Target.MODIFY) >> unsync >> update_subscription_model >> resync >> done

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 {

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 {

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 {

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 {

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 {

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,
 )

test/api/test_subscriptions.py

 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"}

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

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)
 

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)
 

test/workflows/router/test_modify_connection_stratey.py

+import pytest
+
+from gso.products import Router
+from gso.utils.shared_enums import ConnectionStrategy
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_modify_connection_strategy(responses, nokia_router_subscription_factory):
+    subscription_id = nokia_router_subscription_factory(router_access_via_ts=True)
+    subscription = Router.from_subscription(subscription_id)
+    assert subscription.router.router_access_via_ts is True
+    form_data = [
+        {"subscription_id": subscription_id},
+        {"connection_strategy": ConnectionStrategy.IN_BAND},
+    ]
+    result, _, _ = run_workflow("modify_connection_strategy", form_data)
+    assert_complete(result)
+
+    # Fetch the updated subscription after running the workflow
+    updated_subscription = Router.from_subscription(subscription_id)
+    assert updated_subscription.status == "active"
+    assert updated_subscription.router.router_access_via_ts is False