Update unit test for LAN switch interconnect creation

gso/migrations/versions/2024-10-28_2bbe91a01715_update_switch_pop_vlan_and_switch_.py

+"""Update Switch, PoP VLAN, and Switch interconnect product models.
+
+Revision ID: 2bbe91a01715
+Revises: 0e7e7d749617
+Create Date: 2024-10-28 15:18:54.138252
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '2bbe91a01715'
+down_revision = '0e7e7d749617'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+UPDATE resource_types SET resource_type='ts_port' WHERE resource_types.resource_type = 'switch_ts_port'
+    """))
+    conn.execute(sa.text("""
+UPDATE resource_types SET resource_type='fqdn' WHERE resource_types.resource_type = 'switch_hostname'
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('lan_switch_interconnect_ip_network', 'IP resources for a LAN Switch Interconnect') 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 ('lan_switch_interconnect_ip_network')))
+    """))
+
+
+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 ('lan_switch_interconnect_ip_network'))
+    """))
+    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 ('lan_switch_interconnect_ip_network'))
+    """))
+    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 ('lan_switch_interconnect_ip_network'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('lan_switch_interconnect_ip_network')
+    """))
+    conn.execute(sa.text("""
+UPDATE resource_types SET resource_type='switch_ts_port' WHERE resource_types.resource_type = 'ts_port'
+    """))
+    conn.execute(sa.text("""
+UPDATE resource_types SET resource_type='switch_hostname' WHERE resource_types.resource_type = 'fqdn'
+    """))

gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@@ -44,6 +44,7 @@ def _initial_input_form(product_name: str) -> FormGenerator:
         model_config = ConfigDict(title=product_name)
 
         tt_number: TTNumber
+        partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
         router_side: active_router_selector()  # type: ignore[valid-type]
         switch_side: active_switch_selector()  # type: ignore[valid-type]
         address_space: LanSwitchInterconnectAddressSpace
@@ -78,7 +79,7 @@ def _initial_input_form(product_name: str) -> FormGenerator:
     class InterconnectRouterSideForm(FormPage):
         model_config = ConfigDict(title="Please enter interface names and descriptions for the router side.")
 
-        router_side_iface: available_lags_choices(user_input.router_side) or JuniperAEInterface  # type: ignore[valid-type]
+        router_side_iface: available_lags_choices(user_input.router_side) or JuniperAEInterface or str  # type: ignore[valid-type]
         router_side_ae_members: router_side_ae_member_list
 
     router_side_input = yield InterconnectRouterSideForm
@@ -108,7 +109,7 @@ def create_subscription(product: UUIDstr, partner: str) -> State:
     """Create a new subscription object in the database."""
     subscription = LanSwitchInterconnectInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])
 
-    return {"subscription": subscription}
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
 
 
 @step("Initialize subscription")
@@ -134,7 +135,7 @@ def initialize_subscription(
         subscription.lan_switch_interconnect.router_side.ae_members.append(
             LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
         )
-    subscription.lan_switch_interconnect.switch_side.node = Switch.from_subscription(switch_side).switch
+    subscription.lan_switch_interconnect.switch_side.switch = Switch.from_subscription(switch_side).switch
     subscription.lan_switch_interconnect.switch_side.ae_iface = switch_side_iface
     for member in switch_side_ae_members:
         subscription.lan_switch_interconnect.switch_side.ae_members.append(

gso/workflows/switch/create_switch.py

@@ -85,6 +85,7 @@ def initialize_subscription(
     subscription.switch.ts_port = ts_port
     subscription.switch.switch_vendor = vendor
     subscription.switch.switch_model = model
+    subscription.description = f"Switch {subscription.switch.fqdn}"
 
     return {"subscription": subscription}
 

test/conftest.py

@@ -109,7 +109,9 @@ class FakerProvider(BaseProvider):
         return self.generator.random_int(min=1, max=128)
 
     def network_interface(self) -> str:
-        return self.generator.numerify("ge-@#/@#/@#")
+        interface = self.generator.random_choices(elements=("ge", "et", "xe"))[0]
+        number = self.generator.numerify("-%/%/%")
+        return f"{interface}{number}"
 
     def juniper_ae_interface_name(self) -> str:
         return self.generator.numerify("ae@#")

test/fixtures/switch_fixtures.py

@@ -35,7 +35,7 @@ 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 NotImplemented
+            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

@@ -43,11 +43,11 @@ def input_form_data(faker, router_subscription_factory, switch_subscription_fact
                 "vlan_id": 111,  # VLAN ID for new interconnections is always 111
             },
             {
-                "router_side_iface": "lag-4",
+                "router_side_iface": "lag-1",
                 "router_side_ae_members": faker.link_members_nokia()[:2],
             },
             {
-                "switch_side_iface": "ae9",
+                "switch_side_iface": faker.network_interface(),
                 "switch_side_ae_members": faker.link_members_juniper()[:2],
             },
         ]