Add create_switch workflow

gso/workflows/__init__.py

@@ -50,6 +50,9 @@ LazyWorkflowInstance("gso.workflows.router.validate_router", "validate_router")
 LazyWorkflowInstance("gso.workflows.router.promote_p_to_pe", "promote_p_to_pe")
 LazyWorkflowInstance("gso.workflows.router.modify_kentik_license", "modify_router_kentik_license")
 
+#  Switch workflows
+LazyWorkflowInstance("gso.workflows.switch.create_switch", "create_switch")
+
 #  Site workflows
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")

gso/workflows/switch/__init__.py

+"""Workflows for switches."""

gso/workflows/switch/create_switch.py

+"""A creation workflow for adding a new switch to the subscription database."""
+
+from typing import Self
+
+from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
+from orchestrator.workflow import StepList, begin, done, inputstep, 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 ConfigDict, model_validator
+from pydantic_forms.validators import Label, ReadOnlyField
+
+from gso.products.product_blocks.switch import SwitchModel
+from gso.products.product_types.site import Site
+from gso.products.product_types.switch import SwitchInactive
+from gso.services import infoblox
+from gso.services.lso_client import execute_playbook, lso_interaction
+from gso.services.partners import get_partner_by_name
+from gso.services.sharepoint import SharePointClient
+from gso.settings import load_oss_params
+from gso.utils.helpers import active_site_selector, generate_fqdn
+from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.types import TTNumber
+from gso.utils.workflow_steps import prompt_sharepoint_checklist_url
+
+
+def initial_input_form_generator(product_name: str) -> FormGenerator:
+    """Input form for creating a new Switch."""
+
+    class CreateSwitchForm(FormPage):
+        model_config = ConfigDict(title=product_name)
+
+        tt_number: TTNumber
+        switch_site: active_site_selector()  # type: ignore[valid-type]
+        hostname: str
+        ts_port: PortNumber
+        vendor: ReadOnlyField(Vendor.JUNIPER, default_type=Vendor)  # type: ignore[valid-type]
+        model: ReadOnlyField(SwitchModel.EX3400, default_type=SwitchModel)  # type: ignore[valid-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
+
+    user_input = yield CreateSwitchForm
+    return user_input.model_dump()
+
+
+@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}
+
+
+@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.switch_site = Site.from_subscription(switch_site).site
+    subscription.switch.switch_fqdn = generate_fqdn(
+        hostname, subscription.switch.switch_site.site_name, subscription.switch.switch_site.site_country_code
+    )
+    subscription.switch.switch_ts_port = ts_port
+    subscription.switch.switch_vendor = vendor
+    subscription.switch.switch_model = model
+
+    return {"subscription": subscription}
+
+
+@step("[DRY RUN] Deploy base config")
+def deploy_base_config_dry(subscription: dict, tt_number: str, callback_route: str, process_id: UUIDstr) -> None:
+    """Perform a dry run of provisioning base config on a switch."""
+    inventory = subscription["switch"]["switch_fqdn"]
+
+    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",
+    }
+
+    execute_playbook(
+        playbook_name="switch_base_config.yaml",
+        callback_route=callback_route,
+        inventory=inventory,
+        extra_vars=extra_vars,
+    )
+
+
+@step("[FOR REAL] Deploy base config")
+def deploy_base_config_real(subscription: dict, tt_number: str, callback_route: str, process_id: UUIDstr) -> None:
+    """Provision base config on a switch."""
+    inventory = subscription["switch"]["switch_fqdn"]
+
+    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",
+    }
+
+    execute_playbook(
+        playbook_name="switch_base_config.yaml",
+        callback_route=callback_route,
+        inventory=inventory,
+        extra_vars=extra_vars,
+    )
+
+
+@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(FormPage):
+        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 {}
+
+
+@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(FormPage):
+        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 {}
+
+
+@step("Create Netbox device")
+def create_netbox_device() -> State:
+    """Add the switch as a new device in Netbox."""
+    return {"netbox_device": "Not implemented."}
+
+
+@step("Run post-deployment checks")
+def run_post_deploy_checks(subscription: dict, callback_route: str) -> None:
+    """Workflow step for running checks after installing base config."""
+    execute_playbook(
+        playbook_name="switch_base_config_checks.yaml",
+        callback_route=callback_route,
+        inventory=subscription["switch"]["switch_fqdn"],
+        extra_vars={"subscription_json": subscription},
+    )
+
+
+@step("Create a new SharePoint checklist")
+def create_new_sharepoint_checklist(subscription: SwitchInactive, tt_number: str, process_id: UUIDstr) -> State:
+    """Create a new checklist in SharePoint for approving this router."""
+    if not subscription.switch.switch_fqdn:
+        msg = "Switch is missing an FQDN."
+        raise ProcessFailureError(msg, details=subscription.subscription_id)
+
+    new_list_item_url = SharePointClient().add_list_item(
+        list_name="switch",
+        fields={
+            "Title": subscription.switch.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}
+
+
+@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
+    )