gso/workflows/router/modify_connection_strategy.py

+"""Modify connection strategy workflow. Flipping the connection between in-band to out-of-band."""
+
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, UUIDstr
+from orchestrator.workflow import StepList, done, init, 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.router import Router
+from gso.utils.shared_enums import ConnectionStrategy
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    """Modify the connection strategy initial formruff format."""
+    subscription = Router.from_subscription(subscription_id)
+
+    current_connection_strategy = (
+        ConnectionStrategy.OUT_OF_BAND if subscription.router.router_access_via_ts else ConnectionStrategy.IN_BAND
+    )
+
+    class ModifyConnectionStrategyForm(FormPage):
+        class Config:
+            title = f"Modify the connection strategy of {subscription.router.router_fqdn}."
+
+        connection_strategy: ConnectionStrategy = current_connection_strategy
+
+    user_input = yield ModifyConnectionStrategyForm
+
+    return user_input.dict()
+
+
+@step("Update subscription model")
+def update_subscription_model(subscription: Router, connection_strategy: str) -> State:
+    """Update the database model to reflect the new connection strategy.
+
+    If the connection strategy is set to IN-BAND, then access_via_ts should be set to False.
+    Conversely, if the connection strategy is set to OUT-OF-BAND, access_via_ts should be set to True.
+    """
+    subscription.router.router_access_via_ts = connection_strategy == ConnectionStrategy.OUT_OF_BAND
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Modify connection strategy",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def modify_connection_strategy() -> StepList:
+    """Modify the connection strategy."""
+    return init >> store_process_subscription(Target.MODIFY) >> unsync >> update_subscription_model >> resync >> done

gso/workflows/router/redeploy_base_config.py

@@ -9,7 +9,7 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
 from gso.products.product_types.router import Router
-from gso.services.provisioning_proxy import pp_interaction
+from gso.services.lso_client import lso_interaction
 from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real
 
 
@@ -40,8 +40,8 @@ def redeploy_base_config() -> StepList:
         init
         >> store_process_subscription(Target.MODIFY)
         >> unsync
-        >> pp_interaction(deploy_base_config_dry)
-        >> pp_interaction(deploy_base_config_real)
+        >> lso_interaction(deploy_base_config_dry)
+        >> lso_interaction(deploy_base_config_real)
         >> resync
         >> done
     )

gso/workflows/router/terminate_router.py

@@ -20,8 +20,8 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
 from gso.products.product_types.router import Router
 from gso.services import infoblox
+from gso.services.lso_client import execute_playbook, lso_interaction
 from gso.services.netbox_client import NetboxClient
-from gso.services.provisioning_proxy import execute_playbook, pp_interaction
 from gso.utils.shared_enums import Vendor
 
 logger = logging.getLogger(__name__)
@@ -123,8 +123,8 @@ def terminate_router() -> StepList:
         >> store_process_subscription(Target.TERMINATE)
         >> unsync
         >> run_ipam_steps(deprovision_loopback_ips)
-        >> run_config_steps(pp_interaction(remove_config_from_router_dry))
-        >> run_config_steps(pp_interaction(remove_config_from_router_real))
+        >> run_config_steps(lso_interaction(remove_config_from_router_dry))
+        >> run_config_steps(lso_interaction(remove_config_from_router_real))
         >> router_is_nokia(remove_device_from_netbox)
         >> set_status(SubscriptionLifecycle.TERMINATED)
         >> resync

gso/workflows/router/update_ibgp_mesh.py

@@ -14,8 +14,8 @@ from pydantic import root_validator
 
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
-from gso.services import librenms_client, provisioning_proxy, subscriptions
-from gso.services.provisioning_proxy import pp_interaction
+from gso.services import librenms_client, lso_client, subscriptions
+from gso.services.lso_client import lso_interaction
 from gso.services.subscriptions import get_active_trunks_that_terminate_on_router
 from gso.utils.helpers import SNMPVersion
 
