Fix bugs in L3 migration workflow, and update unit tests

gso/workflows/l3_core_service/migrate_l3_core_service.py

 """A modification workflow that migrates a L3 Core Service to a new set of Edge Ports."""
 
-import copy
 import json
 from typing import Any
+from uuid import UUID
 
 from orchestrator import workflow
 from orchestrator.config.assignee import Assignee
 from orchestrator.forms import FormPage, SubmitFormPage
 from orchestrator.targets import Target
+from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, begin, done, inputstep, step
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
@@ -87,29 +88,32 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
 @step("Show BGP neighbors")
 def show_bgp_neighbors(
-    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: AccessPort
+    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: UUIDstr
 ) -> LSOState:
     """List all BGP neighbors on the old router, to present an expected base-line for the new one."""
+    old_access_port_fqdn = AccessPort.from_db(UUID(old_access_port)).sbp.edge_port.node.router_fqdn
+
     return {
         "playbook_name": "gap_ansible/playbooks/manage_bgp_peers.yaml",
-        "inventory": {old_access_port.sbp.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {old_access_port_fqdn: None}}},
         "extra_vars": {
             "dry_run": True,
             "verb": "check",
             "subscription": subscription,
             "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Show BGP neighbors.",
         },
+        "old_access_port_fqdn": old_access_port_fqdn,
     }
 
 
 @step("[DRY RUN] Deactivate BGP session on the old router")
 def deactivate_bgp_dry(
-    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: AccessPort
+    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port_fqdn: str
 ) -> LSOState:
     """Perform a dry run of deactivating the BGP session on the old router."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_bgp_peers.yaml",
-        "inventory": {old_access_port.sbp.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {old_access_port_fqdn: None}}},
         "extra_vars": {
             "dry_run": True,
             "verb": "disable",
@@ -121,12 +125,12 @@ def deactivate_bgp_dry(
 
 @step("[FOR REAL] Deactivate BGP session on the old router")
 def deactivate_bgp_real(
-    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: AccessPort
+    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port_fqdn: str
 ) -> LSOState:
     """Deactivate the BGP session on the old router."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_bgp_peers.yaml",
-        "inventory": {old_access_port.sbp.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {old_access_port_fqdn: None}}},
         "extra_vars": {
             "dry_run": False,
             "verb": "disable",
@@ -158,12 +162,12 @@ def inform_operator_traffic_check() -> FormGenerator:
 
 @step("[DRY RUN] Deactivate SBP config on the old router")
 def deactivate_sbp_dry(
-    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: AccessPort
+    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port_fqdn: str
 ) -> LSOState:
     """Perform a dry run of deactivating SBP config on the old router."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_sbp.yaml",
-        "inventory": {old_access_port.sbp.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {old_access_port_fqdn: None}}},
         "extra_vars": {
             "dry_run": True,
             "verb": "disable",
@@ -175,23 +179,24 @@ def deactivate_sbp_dry(
 
 @step("[FOR REAL] Deactivate SBP config on the old router")
 def deactivate_sbp_real(
-    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port: AccessPort
+    subscription: L3CoreService, process_id: UUIDstr, tt_number: TTNumber, old_access_port_fqdn: str
 ) -> LSOState:
     """Deactivate the BGP session on the old router."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_sbp.yaml",
-        "inventory": {old_access_port.sbp.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {old_access_port_fqdn: None}}},
         "extra_vars": {
             "dry_run": False,
             "verb": "disable",
             "subscription": subscription,
             "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deactivate BGP session.",
         },
+        "__remove_keys": ["old_access_port_fqdn"],
     }
 
 
 @step("Generate updated subscription model")
