Create switch

A creation workflow for adding a new switch to the subscription database.

_initial_input_form_generator(product_name)

Input form for creating a new Switch.

Source code in gso/workflows/switch/create_switch.py

def _initial_input_form_generator(product_name: str) -> FormGenerator:
    """Input form for creating a new Switch."""

    class CreateSwitchForm(SubmitFormPage):
        model_config = ConfigDict(title=product_name)

        tt_number: TTNumber
        partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
        switch_site: active_site_selector() or str  # type: ignore[valid-type]
        hostname: str
        ts_port: PortNumber
        vendor: ReadOnlyField(Vendor.JUNIPER, default_type=Vendor)  # type: ignore[valid-type]
        model: SwitchModel = Choice("Switch model", SwitchModel.values())  # type: ignore[assignment, arg-type]

        @model_validator(mode="after")
        def hostname_must_be_available(self) -> Self:
            if not self.switch_site:
                msg = "Please select a site before setting the hostname."
                raise ValueError(msg)

            selected_site = Site.from_subscription(self.switch_site).site
            input_fqdn = generate_fqdn(self.hostname, selected_site.site_name, selected_site.site_country_code)
            if not infoblox.hostname_available(input_fqdn):
                msg = f'FQDN "{input_fqdn}" is not available.'
                raise ValueError(msg)

            return self

    input_form = yield CreateSwitchForm
    user_input = input_form.model_dump()
    summary_fields = ["tt_number", "partner", "switch_site", "hostname", "ts_port", "vendor", "model"]
    yield from create_summary_form(user_input, product_name, summary_fields)

    return user_input

create_subscription(product, partner)

Create a new subscription object.

Source code in gso/workflows/switch/create_switch.py

