Add unit tests for LAN Switch Interconnect

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

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}

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,

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

test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py

+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

test/workflows/lan_switch_interconnect/test_import_lan_switch_interconnect.py

+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

test/workflows/lan_switch_interconnect/test_terminate_lan_switch_interconnect.py

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