From 5881e93ff6350bf8681feb2507391dcb5322917e Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Thu, 31 Oct 2024 14:29:49 +0100
Subject: [PATCH] Add unit tests for LAN Switch Interconnect
---
gso/workflows/__init__.py | 7 +++
...create_imported_lan_switch_interconnect.py | 25 ++++++++---
test/conftest.py | 1 +
.../lan_switch_interconnect_fixtures.py | 42 +++++++++++++-----
...create_imported_lan_switch_interconnect.py | 44 +++++++++++++++++++
.../test_import_lan_switch_interconnect.py | 20 +++++++++
.../test_terminate_lan_switch_interconnect.py | 17 +++++++
7 files changed, 139 insertions(+), 17 deletions(-)
create mode 100644 test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
create mode 100644 test/workflows/lan_switch_interconnect/test_import_lan_switch_interconnect.py
create mode 100644 test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index 12ab19c6..362f96ab 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -71,6 +71,13 @@ LazyWorkflowInstance(
LazyWorkflowInstance(
"gso.workflows.lan_switch_interconnect.validate_lan_switch_interconnect", "validate_lan_switch_interconnect"
)
+LazyWorkflowInstance(
+ "gso.workflows.lan_switch_interconnect.create_imported_lan_switch_interconnect",
+ "create_imported_lan_switch_interconnect",
+)
+LazyWorkflowInstance(
+ "gso.workflows.lan_switch_interconnect.import_lan_switch_interconnect", "import_lan_switch_interconnect"
+)
# Site workflows
LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
diff --git a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
index 282b0e26..1439c2fe 100644
--- a/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py
@@ -8,14 +8,17 @@ from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, begin, done
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
-from products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnectInactive
from gso.cli.imports import LanSwitchInterconnectRouterSideImportModel, LanSwitchInterconnectSwitchSideImportModel
from gso.products import ProductName
from gso.products.product_blocks.lan_switch_interconnect import (
+ LanSwitchInterconnectInterfaceBlockInactive,
LanSwitchInterconnectRouterSideBlockInactive,
LanSwitchInterconnectSwitchSideBlockInactive,
)
+from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnectInactive
+from gso.products.product_types.router import Router
+from gso.products.product_types.switch import Switch
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.types.ip_address import AddressSpace, IPv4NetworkType
@@ -51,19 +54,31 @@ def initialize_subscription(
lan_switch_interconnect_ip_network: IPv4NetworkType | None,
address_space: AddressSpace,
minimum_links: int,
- lan_switch_interconnect_router_side: dict,
- lan_switch_interconnect_switch_side: dict,
+ router_side: dict,
+ switch_side: dict,
) -> State:
"""Initialize the subscription using input data."""
subscription.lan_switch_interconnect.lan_switch_interconnect_description = lan_switch_interconnect_description
subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = lan_switch_interconnect_ip_network
subscription.lan_switch_interconnect.address_space = address_space
subscription.lan_switch_interconnect.minimum_links = minimum_links
+
+ router_block = Router.from_subscription(router_side.pop("node")).router
+ router_side_interfaces = [
+ LanSwitchInterconnectInterfaceBlockInactive.new(uuid4(), **ae_member)
+ for ae_member in router_side.pop("ae_members")
+ ]
subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
- uuid4(), **lan_switch_interconnect_router_side
+ uuid4(), **router_side, node=router_block, ae_members=router_side_interfaces
)
+
+ switch_block = Switch.from_subscription(switch_side.pop("switch")).switch
+ switch_side_interfaces = [
+ LanSwitchInterconnectInterfaceBlockInactive.new(uuid4(), **ae_member)
+ for ae_member in switch_side.pop("ae_members")
+ ]
subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
- uuid4(), **lan_switch_interconnect_switch_side
+ uuid4(), **switch_side, switch=switch_block, ae_members=switch_side_interfaces
)
return {"subscription": subscription}
diff --git a/test/conftest.py b/test/conftest.py
index c2a9a723..8c8d246c 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -40,6 +40,7 @@ from test.fixtures import ( # noqa: F401
edge_port_subscription_factory,
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
+ lan_switch_interconnect_subscription_factory,
nren_access_port_factory,
nren_l3_core_service_subscription_factory,
office_router_subscription_factory,
diff --git a/test/fixtures/lan_switch_interconnect_fixtures.py b/test/fixtures/lan_switch_interconnect_fixtures.py
index 3f78deb4..fe0cb87f 100644
--- a/test/fixtures/lan_switch_interconnect_fixtures.py
+++ b/test/fixtures/lan_switch_interconnect_fixtures.py
@@ -7,6 +7,7 @@ from orchestrator.types import SubscriptionLifecycle, UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.lan_switch_interconnect import (
+ LanSwitchInterconnectInterfaceBlockInactive,
LanSwitchInterconnectRouterSideBlockInactive,
LanSwitchInterconnectSwitchSideBlockInactive,
)
@@ -14,6 +15,8 @@ from gso.products.product_types.lan_switch_interconnect import (
ImportedLanSwitchInterconnectInactive,
LanSwitchInterconnectInactive,
)
+from gso.products.product_types.router import Router
+from gso.products.product_types.switch import Switch
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
@@ -22,11 +25,11 @@ from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4Networ
def lan_switch_interconnect_subscription_factory(
faker, geant_partner, router_subscription_factory, switch_subscription_factory
):
- def _create_subscription(
+ def create_subscription(
description: str | None = None,
partner: dict | None = None,
status: SubscriptionLifecycle | None = None,
- start_date: str = "2024-10-30T02:12:22+33:33",
+ start_date: str | None = "2024-01-01T10:20:30+01:02",
lan_switch_interconnect_description: str | None = None,
lan_switch_interconnect_ip_network: IPv4NetworkType | None = None,
address_space: AddressSpace | None = None,
@@ -40,18 +43,30 @@ def lan_switch_interconnect_subscription_factory(
switch_side_ae_members: list[dict[str, str]] | None = None,
switch_side_ipv4_address: IPv4AddressType | None = None,
*,
- is_imported: bool = True,
+ is_imported: bool | None = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
-
if is_imported:
product_id = get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT)
- subscription = LanSwitchInterconnectInactive.from_product_id(product_id, partner["patrner_id"])
+ subscription = LanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])
else:
product_id = get_product_id_by_name(ProductName.IMPORTED_LAN_SWITCH_INTERCONNECT)
subscription = ImportedLanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])
+ router_side_ae_members = router_side_ae_members or [
+ LanSwitchInterconnectInterfaceBlockInactive.new(
+ uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
+ )
+ for _ in range(2)
+ ]
+ switch_side_ae_members = switch_side_ae_members or [
+ LanSwitchInterconnectInterfaceBlockInactive.new(
+ uuid4(), interface_name=faker.network_interface(), interface_description=faker.sentence()
+ )
+ for _ in range(2)
+ ]
+
subscription.lan_switch_interconnect.lan_switch_interconnect_description = (
lan_switch_interconnect_description or faker.sentence()
)
@@ -62,27 +77,30 @@ def lan_switch_interconnect_subscription_factory(
subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
uuid4(),
- node=router_side_node or router_subscription_factory(),
+ node=router_side_node or Router.from_subscription(router_subscription_factory()).router,
ae_iface=router_side_ae_iface or faker.network_interface(),
- ae_members=router_side_ae_members or faker.link_members_nokia()[:2],
+ ae_members=router_side_ae_members,
ipv4_address=router_side_ipv4_address or faker.ipv4(),
)
subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
uuid4(),
- switch=switch_side_switch or switch_subscription_factory(),
+ switch=switch_side_switch or Switch.from_subscription(switch_subscription_factory()).switch,
ae_iface=switch_side_ae_iface or faker.network_interface(),
- ae_members=switch_side_ae_members or faker.link_members_juniper()[:2],
+ ae_members=switch_side_ae_members,
ipv4_address=switch_side_ipv4_address or faker.ipv4(),
)
subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
- subscription.description = description or "Generated LAN Switch Interconnect"
+ subscription.insync = True
+ subscription.description = description or faker.sentence()
subscription.start_date = start_date
- subscription.status = status or SubscriptionLifecycle.ACTIVE
+
+ if status:
+ subscription.status = status
subscription.save()
db.session.commit()
return str(subscription.subscription_id)
- return _create_subscription
+ return create_subscription
diff --git a/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
new file mode 100644
index 00000000..69349ae0
--- /dev/null
+++ b/test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py
@@ -0,0 +1,44 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnect
+from gso.utils.types.ip_address import AddressSpace
+from test.workflows import (
+ assert_complete,
+ extract_state,
+ run_workflow,
+)
+
+
+@pytest.fixture()
+def workflow_input_data(faker, router_subscription_factory, switch_subscription_factory):
+ return {
+ "lan_switch_interconnect_description": faker.sentence(),
+ "lan_switch_interconnect_ip_network": faker.ipv4_network(),
+ "address_space": AddressSpace.PUBLIC,
+ "minimum_links": 1,
+ "router_side": {
+ "node": router_subscription_factory(),
+ "ae_iface": faker.nokia_lag_interface_name(),
+ "ae_members": faker.link_members_nokia(),
+ "ipv4_address": faker.ipv4(),
+ },
+ "switch_side": {
+ "switch": switch_subscription_factory(),
+ "ae_iface": faker.juniper_ae_interface_name(),
+ "ae_members": faker.link_members_juniper(),
+ "ipv4_address": faker.ipv4(),
+ },
+ }
+
+
+@pytest.mark.workflow()
+def test_create_imported_lan_switch_interconnect_success(workflow_input_data):
+ result, _, _ = run_workflow("create_imported_lan_switch_interconnect", [workflow_input_data])
+ state = extract_state(result)
+ subscription = ImportedLanSwitchInterconnect.from_subscription(state["subscription_id"])
+
+ assert_complete(result)
+ assert subscription.product.name == ProductName.IMPORTED_LAN_SWITCH_INTERCONNECT
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
diff --git a/test/workflows/lan_switch_interconnect/test_import_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_import_lan_switch_interconnect.py
new file mode 100644
index 00000000..5808299f
--- /dev/null
+++ b/test/workflows/lan_switch_interconnect/test_import_lan_switch_interconnect.py
@@ -0,0 +1,20 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_import_lan_switch_interconnect_success(lan_switch_interconnect_subscription_factory):
+ imported_lan_switch_interconnect = lan_switch_interconnect_subscription_factory(is_imported=False)
+ result, _, _ = run_workflow(
+ "import_lan_switch_interconnect", [{"subscription_id": imported_lan_switch_interconnect}]
+ )
+ subscription = LanSwitchInterconnect.from_subscription(imported_lan_switch_interconnect)
+
+ assert_complete(result)
+ assert subscription.product.name == ProductName.LAN_SWITCH_INTERCONNECT
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.insync is True
diff --git a/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
new file mode 100644
index 00000000..6d5ae538
--- /dev/null
+++ b/test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py
@@ -0,0 +1,17 @@
+import pytest
+
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.mark.workflow()
+def test_terminate_lan_switch_interconnect(lan_switch_interconnect_subscription_factory, faker):
+ subscription_id = lan_switch_interconnect_subscription_factory()
+ initial_lan_switch_interconnect_data = [{"subscription_id": subscription_id}, {"tt_number": faker.tt_number()}]
+ result, _, _ = run_workflow("terminate_lan_switch_interconnect", initial_lan_switch_interconnect_data)
+ assert_complete(result)
+
+ state = extract_state(result)
+ subscription_id = state["subscription_id"]
+ subscription = LanSwitchInterconnect.from_subscription(subscription_id)
+ assert subscription.status == "terminated"
--
GitLab