Add properties for switch and DCN management VLAN IDs to LAN Switch Interconnect

gso/migrations/versions/2024-12-09_818d4ffe65df_add_vlan_ids_to_lan_switch_interconnect.py

+"""Add VLAN IDs to LAN Switch Interconnect.
+
+Revision ID: 818d4ffe65df
+Revises: fc7bd696014e
+Create Date: 2024-12-09 11:11:35.239599
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '818d4ffe65df'
+down_revision = 'fc7bd696014e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('switch_management_vlan_id', 'VLAN ID of the switch management network') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('dcn_management_vlan_id', 'VLAN ID of the DCN management network') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('switch_management_vlan_id')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('dcn_management_vlan_id')))
+    """))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('switch_management_vlan_id'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('switch_management_vlan_id'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('dcn_management_vlan_id'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('dcn_management_vlan_id'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('switch_management_vlan_id', 'dcn_management_vlan_id'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('switch_management_vlan_id', 'dcn_management_vlan_id')
+    """))

gso/products/product_blocks/lan_switch_interconnect.py

@@ -6,6 +6,7 @@ from orchestrator.types import SubscriptionLifecycle
 from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
 from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
 from gso.utils.types.interfaces import LAGMemberList
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
 class LanSwitchInterconnectInterfaceBlockInactive(
@@ -110,6 +111,8 @@ class LanSwitchInterconnectBlockInactive(
 
     lan_switch_interconnect_description: str | None = None
     minimum_links: int | None = None
+    switch_management_vlan_id: VLAN_ID | None = None
+    dcn_management_vlan_id: VLAN_ID | None = None
     router_side: LanSwitchInterconnectRouterSideBlockInactive
     switch_side: LanSwitchInterconnectSwitchSideBlockInactive
 
@@ -121,6 +124,8 @@ class LanSwitchInterconnectBlockProvisioning(
 
     lan_switch_interconnect_description: str | None = None
     minimum_links: int | None = None
+    switch_management_vlan_id: VLAN_ID
+    dcn_management_vlan_id: VLAN_ID | None
     router_side: LanSwitchInterconnectRouterSideBlockProvisioning
     switch_side: LanSwitchInterconnectSwitchSideBlockProvisioning
 
@@ -132,6 +137,10 @@ class LanSwitchInterconnectBlock(LanSwitchInterconnectBlockProvisioning, lifecyc
     lan_switch_interconnect_description: str
     #: The minimum amount of links the LAN Switch Interconnect should consist of.
     minimum_links: int
+    #: VLAN ID for the switch management network.
+    switch_management_vlan_id: VLAN_ID
+    #: VLAN ID for the DCN management network, if the site of this product contains optical equipment.
+    dcn_management_vlan_id: VLAN_ID | None
     #: The router side of the LAN Switch Interconnect.
     router_side: LanSwitchInterconnectRouterSideBlock
     #: The switch side of the LAN Switch Interconnect.

gso/utils/types/virtual_identifiers.py

@@ -13,3 +13,6 @@ VC_ID = Annotated[
         "A Virtual Circuit ID, the upper limit comes from the highest number that a service ID could be in Nokia srOS."
     ),
 ]
+
+DEFAULT_SWITCH_MANAGEMENT_VLAN_ID: VLAN_ID = 998
+DEFAULT_DCN_MANAGEMENT_VLAN_ID: VLAN_ID = 103

gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py

@@ -21,12 +21,19 @@ 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.virtual_identifiers import (
+    DEFAULT_DCN_MANAGEMENT_VLAN_ID,
+    DEFAULT_SWITCH_MANAGEMENT_VLAN_ID,
+    VLAN_ID,
+)
 
 
 def _initial_input_form_generator() -> FormGenerator:
     class ImportLanSwitchInterconnect(FormPage):
         lan_switch_interconnect_description: str
         minimum_links: int
+        switch_management_vlan_id: VLAN_ID | None = DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
+        dcn_management_vlan_id: VLAN_ID | None = DEFAULT_DCN_MANAGEMENT_VLAN_ID
         router_side: LanSwitchInterconnectRouterSideImportModel
         switch_side: LanSwitchInterconnectSwitchSideImportModel
 
@@ -49,12 +56,16 @@ def initialize_subscription(
     subscription: ImportedLanSwitchInterconnectInactive,
     lan_switch_interconnect_description: str,
     minimum_links: int,
+    switch_management_vlan_id: VLAN_ID,
+    dcn_management_vlan_id: VLAN_ID,
     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.minimum_links = minimum_links
+    subscription.lan_switch_interconnect.switch_management_vlan_id = switch_management_vlan_id
+    subscription.lan_switch_interconnect.dcn_management_vlan_id = dcn_management_vlan_id
 
     router_block = Router.from_subscription(router_side.pop("node")).router
     router_side_interfaces = [

gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@@ -13,7 +13,7 @@ from orchestrator.workflow import StepList, begin, done, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import AfterValidator, ConfigDict
-from pydantic_forms.validators import Divider, ReadOnlyField
+from pydantic_forms.validators import ReadOnlyField
 
 from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectInterfaceBlockInactive,
@@ -42,6 +42,7 @@ from gso.utils.types.interfaces import (
     validate_interface_names_are_unique,
 )
 from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.virtual_identifiers import DEFAULT_DCN_MANAGEMENT_VLAN_ID, DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
 from gso.workflows.shared import create_summary_form
 
 
@@ -55,8 +56,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
         switch_side: active_switch_selector()  # type: ignore[valid-type]
         description: str
         minimum_link_count: int
-        divider: Divider
-        vlan_id: ReadOnlyField(111, default_type=int)  # type: ignore[valid-type]
 
     initial_input = yield CreateLANSwitchInterconnectForm
     router = Router.from_subscription(initial_input.router_side)
@@ -113,7 +112,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
         "switch_side",
         "description",
         "minimum_link_count",
-        "vlan_id",
         "router_side_iface",
         "router_side_ae_members",
         "switch_side_iface",
@@ -147,6 +145,7 @@ def initialize_subscription(
     """Update the product model with all input from the operator."""
     subscription.lan_switch_interconnect.lan_switch_interconnect_description = description
     subscription.lan_switch_interconnect.minimum_links = minimum_link_count
+    subscription.lan_switch_interconnect.switch_management_vlan_id = DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
     subscription.lan_switch_interconnect.router_side.node = Router.from_subscription(router_side).router
     subscription.lan_switch_interconnect.router_side.ae_iface = router_side_iface
     for member in router_side_ae_members:
@@ -159,6 +158,8 @@ def initialize_subscription(
         subscription.lan_switch_interconnect.switch_side.ae_members.append(
             LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
         )
+    if subscription.lan_switch_interconnect.router_side.node.router_site.site_contains_optical_equipment:
+        subscription.lan_switch_interconnect.dcn_management_vlan_id = DEFAULT_DCN_MANAGEMENT_VLAN_ID
 
     return {"subscription": subscription}
 

test/fixtures/lan_switch_interconnect_fixtures.py

@@ -18,6 +18,11 @@ from gso.products.product_types.lan_switch_interconnect import (
 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.virtual_identifiers import (
+    DEFAULT_DCN_MANAGEMENT_VLAN_ID,
+    DEFAULT_SWITCH_MANAGEMENT_VLAN_ID,
+    VLAN_ID,
+)
 
 
 @pytest.fixture()
@@ -31,6 +36,8 @@ def lan_switch_interconnect_subscription_factory(
         start_date: str | None = "2024-01-01T10:20:30+01:02",
         lan_switch_interconnect_description: str | None = None,
         minimum_links: int | None = None,
+        switch_management_vlan_id: VLAN_ID | None = None,
+        dcn_management_vlan_id: VLAN_ID | 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,
@@ -78,6 +85,12 @@ def lan_switch_interconnect_subscription_factory(
             ae_iface=switch_side_ae_iface or faker.network_interface(),
             ae_members=switch_side_ae_members,
         )
+        subscription.lan_switch_interconnect.dcn_management_vlan_id = (
+            dcn_management_vlan_id or DEFAULT_DCN_MANAGEMENT_VLAN_ID
+        )
+        subscription.lan_switch_interconnect.switch_management_vlan_id = (
+            switch_management_vlan_id or DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
+        )
 
         subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
         subscription.insync = True

test/workflows/lan_switch_interconnect/test_create_imported_lan_switch_interconnect.py

@@ -3,6 +3,7 @@ from orchestrator.types import SubscriptionLifecycle
 
 from gso.products import ProductName
 from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnect
+from gso.utils.types.virtual_identifiers import DEFAULT_DCN_MANAGEMENT_VLAN_ID, DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
 from test.workflows import (
     assert_complete,
     extract_state,
@@ -37,3 +38,24 @@ def test_create_imported_lan_switch_interconnect_success(workflow_input_data):
     assert_complete(result)
     assert subscription.product.name == ProductName.IMPORTED_LAN_SWITCH_INTERCONNECT
     assert subscription.status == SubscriptionLifecycle.ACTIVE
+    assert subscription.lan_switch_interconnect.dcn_management_vlan_id == DEFAULT_DCN_MANAGEMENT_VLAN_ID
+    assert subscription.lan_switch_interconnect.switch_management_vlan_id == DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
+
+
+@pytest.mark.workflow()
+def test_create_imported_lan_switch_interconnect_custom_vlan_ids(faker, workflow_input_data):
+    custom_switch_vlan_id = faker.vlan_id()
+    custom_dcn_vlan_id = faker.vlan_id()
+    workflow_input_data.update({
+        "switch_management_vlan_id": custom_switch_vlan_id,
+        "dcn_management_vlan_id": custom_dcn_vlan_id,
+    })
+    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
+    assert subscription.lan_switch_interconnect.dcn_management_vlan_id == custom_dcn_vlan_id
+    assert subscription.lan_switch_interconnect.switch_management_vlan_id == custom_switch_vlan_id

test/workflows/lan_switch_interconnect/test_create_lan_switch_interconnect.py

@@ -6,6 +6,7 @@ from orchestrator.types import SubscriptionLifecycle
 from gso.products import ProductName
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.types.virtual_identifiers import DEFAULT_DCN_MANAGEMENT_VLAN_ID
 from test.services.conftest import MockedNetboxClient
 from test.workflows import assert_complete, extract_state, run_workflow
 
@@ -26,43 +27,51 @@ def _netbox_client_mock():
 
 
 @pytest.fixture()
-def input_form_data(faker, router_subscription_factory, switch_subscription_factory):
-    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(),
-            "description": faker.sentence(),
-            "minimum_link_count": 2,
-            "vlan_id": 111,  # VLAN ID for new interconnections is always 111
-        },
-        {
-            "router_side_iface": "lag-1",
-            "router_side_ae_members": faker.link_members_nokia()[:2],
-        },
-        {
-            "switch_side_iface": faker.network_interface(),
-            "switch_side_ae_members": faker.link_members_juniper()[:2],
-        },
-        {},
-    ]
+def input_form_data(faker, router_subscription_factory, switch_subscription_factory, site_subscription_factory):
+    def _input_form_data(site_contains_optical_equipment):
+        return [
+            {
+                "product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
+            },
+            {
+                "tt_number": faker.tt_number(),
+                "router_side": router_subscription_factory(
+                    router_site=site_subscription_factory(
+                        site_contains_optical_equipment=site_contains_optical_equipment
+                    )
+                ),
+                "switch_side": switch_subscription_factory(),
+                "description": faker.sentence(),
+                "minimum_link_count": 2,
+            },
+            {
+                "router_side_iface": "lag-1",
+                "router_side_ae_members": faker.link_members_nokia()[:2],
+            },
+            {
+                "switch_side_iface": faker.network_interface(),
+                "switch_side_ae_members": faker.link_members_juniper()[:2],
+            },
+            {},
+        ]
+
+    return _input_form_data
 
 
 @pytest.mark.workflow()
-@patch("gso.services.infoblox.create_v6_network_by_ip")
-@patch("gso.services.infoblox.create_v4_network_by_ip")
-@patch("gso.services.infoblox.create_host_by_ip")
+@pytest.mark.parametrize("site_contains_optical_equipment", [True, False])
+@patch("gso.workflows.lan_switch_interconnect.create_lan_switch_interconnect.create_v6_network_by_ip")
+@patch("gso.workflows.lan_switch_interconnect.create_lan_switch_interconnect.create_v4_network_by_ip")
+@patch("gso.workflows.lan_switch_interconnect.create_lan_switch_interconnect.create_host_by_ip")
 def test_create_lan_switch_interconnect_success(
     mock_create_host,
     mock_create_v4_network,
     mock_create_v6_network,
+    site_contains_optical_equipment,
     input_form_data,
     _netbox_client_mock,  # noqa: PT019
 ):
-    result, _, _ = run_workflow("create_lan_switch_interconnect", input_form_data)
+    result, _, _ = run_workflow("create_lan_switch_interconnect", input_form_data(site_contains_optical_equipment))
 
     assert_complete(result)
     state = extract_state(result)
@@ -72,3 +81,6 @@ def test_create_lan_switch_interconnect_success(
     assert mock_create_v4_network.call_count == 1
     assert mock_create_v6_network.call_count == 1
     assert mock_create_host.call_count == 4
+    assert subscription.lan_switch_interconnect.dcn_management_vlan_id == (
+        DEFAULT_DCN_MANAGEMENT_VLAN_ID if site_contains_optical_equipment else None
+    )