Added tests for termination workflows.

gso/workflows/router/terminate_router.py

@@ -263,8 +263,8 @@ def terminate_router() -> StepList:
     run_config_steps = conditional(lambda state: state["remove_configuration"])
     update_ibgp_mesh = conditional(lambda state: state["update_ibgp_mesh"])
     router_is_nokia = conditional(lambda state: state["router_is_nokia"])
-    router_is_pe = conditional(lambda state: state["router_role"] == "PE")
-    router_is_p = conditional(lambda state: state["router_role"] == "P")
+    router_is_pe = conditional(lambda state: state["router_role"] == RouterRole.PE)
+    router_is_p = conditional(lambda state: state["router_role"] == RouterRole.P)
 
     return (
         begin

test/workflows/router/test_terminate_router.py

@@ -8,26 +8,84 @@ from test.workflows import assert_complete, assert_lso_interaction_success, extr
 
 @pytest.mark.workflow()
 @pytest.mark.parametrize("remove_configuration", [True, False])
+@pytest.mark.parametrize("update_ibgp_mesh", [True, False])
 @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(
-    mock_delete_host_by_ip,
-    mock_delete_device,
-    mock_execute_playbook,
-    remove_configuration,
-    nokia_router_subscription_factory,
-    faker,
-    data_config_filename,
+@patch("gso.workflows.router.terminate_router.LibreNMSClient.remove_device")
+def test_terminate_pe_router_full_success(
+        mock_librenms_remove_device,
+        mock_delete_host_by_ip,
+        mock_delete_device,
+        mock_execute_playbook,
+        remove_configuration,
+        update_ibgp_mesh,
+        nokia_router_subscription_factory,
+        faker,
+        data_config_filename,
 ):
     #  Prepare mock values and expected results
     product_id = nokia_router_subscription_factory()
     router_termination_input_form_data = {
         "tt_number": faker.tt_number(),
         "remove_configuration": remove_configuration,
+        "update_ibgp_mesh": update_ibgp_mesh,
     }
-    lso_interaction_count = 2 if remove_configuration else 0
+    lso_interaction_count = 0
+    if remove_configuration:
+        lso_interaction_count += 2
+    if update_ibgp_mesh:
+        lso_interaction_count += 4
+    #  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(lso_interaction_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 = Router.from_subscription(subscription_id)
+
+    assert subscription.status == "terminated"
+    assert mock_delete_device.call_count == 1
+    assert mock_delete_host_by_ip.call_count == 1
+    assert mock_librenms_remove_device.call_count == 1
+    assert mock_execute_playbook.call_count == lso_interaction_count
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize("remove_configuration", [True, False])
+@pytest.mark.parametrize("update_ibgp_mesh", [True, False])
+@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")
+@patch("gso.workflows.router.terminate_router.LibreNMSClient.remove_device")
+def test_terminate_p_router_full_success(
+        mock_librenms_remove_device,
+        mock_delete_host_by_ip,
+        mock_delete_device,
+        mock_execute_playbook,
+        remove_configuration,
+        update_ibgp_mesh,
+        nokia_router_subscription_factory,
+        faker,
+        data_config_filename,
+):
+    #  Prepare mock values and expected results
+    product_id = nokia_router_subscription_factory(router_role="p")
+    router_termination_input_form_data = {
+        "tt_number": faker.tt_number(),
+        "remove_configuration": remove_configuration,
+        "update_ibgp_mesh": update_ibgp_mesh,
+    }
+    lso_interaction_count = 0
+    if remove_configuration:
+        lso_interaction_count += 2
+    if update_ibgp_mesh:
+        lso_interaction_count += 2
     #  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)
@@ -44,4 +102,5 @@ 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
+    assert mock_librenms_remove_device.call_count == 1
     assert mock_execute_playbook.call_count == lso_interaction_count