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
 
@@ -36,6 +38,7 @@ def site_subscription_factory(faker, geant_partner):
         site_internal_id=None,
         site_tier=SiteTier.TIER1,
         site_ts_address=None,
+        status: SubscriptionLifecycle | None = None,
         partner: dict | None = None,
     ) -> UUIDstr:
         if partner is None:
@@ -52,7 +55,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
@@ -68,6 +71,9 @@ def site_subscription_factory(faker, geant_partner):
         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()
 
@@ -104,7 +110,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 +165,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 +250,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 +292,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

test/workflows/shared/cancel_subscription.py

+import pytest
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.mark.parametrize(
+    "subscription_factory",
+    [
+        "site_subscription_factory",
+        "juniper_router_subscription_factory",
+        "nokia_router_subscription_factory",
+        "iptrunk_subscription_factory",
+    ],
+)
+@pytest.mark.workflow()
+def test_cancel_workflow_success(subscription_factory, geant_partner, request):
+    subscription_id = request.getfixturevalue(subscription_factory)(
+        status=SubscriptionLifecycle.INITIAL,
+        partner=geant_partner,
+    )
+    initial_site_data = [{"subscription_id": subscription_id}, {}]
+    result, _, _ = run_workflow("cancel_subscription", initial_site_data)
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = SubscriptionModel.from_subscription(subscription_id)
+    assert subscription.status == "terminated"

test/workflows/site/test_create_site.py

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