-def generate_updated_subscription_model(
+def generate_scoped_subscription_model(
     subscription: L3CoreService, old_access_port: UUIDstr, new_edge_port: EdgePort
 ) -> State:
     """Calculate what the updated subscription model will look like, but don't update the actual subscription yet.
@@ -199,45 +204,42 @@ def generate_updated_subscription_model(
     The new subscription is used for running Ansible playbooks remotely, but the updated subscription model is not
     stored yet, to avoid issues recovering when the workflow is aborted.
     """
-    updated_subscription = copy.deepcopy(subscription)
-    for ap in updated_subscription.l3_core_service.ap_list:
-        if str(ap.subscription_instance_id) == str(old_access_port):
-            ap.sbp.edge_port = new_edge_port.edge_port
+    updated_subscription = json.loads(json_dumps(subscription))
+    for index, ap in enumerate(updated_subscription["l3_core_service"]["ap_list"]):
+        if ap["subscription_instance_id"] == old_access_port:
+            #  We have found the AP that is to be replaced, we can return all the necessary information to the state.
+            #  First, remove all unneeded unchanged APs that should not be included when executing a playbook.
+            updated_subscription["l3_core_service"]["ap_list"] = [
+                updated_subscription["l3_core_service"]["ap_list"][index]
+            ]
+            #  Second, replace the AP that is migrated such that it includes the new EP instead of the old one.
+            updated_subscription["l3_core_service"]["ap_list"][0]["sbp"]["edge_port"] = json.loads(
+                json_dumps(new_edge_port.edge_port)
+            )
+            return {"scoped_subscription": updated_subscription, "replaced_ap_index": index}
 
-    return {"updated_subscription": updated_subscription}
+    msg = "Failed to find selected AP in current subscription."
+    raise ProcessFailureError(msg, details=old_access_port)
 
 
 @step("[DRY RUN] Configure service on new Edge Port")
 def deploy_new_ep_dry(
-    updated_subscription: L3CoreService,
-    process_id: UUIDstr,
-    tt_number: TTNumber,
-    old_access_port: UUIDstr,
-    new_edge_port: EdgePort,
+    scoped_subscription: dict[str, Any], process_id: UUIDstr, tt_number: TTNumber, new_edge_port: EdgePort
 ) -> LSOState:
     """Deploy Access Port on the destination Edge Port, as a dry run.
 
     Only the updated Access Port is sent as part of the subscription model, to reduce the scope of the playbook.
     """
-    scoped_subscription = copy.deepcopy(updated_subscription)
-    scoped_subscription.l3_core_service.ap_list = list(
-        filter(
-            lambda ap: str(ap.subscription_instance_id) == str(old_access_port),
-            scoped_subscription.l3_core_service.ap_list,
-        )
-    )
-
     return {
         "playbook_name": "gap_ansible/playbooks/l3_core_service.yaml",
-        "inventory": {new_edge_port.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {new_edge_port.edge_port.node.router_fqdn: None}}},
         "extra_vars": {
             "dry_run": True,
             "verb": "deploy",
-            "subscription": json.loads(json_dumps(scoped_subscription)),
+            "subscription": scoped_subscription,
             "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
             "Deploying SBP and standard IDs.",
         },
-        "scoped_subscription": scoped_subscription,
     }
 
 
