diff --git a/gso/migrations/versions/2025-03-27_3541c7e57284_add_l2circuit_migration_workflow.py b/gso/migrations/versions/2025-03-27_3541c7e57284_add_l2circuit_migration_workflow.py
new file mode 100644
index 0000000000000000000000000000000000000000..0dfcd2cbee3f627b93c2d1d38247cd3a2ca72699
--- /dev/null
+++ b/gso/migrations/versions/2025-03-27_3541c7e57284_add_l2circuit_migration_workflow.py
@@ -0,0 +1,39 @@
+"""Add L2Circuit migration workflow.
+
+Revision ID: 3541c7e57284
+Revises: b14f71db2b58
+Create Date: 2025-03-27 15:35:39.351436
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '3541c7e57284'
+down_revision = 'b14f71db2b58'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+ {
+ "name": "migrate_layer_2_circuit",
+ "target": "MODIFY",
+ "description": "Migrate Layer 2 Circuit",
+ "product_type": "Layer2Circuit"
+ }
+]
+
+
+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"])
diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index b1a54748dbd43be1fc98ca3145b467e097f9f14a..02ea8449fd9eefba694548940ab8b83a6a61a85d 100644
--- a/gso/translations/en-GB.json
+++ b/gso/translations/en-GB.json
@@ -114,6 +114,7 @@
"import_switch": "NOT FOR HUMANS -- Finalize import into a Switch",
"migrate_edge_port": "Migrate Edge Port",
"migrate_iptrunk": "Migrate IP Trunk",
+ "migrate_layer_2_circuit": "Migrate Layer 2 Circuit",
"migrate_l3_core_service": "Migrate L3 Core Service",
"modify_connection_strategy": "Modify connection strategy",
"modify_edge_port": "Modify Edge Port",
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 15b9fcd42b193127797d06f7e771519efc3ff191..c20e8a4e780daff69688fa1b83b7cf57787ce99a 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -3,7 +3,7 @@
import random
import re
from ipaddress import IPv4Network, IPv6Network
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypeAlias, cast
from uuid import UUID
from orchestrator.types import SubscriptionLifecycle
@@ -11,6 +11,7 @@ from pydantic_forms.types import UUIDstr
from pydantic_forms.validators import Choice
from gso import settings
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import Router
from gso.services.netbox_client import NetboxClient
@@ -267,15 +268,18 @@ def active_switch_selector() -> Choice:
return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)) # type: ignore[arg-type]
-def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> Choice:
+def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> TypeAlias:
"""Generate a dropdown selector for choosing an active Edge Port in an input form."""
edge_ports = get_active_edge_port_subscriptions(partner_id=partner_id)
options = {str(edge_port.subscription_id): edge_port.description for edge_port in edge_ports}
- return Choice(
- "Select an Edge Port",
- zip(options.keys(), options.items(), strict=True), # type: ignore[arg-type]
+ return cast(
+ type[Choice],
+ Choice.__call__(
+ "Select an Edge Port",
+ zip(options.keys(), options.items(), strict=True),
+ ),
)
@@ -301,14 +305,17 @@ def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int
raise ValueError(err_msg)
-def generate_unique_vc_id(max_attempts: int = 100) -> VC_ID | None:
- """Generate a unique 8-digit VC_ID starting with '11'.
+def generate_unique_vc_id(l2c_type: str, max_attempts: int = 100) -> VC_ID | None:
+ """Generate a unique 8-digit VC_ID.
- This function attempts to generate an 8-digit VC_ID beginning with '11',
- checking its uniqueness before returning it. A maximum attempt limit is
- set to prevent infinite loops in case the ID space is saturated.
+ This function attempts to generate a ``VC_ID`` based on their circuit type,
+ and ensures uniqueness before returning it. A maximum attempt limit is
+ set to prevent an infinite loop in case the ID space is saturated.
+ The range used for generating a ``VC_ID`` depends on the circuit type:
+ ``Ethernet`` and ``VLAN`` type circuits get their IDs from different ranges.
Args:
+ l2c_type: type of l2circuit.
max_attempts: The maximum number of attempts to generate a unique ID.
Returns:
@@ -317,7 +324,9 @@ def generate_unique_vc_id(max_attempts: int = 100) -> VC_ID | None:
def create_vc_id() -> str:
"""Generate an 8-digit VC_ID starting with '11'."""
- return f"11{random.randint(100000, 999999)}" # noqa: S311
+ if l2c_type == Layer2CircuitType.ETHERNET:
+ return str(random.randint(30001, 39999))
+ return str(random.randint(6000, 6999))
for _ in range(max_attempts):
vc_id = create_vc_id()
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index 0f1abab7165f2db22138765e51591705cf04ae55..e977c62c25e434852c999d91cd0b2239a3da3e6b 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -132,6 +132,7 @@ LazyWorkflowInstance("gso.workflows.l3_core_service.validate_prefix_list", "vali
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")
+LazyWorkflowInstance("gso.workflows.l2_circuit.migrate_layer_2_circuit", "migrate_layer_2_circuit")
LazyWorkflowInstance("gso.workflows.l2_circuit.create_imported_layer_2_circuit", "create_imported_layer_2_circuit")
LazyWorkflowInstance("gso.workflows.l2_circuit.import_layer_2_circuit", "import_layer_2_circuit")
diff --git a/gso/workflows/edge_port/migrate_edge_port.py b/gso/workflows/edge_port/migrate_edge_port.py
index afd2d6d0f46e9dd6a81f9fe1e290af0ff3b217d0..ecdabd6f3460343249e98b59bd87557514ebdeb5 100644
--- a/gso/workflows/edge_port/migrate_edge_port.py
+++ b/gso/workflows/edge_port/migrate_edge_port.py
@@ -292,7 +292,7 @@ def migrate_l3_core_services_to_new_node(subscription_id: UUIDstr, tt_number: TT
},
],
],
- countdown=random.choice([2, 3, 4, 5]), # noqa: S311
+ countdown=random.choice([2, 3, 4, 5]),
)
return {"l3_core_services": l3_core_services}
@@ -319,7 +319,7 @@ def migrate_l2_circuits_to_new_node(subscription_id: UUIDstr, tt_number: TTNumbe
},
],
],
- countdown=random.choice([2, 3, 4, 5]), # noqa: S311
+ countdown=random.choice([2, 3, 4, 5]),
)
return {"layer2_circuits": layer2_circuits}
diff --git a/gso/workflows/l2_circuit/create_layer_2_circuit.py b/gso/workflows/l2_circuit/create_layer_2_circuit.py
index 5138aef1a167fe90a39c96bc0abaf557aa9ec875..ce70f215fa60ee80cc9a7a4f0240bec8729a370e 100644
--- a/gso/workflows/l2_circuit/create_layer_2_circuit.py
+++ b/gso/workflows/l2_circuit/create_layer_2_circuit.py
@@ -18,6 +18,7 @@ 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 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
@@ -25,6 +26,11 @@ 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:
@@ -125,8 +131,10 @@ def initialize_subscription(
layer2_circuit_side = Layer2CircuitSideBlockInactive.new(uuid4(), sbp=sbp)
layer_2_circuit_sides.append(layer2_circuit_side)
subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
- subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id()
subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
+ subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id(
+ l2c_type=subscription.layer_2_circuit.layer_2_circuit_type
+ )
subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
subscription.layer_2_circuit.policer_enabled = policer_enabled
@@ -136,8 +144,9 @@ def initialize_subscription(
subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"
subscription = Layer2Circuit.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
+ fqdn_list = [side.sbp.edge_port.node.router_fqdn for side in subscription.layer_2_circuit.layer_2_circuit_sides]
- return {"subscription": subscription}
+ return {"subscription": subscription, "fqdn_list": fqdn_list}
@workflow(
@@ -152,6 +161,9 @@ def create_layer_2_circuit() -> StepList:
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
+ >> extract_partner_name_from_edge_port
+ >> lso_interaction(provision_l2circuit_dry)
+ >> lso_interaction(provision_l2circuit_real)
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> done
diff --git a/gso/workflows/l2_circuit/migrate_layer2_circuit.py b/gso/workflows/l2_circuit/migrate_layer2_circuit.py
deleted file mode 100644
index e7d4d9b294cce4bd0ab5d8b2b0926fa28567a967..0000000000000000000000000000000000000000
--- a/gso/workflows/l2_circuit/migrate_layer2_circuit.py
+++ /dev/null
@@ -1,20 +0,0 @@
-"""This workflow migrates an L2 Core Service to a new Edge Port.
-
-It can be triggered by an operator or automatically by the system during Edge Port migration which is a separate
-workflow.
-
-System-triggered migration:
-When the system migrates an Edge Port, it runs the workflow automatically. The source and destination Edge Ports are
-set to the same values. Then here migration only applies the configuration to the router and fill the drift between
-core DB as source of truth and the actual network since the intent of network has changed in the previous workflow
-even though the L2 Circuit Service is not changed.
-
-Operator-triggered migration:
-When an operator initiates the workflow, they are required to specify both the source and destination EdgePorts.
-During the migration process, the system updates the related edge_port reference to replace the source
-EdgePort with the destination EdgePort and applies the necessary configuration changes to the router.
-
-Important Note:
-Since an L2 Circuit Service has multiple side, the workflow must be run separately for each side to fully
-migrate the service.
-"""
diff --git a/gso/workflows/l2_circuit/migrate_layer_2_circuit.py b/gso/workflows/l2_circuit/migrate_layer_2_circuit.py
new file mode 100644
index 0000000000000000000000000000000000000000..70a91e5aaaf24c274ca510bb9121e3b2ac37a882
--- /dev/null
+++ b/gso/workflows/l2_circuit/migrate_layer_2_circuit.py
@@ -0,0 +1,152 @@
+"""This workflow migrates an L2 Core Service to a new Edge Port.
+
+It can be triggered by an operator or automatically by the system during Edge Port migration which is a separate
+workflow.
+
+System-triggered migration:
+When the system migrates an Edge Port, it runs the workflow automatically. The source and destination Edge Ports are
+set to the same values. Then here migration only applies the configuration to the router and fill the drift between
+core DB as source of truth and the actual network since the intent of network has changed in the previous workflow
+even though the L2 Circuit Service is not changed.
+
+Operator-triggered migration:
+When an operator initiates the workflow, they are required to specify both the source and destination EdgePorts.
+During the migration process, the system updates the related edge_port reference to replace the source
+EdgePort with the destination EdgePort and applies the necessary configuration changes to the router.
+
+!!! info
+ Since an L2 Circuit Service has two sides, the workflow must be run separately for each side to fully
+ migrate the service.
+"""
+
+from typing import TypeAlias, cast
+
+from orchestrator import step, workflow
+from orchestrator.forms import FormPage, SubmitFormPage
+from orchestrator.forms.validators import Choice, Label
+from orchestrator.targets import Target
+from orchestrator.utils.errors import ProcessFailureError
+from orchestrator.workflow import StepList, begin, conditional, done
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import ConfigDict, Field
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+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 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.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:
+ """Generate an input form for migrating a Layer 2 Circuit."""
+ subscription = Layer2Circuit.from_subscription(subscription_id)
+
+ def circuit_side_selector() -> TypeAlias:
+ sides_dict = {
+ str(side.sbp.edge_port.owner_subscription_id): EdgePort.from_subscription(
+ side.sbp.edge_port.owner_subscription_id
+ ).description
+ for side in subscription.layer_2_circuit.layer_2_circuit_sides
+ }
+ return cast(
+ type[Choice],
+ Choice.__call__("Select one side of the circuit", zip(sides_dict.keys(), sides_dict.items(), strict=True)),
+ )
+
+ class MigrateL2CircuitForm(FormPage):
+ model_config = ConfigDict(title="Migrating Layer 2 Circuit")
+
+ tt_number: TTNumber
+ replace_side: circuit_side_selector() # type: ignore[valid-type]
+ divider: Divider = Field(None, exclude=True)
+
+ label_a: Label = Field("Are we migrating to a different site?", exclude=True)
+ migrate_to_different_site: bool = False
+
+ label_b: Label = Field("Execute Ansible playbooks on the OLD side of the circuit?", exclude=True)
+ run_old_side_ansible: bool = True
+ label_c: Label = Field("Execute Ansible playbooks on the NEW side of the circuit?", exclude=True)
+ run_new_side_ansible: bool = True
+
+ initial_user_input = yield MigrateL2CircuitForm
+ replace_side_partner = get_partner_by_id(EdgePort.from_subscription(initial_user_input.replace_side).customer_id)
+
+ class SelectNewEdgePortForm(SubmitFormPage):
+ model_config = ConfigDict(title="Migrating Layer 2 Circuit")
+
+ new_edge_port: active_edge_port_selector(partner_id=replace_side_partner.partner_id) # type: ignore[valid-type]
+
+ user_input = yield SelectNewEdgePortForm
+
+ return {
+ "tt_number": initial_user_input.tt_number,
+ "run_old_side_ansible": initial_user_input.run_old_side_ansible,
+ "run_new_side_ansible": initial_user_input.run_new_side_ansible,
+ "subscription": subscription,
+ "subscription_id": subscription_id,
+ "old_edge_port": initial_user_input.replace_side,
+ "new_edge_port": user_input.new_edge_port,
+ }
+
+
+@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."""
+ replace_index = (
+ 0
+ if str(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id)
+ == old_edge_port
+ else 1
+ )
+ subscription.layer_2_circuit.layer_2_circuit_sides[replace_index].sbp.edge_port = EdgePort.from_subscription(
+ new_edge_port
+ ).edge_port
+
+ vc_id = generate_unique_vc_id(l2c_type=subscription.layer_2_circuit.layer_2_circuit_type)
+ if not vc_id:
+ msg = "Failed to generate unique Virtual Circuit ID."
+ raise ProcessFailureError(msg)
+
+ subscription.layer_2_circuit.virtual_circuit_id = vc_id
+
+ return {"subscription": subscription}
+
+
+@workflow(
+ "Migrate Layer 2 Circuit",
+ initial_input_form=wrap_modify_initial_input_form(input_form_generator),
+ target=Target.MODIFY,
+)
+def migrate_layer_2_circuit() -> StepList:
+ """Migrate a Layer 2 Circuit."""
+ run_old_side_ansible = conditional(lambda state: state["run_old_side_ansible"])
+ run_new_side_ansible = conditional(lambda state: state["run_new_side_ansible"])
+
+ return (
+ begin
+ >> store_process_subscription(Target.MODIFY)
+ >> unsync
+ >> run_old_side_ansible(generate_fqdn_list)
+ >> run_old_side_ansible(extract_partner_name_from_edge_port)
+ >> run_old_side_ansible(lso_interaction(terminate_l2circuit_dry))
+ >> run_old_side_ansible(lso_interaction(terminate_l2circuit_real))
+ >> update_subscription_model
+ >> run_new_side_ansible(generate_fqdn_list)
+ >> run_new_side_ansible(extract_partner_name_from_edge_port)
+ >> run_new_side_ansible(lso_interaction(provision_l2circuit_dry))
+ >> run_new_side_ansible(lso_interaction(provision_l2circuit_real))
+ >> resync
+ >> done
+ )
diff --git a/gso/workflows/l2_circuit/modify_layer_2_circuit.py b/gso/workflows/l2_circuit/modify_layer_2_circuit.py
index 91c791359b99194e9e089e7a0823b3b857aa37af..ae879037c0700a515ca7d2507fbad6eba7d83715 100644
--- a/gso/workflows/l2_circuit/modify_layer_2_circuit.py
+++ b/gso/workflows/l2_circuit/modify_layer_2_circuit.py
@@ -3,7 +3,7 @@
from orchestrator import begin, done, workflow
from orchestrator.forms import FormPage, SubmitFormPage
from orchestrator.targets import Target
-from orchestrator.workflow import StepList, step
+from orchestrator.workflow import StepList, conditional, 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
@@ -13,10 +13,17 @@ 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 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.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:
@@ -36,6 +43,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
policer_enabled: bool = subscription.layer_2_circuit.policer_enabled
custom_service_name: str | None = subscription.layer_2_circuit.custom_service_name
+ label: Label = Field("Should this workflow execute Ansible playbooks on routers?", exclude=True)
+ run_ansible_steps: bool = True
+
layer_2_circuit_input = yield ModifyL2CircuitForm
class ModifyLayer2CircuitServiceSidesPage(SubmitFormPage):
@@ -127,11 +137,17 @@ def modify_layer_2_circuit_subscription(
)
def modify_layer_2_circuit() -> StepList:
"""Modify an existing Layer 2 Circuit service subscription."""
+ run_ansible_steps = conditional(lambda state: state["run_ansible_steps"])
+
return (
begin
>> store_process_subscription(Target.MODIFY)
>> unsync
+ >> run_ansible_steps(generate_fqdn_list)
+ >> run_ansible_steps(extract_partner_name_from_edge_port)
>> modify_layer_2_circuit_subscription
+ >> run_ansible_steps(lso_interaction(provision_l2circuit_dry))
+ >> run_ansible_steps(lso_interaction(provision_l2circuit_real))
>> resync
>> done
)
diff --git a/gso/workflows/l2_circuit/shared_steps.py b/gso/workflows/l2_circuit/shared_steps.py
new file mode 100644
index 0000000000000000000000000000000000000000..48b6429b6405afdfc1f46fcbed48f4dfe4a99c3c
--- /dev/null
+++ b/gso/workflows/l2_circuit/shared_steps.py
@@ -0,0 +1,121 @@
+"""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,
+ }
diff --git a/gso/workflows/l2_circuit/terminate_layer_2_circuit.py b/gso/workflows/l2_circuit/terminate_layer_2_circuit.py
index e8773a678ea5ce19c0dddef11e19d036f622460e..bab755eb824c8d92f0ed1e91d815eaf6a9df1fb4 100644
--- a/gso/workflows/l2_circuit/terminate_layer_2_circuit.py
+++ b/gso/workflows/l2_circuit/terminate_layer_2_circuit.py
@@ -4,13 +4,22 @@ 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
+from orchestrator.workflow import StepList, conditional, 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 import Field
from pydantic_forms.types import FormGenerator, UUIDstr
+from pydantic_forms.validators import Label
from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.services.lso_client import lso_interaction
from gso.utils.types.tt_number import TTNumber
+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:
@@ -19,8 +28,11 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class TerminateForm(SubmitFormPage):
tt_number: TTNumber
- yield TerminateForm
- return {"subscription": layer_2_circuit}
+ label: Label = Field("Should this workflow run Ansible playbooks to remove configuration from routers?")
+ run_ansible_steps: bool = True
+
+ user_input = yield TerminateForm
+ return {"subscription": layer_2_circuit} | user_input.model_dump()
@workflow(
@@ -30,10 +42,16 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
def terminate_layer_2_circuit() -> StepList:
"""Terminate a Layer 2 Circuit subscription."""
+ run_ansible_steps = conditional(lambda state: state["run_ansible_steps"])
+
return (
begin
>> store_process_subscription(Target.TERMINATE)
>> unsync
+ >> run_ansible_steps(generate_fqdn_list)
+ >> run_ansible_steps(extract_partner_name_from_edge_port)
+ >> run_ansible_steps(lso_interaction(terminate_l2circuit_dry))
+ >> run_ansible_steps(lso_interaction(terminate_l2circuit_real))
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
>> done
diff --git a/pyproject.toml b/pyproject.toml
index 6fb91172edfe7f78200037184ba51e1f77cad78f..34e477ffd71d29f1bd31b5d2ea0282e99d0ed05a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,6 +46,7 @@ ignore = [
"PLR0913",
"PLR0904",
"PLW1514",
+ "S311",
]
select = [
"A",
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index bc882410526e73dde39025e5da0adb91bd8881a5..26ab3bdd918da528e4c2f1564a8ced089caf803c 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -405,11 +405,12 @@ def l3_core_service_data(temp_file, faker, partner_factory, edge_port_subscripti
@pytest.fixture()
def layer_2_circuit_data(temp_file, faker, partner_factory, edge_port_subscription_factory):
def _layer_2_circuit_data(**kwargs):
+ circuit_type = Layer2CircuitType.VLAN
layer_2_circuit_input_data = {
"partner": partner_factory()["name"],
"service_type": Layer2CircuitServiceType.GEANT_PLUS,
"gs_id": faker.imported_gs_id(),
- "vc_id": generate_unique_vc_id(),
+ "vc_id": generate_unique_vc_id(circuit_type),
"layer_2_circuit_side_a": {
"edge_port": str(edge_port_subscription_factory().subscription_id),
"vlan_id": faker.vlan_id(),
@@ -418,7 +419,7 @@ def layer_2_circuit_data(temp_file, faker, partner_factory, edge_port_subscripti
"edge_port": str(edge_port_subscription_factory().subscription_id),
"vlan_id": faker.vlan_id(),
},
- "layer_2_circuit_type": Layer2CircuitType.VLAN,
+ "layer_2_circuit_type": circuit_type,
"vlan_range_lower_bound": faker.vlan_id(),
"vlan_range_upper_bound": faker.vlan_id(),
"policer_enabled": False,
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index ad6da994a083b5746107d381d831de6dc1a291ff..7ec1e0d9357e1e6174116c409f4e400a4ad78b95 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/test/fixtures/edge_port_fixtures.py
@@ -16,7 +16,7 @@ from gso.utils.types.interfaces import PhysicalPortCapacity
@pytest.fixture()
-def edge_port_subscription_factory(faker, partner_factory, router_subscription_factory):
+def edge_port_subscription_factory(faker, geant_partner, router_subscription_factory):
def subscription_create(
description=None,
partner: dict | None = None,
@@ -38,7 +38,8 @@ def edge_port_subscription_factory(faker, partner_factory, router_subscription_f
ignore_if_down=False,
is_imported=False,
) -> SubscriptionModel:
- partner = partner or partner_factory()
+ # Use default partner if none defined
+ partner = partner or geant_partner
node = node or router_subscription_factory(vendor=Vendor.NOKIA).router
if is_imported:
diff --git a/test/fixtures/l3_core_service_fixtures.py b/test/fixtures/l3_core_service_fixtures.py
index bffe03dae1e11dfe8e06e5b2cd4bd02543cc4627..fc72ca3998ccec8c03e893f4178c452fbabf6e6f 100644
--- a/test/fixtures/l3_core_service_fixtures.py
+++ b/test/fixtures/l3_core_service_fixtures.py
@@ -133,7 +133,7 @@ def access_port_factory(faker, service_binding_port_factory):
):
return AccessPort.new(
subscription_id=uuid4(),
- ap_type=ap_type or random.choice(list(APType)), # noqa: S311
+ ap_type=ap_type or random.choice(list(APType)),
sbp=service_binding_port or service_binding_port_factory(edge_port=edge_port, partner=partner),
)
diff --git a/test/fixtures/layer_2_circuit_fixtures.py b/test/fixtures/layer_2_circuit_fixtures.py
index f00b3375ce9da225e07e0259991019aa137066fc..584df642c94b7f010f0d219fa95498a5aafc58e1 100644
--- a/test/fixtures/layer_2_circuit_fixtures.py
+++ b/test/fixtures/layer_2_circuit_fixtures.py
@@ -96,22 +96,16 @@ def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscri
layer_2_circuit_sides.append(layer_2_circuit_side)
subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
- subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id()
+ subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id(layer_2_circuit_type)
subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
if layer_2_circuit_type == Layer2CircuitType.VLAN:
subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound or faker.vlan_id()
subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound or faker.vlan_id()
- else:
- subscription.layer_2_circuit.vlan_range_lower_bound = None
- subscription.layer_2_circuit.vlan_range_upper_bound = None
subscription.layer_2_circuit.policer_enabled = policer_enabled
if policer_enabled:
subscription.layer_2_circuit.bandwidth = policer_bandwidth or faker.bandwidth()
subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate or faker.bandwidth()
- else:
- subscription.layer_2_circuit.bandwidth = None
- subscription.layer_2_circuit.policer_burst_rate = None
subscription.description = description or (
f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"
)
diff --git a/test/workflows/l2_circuit/test_create_imported_layer_2_circuit.py b/test/workflows/l2_circuit/test_create_imported_layer_2_circuit.py
index 49e19bc7c693e3a1f9d5be14cbab6898a439f18b..84d894d98c25784406b86d032575991676acda74 100644
--- a/test/workflows/l2_circuit/test_create_imported_layer_2_circuit.py
+++ b/test/workflows/l2_circuit/test_create_imported_layer_2_circuit.py
@@ -17,15 +17,16 @@ def test_create_imported_layer_2_circuit_success(
edge_port_a = str(edge_port_subscription_factory(partner=partner).subscription_id)
edge_port_b = str(edge_port_subscription_factory(partner=partner).subscription_id)
policer_enabled = faker.boolean()
+ circuit_type = Layer2CircuitType.VLAN
creation_form_input_data = [
{
"service_type": layer_2_service_type,
"partner": partner["name"],
- "layer_2_circuit_type": Layer2CircuitType.VLAN,
+ "layer_2_circuit_type": circuit_type,
"policer_enabled": policer_enabled,
"vlan_range_lower_bound": faker.vlan_id(),
"vlan_range_upper_bound": faker.vlan_id(),
- "vc_id": generate_unique_vc_id(),
+ "vc_id": generate_unique_vc_id(circuit_type),
"policer_bandwidth": faker.bandwidth() if policer_enabled else None,
"policer_burst_rate": faker.bandwidth() if policer_enabled else None,
"gs_id": faker.imported_gs_id(),
diff --git a/test/workflows/l2_circuit/test_create_layer_2_circuit.py b/test/workflows/l2_circuit/test_create_layer_2_circuit.py
index ecec066a438b591d14b138bf16a3be93a4b1735d..4d282f6f73f5f2df113a069c75a2f3126773c2d6 100644
--- a/test/workflows/l2_circuit/test_create_layer_2_circuit.py
+++ b/test/workflows/l2_circuit/test_create_layer_2_circuit.py
@@ -1,10 +1,12 @@
+from unittest.mock import patch
+
import pytest
from orchestrator.types import SubscriptionLifecycle
from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
from gso.products.product_types.layer_2_circuit import LAYER_2_CIRCUIT_SERVICE_TYPES, Layer2Circuit
from gso.services.subscriptions import get_product_id_by_name
-from test.workflows import assert_complete, extract_state, run_workflow
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
def generate_layer_2_circuit_input(
@@ -58,16 +60,23 @@ def layer_2_circuit_ethernet_input(
@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
def test_create_layer_2_circuit_success(
+ mock_lso_interaction,
layer_2_circuit_service_type,
layer_2_circuit_input,
faker,
partner_factory,
):
- result, _, _ = run_workflow("create_layer_2_circuit", layer_2_circuit_input)
+ result, process_stat, step_log = run_workflow("create_layer_2_circuit", layer_2_circuit_input)
+
+ for _ in range(2):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
state = extract_state(result)
subscription = Layer2Circuit.from_subscription(state["subscription_id"])
+ assert mock_lso_interaction.call_count == 2
assert subscription.status == SubscriptionLifecycle.ACTIVE
assert subscription.layer_2_circuit.virtual_circuit_id is not None
assert len(subscription.layer_2_circuit.layer_2_circuit_sides) == 2
@@ -104,16 +113,23 @@ def test_create_layer_2_circuit_success(
@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
def test_create_layer_2_circuit_with_ethernet_type(
+ mock_lso_interaction,
layer_2_circuit_service_type,
layer_2_circuit_ethernet_input,
faker,
partner_factory,
):
- result, _, _ = run_workflow("create_layer_2_circuit", layer_2_circuit_ethernet_input)
+ result, process_stat, step_log = run_workflow("create_layer_2_circuit", layer_2_circuit_ethernet_input)
+
+ for _ in range(2):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
state = extract_state(result)
subscription = Layer2Circuit.from_subscription(state["subscription_id"])
+ assert mock_lso_interaction.call_count == 2
assert subscription.status == SubscriptionLifecycle.ACTIVE
assert subscription.layer_2_circuit.virtual_circuit_id is not None
assert len(subscription.layer_2_circuit.layer_2_circuit_sides) == 2
diff --git a/test/workflows/l2_circuit/test_migrate_layer_2_circuit.py b/test/workflows/l2_circuit/test_migrate_layer_2_circuit.py
new file mode 100644
index 0000000000000000000000000000000000000000..9c016cd28a5abbbd1a3d9cfd8572eeb78c5fedff
--- /dev/null
+++ b/test/workflows/l2_circuit/test_migrate_layer_2_circuit.py
@@ -0,0 +1,72 @@
+from unittest.mock import patch
+
+import pytest
+
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.layer_2_circuit import LAYER_2_CIRCUIT_SERVICE_TYPES, Layer2Circuit
+from gso.products.product_types.router import Router
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
+@pytest.mark.parametrize("run_old_side_ansible", [False, True])
+@pytest.mark.parametrize("run_new_side_ansible", [False, True])
+@patch("gso.services.lso_client._send_request")
+def test_migrate_layer_2_circuit(
+ mock_lso_interaction,
+ run_new_side_ansible,
+ run_old_side_ansible,
+ layer_2_circuit_service_type,
+ layer_2_circuit_subscription_factory,
+ edge_port_subscription_factory,
+ faker,
+):
+ subscription = layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type)
+ side_b_router = Router.from_subscription(
+ subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.node.owner_subscription_id
+ )
+ old_edge_port = EdgePort.from_subscription(
+ subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id
+ )
+ new_edge_port = edge_port_subscription_factory(node=side_b_router.router)
+
+ initial_layer_2_circuit_data = [
+ {"subscription_id": subscription.subscription_id},
+ {
+ "tt_number": faker.tt_number(),
+ "replace_side": old_edge_port.subscription_id,
+ "migrate_to_different_site": False,
+ "run_old_side_ansible": run_old_side_ansible,
+ "run_new_side_ansible": run_new_side_ansible,
+ },
+ {"new_edge_port": new_edge_port.subscription_id},
+ ]
+
+ result, process_stat, step_log = run_workflow("migrate_layer_2_circuit", initial_layer_2_circuit_data)
+
+ lso_step_count = 0
+ lso_step_count += 2 if run_old_side_ansible else 0
+ lso_step_count += 2 if run_new_side_ansible else 0
+
+ for _ in range(lso_step_count):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+ assert_complete(result)
+
+ state = extract_state(result)
+ subscription_id = state["subscription_id"]
+ subscription = Layer2Circuit.from_subscription(subscription_id)
+ assert subscription.status == "active"
+ assert mock_lso_interaction.call_count == lso_step_count
+
+ replaced_edge_port = subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port
+ assert replaced_edge_port.model_dump(exclude="edge_port_ae_members") == new_edge_port.edge_port.model_dump(
+ exclude="edge_port_ae_members"
+ )
+ assert replaced_edge_port.edge_port_ae_members[0].model_dump(
+ exclude="owner_subscription_id"
+ ) == new_edge_port.edge_port.edge_port_ae_members[0].model_dump(exclude="owner_subscription_id")
+ assert replaced_edge_port.edge_port_ae_members[1].model_dump(
+ exclude="owner_subscription_id"
+ ) == new_edge_port.edge_port.edge_port_ae_members[1].model_dump(exclude="owner_subscription_id")
diff --git a/test/workflows/l2_circuit/test_modify_layer_2_circuit.py b/test/workflows/l2_circuit/test_modify_layer_2_circuit.py
index cc6f773c85c8cb3ececb9e86d240fa5fd5924c44..e4f4dcce33efeabe2dac11c54a41b014577ab0de 100644
--- a/test/workflows/l2_circuit/test_modify_layer_2_circuit.py
+++ b/test/workflows/l2_circuit/test_modify_layer_2_circuit.py
@@ -1,15 +1,21 @@
+from unittest.mock import patch
+
import pytest
from orchestrator.types import SubscriptionLifecycle
from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
from gso.products.product_types.layer_2_circuit import LAYER_2_CIRCUIT_SERVICE_TYPES, Layer2Circuit
-from test.workflows import assert_complete, extract_state, run_workflow
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
+@pytest.mark.parametrize("run_ansible_steps", [False, True])
@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
def test_modify_layer_2_circuit_change_policer_bandwidth(
+ mock_lso_interaction,
layer_2_circuit_service_type,
+ run_ansible_steps,
layer_2_circuit_subscription_factory,
faker,
partner_factory,
@@ -22,6 +28,7 @@ def test_modify_layer_2_circuit_change_policer_bandwidth(
"layer_2_circuit_type": Layer2CircuitType.VLAN,
"policer_enabled": False,
"custom_service_name": faker.sentence(),
+ "run_ansible_steps": run_ansible_steps,
},
{
"vlan_range_lower_bound": subscription.layer_2_circuit.vlan_range_lower_bound,
@@ -32,9 +39,16 @@ def test_modify_layer_2_circuit_change_policer_bandwidth(
"layer_2_circuit_side_b": {},
},
]
- result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data)
+ result, process_stat, step_log = run_workflow("modify_layer_2_circuit", input_form_data)
+
+ lso_step_count = 2 if run_ansible_steps else 0
+
+ for _ in range(lso_step_count):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
subscription = Layer2Circuit.from_subscription(str(subscription.subscription_id))
assert_complete(result)
+ assert mock_lso_interaction.call_count == lso_step_count
assert subscription.status == SubscriptionLifecycle.ACTIVE
assert subscription.layer_2_circuit.policer_enabled is False
assert subscription.layer_2_circuit.bandwidth is None
@@ -54,9 +68,13 @@ def test_modify_layer_2_circuit_change_policer_bandwidth(
@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
+@pytest.mark.parametrize("run_ansible_steps", [False, True])
@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
def test_modify_layer_2_circuit_change_circuit_type(
+ mock_lso_interaction,
layer_2_circuit_service_type,
+ run_ansible_steps,
layer_2_circuit_subscription_factory,
faker,
partner_factory,
@@ -67,6 +85,7 @@ def test_modify_layer_2_circuit_change_circuit_type(
{
"tt_number": faker.tt_number(),
"layer_2_circuit_type": Layer2CircuitType.ETHERNET,
+ "run_ansible_steps": run_ansible_steps,
},
{
"vlan_range_lower_bound": None,
@@ -76,10 +95,17 @@ def test_modify_layer_2_circuit_change_circuit_type(
"layer_2_circuit_side_b": {"vlan_id": faker.vlan_id()},
},
]
- result, _, _ = run_workflow("modify_layer_2_circuit", input_form_data)
+ result, process_stat, step_log = run_workflow("modify_layer_2_circuit", input_form_data)
+
+ lso_step_count = 2 if run_ansible_steps else 0
+
+ for _ in range(lso_step_count):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
state = extract_state(result)
subscription = Layer2Circuit.from_subscription(state["subscription_id"])
+ assert mock_lso_interaction.call_count == lso_step_count
assert subscription.status == SubscriptionLifecycle.ACTIVE
assert subscription.layer_2_circuit.vlan_range_lower_bound is None
assert subscription.layer_2_circuit.vlan_range_upper_bound is None
diff --git a/test/workflows/l2_circuit/test_terminate_layer_2_circuit.py b/test/workflows/l2_circuit/test_terminate_layer_2_circuit.py
index 750c7b06c06984229940dca59bfaa0751a085d17..55deedededc9c3c43ef3ca56d6145508cbf7f5d4 100644
--- a/test/workflows/l2_circuit/test_terminate_layer_2_circuit.py
+++ b/test/workflows/l2_circuit/test_terminate_layer_2_circuit.py
@@ -1,20 +1,36 @@
+from unittest.mock import patch
+
import pytest
from gso.products.product_types.layer_2_circuit import LAYER_2_CIRCUIT_SERVICE_TYPES, Layer2Circuit
-from test.workflows import assert_complete, extract_state, run_workflow
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
@pytest.mark.workflow()
@pytest.mark.parametrize("layer_2_circuit_service_type", LAYER_2_CIRCUIT_SERVICE_TYPES)
-def test_terminate_layer_2_circuit(layer_2_circuit_service_type, layer_2_circuit_subscription_factory, faker):
+@pytest.mark.parametrize("run_ansible_steps", [True, False])
+@patch("gso.services.lso_client._send_request")
+def test_terminate_layer_2_circuit(
+ mock_lso_interaction, layer_2_circuit_service_type, run_ansible_steps, layer_2_circuit_subscription_factory, faker
+):
subscription_id = str(
layer_2_circuit_subscription_factory(layer_2_circuit_service_type=layer_2_circuit_service_type).subscription_id
)
- initialt_layer_2_circuit_data = [{"subscription_id": subscription_id}, {"tt_number": faker.tt_number()}]
- result, _, _ = run_workflow("terminate_layer_2_circuit", initialt_layer_2_circuit_data)
+ initial_layer_2_circuit_data = [
+ {"subscription_id": subscription_id},
+ {"tt_number": faker.tt_number(), "run_ansible_steps": run_ansible_steps},
+ ]
+ result, process_stat, step_log = run_workflow("terminate_layer_2_circuit", initial_layer_2_circuit_data)
+
+ lso_step_count = 2 if run_ansible_steps else 0
+
+ for _ in range(lso_step_count):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
state = extract_state(result)
subscription_id = state["subscription_id"]
subscription = Layer2Circuit.from_subscription(subscription_id)
assert subscription.status == "terminated"
+ assert mock_lso_interaction.call_count == lso_step_count