Added import endpoints for SRXes and QFXes

gso/api/v1/imports.py

@@ -11,11 +11,12 @@ from pydantic import BaseModel, root_validator, validator
 
 from gso.auth.security import opa_security_default
 from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
-from gso.products.product_blocks.router import RouterRole, RouterVendor
+from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.site import SiteTier
 from gso.services import subscriptions
 from gso.services.crm import CustomerNotFoundError, get_customer_by_name
 from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
+from gso.utils.shared_choices import PortNumber, Vendor
 
 router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
 
@@ -50,7 +51,7 @@ class RouterImportModel(BaseModel):
     router_site: str
     hostname: str
     ts_port: int
-    router_vendor: RouterVendor
+    router_vendor: Vendor
     router_role: RouterRole
     router_lo_ipv4_address: ipaddress.IPv4Address
     router_lo_ipv6_address: ipaddress.IPv6Address
@@ -136,6 +137,27 @@ class IptrunkImportModel(BaseModel):
         return values
 
 
+class SuperPopSwitchImportModel(BaseModel):
+    """Required fields for importing an existing :class:`gso.product.product_types.super_pop_switch`."""
+
+    customer: str
+    super_pop_switch_site: str
+    hostname: str
+    super_pop_switch_ts_port: PortNumber
+    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+
+
+class OfficeRouterImportModel(BaseModel):
+    """Required fields for importing an existing :class:`gso.product.product_types.office_router`."""
+
+    customer: str
+    office_router_site: str
+    office_router_fqdn: str
+    office_router_ts_port: PortNumber
+    office_router_lo_ipv4_address: ipaddress.IPv4Address
+    office_router_lo_ipv6_address: ipaddress.IPv6Address
+
+
 def _start_process(process_name: str, data: dict) -> UUID:
     """Start a process and handle common exceptions."""
     pid: UUID = processes.start_process(process_name, [data])
@@ -184,7 +206,7 @@ def import_router(router_data: RouterImportModel) -> dict[str, Any]:
     :raises HTTPException: If there's an error in the process.
     """
     pid = _start_process("import_router", router_data.dict())
-    return {"detail": "Router added successfully", "pid": pid}
+    return {"detail": "Router has been added successfully", "pid": pid}
 
 
 @router.post("/iptrunks", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
@@ -200,4 +222,36 @@ def import_iptrunk(iptrunk_data: IptrunkImportModel) -> dict[str, Any]:
     :raises HTTPException: If there's an error in the process.
     """
     pid = _start_process("import_iptrunk", iptrunk_data.dict())
