Add checklist to trunk migration, and add ACTIVITY_TYPE column as discussed with SM

gso/workflows/iptrunk/create_iptrunk.py

@@ -582,6 +582,7 @@ def create_new_sharepoint_checklist(subscription: IptrunkProvisioning, tt_number
             "Title": f"{subscription.description} - {subscription.iptrunk.geant_s_sid}",
             "TT_NUMBER": tt_number,
             "GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
+            "ACTIVITY_TYPE": "Creation",
         },
     )
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -31,7 +31,9 @@ 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.sharepoint import SharePointClient
 from gso.services.subscriptions import get_active_router_subscriptions
+from gso.settings import load_oss_params
 from gso.utils.helpers import (
     LAGMember,
     available_interfaces_choices,
@@ -41,7 +43,7 @@ from gso.utils.helpers import (
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types import TTNumber
-from gso.utils.workflow_steps import set_isis_to_max
+from gso.utils.workflow_steps import prompt_sharepoint_checklist_url, set_isis_to_max
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -807,6 +809,22 @@ def netbox_allocate_new_interfaces(subscription: Iptrunk, replace_index: int) ->
     return {"subscription": subscription}
 
 
+@step("Create a new SharePoint checklist item")
+def create_new_sharepoint_checklist(subscription: Iptrunk, tt_number: str, process_id: UUIDstr) -> State:
+    """Create a new checklist item in SharePoint for approving this migrated IPtrunk."""
+    new_list_item_url = SharePointClient().add_list_item(
+        list_name="ip_trunk",
+        fields={
+            "Title": f"{subscription.description} - {subscription.iptrunk.geant_s_sid}",
+            "TT_NUMBER": tt_number,
+            "GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
+            "ACTIVITY_TYPE": "Migration",
+        },
+    )
+
+    return {"checklist_url": new_list_item_url}
+
+
 @workflow(
     "Migrate an IP Trunk",
     initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
@@ -868,5 +886,7 @@ def migrate_iptrunk() -> StepList:
         >> old_side_is_nokia(netbox_remove_old_interfaces)
         >> new_side_is_nokia(netbox_allocate_new_interfaces)
         >> resync
+        >> create_new_sharepoint_checklist
+        >> prompt_sharepoint_checklist_url
         >> done
     )

test/workflows/iptrunk/test_migrate_iptrunk.py

@@ -3,11 +3,12 @@ from unittest.mock import patch
 
 import pytest
 
-from gso.products import Iptrunk
+from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
 from gso.utils.shared_enums import Vendor
 from test import USER_CONFIRM_EMPTY_FORM
 from test.conftest import UseJuniperSide
+from test.services.conftest import MockedSharePointClient
 from test.workflows import (
     assert_complete,
     assert_lso_interaction_success,
@@ -121,7 +122,9 @@ def interface_lists_are_equal(list1, list2):
 @patch("gso.services.netbox_client.NetboxClient.allocate_interface")
 @patch("gso.services.netbox_client.NetboxClient.free_interface")
 @patch("gso.services.netbox_client.NetboxClient.delete_interface")
+@patch("gso.workflows.iptrunk.migrate_iptrunk.SharePointClient")
 def test_migrate_iptrunk_success(
+    mock_sharepoint_client,
     mocked_delete_interface,
     mocked_free_interface,
     mocked_allocate_interface,
@@ -146,6 +149,7 @@ def test_migrate_iptrunk_success(
     mocked_create_interface.return_value = mocked_netbox.create_interface()
     mocked_get_available_lags.return_value = mocked_netbox.get_available_lags()
     mocked_delete_interface.return_value = mocked_netbox.delete_interface()
+    mock_sharepoint_client.return_value = MockedSharePointClient
 
     result, process_stat, step_log = run_workflow("migrate_iptrunk", migrate_form_input)
 
@@ -164,6 +168,10 @@ def test_migrate_iptrunk_success(
     for _ in range(1):
         result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
 
+    #  Continue workflow after it has displayed a checklist URL.
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
     assert_complete(result)
 
     state = extract_state(result)