@@ -86,7 +86,7 @@ def add_p_to_mesh_dry(
         "verb": "add_p_to_pe",
     }
 
-    provisioning_proxy.execute_playbook(
+    lso_client.execute_playbook(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=_generate_pe_inventory(pe_router_list),
@@ -106,7 +106,7 @@ def add_p_to_mesh_real(
         "verb": "add_p_to_pe",
     }
 
-    provisioning_proxy.execute_playbook(
+    lso_client.execute_playbook(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=_generate_pe_inventory(pe_router_list),
@@ -134,7 +134,7 @@ def add_all_pe_to_p_dry(
         "verb": "add_pe_to_p",
     }
 
-    provisioning_proxy.execute_playbook(
+    lso_client.execute_playbook(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=subscription["router"]["router_fqdn"],
@@ -162,7 +162,7 @@ def add_all_pe_to_p_real(
         "verb": "add_pe_to_p",
     }
 
-    provisioning_proxy.execute_playbook(
+    lso_client.execute_playbook(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=subscription["router"]["router_fqdn"],
@@ -173,7 +173,7 @@ def add_all_pe_to_p_real(
 @step("Verify iBGP session health")
 def check_ibgp_session(subscription: Router, callback_route: str) -> None:
     """Run a playbook using the provisioning proxy, to check the health of the new iBGP session."""
-    provisioning_proxy.execute_playbook(
+    lso_client.execute_playbook(
         playbook_name="check_ibgp.yaml",
         callback_route=callback_route,
         inventory=subscription.router.router_fqdn,
@@ -247,11 +247,11 @@ def update_ibgp_mesh() -> StepList:
         >> store_process_subscription(Target.MODIFY)
         >> unsync
         >> calculate_pe_router_list
-        >> pp_interaction(add_p_to_mesh_dry)
-        >> pp_interaction(add_p_to_mesh_real)
-        >> pp_interaction(add_all_pe_to_p_dry)
-        >> pp_interaction(add_all_pe_to_p_real)
-        >> pp_interaction(check_ibgp_session)
+        >> lso_interaction(add_p_to_mesh_dry)
+        >> lso_interaction(add_p_to_mesh_real)
+        >> lso_interaction(add_all_pe_to_p_dry)
+        >> lso_interaction(add_all_pe_to_p_real)
+        >> lso_interaction(check_ibgp_session)
         >> add_device_to_librenms
         >> prompt_insert_in_radius
         >> prompt_radius_login

setup.py

@@ -2,7 +2,7 @@ from setuptools import find_packages, setup
 
 setup(
     name="geant-service-orchestrator",
-    version="0.9",
+    version="0.10",
     author="GÉANT Orchestration and Automation Team",
     author_email="goat@geant.org",
     description="GÉANT Service Orchestrator",

test/conftest.py

@@ -246,4 +246,4 @@ def test_client(fastapi_app):
 
 @pytest.fixture(scope="session")
 def geant_partner():
-    return create_partner(PartnerCreate(name="GEANT-TEST", partner_type=PartnerType.GEANT))
+    return create_partner(PartnerCreate(name="GEANT-TEST", partner_type=PartnerType.GEANT, email="goat-test@geant.org"))

test/db/test_migrations.py

+import pytest
+from alembic.migration import MigrationContext
+from sqlalchemy import create_engine
+
+
+@pytest.mark.noautofixt()
+def test_migrations_no_branches(db_uri):
+    engine = create_engine(db_uri)
+    conn = engine.connect()
+    context = MigrationContext.configure(conn)
+
+    assert len(context.get_current_heads()) == 1, (
+        "\n\n!!!\n\nDetected a branching of db migrations. This is caused by multiple `down_revision`s referring to the"
+        " same migration.\n\nPlease resolve this by updating either one of these migrations: "
+        f"{context.get_current_heads()}.\n\n!!!\n\n"
+    )

test/fixtures.py

@@ -250,8 +250,8 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker, geant
         geant_s_sid = geant_s_sid or faker.geant_sid()
         iptrunk_description = iptrunk_description or faker.sentence()
         iptrunk_isis_metric = iptrunk_isis_metric or faker.pyint()
-        iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4(network=True)
-        iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6(network=True)
+        iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4_network(max_subnet=31)
+        iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6_network(max_subnet=126)
         iptrunk_minimum_links = 1
         iptrunk_side_a = iptrunk_side_subscription_factory()
         iptrunk_side_b = iptrunk_side_subscription_factory()

test/workflows/__init__.py

@@ -318,8 +318,8 @@ def user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
     return result, step_log
 
 
-def assert_pp_interaction_success(result: Process, process_stat: ProcessStat, step_log: list):
-    """Assert a successful pp interaction in a workflow.
+def assert_lso_interaction_success(result: Process, process_stat: ProcessStat, step_log: list):
+    """Assert a successful LSO interaction in a workflow.
 
     First, the workflow is awaiting callback. It is resumed but a result from LSO, after which the user submits the
     confirmation input step. Two assertions are made: the workflow is awaiting callback at first, and suspended when
@@ -332,8 +332,8 @@ def assert_pp_interaction_success(result: Process, process_stat: ProcessStat, st
     return resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
 
 
-def assert_pp_interaction_failure(result: Process, process_stat: ProcessStat, step_log: list):
-    """Assert a failed pp interaction in a workflow.
+def assert_lso_interaction_failure(result: Process, process_stat: ProcessStat, step_log: list):
+    """Assert a failed LSO interaction in a workflow.
 
     First, the workflow is awaiting callback. It is resumed by a "failure" result from LSO, after which the workflow is
     in a failed state. This failed state is also returned. Two assertions are made: the workflow is awaiting callback at

test/workflows/iptrunk/test_create_iptrunk.py

@@ -12,8 +12,8 @@ from test import USER_CONFIRM_EMPTY_FORM
 from test.services.conftest import MockedNetboxClient
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_failure,
-    assert_pp_interaction_success,
+    assert_lso_interaction_failure,
+    assert_lso_interaction_success,
     assert_suspended,
     extract_state,
     resume_workflow,
@@ -114,14 +114,14 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     test_client,
 ):
     mock_create_host.return_value = None
-    mock_allocate_v4_network.return_value = faker.ipv4(network=True)
-    mock_allocate_v6_network.return_value = faker.ipv6(network=True)
+    mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
     product_id = get_product_id_by_name(ProductType.IP_TRUNK)
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
     result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
 
     for _ in range(6):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_suspended(result)
     result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
@@ -160,16 +160,16 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
     _netbox_client_mock,  # noqa: PT019
     data_config_filename: PathLike,
 ):
-    mock_allocate_v4_network.return_value = faker.ipv4(network=True)
-    mock_allocate_v6_network.return_value = faker.ipv6(network=True)
+    mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
     product_id = get_product_id_by_name(ProductType.IP_TRUNK)
 
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
     result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
 
-    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
-    assert_pp_interaction_failure(result, process_stat, step_log)
+    assert_lso_interaction_failure(result, process_stat, step_log)
 
     assert mock_execute_playbook.call_count == 2
 
@@ -194,13 +194,13 @@ def test_successful_iptrunk_creation_with_juniper_interface_names(
 ):
     mock_create_host.return_value = None
     mock_allocate_v4_network.return_value = faker.ipv4_network(max_subnet=31)
-    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=127)
+    mock_allocate_v6_network.return_value = faker.ipv6_network(max_subnet=126)
     product_id = get_product_id_by_name(ProductType.IP_TRUNK)
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
     result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
 
     for _ in range(6):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_suspended(result)
     result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)

test/workflows/iptrunk/test_deploy_twamp.py

@@ -5,7 +5,7 @@ import pytest
 from gso.products import Iptrunk
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     extract_state,
     run_workflow,
 )
@@ -26,7 +26,7 @@ def test_iptrunk_deploy_twamp_success(
     result, process_stat, step_log = run_workflow("deploy_twamp", initial_input_data)
 
     for _ in range(2):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 

test/workflows/iptrunk/test_migrate_iptrunk.py

@@ -10,7 +10,7 @@ from test import USER_CONFIRM_EMPTY_FORM
 from test.conftest import UseJuniperSide
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     assert_suspended,
     extract_state,
     resume_workflow,
@@ -110,7 +110,7 @@ def interface_lists_are_equal(list1, list2):
 @pytest.mark.workflow()
 @patch("gso.services.infoblox.create_host_by_ip")
 @patch("gso.services.infoblox.delete_host_by_ip")
-@patch("gso.services.provisioning_proxy._send_request")
+@patch("gso.services.lso_client._send_request")
 @patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
 @patch("gso.services.netbox_client.NetboxClient.get_available_lags")
 @patch("gso.services.netbox_client.NetboxClient.create_interface")
@@ -147,19 +147,20 @@ def test_migrate_iptrunk_success(
 
     result, process_stat, step_log = run_workflow("migrate_iptrunk", migrate_form_input)
 
-    for _ in range(5):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    for _ in range(6):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_suspended(result)
     result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
 
-    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    for _ in range(4):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_suspended(result)
     result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
 
     for _ in range(3):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 
@@ -168,7 +169,7 @@ def test_migrate_iptrunk_success(
     subscription = Iptrunk.from_subscription(subscription_id)
 
     assert subscription.status == "active"
-    assert mock_execute_playbook.call_count == 9
+    assert mock_execute_playbook.call_count == 13
     assert mock_create_host_by_ip.call_count == 1
     assert mock_delete_host_by_ip.call_count == 1
 

test/workflows/iptrunk/test_modify_isis_metric.py

@@ -5,14 +5,14 @@ import pytest
 from gso.products import Iptrunk
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     extract_state,
     run_workflow,
 )
 
 
 @pytest.mark.workflow()
-@patch("gso.services.provisioning_proxy.execute_playbook")
+@patch("gso.services.lso_client.execute_playbook")
 def test_iptrunk_modify_isis_metric_success(
     mock_provision_ip_trunk,
     iptrunk_subscription_factory,
@@ -30,7 +30,7 @@ def test_iptrunk_modify_isis_metric_success(
     result, process_stat, step_log = run_workflow("modify_isis_metric", initial_iptrunk_data)
 
     for _ in range(2):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 

test/workflows/iptrunk/test_modify_trunk_interface.py

@@ -8,7 +8,7 @@ from gso.utils.shared_enums import Vendor
 from test.conftest import UseJuniperSide
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     extract_state,
     run_workflow,
 )
@@ -121,7 +121,7 @@ def test_iptrunk_modify_trunk_interface_success(
     result, process_stat, step_log = run_workflow("modify_trunk_interface", input_form_iptrunk_data)
 
     for _ in range(2):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 

test/workflows/iptrunk/test_terminate_iptrunk.py

@@ -7,7 +7,7 @@ from gso.settings import load_oss_params
 from test.services.conftest import MockedNetboxClient
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     extract_state,
     run_workflow,
 )
@@ -48,7 +48,7 @@ def test_successful_iptrunk_termination(
     result, process_stat, step_log = run_workflow("terminate_iptrunk", initial_iptrunk_data)
 
     for _ in range(3):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 

test/workflows/router/test_create_router.py

@@ -11,8 +11,8 @@ from gso.utils.shared_enums import Vendor
 from test import USER_CONFIRM_EMPTY_FORM
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_failure,
-    assert_pp_interaction_success,
+    assert_lso_interaction_failure,
+    assert_lso_interaction_success,
     assert_suspended,
     extract_state,
     resume_workflow,
@@ -85,14 +85,14 @@ def test_create_nokia_router_success(
     )
 
     for _ in range(2):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     # Handle four consecutive user input steps
     for _ in range(4):
         assert_suspended(result)
         result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
 
-    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_suspended(result)
     result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
@@ -173,13 +173,13 @@ def test_create_nokia_router_lso_failure(
     initial_router_data = [{"product": product_id}, router_creation_input_form_data]
     result, process_stat, step_log = run_workflow("create_router", initial_router_data)
 
-    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     state = extract_state(result)
     subscription_id = state["subscription_id"]
     subscription = Router.from_subscription(subscription_id)
 
-    assert_pp_interaction_failure(result, process_stat, step_log)
+    assert_lso_interaction_failure(result, process_stat, step_log)
 
     assert subscription.status == "initial"
     assert subscription.description == f"Router {mock_fqdn}"

test/workflows/router/test_modify_connection_stratey.py

+import pytest
+
+from gso.products import Router
+from gso.utils.shared_enums import ConnectionStrategy
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+def test_modify_connection_strategy(responses, nokia_router_subscription_factory):
+    subscription_id = nokia_router_subscription_factory(router_access_via_ts=True)
+    subscription = Router.from_subscription(subscription_id)
+    assert subscription.router.router_access_via_ts is True
+    form_data = [
+        {"subscription_id": subscription_id},
+        {"connection_strategy": ConnectionStrategy.IN_BAND},
+    ]
+    result, _, _ = run_workflow("modify_connection_strategy", form_data)
+    assert_complete(result)
+
+    # Fetch the updated subscription after running the workflow
+    updated_subscription = Router.from_subscription(subscription_id)
+    assert updated_subscription.status == "active"
+    assert updated_subscription.router.router_access_via_ts is False

test/workflows/router/test_redeploy_base_config.py

@@ -5,14 +5,14 @@ import pytest
 from gso.products import Router
 from test.workflows import (
     assert_complete,
-    assert_pp_interaction_success,
+    assert_lso_interaction_success,
     extract_state,
     run_workflow,
 )
 
 
 @pytest.mark.workflow()
-@patch("gso.services.provisioning_proxy._send_request")
+@patch("gso.services.lso_client._send_request")
 def test_redeploy_base_config_success(
     mock_provision_router,
     nokia_router_subscription_factory,
@@ -26,7 +26,7 @@ def test_redeploy_base_config_success(
     result, process_stat, step_log = run_workflow("redeploy_base_config", initial_input_data)
 
     for _ in range(2):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 

test/workflows/router/test_terminate_router.py

@@ -3,7 +3,7 @@ from unittest.mock import patch
 import pytest
 
 from gso.products import Router
-from test.workflows import assert_complete, assert_pp_interaction_success, extract_state, run_workflow
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
 
 
 @pytest.mark.workflow()
@@ -16,7 +16,7 @@ from test.workflows import assert_complete, assert_pp_interaction_success, extra
         (False, False),
     ],
 )
-@patch("gso.services.provisioning_proxy._send_request")
+@patch("gso.services.lso_client._send_request")
 @patch("gso.workflows.router.terminate_router.NetboxClient.delete_device")
 @patch("gso.workflows.router.terminate_router.infoblox.delete_host_by_ip")
 def test_terminate_router_full_success(
@@ -36,14 +36,14 @@ def test_terminate_router_full_success(
         "remove_configuration": remove_configuration,
         "clean_up_ipam": clean_up_ipam,
     }
-    pp_interaction_count = 2 if remove_configuration else 0
+    lso_interaction_count = 2 if remove_configuration else 0
 
     #  Run workflow
     initial_router_data = [{"subscription_id": product_id}, router_termination_input_form_data]
     result, process_stat, step_log = run_workflow("terminate_router", initial_router_data)
 
-    for _ in range(pp_interaction_count):
-        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+    for _ in range(lso_interaction_count):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 
@@ -54,4 +54,4 @@ def test_terminate_router_full_success(
     assert subscription.status == "terminated"
     assert mock_delete_device.call_count == 1
     assert mock_delete_host_by_ip.call_count == (1 if clean_up_ipam else 0)
-    assert mock_execute_playbook.call_count == pp_interaction_count
+    assert mock_execute_playbook.call_count == lso_interaction_count