Add workflows for Switch and LAN switch interconnect: create, import, terminate, validate, modify

gso/workflows/switch/validate_switch.py

@@ -8,30 +8,29 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
 from gso.products.product_types.switch import Switch
-from gso.services.lso_client import anonymous_lso_interaction, execute_playbook
+from gso.services.lso_client import LSOState, anonymous_lso_interaction
 from gso.services.netbox_client import NetboxClient
 
 
 @step("Validate switch in Netbox")
 def check_netbox_device(subscription: Switch) -> None:
     """Fetch the device in Netbox. Will raise an exception when it is not found."""
-    NetboxClient().get_device_by_name(subscription.switch.switch_fqdn)
+    NetboxClient().get_device_by_name(subscription.switch.fqdn)
 
 
 @step("Check base config for drift")
-def verify_base_config(subscription: dict[str, Any], callback_route: str) -> None:
+def verify_base_config(subscription: dict[str, Any]) -> LSOState:
     """Workflow step for running a playbook that checks whether base config has drifted."""
-    execute_playbook(
-        playbook_name="switch_base_config.yaml",
-        callback_route=callback_route,
-        inventory=subscription["switch"]["switch_fqdn"],
-        extra_vars={
+    return {
+        "playbook_name": "switch_base_config.yaml",
+        "inventory": {"all": {"hosts": {subscription["switch"]["fqdn"]: None}}},
+        "extra_vars": {
             "subscription_json": subscription,
             "verb": "deploy",
             "dry_run": "true",
             "is_verification_workflow": "true",
         },
-    )
+    }
 
 
 @workflow(

test/cli/test_imports.py

@@ -7,23 +7,27 @@ import pytest
 from gso.cli.imports import (
     import_edge_port,
     import_iptrunks,
+    import_lan_switch_interconnect,
     import_nren_l3_core_service,
     import_office_routers,
     import_opengear,
     import_routers,
     import_sites,
     import_super_pop_switches,
+    import_switches,
 )
 from gso.products.product_blocks.bgp_session import IPFamily
 from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
+from gso.products.product_blocks.switch import SwitchModel
 from gso.products.product_types.router import Router
 from gso.products.product_types.site import Site
 from gso.utils.helpers import iso_from_ipv4
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.utils.types.ip_address import AddressSpace
 
 
 ##############
@@ -200,6 +204,59 @@ def opengear_data(temp_file, faker, site_subscription_factory):
     return _opengear_data
 
 
+@pytest.fixture()
+def switch_data(temp_file, faker, site_subscription_factory):
+    def _switch_data(**kwargs):
+        switch_data = {
+            "fqdn": faker.domain_name(levels=4),
+            "ts_port": faker.port_number(is_user=True),
+            "site": site_subscription_factory(),
+            "switch_vendor": Vendor.JUNIPER,
+            "switch_model": SwitchModel.EX3400,
+        }
+        switch_data.update(**kwargs)
+
+        temp_file.write_text(json.dumps([switch_data]))
+        return {"path": str(temp_file), "data": switch_data}
+
+    return _switch_data
+
+
+@pytest.fixture()
+def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory, router_subscription_factory):
+    def _lan_switch_interconnect_data(**kwargs):
+        lan_switch_interconnect_data = {
+            "lan_switch_interconnect_description": faker.sentence(),
+            "lan_switch_interconnect_ip_network": str(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": [
+                    {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
+                    for _ in range(2)
+                ],
+                "ipv4_address": faker.ipv4(),
+            },
+            "switch_side": {
+                "switch": switch_subscription_factory(),
+                "ae_iface": faker.juniper_ae_interface_name(),
+                "ae_members": [
+                    {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
+                    for _ in range(2)
+                ],
+                "ipv4_address": faker.ipv4(),
+            },
+        }
+        lan_switch_interconnect_data.update(**kwargs)
+
+        temp_file.write_text(json.dumps([lan_switch_interconnect_data]))
+        return {"path": str(temp_file), "data": lan_switch_interconnect_data}
+
+    return _lan_switch_interconnect_data
+
+
 @pytest.fixture()
 def edge_port_data(temp_file, faker, router_subscription_factory, partner_factory):
     def _edge_port_data(**kwargs):
@@ -425,6 +482,20 @@ router_lo_ipv6_address
     assert mock_start_process.call_count == 0
 
 
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_switch_success(mock_start_process, mock_sleep, switch_data):
+    import_switches(switch_data()["path"])
+    assert mock_start_process.call_count == 1
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_lan_switch_interconnect(mock_start_process, mock_sleep, lan_switch_interconnect_data):
+    import_lan_switch_interconnect(lan_switch_interconnect_data()["path"])
+    assert mock_start_process.call_count == 1
+
+
 @patch("gso.cli.imports.time.sleep")
 @patch("gso.cli.imports.start_process")
 def test_import_iptrunk_successful(mock_start_process, mock_sleep, iptrunk_data):

test/fixtures/__init__.py

 from test.fixtures.edge_port_fixtures import edge_port_subscription_factory
 from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
+from test.fixtures.lan_switch_interconnect_fixtures import lan_switch_interconnect_subscription_factory
 from test.fixtures.nren_l3_core_service_fixtures import (
     bgp_session_subscription_factory,
     nren_access_port_factory,
@@ -18,6 +19,7 @@ __all__ = [
     "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

+from uuid import uuid4
+
+import pytest
+from orchestrator.db import db
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+
+from gso.products import ProductName
+from gso.products.product_blocks.lan_switch_interconnect import (
+    LanSwitchInterconnectRouterSideBlockInactive,
+    LanSwitchInterconnectSwitchSideBlockInactive,
+)
+from gso.products.product_types.lan_switch_interconnect import (
+    ImportedLanSwitchInterconnectInactive,
+    LanSwitchInterconnectInactive,
+)
+from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
+
+
+@pytest.fixture()
+def lan_switch_interconnect_subscription_factory(
+    faker, geant_partner, router_subscription_factory, switch_subscription_factory
+):
+    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",
+        lan_switch_interconnect_description: str | None = None,
+        lan_switch_interconnect_ip_network: IPv4NetworkType | None = None,
+        address_space: AddressSpace | None = None,
+        minimum_links: int | None = None,
+        router_side_node: UUIDstr | None = None,
+        router_side_ae_iface: str | None = None,
+        router_side_ae_members: list[dict[str, str]] | None = None,
+        router_side_ipv4_address: IPv4AddressType | None = None,
+        switch_side_switch: UUIDstr | None = None,
+        switch_side_ae_iface: str | None = None,
+        switch_side_ae_members: list[dict[str, str]] | None = None,
+        switch_side_ipv4_address: IPv4AddressType | None = None,
+        *,
+        is_imported: bool = 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"])
+        else:
+            product_id = get_product_id_by_name(ProductName.IMPORTED_LAN_SWITCH_INTERCONNECT)
+            subscription = ImportedLanSwitchInterconnectInactive.from_product_id(product_id, partner["partner_id"])
+
+        subscription.lan_switch_interconnect.lan_switch_interconnect_description = (
+            lan_switch_interconnect_description or faker.sentence()
+        )
+        subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = (
+            lan_switch_interconnect_ip_network or faker.ipv4_network()
+        )
+        subscription.lan_switch_interconnect.address_space = address_space or AddressSpace.PRIVATE
+        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(),
+            ae_iface=router_side_ae_iface or faker.network_interface(),
+            ae_members=router_side_ae_members or faker.link_members_nokia()[:2],
+            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(),
+            ae_iface=switch_side_ae_iface or faker.network_interface(),
+            ae_members=switch_side_ae_members or faker.link_members_juniper()[:2],
+            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.start_date = start_date
+        subscription.status = status or SubscriptionLifecycle.ACTIVE
+
+        subscription.save()
+        db.session.commit()
+
+        return str(subscription.subscription_id)
+
+    return _create_subscription

test/fixtures/switch_fixtures.py

@@ -7,7 +7,7 @@ from pydantic_forms.types import UUIDstr
 from gso.products import ProductName
 from gso.products.product_blocks.switch import SwitchModel
 from gso.products.product_types.site import Site
-from gso.products.product_types.switch import SwitchInactive
+from gso.products.product_types.switch import ImportedSwitchInactive, SwitchInactive
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.ip_address import PortNumber
@@ -35,7 +35,6 @@ def switch_subscription_factory(faker, geant_partner, site_subscription_factory)
             switch_subscription = SwitchInactive.from_product_id(product_id, partner["partner_id"])
         else:
             product_id = get_product_id_by_name(ProductName.IMPORTED_SWITCH)
-            raise NotImplementedError
             switch_subscription = ImportedSwitchInactive.from_product_id(product_id, partner["partner_id"])
 
         switch_subscription.switch.fqdn = fqdn or faker.domain_name(levels=4)

test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py

@@ -4,9 +4,9 @@ import pytest
 from orchestrator.types import SubscriptionLifecycle
 
 from gso.products import ProductName
-from gso.products.product_blocks.lan_switch_interconnect import LanSwitchInterconnectAddressSpace
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.types.ip_address import AddressSpace
 from test.services.conftest import MockedNetboxClient
 from test.workflows import assert_complete, extract_state, run_workflow
 
@@ -28,7 +28,7 @@ def _netbox_client_mock():
 
 @pytest.fixture()
 def input_form_data(faker, router_subscription_factory, switch_subscription_factory):
-    def _generate_form_data(address_space: LanSwitchInterconnectAddressSpace):
+    def _generate_form_data(address_space: AddressSpace):
         return [
             {
                 "product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
@@ -55,9 +55,7 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact
     return _generate_form_data
 
 
-@pytest.mark.parametrize(
-    "address_space", [LanSwitchInterconnectAddressSpace.PRIVATE, LanSwitchInterconnectAddressSpace.PUBLIC]
-)
+@pytest.mark.parametrize("address_space", [AddressSpace.PRIVATE, AddressSpace.PUBLIC])
 @pytest.mark.workflow()
 def test_create_lan_switch_interconnect_success(
     address_space,