Add cancel_subscription workflow

gso/workflows/__init__.py

@@ -5,6 +5,7 @@ from orchestrator.workflows import LazyWorkflowInstance
 
 WF_USABLE_MAP.update(
     {
+        "cancel_subscription": ["initial"],
         "redeploy_base_config": ["provisioning", "active"],
         "update_ibgp_mesh": ["provisioning", "active"],
         "activate_router": ["provisioning"],
@@ -27,6 +28,7 @@ LazyWorkflowInstance("gso.workflows.router.redeploy_base_config", "redeploy_base
 LazyWorkflowInstance("gso.workflows.router.terminate_router", "terminate_router")
 LazyWorkflowInstance("gso.workflows.router.update_ibgp_mesh", "update_ibgp_mesh")
 LazyWorkflowInstance("gso.workflows.router.modify_connection_strategy", "modify_connection_strategy")
+LazyWorkflowInstance("gso.workflows.shared.cancel_subscription", "cancel_subscription")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")
 LazyWorkflowInstance("gso.workflows.site.terminate_site", "terminate_site")

gso/workflows/shared/__init__.py

+"""Workflows that are shared across multiple products."""

gso/workflows/shared/cancel_subscription.py

+"""Cancel a subscription that is in the initial lifecycle state."""
+
+from orchestrator.forms import FormPage
+from orchestrator.forms.validators import Label
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, done, init, workflow
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+
+def _initial_input_form(subscription_id: UUIDstr) -> FormGenerator:
+    class CancelSubscriptionForm(FormPage):
+        info_label: Label = f"Canceling subscription with ID {subscription_id}"  # type:ignore[assignment]
+        info_label_2: Label = (
+            "This will immediately mark the subscription as terminated, preventing any other workflows from interacting"
+            " with this product subscription."
+        )
+        info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
+        info_label_4: Label = "THIS WORKFLOW IS IRREVERSIBLE AND MAY HAVE UNFORESEEN CONSEQUENCES."
+
+    yield CancelSubscriptionForm
+
+    return {"subscription_id": subscription_id}
+
+
+@workflow(
+    "Cancel an initial subscription",
+    initial_input_form=wrap_modify_initial_input_form(_initial_input_form),
+    target=Target.TERMINATE,
+)
+def cancel_subscription() -> StepList:
+    """Cancel an initial subscription, taking it from the ``INITIAL`` state straight to ``TERMINATED``.
+
+    This workflow can be used when a creation workflow has failed, and the process needs to be restarted. This workflow
+    will prevent a stray subscription, forever stuck in the initial state, to stick around.
+
+    * Update the subscription lifecycle state to ``TERMINATED``.
+    """
+    return (
+        init
+        >> store_process_subscription(Target.TERMINATE)
+        >> unsync
+        >> set_status(SubscriptionLifecycle.TERMINATED)
+        >> resync
+        >> done
+    )