-    return {"detail": "Iptrunk added successfully", "pid": pid}
+    return {"detail": "Iptrunk has been added successfully", "pid": pid}
+
+
+@router.post("/super-pop-switches", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def import_super_pop_switch(super_pop_switch_data: SuperPopSwitchImportModel) -> dict[str, Any]:
+    """Import a Super PoP switch by running the import_super_pop_switch workflow.
+
+    :param super_pop_switch_data: The Super PoP switch information to be imported.
+    :type super_pop_switch_data: SuperPopSwitchImportModel
+
+    :return: A dictionary containing the process id of the started process and detail message.
+    :rtype: dict[str, Any]
+
+    :raises HTTPException: If there's an error in the process.
+    """
+    pid = _start_process("import_super_pop_switch", super_pop_switch_data.dict())
+    return {"detail": "Super PoP switch has been added successfully", "pid": pid}
+
+
+@router.post("/office-routers", status_code=status.HTTP_201_CREATED, response_model=ImportResponseModel)
+def import_office_router(office_router_data: OfficeRouterImportModel) -> dict[str, Any]:
+    """Import a office router by running the import_office_router workflow.
+
+    :param office_router_data: The office router information to be imported.
+    :type office_router_data: OfficeRouterImportModel
+
+    :return: A dictionary containing the process id of the started process and detail message.
+    :rtype: dict[str, Any]
+
+    :raises HTTPException: If there's an error in the process.
+    """
+    pid = _start_process("import_office_router", office_router_data.dict())
+    return {"detail": "Office router has been added successfully", "pid": pid}

gso/cli/imports.py

@@ -11,17 +11,23 @@ from pydantic import ValidationError
 
 from gso.api.v1.imports import (
     IptrunkImportModel,
+    OfficeRouterImportModel,
     RouterImportModel,
     SiteImportModel,
+    SuperPopSwitchImportModel,
     import_iptrunk,
+    import_office_router,
     import_router,
     import_site,
+    import_super_pop_switch,
 )
 from gso.services.subscriptions import get_active_subscriptions_by_field_and_value
 
 app: typer.Typer = typer.Typer()
 
-T = TypeVar("T", SiteImportModel, RouterImportModel, IptrunkImportModel)
+T = TypeVar(
+    "T", SiteImportModel, RouterImportModel, IptrunkImportModel, SuperPopSwitchImportModel, OfficeRouterImportModel
+)
 
 common_filepath_option = typer.Option(
     default="data.json",
@@ -89,6 +95,20 @@ def import_routers(filepath: str = common_filepath_option) -> None:
     generic_import_data(filepath, RouterImportModel, import_router, "hostname")
 
 
+@app.command()
+def import_super_pop_switches(filepath: str = common_filepath_option) -> None:
+    """Import Super PoP Switches into GSO."""
+    # Use the import_data function to handle common import logic
+    generic_import_data(filepath, SuperPopSwitchImportModel, import_super_pop_switch, "hostname")
+
+
+@app.command()
+def import_office_routers(filepath: str = common_filepath_option) -> None:
+    """Import office routers into GSO."""
+    # Use the import_data function to handle common import logic
+    generic_import_data(filepath, OfficeRouterImportModel, import_office_router, "office_router_fqdn")
+
+
 def get_router_subscription_id(node_name: str) -> str | None:
     """Get the subscription id for a router by its node name."""
     subscriptions = get_active_subscriptions_by_field_and_value(

gso/migrations/versions/2024-02-14_1c3c90ea1d8c_add_super_pop_switch_product.py

+"""Add Super PoP switch product..
+
+Revision ID: 1c3c90ea1d8c
+Revises: bacd55c26106
+Create Date: 2024-02-14 11:00:11.858023
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '1c3c90ea1d8c'
+down_revision = 'bacd55c26106'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Super PoP switch', 'Super PoP switch', 'SuperPopSwitch', 'Super_POP_SWITCH', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('SuperPopSwitchBlock', 'Super PoP switch block', 'SUPER_POP_SWITCH_BLK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('super_pop_switch_ts_port', 'The port of the terminal server that this Super PoP switch is connected to. Used to offer out of band access.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('super_pop_switch_fqdn', 'Super PoP switch FQDN.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('super_pop_switch_mgmt_ipv4_address', 'The IPv4 management address of the Super PoP switch.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_product_blocks (product_id, product_block_id) VALUES ((SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock')))
+    """))
+    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 ('SuperPopSwitchBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_fqdn')))
+    """))
+    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 ('SuperPopSwitchBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_ts_port')))
+    """))
+    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 ('SuperPopSwitchBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_mgmt_ipv4_address')))
+    """))
+    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 ('SuperPopSwitchBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor')))
+    """))
+
+
+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 ('SuperPopSwitchBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_fqdn'))
+    """))
+    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 ('SuperPopSwitchBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_fqdn'))
+    """))
+    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 ('SuperPopSwitchBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_ts_port'))
+    """))
+    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 ('SuperPopSwitchBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_ts_port'))
+    """))
+    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 ('SuperPopSwitchBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_mgmt_ipv4_address'))
+    """))
+    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 ('SuperPopSwitchBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_mgmt_ipv4_address'))
+    """))
+    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 ('SuperPopSwitchBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor'))
+    """))
+    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 ('SuperPopSwitchBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor'))
+    """))
+    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 ('super_pop_switch_ts_port', 'super_pop_switch_fqdn', 'super_pop_switch_mgmt_ipv4_address'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('super_pop_switch_ts_port', 'super_pop_switch_fqdn', 'super_pop_switch_mgmt_ipv4_address')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_product_blocks WHERE product_product_blocks.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch')) AND product_product_blocks.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_blocks WHERE product_blocks.name IN ('SuperPopSwitchBlock')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes WHERE processes.pid IN (SELECT processes_subscriptions.pid FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch'))))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Super PoP switch'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM products WHERE products.name IN ('Super PoP switch')
+    """))

gso/migrations/versions/2024-02-20_6e4952687205_add_office_router_product.py

+"""Add office router product..
+
+Revision ID: 6e4952687205
+Revises: 1c3c90ea1d8c
+Create Date: 2024-02-20 11:22:30.804258
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '6e4952687205'
+down_revision = '1c3c90ea1d8c'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Office router', 'Office router product', 'OfficeRouter', 'OFFICE_ROUTER', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('OfficeRouterBlock', 'Office router product block', 'OFFICE_ROUTER_BLOCK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('office_router_fqdn', 'Office router FQDN') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('office_router_ts_port', 'The port of the terminal server that this office router is connected to') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('office_router_lo_ipv4_address', 'The IPv4 loopback address of the office router.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('office_router_lo_ipv6_address', 'The IPv6 loopback address of the office router.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_product_blocks (product_id, product_block_id) VALUES ((SELECT products.product_id FROM products WHERE products.name IN ('Office router')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock')))
+    """))
+    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 ('OfficeRouterBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_fqdn')))
+    """))
+    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 ('OfficeRouterBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_ts_port')))
+    """))
+    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 ('OfficeRouterBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv4_address')))
+    """))
+    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 ('OfficeRouterBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv6_address')))
+    """))
+    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 ('OfficeRouterBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor')))
+    """))
+
+
+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 ('OfficeRouterBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_fqdn'))
+    """))
+    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 ('OfficeRouterBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_fqdn'))
+    """))
+    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 ('OfficeRouterBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_ts_port'))
+    """))
+    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 ('OfficeRouterBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_ts_port'))
+    """))
+    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 ('OfficeRouterBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv4_address'))
+    """))
+    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 ('OfficeRouterBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv4_address'))
+    """))
+    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 ('OfficeRouterBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv6_address'))
+    """))
+    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 ('OfficeRouterBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('office_router_lo_ipv6_address'))
+    """))
+    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 ('OfficeRouterBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor'))
+    """))
+    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 ('OfficeRouterBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vendor'))
+    """))
+    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 ('office_router_fqdn', 'office_router_ts_port', 'office_router_lo_ipv4_address', 'office_router_lo_ipv6_address'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('office_router_fqdn', 'office_router_ts_port', 'office_router_lo_ipv4_address', 'office_router_lo_ipv6_address')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_product_blocks WHERE product_product_blocks.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Office router')) AND product_product_blocks.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('SiteBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_blocks WHERE product_blocks.name IN ('OfficeRouterBlock')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes WHERE processes.pid IN (SELECT processes_subscriptions.pid FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Office router'))))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Office router')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Office router')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Office router'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM products WHERE products.name IN ('Office router')
+    """))

gso/products/__init__.py

@@ -9,8 +9,10 @@ from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
 from pydantic_forms.types import strEnum
 
 from gso.products.product_types.iptrunk import Iptrunk
+from gso.products.product_types.office_router import OfficeRouter
 from gso.products.product_types.router import Router
 from gso.products.product_types.site import Site
+from gso.products.product_types.super_pop_switch import SuperPopSwitch
 
 
 class ProductType(strEnum):
@@ -19,6 +21,8 @@ class ProductType(strEnum):
     IP_TRUNK = "IP trunk"
     ROUTER = "Router"
     SITE = "Site"
+    SUPER_POP_SWITCH = "Super PoP switch"
+    OFFICE_ROUTER = "Office router"
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -26,5 +30,7 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         "IP trunk": Iptrunk,
         "Router": Router,
         "Site": Site,
+        "Super PoP switch": SuperPopSwitch,
+        "Office router": OfficeRouter,
     },
 )

gso/products/product_blocks/office_router.py

+"""Product block for :class:`office router` products."""
+
+import ipaddress
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.site import (
+    SiteBlock,
+    SiteBlockInactive,
+    SiteBlockProvisioning,
+)
+from gso.utils.shared_choices import PortNumber, Vendor
+
+
+class OfficeRouterBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="OfficeRouterBlock",
+):
+    """An office router that's being currently inactive. See :class:`OfficeRouterBlock`."""
+
+    office_router_fqdn: str | None = None
+    office_router_ts_port: PortNumber | None = None
+    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
+    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_site: SiteBlockInactive | None
+    vendor: Vendor | None = None
+
+
+class OfficeRouterBlockProvisioning(OfficeRouterBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An office router that's being provisioned. See :class:`RouterBlock`."""
+
+    office_router_fqdn: str | None = None
+    office_router_ts_port: PortNumber | None = None
+    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
+    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_site: SiteBlockProvisioning | None
+    vendor: Vendor | None = None
+
+
+class OfficeRouterBlock(OfficeRouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An office router that's currently deployed in the network."""
+
+    #:  Office router FQDN.
+    office_router_fqdn: str
+    #:  The port of the terminal server that this office router is connected to. Used to offer out of band access.
+    office_router_ts_port: PortNumber
+    #:  The IPv4 loopback address of the office router.
+    office_router_lo_ipv4_address: ipaddress.IPv4Address
+    #:  The IPv6 loopback address of the office router.
+    office_router_lo_ipv6_address: ipaddress.IPv6Address
+    #:  The :class:`Site` that this office router resides in. Both physically and computationally.
+    office_router_site: SiteBlock
+    #:  The vendor of an office router. Defaults to Juniper.
+    vendor: Vendor = Vendor.JUNIPER

gso/products/product_blocks/router.py

@@ -4,14 +4,13 @@ import ipaddress
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
-from pydantic import ConstrainedInt
 
-from gso import settings
 from gso.products.product_blocks.site import (
     SiteBlock,
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
+from gso.utils.shared_choices import PortNumber, Vendor
 
 
 class RouterRole(strEnum):
@@ -22,23 +21,6 @@ class RouterRole(strEnum):
     AMT = "amt"
 
 
-class RouterVendor(strEnum):
-    """Enumerator for the different product vendors that are supported."""
-
-    JUNIPER = "juniper"
-    NOKIA = "nokia"
-
-
-class PortNumber(ConstrainedInt):
-    """Constrained integer for valid port numbers.
-
-    The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
-    """
-
-    gt = 0
-    le = 49151
-
-
 class RouterBlockInactive(
     ProductBlockModel,
     lifecycle=[SubscriptionLifecycle.INITIAL],
@@ -54,13 +36,7 @@ class RouterBlockInactive(
     router_lo_iso_address: str | None = None
     router_role: RouterRole | None = None
     router_site: SiteBlockInactive | None
-    vendor: RouterVendor | None = None
-
-
-def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
-    """Generate an :term:`FQDN` from a hostname, site name, and a country code."""
-    oss = settings.load_oss_params()
-    return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"
+    vendor: Vendor | None = None
 
 
 class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -74,8 +50,7 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
     router_lo_iso_address: str
     router_role: RouterRole
     router_site: SiteBlockProvisioning
-    vendor: RouterVendor
-
+    vendor: Vendor
 
 class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     """A router that's currently deployed in the network."""
@@ -97,4 +72,4 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
     #:  The :class:`Site` that this router resides in. Both physically and computationally.
     router_site: SiteBlock
     #:  The vendor of a router.
-    vendor: RouterVendor
+    vendor: Vendor

gso/products/product_blocks/super_pop_switch.py

+"""Product block for :class:`Super PoP Switch` products."""
+
+import ipaddress
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.site import (
+    SiteBlock,
+    SiteBlockInactive,
+    SiteBlockProvisioning,
+)
+from gso.utils.shared_choices import PortNumber, Vendor
+
+
+class SuperPopSwitchBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="SuperPopSwitchBlock",
+):
+    """A Super PoP switch that's being currently inactive. See :class:`SuperPopSwitchBlock`."""
+
+    super_pop_switch_fqdn: str | None = None
+    super_pop_switch_ts_port: PortNumber | None = None
+    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_site: SiteBlockInactive | None
+    vendor: Vendor | None = None
+
+
+class SuperPopSwitchBlockProvisioning(SuperPopSwitchBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Super PoP switch that's being provisioned. See :class:`SuperPopSwitchBlock`."""
+
+    super_pop_switch_fqdn: str | None = None
+    super_pop_switch_ts_port: PortNumber | None = None
+    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_site: SiteBlockProvisioning | None
+    vendor: Vendor | None = None
+
+
+class SuperPopSwitchBlock(SuperPopSwitchBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A Super PoP switch that's currently deployed in the network."""
+
+    #:  Super PoP switch FQDN.
+    super_pop_switch_fqdn: str
+    #:  The port of the terminal server that this Super PoP switch is connected to. Used to offer out of band access.
+    super_pop_switch_ts_port: PortNumber
+    #:  The IPv4 management address of the Super PoP switch.
+    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+    #:  The :class:`Site` that this Super PoP switch resides in. Both physically and computationally.
+    super_pop_switch_site: SiteBlock
+    #:  The vendor of a Super PoP switch. Defaults to Juniper.
+    vendor: Vendor = Vendor.JUNIPER

gso/products/product_types/office_router.py

+"""An office router product type."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.office_router import (
+    OfficeRouterBlock,
+    OfficeRouterBlockInactive,
+    OfficeRouterBlockProvisioning,
+)
+
+
+class OfficeRouterInactive(SubscriptionModel, is_base=True):
+    """An inactive office router."""
+
+    office_router: OfficeRouterBlockInactive
+
+
+class OfficeRouterProvisioning(OfficeRouterInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An office router that is being provisioned."""
+
+    office_router: OfficeRouterBlockProvisioning
+
+
+class OfficeRouter(OfficeRouterProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An office router that is currently active."""
+
+    office_router: OfficeRouterBlock

gso/products/product_types/super_pop_switch.py

+"""A Super PoP switch product type."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.super_pop_switch import (
+    SuperPopSwitchBlock,
+    SuperPopSwitchBlockInactive,
+    SuperPopSwitchBlockProvisioning,
+)
+
+
+class SuperPopSwitchInactive(SubscriptionModel, is_base=True):
+    """An inactive Super PoP switch."""
+
+    super_pop_switch: SuperPopSwitchBlockInactive
+
+
+class SuperPopSwitchProvisioning(SuperPopSwitchInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Super PoP switch that is being provisioned."""
+
+    super_pop_switch: SuperPopSwitchBlockProvisioning
+
+
+class SuperPopSwitch(SuperPopSwitchProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A Super PoP switch that is currently active."""
+
+    super_pop_switch: SuperPopSwitchBlock

gso/services/subscriptions.py

@@ -20,6 +20,7 @@ from orchestrator.types import SubscriptionLifecycle
 from pydantic_forms.types import UUIDstr
 
 from gso.products import ProductType
+from gso.products.product_types.site import Site
 
 SubscriptionType = dict[str, Any]
 
@@ -175,3 +176,17 @@ def count_incomplete_validate_products() -> int:
 def get_insync_subscriptions() -> list[SubscriptionTable]:
     """Retrieve all subscriptions that are currently in sync."""
     return SubscriptionTable.query.join(ProductTable).filter(SubscriptionTable.insync.is_(True)).all()
+
+
+def get_site_by_name(site_name: str) -> Site:
+    """Get a site by its name.
+
+    :param site_name: The name of the site.
+    :type site_name: str
+    """
+    subscription = get_active_subscriptions_by_field_and_value("site_name", site_name)
+    if not subscription:
+        msg = f"Site with name {site_name} not found."
+        raise ValueError(msg)
+
+    return Site.from_subscription(subscription[0].subscription_id)

gso/utils/helpers.py

@@ -12,12 +12,13 @@ from pydantic import BaseModel, validator
 from pydantic.fields import ModelField
 from pydantic_forms.validators import Choice
 
+from gso import settings
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
-from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_blocks.site import SiteTier
 from gso.products.product_types.router import Router
 from gso.services.netbox_client import NetboxClient
 from gso.services.subscriptions import get_active_subscriptions_by_field_and_value
+from gso.utils.shared_choices import Vendor
 
 
 class LAGMember(BaseModel):
@@ -44,7 +45,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
     For Nokia routers, return a list of available interfaces.
     For Juniper routers, return a string.
     """
-    if get_router_vendor(router_id) != RouterVendor.NOKIA:
+    if get_router_vendor(router_id) != Vendor.NOKIA:
         return None
     interfaces = {
         interface["name"]: f"{interface['name']}  {interface['description']}"
@@ -54,16 +55,16 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
 
 
 def available_interfaces_choices_including_current_members(
-    router_id: UUID,
-    speed: str,
-    interfaces: list[IptrunkInterfaceBlock],
+        router_id: UUID,
+        speed: str,
+        interfaces: list[IptrunkInterfaceBlock],
 ) -> Choice | None:
     """Return a list of available interfaces for a given router and speed including the current members.
 
     For Nokia routers, return a list of available interfaces.
     For Juniper routers, return a string.
     """
-    if get_router_vendor(router_id) != RouterVendor.NOKIA:
+    if get_router_vendor(router_id) != Vendor.NOKIA:
         return None
 
     available_interfaces = list(NetboxClient().get_available_interfaces(router_id, speed))
@@ -88,20 +89,20 @@ def available_lags_choices(router_id: UUID) -> Choice | None:
     For Nokia routers, return a list of available lags.
     For Juniper routers, return ``None``.
     """
-    if get_router_vendor(router_id) != RouterVendor.NOKIA:
+    if get_router_vendor(router_id) != Vendor.NOKIA:
         return None
     side_a_ae_iface_list = NetboxClient().get_available_lags(router_id)
     return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True))  # type: ignore[arg-type]
 
 
-def get_router_vendor(router_id: UUID) -> RouterVendor:
+def get_router_vendor(router_id: UUID) -> Vendor:
     """Retrieve the vendor of a router.
 
     :param router_id: The :term:`UUID` of the router.
     :type router_id: :class:`uuid.UUID`
 
     :return: The vendor of the router.
-    :rtype: RouterVendor:
+    :rtype: Vendor:
     """
     return Router.from_subscription(router_id).router.vendor
 
@@ -130,7 +131,7 @@ def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
     :rtype: :class:`UUIDstr`
     """
     router_type = Router.from_subscription(subscription_id)
-    if router_type.router.vendor == RouterVendor.NOKIA:
+    if router_type.router.vendor == Vendor.NOKIA:
         device = NetboxClient().get_device_by_name(router_type.router.router_fqdn)
         if not device:
             msg = "The selected router does not exist in Netbox."
@@ -259,7 +260,7 @@ def validate_interface_name_list(interface_name_list: list, vendor: str) -> list
                   exception.
     """
     # For Nokia nothing to do
-    if vendor == RouterVendor.NOKIA:
+    if vendor == Vendor.NOKIA:
         return interface_name_list
     pattern = re.compile(r"^(ge|et|xe)-[0-9]/[0-9]/[0-9]$")
     for interface in interface_name_list:
@@ -290,3 +291,9 @@ def validate_tt_number(tt_number: str) -> str:
         raise ValueError(err_msg)
 
     return tt_number
+
+
+def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
+    """Generate an :term:`FQDN` from a hostname, site name, and a country code."""
+    oss = settings.load_oss_params()
+    return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"

gso/utils/shared_choices.py

+"""Shared choices for the different models."""
+
+from pydantic import ConstrainedInt
+from pydantic_forms.types import strEnum
+
+
+class Vendor(strEnum):
+    """Enumerator for the different product vendors that are supported."""
+
+    JUNIPER = "juniper"
+    NOKIA = "nokia"
+
+
+class PortNumber(ConstrainedInt):
+    """Constrained integer for valid port numbers.
+
+    The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
+    """
+
+    gt = 0
+    le = 49151

gso/workflows/__init__.py

@@ -18,3 +18,5 @@ LazyWorkflowInstance("gso.workflows.site.terminate_site", "terminate_site")
 LazyWorkflowInstance("gso.workflows.tasks.import_site", "import_site")
 LazyWorkflowInstance("gso.workflows.tasks.import_router", "import_router")
 LazyWorkflowInstance("gso.workflows.tasks.import_iptrunk", "import_iptrunk")
+LazyWorkflowInstance("gso.workflows.tasks.import_super_pop_switch", "import_super_pop_switch")
+LazyWorkflowInstance("gso.workflows.tasks.import_office_router", "import_office_router")

gso/workflows/iptrunk/create_iptrunk.py

@@ -21,7 +21,7 @@ from gso.products.product_blocks.iptrunk import (
     IptrunkType,
     PhyPortCapacity,
 )
-from gso.products.product_blocks.router import RouterVendor
+
 from gso.products.product_types.iptrunk import IptrunkInactive
 from gso.products.product_types.router import Router
 from gso.services import infoblox, subscriptions
@@ -38,6 +38,7 @@ from gso.utils.helpers import (
     validate_router_in_netbox,
     validate_tt_number,
 )
+from gso.utils.shared_choices import Vendor
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -82,7 +83,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     class JuniperAeMembers(UniqueConstrainedList[LAGMember]):
         min_items = initial_user_input.iptrunk_minimum_links
 
-    if get_router_vendor(router_a) == RouterVendor.NOKIA:
+    if get_router_vendor(router_a) == Vendor.NOKIA:
 
         class NokiaLAGMemberA(LAGMember):
             interface_name: available_interfaces_choices(  # type: ignore[valid-type]
@@ -130,7 +131,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     user_input_router_side_b = yield SelectRouterSideB
     router_b = user_input_router_side_b.side_b_node_id.name
 
-    if get_router_vendor(router_b) == RouterVendor.NOKIA:
+    if get_router_vendor(router_b) == Vendor.NOKIA:
 
         class NokiaLAGMemberB(LAGMember):
             interface_name: available_interfaces_choices(  # type: ignore[valid-type]
@@ -400,7 +401,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
     """Create the :term:`LAG` interfaces in NetBox and attach the lag interfaces to the physical interfaces."""
     nbclient = NetboxClient()
     for trunk_side in subscription.iptrunk.iptrunk_sides:
-        if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
+        if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == Vendor.NOKIA:
             # Create :term:`LAG` interfaces
             lag_interface: Interfaces = nbclient.create_interface(
                 iface_name=trunk_side.iptrunk_side_ae_iface,
@@ -469,8 +470,8 @@ def create_iptrunk() -> StepList:
     * Allocate the interfaces in Netbox
     * Set the subscription to active in the database
     """
-    side_a_is_nokia = conditional(lambda state: get_router_vendor(state["side_a_node_id"]) == RouterVendor.NOKIA)
-    side_b_is_nokia = conditional(lambda state: get_router_vendor(state["side_b_node_id"]) == RouterVendor.NOKIA)
+    side_a_is_nokia = conditional(lambda state: get_router_vendor(state["side_a_node_id"]) == Vendor.NOKIA)
+    side_b_is_nokia = conditional(lambda state: get_router_vendor(state["side_b_node_id"]) == Vendor.NOKIA)
 
     return (
         init

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -25,7 +25,6 @@ from pydantic_forms.core import ReadOnlyField
 from pynetbox.models.dcim import Interfaces
 
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
-from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
 from gso.services import infoblox
@@ -41,6 +40,7 @@ from gso.utils.helpers import (
     validate_interface_name_list,
     validate_tt_number,
 )
+from gso.utils.shared_choices import Vendor
 from gso.utils.workflow_steps import set_isis_to_90000
 
 
@@ -109,7 +109,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     new_router = new_side_iptrunk_router_input.new_node
     side_a_ae_iface = available_lags_choices(new_router) or str
 
-    new_side_is_nokia = get_router_vendor(new_router) == RouterVendor.NOKIA
+    new_side_is_nokia = get_router_vendor(new_router) == Vendor.NOKIA
     if new_side_is_nokia:
 
         class NokiaLAGMember(LAGMember):
@@ -155,7 +155,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
         @validator("new_lag_interface", allow_reuse=True, pre=True, always=True)
         def lag_interface_proper_name(cls, new_lag_interface: str) -> str:
-            if get_router_vendor(new_router) == RouterVendor.JUNIPER:
+            if get_router_vendor(new_router) == Vendor.JUNIPER:
                 juniper_lag_re = re.compile("^ae\\d{1,2}$")
                 if not juniper_lag_re.match(new_lag_interface):
                     msg = "Invalid LAG name, please try again."
@@ -643,10 +643,10 @@ def migrate_iptrunk() -> StepList:
     * Update the subscription model in the database
     * Update the reserved interfaces in Netbox
     """
-    new_side_is_nokia = conditional(lambda state: get_router_vendor(state["new_node"]) == RouterVendor.NOKIA)
+    new_side_is_nokia = conditional(lambda state: get_router_vendor(state["new_node"]) == Vendor.NOKIA)
     old_side_is_nokia = conditional(
         lambda state: get_router_vendor(state["old_side_data"]["iptrunk_side_node"]["owner_subscription_id"])
-        == RouterVendor.NOKIA
+        == Vendor.NOKIA
     )
     should_restore_isis_metric = conditional(lambda state: state["restore_isis_metric"])
 

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -21,7 +21,6 @@ from gso.products.product_blocks.iptrunk import (
     IptrunkType,
     PhyPortCapacity,
 )
-from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import execute_playbook, pp_interaction
@@ -34,6 +33,7 @@ from gso.utils.helpers import (
     validate_iptrunk_unique_interface,
     validate_tt_number,
 )
+from gso.utils.shared_choices import Vendor
 
 
 def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_index: int) -> type[LAGMember]:
@@ -41,7 +41,7 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
     router = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_node
     router_vendor = get_router_vendor(router.owner_subscription_id)
     iptrunk_minimum_link = initial_user_input["iptrunk_minimum_links"]
-    if router_vendor == RouterVendor.NOKIA:
+    if router_vendor == Vendor.NOKIA:
         iptrunk_speed = initial_user_input["iptrunk_speed"]
 
         class NokiaLAGMember(LAGMember):
@@ -386,13 +386,13 @@ def modify_trunk_interface() -> StepList:
         lambda state: get_router_vendor(
             state["subscription"]["iptrunk"]["iptrunk_sides"][0]["iptrunk_side_node"]["owner_subscription_id"]
         )
-        == RouterVendor.NOKIA
+        == Vendor.NOKIA
     )
     side_b_is_nokia = conditional(
         lambda state: get_router_vendor(
             state["subscription"]["iptrunk"]["iptrunk_sides"][1]["iptrunk_side_node"]["owner_subscription_id"]
         )
-        == RouterVendor.NOKIA
+        == Vendor.NOKIA
     )
     return (
         init

gso/workflows/iptrunk/terminate_iptrunk.py

@@ -19,12 +19,12 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import validator
 
 from gso.products.product_blocks.iptrunk import IptrunkSideBlock
-from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import infoblox
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import execute_playbook, pp_interaction
 from gso.utils.helpers import get_router_vendor, validate_tt_number
+from gso.utils.shared_choices import Vendor
 from gso.utils.workflow_steps import set_isis_to_90000
 
 
@@ -159,14 +159,14 @@ def terminate_iptrunk() -> StepList:
         and get_router_vendor(
             state["subscription"]["iptrunk"]["iptrunk_sides"][0]["iptrunk_side_node"]["owner_subscription_id"]
         )
-        == RouterVendor.NOKIA
+        == Vendor.NOKIA
     )
     side_b_is_nokia = conditional(
         lambda state: state["clean_up_netbox"]
         and get_router_vendor(
             state["subscription"]["iptrunk"]["iptrunk_sides"][1]["iptrunk_side_node"]["owner_subscription_id"]
         )
-        == RouterVendor.NOKIA
+        == Vendor.NOKIA
     )
 
     config_steps = (

gso/workflows/router/create_router.py

@@ -13,19 +13,15 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
 from pydantic_forms.core import ReadOnlyField
 
-from gso.products.product_blocks.router import (
-    PortNumber,
-    RouterRole,
-    RouterVendor,
-    generate_fqdn,
-)
+from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import RouterInactive
 from gso.products.product_types.site import Site
 from gso.services import infoblox, subscriptions
 from gso.services.crm import get_customer_by_name
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
-from gso.utils.helpers import iso_from_ipv4
+from gso.utils.helpers import generate_fqdn, iso_from_ipv4
+from gso.utils.shared_choices import PortNumber, Vendor
 from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real, run_checks_after_base_config
 
 
@@ -47,7 +43,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
         tt_number: str
         customer: str = ReadOnlyField("GÉANT")
-        vendor: RouterVendor
+        vendor: Vendor
         router_site: _site_selector()  # type: ignore[valid-type]
         hostname: str
         ts_port: PortNumber
@@ -91,7 +87,7 @@ def initialize_subscription(
     ts_port: PortNumber,
     router_site: str,
     router_role: RouterRole,
-    vendor: RouterVendor,
+    vendor: Vendor,
 ) -> State:
     """Initialise the subscription object in the service database."""
     subscription.router.router_ts_port = ts_port
@@ -221,7 +217,7 @@ def create_router() -> StepList:
     * Validate :term:`IPAM` resources
     * Create a new device in Netbox
     """
-    router_is_nokia = conditional(lambda state: state["vendor"] == RouterVendor.NOKIA)
+    router_is_nokia = conditional(lambda state: state["vendor"] == Vendor.NOKIA)
 
     return (
         init

gso/workflows/router/terminate_router.py

@@ -18,11 +18,11 @@ from orchestrator.workflows.steps import (
 )
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
-from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.router import Router
 from gso.services import infoblox
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import execute_playbook, pp_interaction
+from gso.utils.shared_choices import Vendor
 
 logger = logging.getLogger(__name__)
 
@@ -41,7 +41,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         clean_up_ipam: bool = True
 
     user_input = yield TerminateForm
-    return user_input.dict() | {"router_is_nokia": router.router.vendor == RouterVendor.NOKIA}
+    return user_input.dict() | {"router_is_nokia": router.router.vendor == Vendor.NOKIA}
 
 
 @step("Deprovision loopback IPs from IPAM")