Add validate_switch workflow and migration

gso/migrations/versions/2024-08-29_0e7e7d749617_add_switch_workflows.py

+"""Add switch workflows.
+
+Revision ID: 0e7e7d749617
+Revises: bd9be532b3f9
+Create Date: 2024-08-29 15:45:57.581710
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '0e7e7d749617'
+down_revision = 'bd9be532b3f9'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "create_switch",
+        "target": "CREATE",
+        "description": "Create Switch",
+        "product_type": "Switch"
+    },
+    {
+        "name": "activate_switch",
+        "target": "MODIFY",
+        "description": "Activate switch",
+        "product_type": "Switch"
+    },
+    {
+        "name": "terminate_switch",
+        "target": "TERMINATE",
+        "description": "Terminate switch",
+        "product_type": "Switch"
+    },
+    {
+        "name": "validate_switch",
+        "target": "SYSTEM",
+        "description": "Validate switch subscription",
+        "product_type": "Switch"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/workflows/__init__.py

@@ -52,6 +52,9 @@ LazyWorkflowInstance("gso.workflows.router.modify_kentik_license", "modify_route
 
 #  Switch workflows
 LazyWorkflowInstance("gso.workflows.switch.create_switch", "create_switch")
+LazyWorkflowInstance("gso.workflows.switch.activate_switch", "activate_switch")
+LazyWorkflowInstance("gso.workflows.switch.terminate_switch", "terminate_switch")
+LazyWorkflowInstance("gso.workflows.switch.validate_switch", "validate_switch")
 
 #  Site workflows
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")

gso/workflows/switch/validate_switch.py

+"""Validation workflow for switch subscription objects."""
+
+from typing import Any
+
+from orchestrator.targets import Target
+from orchestrator.workflow import StepList, begin, done, step, workflow
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products.product_types.switch import Switch
+from gso.services.lso_client import anonymous_lso_interaction, execute_playbook
+from gso.services.netbox_client import NetboxClient
+
+
+@step("Validate switch in Netbox")
+def check_netbox_device(subscription: Switch) -> None:
+    """Fetch the device in Netbox. Will raise an exception when it is not found."""
+    NetboxClient().get_device_by_name(subscription.switch.switch_fqdn)
+
+
+@step("Check base config for drift")
+def verify_base_config(subscription: dict[str, Any], callback_route: str) -> None:
+    """Workflow step for running a playbook that checks whether base config has drifted."""
+    execute_playbook(
+        playbook_name="switch_base_config.yaml",
+        callback_route=callback_route,
+        inventory=subscription["switch"]["switch_fqdn"],
+        extra_vars={
+            "subscription_json": subscription,
+            "verb": "deploy",
+            "dry_run": "true",
+            "is_verification_workflow": "true",
+        },
+    )
+
+
+@workflow(
+    "Validate switch subscription", target=Target.SYSTEM, initial_input_form=(wrap_modify_initial_input_form(None))
+)
+def validate_switch() -> StepList:
+    """Validate an existing, active switch subscription."""
+    return (
+        begin
+        >> store_process_subscription(Target.SYSTEM)
+        >> unsync
+        >> check_netbox_device
+        >> anonymous_lso_interaction(verify_base_config)
+        >> resync
+        >> done
+    )