Add switch fixture and unit test for lan interconnect product

gso/workflows/__init__.py

@@ -56,6 +56,11 @@ LazyWorkflowInstance("gso.workflows.switch.activate_switch", "activate_switch")
 LazyWorkflowInstance("gso.workflows.switch.terminate_switch", "terminate_switch")
 LazyWorkflowInstance("gso.workflows.switch.validate_switch", "validate_switch")
 
+#  LAN Switch Interconnect workflows
+LazyWorkflowInstance(
+    "gso.workflows.lan_switch_interconnect.create_lan_switch_interconnect", "create_lan_switch_interconnect"
+)
+
 #  Site workflows
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")

test/conftest.py

@@ -48,6 +48,7 @@ from test.fixtures import (  # noqa: F401
     service_binding_port_factory,
     site_subscription_factory,
     super_pop_switch_subscription_factory,
+    switch_subscription_factory,
 )
 
 logging.getLogger("faker.factory").setLevel(logging.WARNING)
@@ -110,6 +111,12 @@ class FakerProvider(BaseProvider):
     def network_interface(self) -> str:
         return self.generator.numerify("ge-@#/@#/@#")
 
+    def juniper_ae_interface_name(self) -> str:
+        return self.generator.numerify("ae@#")
+
+    def nokia_lag_interface_name(self) -> str:
+        return self.generator.numerify("lag-@#")
+
     def link_members_juniper(self) -> LAGMemberList[LAGMember]:
         iface_amount = self.generator.random_int(min=2, max=5)
         interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]

test/fixtures/__init__.py

@@ -11,6 +11,7 @@ from test.fixtures.opengear_fixtures import opengear_subscription_factory
 from test.fixtures.router_fixtures import router_subscription_factory
 from test.fixtures.site_fixtures import site_subscription_factory
 from test.fixtures.super_pop_switch_fixtures import super_pop_switch_subscription_factory
+from test.fixtures.switch_fixtures import switch_subscription_factory
 
 __all__ = [
     "bgp_session_subscription_factory",
@@ -25,4 +26,5 @@ __all__ = [
     "service_binding_port_factory",
     "site_subscription_factory",
     "super_pop_switch_subscription_factory",
+    "switch_subscription_factory",
 ]

test/fixtures/switch_fixtures.py

+import pytest
+from orchestrator.db import db
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+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.services.subscriptions import get_product_id_by_name
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import PortNumber
+
+
+@pytest.fixture()
+def switch_subscription_factory(faker, geant_partner, site_subscription_factory):
+    def subscription_create(
+        partner: dict | None = None,
+        description: str | None = None,
+        start_date: str | None = "2024-01-01T10:20:30+01:02",
+        fqdn: str | None = None,
+        ts_port: PortNumber | None = None,
+        site: UUIDstr | None = None,
+        switch_vendor: Vendor | None = None,
+        switch_model: SwitchModel | None = None,
+        status: SubscriptionLifecycle | 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.SWITCH)
+            switch_subscription = SwitchInactive.from_product_id(product_id, partner["partner_id"])
+        else:
+            product_id = get_product_id_by_name(ProductName.IMPORTED_SWITCH)
+            raise NotImplemented
+            switch_subscription = ImportedSwitchInactive.from_product_id(product_id, partner["partner_id"])
+
+        switch_subscription.switch.fqdn = fqdn or faker.domain_name(levels=4)
+        switch_subscription.switch.ts_port = ts_port or faker.port_number(is_user=True)
+        switch_subscription.switch.site = site or Site.from_subscription(site_subscription_factory()).site
+        switch_subscription.switch.switch_vendor = switch_vendor or Vendor.JUNIPER
+        switch_subscription.switch.switch_model = switch_model or SwitchModel.EX3400
+
+        switch_subscription = SubscriptionModel.from_other_lifecycle(switch_subscription, SubscriptionLifecycle.ACTIVE)
+        switch_subscription.insync = True
+        switch_subscription.description = description or faker.sentence()
+        switch_subscription.start_date = start_date
+
+        if status:
+            switch_subscription.status = status
+
+        switch_subscription.save()
+        db.session.commit()
+
+        return str(switch_subscription.subscription_id)
+
+    return subscription_create

test/services/conftest.py

@@ -64,6 +64,9 @@ class MockedNetboxClient:
     def delete_interface():
         return None
 
+    def get_available_lags_in_range(self):
+        return self.get_available_lags()
+
 
 class MockedSharePointClient:
     class BaseMockObject:

test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py

+from unittest.mock import patch
+
+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 test.services.conftest import MockedNetboxClient
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.fixture()
+def _netbox_client_mock():
+    # Mock NetboxClient methods
+    with (
+        patch("gso.services.netbox_client.NetboxClient.get_device_by_name") as mock_get_device_by_name,
+        patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") as mock_get_available_interfaces,
+        patch("gso.services.netbox_client.NetboxClient.get_available_lags_in_range") as mock_available_lags_in_range,
+    ):
+        mock_get_device_by_name.return_value = MockedNetboxClient().get_device_by_name()
+        mock_get_available_interfaces.return_value = MockedNetboxClient().get_available_interfaces()
+        mock_available_lags_in_range.return_value = MockedNetboxClient().get_available_lags_in_range()
+
+        yield
+
+
+@pytest.fixture()
+def input_form_data(faker, router_subscription_factory, switch_subscription_factory):
+    def _generate_form_data(address_space: LanSwitchInterconnectAddressSpace):
+        return [
+            {
+                "product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
+            },
+            {
+                "tt_number": faker.tt_number(),
+                "router_side": router_subscription_factory(),
+                "switch_side": switch_subscription_factory(),
+                "address_space": address_space,
+                "description": faker.sentence(),
+                "minimum_link_count": 2,
+                "vlan_id": 111,  # VLAN ID for new interconnections is always 111
+            },
+            {
+                "router_side_iface": "lag-4",
+                "router_side_ae_members": faker.link_members_nokia()[:2],
+            },
+            {
+                "switch_side_iface": "ae9",
+                "switch_side_ae_members": faker.link_members_juniper()[:2],
+            },
+        ]
+
+    return _generate_form_data
+
+
+@pytest.mark.parametrize(
+    "address_space", [LanSwitchInterconnectAddressSpace.PRIVATE, LanSwitchInterconnectAddressSpace.PUBLIC]
+)
+@pytest.mark.workflow()
+def test_create_lan_switch_interconnect_success(
+    address_space,
+    input_form_data,
+    _netbox_client_mock,  # noqa: PT019
+):
+    initial_data = input_form_data(address_space)
+
+    result, _, _ = run_workflow("create_lan_switch_interconnect", initial_data)
+
+    assert_complete(result)
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = LanSwitchInterconnect.from_subscription(subscription_id)
+    assert subscription.status == SubscriptionLifecycle.ACTIVE