Add terminate WF for Layer 2 Circuit.

gso/migrations/versions/2024-10-29_72a4f7aa499d_add_l2circuit_workflows.py

@@ -29,6 +29,12 @@ new_workflows = [
         "target": "MODIFY",
         "description": "Modify Layer 2 Circuit Service",
         "product_type": "Layer2Circuit"
+    },
+    {
+        "name": "terminate_layer_2_circuit",
+        "target": "TERMINATE",
+        "description": "Terminate Layer 2 Circuit Service",
+        "product_type": "Layer2Circuit"
     }
 ]
 

gso/workflows/__init__.py

@@ -128,3 +128,4 @@ LazyWorkflowInstance("gso.workflows.nren_l3_core_service.migrate_nren_l3_core_se
 # Layer 2 Circuit workflows
 LazyWorkflowInstance("gso.workflows.l2_circuit.create_layer_2_circuit", "create_layer_2_circuit")
 LazyWorkflowInstance("gso.workflows.l2_circuit.modify_layer_2_circuit", "modify_layer_2_circuit")
+LazyWorkflowInstance("gso.workflows.l2_circuit.terminate_layer_2_circuit", "terminate_layer_2_circuit")

gso/workflows/l2_circuit/terminate_layer_2_circuit.py

+"""Workflow for terminating a Layer 2 Circuit."""
+
+from orchestrator import begin, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, done
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic_forms.types import FormGenerator
+
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.utils.types.tt_number import TTNumber
+
+
+def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    layer_2_circuit = Layer2Circuit.from_subscription(subscription_id)
+
+    class TerminateForm(FormPage):
+        tt_number: TTNumber
+
+    yield TerminateForm
+    return {"subscription": layer_2_circuit}
+
+
+@workflow(
+    "Terminate Layer 2 Circuit",
+    initial_input_form=wrap_modify_initial_input_form(_input_form_generator),
+    target=Target.TERMINATE,
+)
+def terminate_layer_2_circuit() -> StepList:
+    """Terminate a Layer 2 Circuit subscription."""
+    return (
+        begin
+        >> store_process_subscription(Target.TERMINATE)
+        >> unsync
+        >> set_status(SubscriptionLifecycle.TERMINATED)
+        >> resync
+        >> done
+    )