@@ -248,7 +250,7 @@ def deploy_new_ep_real(
     """Deploy Access Port on the destination Edge Port."""
     return {
         "playbook_name": "gap_ansible/playbooks/l3_core_service.yaml",
-        "inventory": {new_edge_port.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {new_edge_port.edge_port.node.router_fqdn: None}}},
         "extra_vars": {
             "dry_run": False,
             "verb": "deploy",
@@ -266,7 +268,7 @@ def deploy_bgp_session_dry(
     """Perform a dry run of deploying the new BGP session."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_bgp_peers.yaml",
-        "inventory": {new_edge_port.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {new_edge_port.edge_port.node.router_fqdn: None}}},
         "extra_vars": {
             "dry_run": True,
             "verb": "deploy",
@@ -283,7 +285,7 @@ def deploy_bgp_session_real(
     """Deploy the new BGP session."""
     return {
         "playbook_name": "gap_ansible/playbooks/manage_bgp_peers.yaml",
-        "inventory": {new_edge_port.edge_port.node.router_fqdn: None},
+        "inventory": {"all": {"hosts": {new_edge_port.edge_port.node.router_fqdn: None}}},
         "extra_vars": {
             "dry_run": False,
             "verb": "deploy",
@@ -295,9 +297,11 @@ def deploy_bgp_session_real(
 
 
 @step("Update subscription model")
-def update_subscription_model(updated_subscription: L3CoreService) -> State:
+def update_subscription_model(subscription: L3CoreService, new_edge_port: EdgePort, replaced_ap_index: int) -> State:
     """Update the subscription model with the new Edge Port attached to the Access Port that is currently migrated."""
-    return {"subscription": updated_subscription, "__remove_keys": ["updated_subscription"]}
+    subscription.l3_core_service.ap_list[replaced_ap_index].sbp.edge_port = new_edge_port.edge_port
+
+    return {"subscription": subscription, "__remove_keys": ["replaced_ap_index"]}
 
 
 @workflow(
@@ -318,7 +322,7 @@ def migrate_l3_core_service() -> StepList:
         >> inform_operator_traffic_check
         >> unsync
         >> start_moodi()  # TODO: include results from first LSO run
-        >> generate_updated_subscription_model
+        >> generate_scoped_subscription_model
         >> lso_interaction(deploy_new_ep_dry)
         >> lso_interaction(deploy_new_ep_real)
         >> lso_interaction(deploy_bgp_session_dry)

test/workflows/l3_core_service/test_migrate_l3_core_service.py

+from unittest.mock import patch
+
 import pytest
 
+from gso.products.product_blocks.l3_core_service import AccessPort
+from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.l3_core_service import L3_CORE_SERVICE_TYPES, L3CoreService
-from test.workflows import assert_complete, extract_state, run_workflow
+from gso.utils.shared_enums import APType
+from test import USER_CONFIRM_EMPTY_FORM
+from test.workflows import (
+    assert_complete,
+    assert_lso_interaction_success,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+)
 
 
-@pytest.mark.parametrize("l3_core_service_type", L3_CORE_SERVICE_TYPES)
 @pytest.mark.workflow()
+@pytest.mark.parametrize("l3_core_service_type", L3_CORE_SERVICE_TYPES)
+@patch("gso.services.lso_client._send_request")
 def test_migrate_l3_core_service_success(
+    mock_execute_playbook,
     faker,
     edge_port_subscription_factory,
     partner_factory,
     l3_core_service_subscription_factory,
     l3_core_service_type,
+    access_port_factory,
 ):
     partner = partner_factory()
     subscription_id = str(
-        l3_core_service_subscription_factory(partner=partner, l3_core_service_type=l3_core_service_type).subscription_id
+        l3_core_service_subscription_factory(
+            partner=partner, l3_core_service_type=l3_core_service_type, ap_list=[access_port_factory()]
+        ).subscription_id
     )
-    new_edge_port_1 = str(edge_port_subscription_factory(partner=partner).subscription_id)
-    new_edge_port_2 = str(edge_port_subscription_factory(partner=partner).subscription_id)
+    new_edge_port = str(edge_port_subscription_factory(partner=partner).subscription_id)
     subscription = L3CoreService.from_subscription(subscription_id)
 
     form_input_data = [
         {"subscription_id": subscription_id},
         {
             "tt_number": faker.tt_number(),
-            "edge_port_selection": [
-                {
-                    "old_edge_port": subscription.l3_core_service.ap_list[0].sbp.edge_port.description,
-                    "new_edge_port": new_edge_port_1,
-                },
-                {
-                    "old_edge_port": subscription.l3_core_service.ap_list[1].sbp.edge_port.description,
-                    "new_edge_port": new_edge_port_2,
-                },
-            ],
+            "old_access_port": subscription.l3_core_service.ap_list[0].subscription_instance_id,
         },
+        {"new_edge_port": new_edge_port},
+        {},
     ]
 
-    result, _, _ = run_workflow("migrate_l3_core_service", form_input_data)
+    result, process_stat, step_log = run_workflow("migrate_l3_core_service", form_input_data)
+
+    for _ in range(5):
+        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(4):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    assert_complete(result)
+    state = extract_state(result)
+    subscription = L3CoreService.from_subscription(state["subscription_id"])
+    assert mock_execute_playbook.call_count == 9
+    assert subscription.insync
+    assert len(subscription.l3_core_service.ap_list) == 1
+    assert str(subscription.l3_core_service.ap_list[0].sbp.edge_port.owner_subscription_id) == new_edge_port
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize("l3_core_service_type", L3_CORE_SERVICE_TYPES)
+@patch("gso.services.lso_client._send_request")
+def test_migrate_l3_core_service_scoped_emission(
+    mock_execute_playbook,
+    faker,
+    edge_port_subscription_factory,
+    access_port_factory,
+    partner_factory,
+    l3_core_service_subscription_factory,
+    l3_core_service_type,
+):
+    partner = partner_factory()
+    custom_ap_list = [access_port_factory(ap_type=APType.LOAD_BALANCED) for _ in range(5)]
+    subscription_id = str(
+        l3_core_service_subscription_factory(
+            partner=partner, l3_core_service_type=l3_core_service_type, ap_list=custom_ap_list
+        ).subscription_id
+    )
+    new_edge_port = str(edge_port_subscription_factory(partner=partner).subscription_id)
+    subscription = L3CoreService.from_subscription(subscription_id)
+    old_access_port = subscription.l3_core_service.ap_list[3].subscription_instance_id
+
+    form_input_data = [
+        {"subscription_id": subscription_id},
+        {
+            "tt_number": faker.tt_number(),
+            "old_access_port": old_access_port,
+        },
+        {"new_edge_port": new_edge_port},
+        {},
+    ]
+
+    result, process_stat, step_log = run_workflow("migrate_l3_core_service", form_input_data)
+
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    # In the first set of playbook runs, the targeted host should be the old EP that is removed from the selected AP.
+    state = extract_state(result)
+    assert len(state["inventory"]["all"]["hosts"].keys()) == 1
+    transmitted_old_ap_fqdn = next(iter(state["inventory"]["all"]["hosts"].keys()))
+    assert AccessPort.from_db(old_access_port).sbp.edge_port.node.router_fqdn == transmitted_old_ap_fqdn
+
+    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)
+
+    result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
+    # In the second set of playbook runs, the only targeted host should be the new EP, with the subscription object
+    # still containing the old EP.
+    state = extract_state(result)
+    assert len(state["inventory"]["all"]["hosts"].keys()) == 1
+    transmitted_new_ep_fqdn = next(iter(state["inventory"]["all"]["hosts"].keys()))
+    assert EdgePort.from_subscription(new_edge_port).edge_port.node.router_fqdn == transmitted_new_ep_fqdn
+    assert (
+        state["subscription"]["l3_core_service"] == state["__old_subscriptions__"][subscription_id]["l3_core_service"]
+    )  # Subscription is unchanged for now
+
+    for _ in range(3):
+        result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
     state = extract_state(result)
     subscription = L3CoreService.from_subscription(state["subscription_id"])
-    assert subscription.insync is True
-    assert len(subscription.l3_core_service.ap_list) == 2
-    assert str(subscription.l3_core_service.ap_list[0].sbp.edge_port.owner_subscription_id) == new_edge_port_1
-    assert str(subscription.l3_core_service.ap_list[1].sbp.edge_port.owner_subscription_id) == new_edge_port_2
+    assert mock_execute_playbook.call_count == 9
+    assert subscription.insync
+    assert len(subscription.l3_core_service.ap_list) == 5
+    assert str(subscription.l3_core_service.ap_list[3].sbp.edge_port.owner_subscription_id) == new_edge_port