Move duplicate Layer 2 Circuit steps into common file

gso/workflows/l2_circuit/create_layer_2_circuit.py

 """Workflow for creating a new Layer 2 Circuit."""
 
-from copy import deepcopy
 from typing import Any, Self
 from uuid import uuid4
 
@@ -19,14 +18,19 @@ from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockIn
 from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitInactive
-from gso.services.lso_client import LSOState, lso_interaction
-from gso.services.partners import get_partner_by_id, get_partner_by_name
+from gso.services.lso_client import lso_interaction
+from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import generate_unique_id
 from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id, partner_choice
 from gso.utils.shared_enums import SBPType
 from gso.utils.types.interfaces import BandwidthString
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.types.virtual_identifiers import VLAN_ID
+from gso.workflows.l2_circuit.shared_steps import (
+    extract_partner_name_from_edge_port,
+    provision_l2circuit_dry,
+    provision_l2circuit_real,
+)
 
 
 def initial_input_generator(product_name: str) -> FormGenerator:
@@ -145,61 +149,6 @@ def initialize_subscription(
     return {"subscription": subscription, "fqdn_list": fqdn_list}
 
 
-@step("Expand subscription dictionary")
-def extract_partner_name_from_edge_port(subscription: dict[str, Any]) -> State:
-    """Expand a subscription model of a Layer 2 Circuit.
-
-    This method will include the name of each Edge Port's partner to be used in Ansible playbooks.
-    """
-    modified_subscription = deepcopy(subscription)
-    for side in modified_subscription["layer_2_circuit"]["layer_2_circuit_sides"]:
-        side["sbp"]["edge_port"]["partner_name"] = get_partner_by_id(
-            EdgePort.from_subscription(side["sbp"]["edge_port"]["owner_subscription_id"]).customer_id
-        ).name
-
-    return {"modified_subscription": modified_subscription}
-
-
-@step("[DRY RUN] Deploy L2circuit")
-def provision_l2circuit_dry(
-    modified_subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of deploying L2circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": True,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[REAL RUN] Deploy L2circuit")
-def provision_l2circuit_real(
-    modified_subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of deploying L2circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": False,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
 @workflow(
     "Create Layer 2 Circuit Service",
     initial_input_form=wrap_create_initial_input_form(initial_input_generator),

gso/workflows/l2_circuit/migrate_layer_2_circuit.py

@@ -19,7 +19,7 @@ EdgePort with the destination EdgePort and applies the necessary configuration c
     migrate the service.
 """
 
-from typing import Any, TypeAlias, cast
+from typing import TypeAlias, cast
 
 from orchestrator import step, workflow
 from orchestrator.forms import FormPage, SubmitFormPage
@@ -35,11 +35,18 @@ from pydantic_forms.validators import Divider
 
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.layer_2_circuit import Layer2Circuit
-from gso.services.lso_client import LSOState, lso_interaction
+from gso.services.lso_client import lso_interaction
 from gso.services.partners import get_partner_by_id
 from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id
 from gso.utils.types.tt_number import TTNumber
-from gso.workflows.l2_circuit.create_layer_2_circuit import extract_partner_name_from_edge_port
+from gso.workflows.l2_circuit.shared_steps import (
+    extract_partner_name_from_edge_port,
+    generate_fqdn_list,
+    provision_l2circuit_dry,
+    provision_l2circuit_real,
+    terminate_l2circuit_dry,
+    terminate_l2circuit_real,
+)
 
 
 def input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -87,19 +94,6 @@ def input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     }
 
 
-@step("Generate FQDN list")
-def generate_fqdn_list(subscription: Layer2Circuit) -> State:
-    """Generate the list of FQDNs that this workflow should target.
-
-    This list will consist of two elements, one for each far end of the circuit.
-    """
-    return {
-        "fqdn_list": [
-            side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides
-        ]
-    }
-
-
 @step("Update subscription model")
 def update_subscription_model(subscription: Layer2Circuit, old_edge_port: UUIDstr, new_edge_port: UUIDstr) -> State:
     """Replace the old Edge Port with the newly selected one in the subscription model."""
@@ -123,86 +117,6 @@ def update_subscription_model(subscription: Layer2Circuit, old_edge_port: UUIDst
     return {"subscription": subscription}
 
 
-@step("[DRY RUN] Remove old config")
-def remove_old_config_dry(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of removing old configuration of a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": True,
-        "verb": "terminate",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Remove config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[FOR REAL] Remove old config")
-def remove_old_config_real(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Remove old configuration of a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": False,
-        "verb": "terminate",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Remove config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[DRY RUN] Deploy new config")
-def deploy_new_config_dry(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of deploying new configuration for a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": True,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[FOR REAL] Deploy new config")
-def deploy_new_config_real(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Deploy configuration for the new Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": False,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
 @workflow(
     "Migrate Layer 2 Circuit",
     initial_input_form=wrap_modify_initial_input_form(input_form_generator),
@@ -216,13 +130,13 @@ def migrate_layer_2_circuit() -> StepList:
         >> unsync
         >> generate_fqdn_list
         >> extract_partner_name_from_edge_port
-        >> lso_interaction(remove_old_config_dry)
-        >> lso_interaction(remove_old_config_real)
+        >> lso_interaction(terminate_l2circuit_dry)
+        >> lso_interaction(terminate_l2circuit_real)
         >> update_subscription_model
         >> generate_fqdn_list
         >> extract_partner_name_from_edge_port
-        >> lso_interaction(deploy_new_config_dry)
-        >> lso_interaction(deploy_new_config_real)
+        >> lso_interaction(provision_l2circuit_dry)
+        >> lso_interaction(provision_l2circuit_real)
         >> resync
         >> done
     )

gso/workflows/l2_circuit/modify_layer_2_circuit.py

 """A modification workflow for a Layer 2 Circuit subscription."""
 
-from typing import Any
-
 from orchestrator import begin, done, workflow
 from orchestrator.forms import FormPage, SubmitFormPage
 from orchestrator.targets import Target
@@ -9,18 +7,23 @@ from orchestrator.workflow import StepList, step
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import BaseModel, ConfigDict, Field
-from pydantic_forms.types import FormGenerator, State, UUIDstr
+from pydantic_forms.types import FormGenerator, UUIDstr
 from pydantic_forms.validators import Divider, Label, ReadOnlyField
 
 from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.layer_2_circuit import Layer2Circuit
-from gso.services.lso_client import LSOState, lso_interaction
+from gso.services.lso_client import lso_interaction
 from gso.services.partners import get_partner_by_id
 from gso.utils.types.interfaces import BandwidthString
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.types.virtual_identifiers import VLAN_ID
-from gso.workflows.l2_circuit.create_layer_2_circuit import extract_partner_name_from_edge_port
+from gso.workflows.l2_circuit.shared_steps import (
+    extract_partner_name_from_edge_port,
+    generate_fqdn_list,
+    provision_l2circuit_dry,
+    provision_l2circuit_real,
+)
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -124,59 +127,6 @@ def modify_layer_2_circuit_subscription(
     return {"subscription": subscription}
 
 
-@step("Generate FQDN list")
-def generate_fqdn_list(subscription: Layer2Circuit) -> State:
-    """Generate the list of FQDNs that this workflow should target.
-
-    This list will consist of two elements, one for each far end of the circuit.
-    """
-    return {
-        "fqdn_list": [
-            side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides
-        ]
-    }
-
-
-@step("[DRY RUN] Deploy new config")
-def deploy_new_config_dry(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of deploying new configuration for a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": True,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[FOR REAL] Deploy new config")
-def deploy_new_config_real(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Deploy configuration for the new Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": False,
-        "verb": "deploy",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Deploy config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
 @workflow(
     "Modify Layer 2 Circuit Service",
     initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
@@ -191,8 +141,8 @@ def modify_layer_2_circuit() -> StepList:
         >> generate_fqdn_list
         >> extract_partner_name_from_edge_port
         >> modify_layer_2_circuit_subscription
-        >> lso_interaction(deploy_new_config_dry)
-        >> lso_interaction(deploy_new_config_real)
+        >> lso_interaction(provision_l2circuit_dry)
+        >> lso_interaction(provision_l2circuit_real)
         >> resync
         >> done
     )

gso/workflows/l2_circuit/shared_steps.py

+"""Workflow steps that are used in multiple Layer 2 Circuit workflows."""
+
+from copy import deepcopy
+from typing import Any
+
+from orchestrator import step
+from pydantic_forms.types import State, UUIDstr
+
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.services.lso_client import LSOState
+from gso.services.partners import get_partner_by_id
+from gso.utils.types.tt_number import TTNumber
+
+
+@step("Generate FQDN list")
+def generate_fqdn_list(subscription: Layer2Circuit) -> State:
+    """Generate the list of FQDNs that this workflow should target.
+
+    This list will consist of two elements, one for each far end of the circuit.
+    """
+    return {
+        "fqdn_list": [
+            side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides
+        ]
+    }
+
+
+@step("Expand subscription dictionary")
+def extract_partner_name_from_edge_port(subscription: dict[str, Any]) -> State:
+    """Expand a subscription model of a Layer 2 Circuit.
+
+    This method will include the name of each Edge Port's partner to be used in Ansible playbooks.
+    """
+    modified_subscription = deepcopy(subscription)
+    for side in modified_subscription["layer_2_circuit"]["layer_2_circuit_sides"]:
+        side["sbp"]["edge_port"]["partner_name"] = get_partner_by_id(
+            EdgePort.from_subscription(side["sbp"]["edge_port"]["owner_subscription_id"]).customer_id
+        ).name
+
+    return {"modified_subscription": modified_subscription}
+
+
+@step("[DRY RUN] Deploy L2circuit")
+def provision_l2circuit_dry(
+    modified_subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, fqdn_list: list[str]
+) -> LSOState:
+    """Perform a dry run of deploying a Layer 2 Circuit."""
+    extra_vars = {
+        "subscription": modified_subscription,
+        "dry_run": True,
+        "verb": "deploy",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
+        f"Deploy config for {modified_subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
+
+
+@step("[REAL RUN] Deploy L2circuit")
+def provision_l2circuit_real(
+    modified_subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, fqdn_list: list[str]
+) -> LSOState:
+    """Perform a dry run of deploying a Layer 2 Circuit."""
+    extra_vars = {
+        "subscription": modified_subscription,
+        "dry_run": False,
+        "verb": "deploy",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
+        f"Deploy config for {modified_subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
+
+
+@step("[DRY RUN] Remove old config")
+def terminate_l2circuit_dry(
+    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
+) -> LSOState:
+    """Perform a dry run of removing old configuration of a Layer 2 Circuit."""
+    extra_vars = {
+        "subscription": modified_subscription,
+        "dry_run": True,
+        "verb": "terminate",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
+        f"Remove config for {modified_subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
+
+
+@step("[FOR REAL] Remove old config")
+def terminate_l2circuit_real(
+    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
+) -> LSOState:
+    """Remove old configuration of a Layer 2 Circuit."""
+    extra_vars = {
+        "subscription": modified_subscription,
+        "dry_run": False,
+        "verb": "terminate",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
+        f"Remove config for {modified_subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
+        "extra_vars": extra_vars,
+    }

gso/workflows/l2_circuit/terminate_layer_2_circuit.py

 """Workflow for terminating a Layer 2 Circuit."""
 
-from typing import Any
-
 from orchestrator import begin, workflow
 from orchestrator.forms import SubmitFormPage
 from orchestrator.targets import Target
 from orchestrator.types import SubscriptionLifecycle
-from orchestrator.workflow import StepList, done, step
+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, State, UUIDstr
+from pydantic_forms.types import FormGenerator, UUIDstr
 
 from gso.products.product_types.layer_2_circuit import Layer2Circuit
-from gso.services.lso_client import LSOState, lso_interaction
+from gso.services.lso_client import lso_interaction
 from gso.utils.types.tt_number import TTNumber
-from gso.workflows.l2_circuit.create_layer_2_circuit import extract_partner_name_from_edge_port
+from gso.workflows.l2_circuit.shared_steps import (
+    extract_partner_name_from_edge_port,
+    generate_fqdn_list,
+    terminate_l2circuit_dry,
+    terminate_l2circuit_real,
+)
 
 
 def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -27,59 +30,6 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     return {"subscription": layer_2_circuit} | user_input.model_dump()
 
 
-@step("Generate FQDN list")
-def generate_fqdn_list(subscription: Layer2Circuit) -> State:
-    """Generate the list of FQDNs that this workflow should target.
-
-    This list will consist of two elements, one for each far end of the circuit.
-    """
-    return {
-        "fqdn_list": [
-            side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides
-        ]
-    }
-
-
-@step("[DRY RUN] Remove old config")
-def remove_old_config_dry(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Perform a dry run of removing old configuration of a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": True,
-        "verb": "terminate",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Remove config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[FOR REAL] Remove old config")
-def remove_old_config_real(
-    process_id: UUIDstr, tt_number: TTNumber, modified_subscription: dict[str, Any], fqdn_list: list[str]
-) -> LSOState:
-    """Remove old configuration of a Layer 2 Circuit."""
-    extra_vars = {
-        "subscription": modified_subscription,
-        "dry_run": False,
-        "verb": "terminate",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
-        f"Remove config for {modified_subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/l2circuit.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
 @workflow(
     "Terminate Layer 2 Circuit Service",
     initial_input_form=wrap_modify_initial_input_form(_input_form_generator),
@@ -93,8 +43,8 @@ def terminate_layer_2_circuit() -> StepList:
         >> unsync
         >> generate_fqdn_list
         >> extract_partner_name_from_edge_port
-        >> lso_interaction(remove_old_config_dry)
-        >> lso_interaction(remove_old_config_real)
+        >> lso_interaction(terminate_l2circuit_dry)
+        >> lso_interaction(terminate_l2circuit_real)
         >> set_status(SubscriptionLifecycle.TERMINATED)
         >> resync
         >> done