@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object."""
    subscription = SwitchInactive.from_product_id(product, get_partner_by_name(partner).partner_id)

    return {"subscription": subscription, "subscription_id": subscription.subscription_id}

initialize_subscription(subscription, switch_site, ts_port, vendor, model, hostname)

Initialize the subscription with user input.

Source code in gso/workflows/switch/create_switch.py

@step("Initialize subscription")
def initialize_subscription(
    subscription: SwitchInactive,
    switch_site: str,
    ts_port: PortNumber,
    vendor: Vendor,
    model: SwitchModel,
    hostname: str,
) -> State:
    """Initialize the subscription with user input."""
    subscription.switch.site = Site.from_subscription(switch_site).site
    subscription.switch.fqdn = generate_fqdn(
        hostname, subscription.switch.site.site_name, subscription.switch.site.site_country_code
    )
    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}

deploy_base_config_dry(subscription, tt_number, process_id)

Perform a dry run of provisioning base config on a switch.

Source code in gso/workflows/switch/create_switch.py

@step("[DRY RUN] Deploy base config")
def deploy_base_config_dry(subscription: dict, tt_number: str, process_id: UUIDstr) -> LSOState:
    """Perform a dry run of provisioning base config on a switch."""
    extra_vars = {
        "subscription_json": subscription,
        "dry_run": True,
        "verb": "deploy",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy base config",
    }

    return {
        "playbook_name": "switch_base_config.yaml",
        "extra_vars": extra_vars,
        "inventory": {"all": {"hosts": {subscription["switch"]["fqdn"]: None}}},
    }

deploy_base_config_real(subscription, tt_number, process_id)

Provision base config on a switch.

Source code in gso/workflows/switch/create_switch.py

@step("[FOR REAL] Deploy base config")
def deploy_base_config_real(subscription: dict, tt_number: str, process_id: UUIDstr) -> LSOState:
    """Provision base config on a switch."""
    extra_vars = {
        "subscription_json": subscription,
        "dry_run": False,
        "verb": "deploy",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy base config",
    }

    return {
        "playbook_name": "switch_base_config.yaml",
        "extra_vars": extra_vars,
        "inventory": {"all": {"hosts": {subscription["switch"]["fqdn"]: None}}},
    }

prompt_console_login()

Wait for confirmation from an operator that console login is possible.

Source code in gso/workflows/switch/create_switch.py

@inputstep("Prompt for console login", assignee=Assignee.SYSTEM)
def prompt_console_login() -> FormGenerator:
    """Wait for confirmation from an operator that console login is possible."""

    class ConsoleLoginPage(SubmitFormPage):
        model_config = ConfigDict(title="Please confirm before continuing")

        info_label: Label = "Please confirm you are able to log in to the switch using out of band connectivity."

    yield ConsoleLoginPage
    return {}

prompt_insert_in_ims()

Wait for confirmation from an operator that the switch has been inserted in IMS.

Source code in gso/workflows/switch/create_switch.py

@inputstep("Prompt IMS insertion", assignee=Assignee.SYSTEM)
def prompt_insert_in_ims() -> FormGenerator:
    """Wait for confirmation from an operator that the switch has been inserted in IMS."""

    class IMSPrompt(SubmitFormPage):
        model_config = ConfigDict(title="Update IMS mediation server")

        info_label_1: Label = "Insert the switch into IMS."
        info_label_2: Label = "Once this is done, press submit to continue the workflow."

    yield IMSPrompt
    return {}

create_netbox_device()

Add the switch as a new device in Netbox.

Source code in gso/workflows/switch/create_switch.py

@step("Create Netbox device")
def create_netbox_device() -> State:
    """Add the switch as a new device in Netbox."""
    return {"netbox_device": "Not implemented."}

run_post_deploy_checks(subscription)

Workflow step for running checks after installing base config.

Source code in gso/workflows/switch/create_switch.py

@step("Run post-deployment checks")
def run_post_deploy_checks(subscription: dict) -> LSOState:
    """Workflow step for running checks after installing base config."""
    return {
        "playbook_name": "switch_base_config_checks.yaml",
        "extra_vars": {"subscription_json": subscription},
        "inventory": {"all": {"hosts": {subscription["switch"]["fqdn"]: None}}},
    }

create_new_sharepoint_checklist(subscription, tt_number, process_id)

Create a new checklist in SharePoint for approving this router.

Source code in gso/workflows/switch/create_switch.py

@step("Create a new SharePoint checklist")
def create_new_sharepoint_checklist(subscription: SwitchProvisioning, tt_number: str, process_id: UUIDstr) -> State:
    """Create a new checklist in SharePoint for approving this router."""
    new_list_item_url = SharePointClient().add_list_item(
        list_name="switch",
        fields={
            "Title": subscription.switch.fqdn,
            "TT_NUMBER": tt_number,
            "GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
        },
    )

    return {"checklist_url": new_list_item_url}

create_switch()

Create a new Switch.

• Create a subscription object in the service database
• Deploy base configuration on the switch
• Add the switch to Netbox
• Run a check playbook after deploying base configuration
• Create a new checklist in SharePoint

Source code in gso/workflows/switch/create_switch.py

@workflow(
    "Create Switch",
    initial_input_form=wrap_create_initial_input_form(_initial_input_form_generator),
    target=Target.CREATE,
)
def create_switch() -> StepList:
    """Create a new Switch.

    * Create a subscription object in the service database
    * Deploy base configuration on the switch
    * Add the switch to Netbox
    * Run a check playbook after deploying base configuration
    * Create a new checklist in SharePoint
    """
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> lso_interaction(deploy_base_config_dry)
        >> lso_interaction(deploy_base_config_real)
        >> prompt_console_login
        >> prompt_insert_in_ims
        >> create_netbox_device
        >> lso_interaction(run_post_deploy_checks)
        >> set_status(SubscriptionLifecycle.PROVISIONING)
        >> create_new_sharepoint_checklist
        >> prompt_sharepoint_checklist_url
        >> resync
        >> done
    )