Added test for migrate IP trunk workflow.

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -69,7 +69,8 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     ]
 
     routers = {}
-    for router_id, router_description in get_active_router_subscriptions(includes=["subscription_id", "description"]):
+    for router in get_active_router_subscriptions(includes=["subscription_id", "description"]):
+        router_id = router["subscription_id"]
         if router_id not in current_routers:
             current_router_site = Router.from_subscription(router_id).router.router_site.subscription
             old_side_site = Router.from_subscription(migrate_form_input.replace_side).router.router_site
@@ -78,7 +79,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
                 and current_router_site.subscription_id == old_side_site.owner_subscription_id
             ):
                 continue
-            routers[str(router_id)] = router_description
+            routers[str(router_id)] = router["description"]
 
     new_router_enum = Choice("Select a new router", zip(routers.keys(), routers.items()))  # type: ignore[arg-type]
 
@@ -403,8 +404,6 @@ def delete_old_config_real(
 
 @step("Update IPAM")
 def update_ipam(subscription: Iptrunk) -> State:
-    pass
-
     return {"subscription": subscription}
 
 

test/services/conftest.py

@@ -38,3 +38,6 @@ class MockedNetboxClient:
 
     def detach_interfaces_from_lag(self):
         return None
+
+    def delete_interface(self):
+        return None

test/workflows/iptrunk/test_migrate_iptrunk.py

+from unittest.mock import patch
+
+import pytest
+
+from gso.products import Iptrunk
+from gso.utils.helpers import LAGMember
+from test.workflows import (
+    assert_complete,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+    user_accept_and_assert_suspended,
+)
+from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
+
+
+@pytest.mark.workflow
+@patch("gso.workflows.iptrunk.migrate_iptrunk.provisioning_proxy.migrate_ip_trunk")
+@patch("gso.workflows.iptrunk.migrate_iptrunk.provisioning_proxy.provision_ip_trunk")
+@patch("gso.services.netbox_client.NetboxClient.get_device_by_name")
+@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")
+@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag")
+@patch("gso.services.netbox_client.NetboxClient.reserve_interface")
+@patch("gso.services.netbox_client.NetboxClient.allocate_interface")
+@patch("gso.services.netbox_client.NetboxClient.free_interface")
+@patch("gso.services.netbox_client.NetboxClient.delete_interface")
+def test_migrate_iptrunk_success(
+    mocked_delete_interface,
+    mocked_free_interface,
+    mocked_allocate_interface,
+    mocked_reserve_interface,
+    mocked_attach_interface_to_lag,
+    mocked_create_interface,
+    mocked_get_available_lags,
+    mocked_get_available_interfaces,
+    mocked_get_device_by_name,
+    mock_provision_ip_trunk,
+    mock_migrate_ip_trunk,
+    iptrunk_subscription_factory,
+    router_subscription_factory,
+    faker,
+):
+    #  Set up mock return values
+    mocked_netbox = MockedNetboxClient()
+    mocked_get_device_by_name.return_value = mocked_netbox.get_device_by_name()
+    mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
+    mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
+    mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
+    mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
+    mocked_free_interface.return_value = mocked_netbox.free_interface()
+    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()
+
+    product_id = iptrunk_subscription_factory()
+    old_subscription = Iptrunk.from_subscription(product_id)
+    new_router = router_subscription_factory()
+
+    #  Run workflow
+    migrate_form_input = [
+        {"subscription_id": product_id},
+        {
+            "tt_number": faker.tt_number(),
+            "replace_side": str(
+                old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id
+            ),
+        },
+        {
+            "new_node": new_router,
+        },
+        {
+            "new_lag_interface": "LAG1",
+            "new_lag_member_interfaces": [
+                LAGMember(interface_name=f"Interface{interface}", interface_description=faker.sentence())
+                for interface in range(2)
+            ],
+        },
+    ]
+
+    result, process_stat, step_log = run_workflow("migrate_iptrunk", migrate_form_input)
+    assert_suspended(result)
+
+    lso_return = {
+        "pp_run_results": {
+            "status": "ok",
+            "job_id": faker.uuid4(),
+            "output": "parsed_output",
+            "return_code": 0,
+        },
+        "confirm": "ACCEPTED",
+    }
+    # Resume steps
+    for _ in range(5):
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
+    for _ in range(2):
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log)
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
+
+    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
+    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
+    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
+    result, step_log = resume_workflow(process_stat, step_log, [{}, {}])
+
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Iptrunk.from_subscription(subscription_id)
+
+    assert "active" == subscription.status
+    assert mock_provision_ip_trunk.call_count == 2
+    assert mock_migrate_ip_trunk.call_count == 7
+    # Assert all Netbox calls have been made
+    # This test case is only for migrating Nokia to Nokia.
+    # For Juniper to Nokia and Nokia to Juniper, the workflow is different.
+    assert mocked_create_interface.call_count == 1  # once for creating the LAG on the newly replaced side
+    assert mocked_reserve_interface.call_count == 2  # Twice for the new interfaces
+    assert mocked_attach_interface_to_lag.call_count == 2  # Twice for the new interfaces
+    assert mocked_allocate_interface.call_count == 2  # Twice for the new interfaces
+    assert mocked_free_interface.call_count == 2  # Twice for the old interfaces
+    assert mocked_delete_interface.call_count == 1  # once for deleting the LAG on the old replaced side
+
+    # Assert the new side is replaced
+    assert str(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id) == new_router
+    assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface == "LAG1"
+    assert len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members) == 2
+    assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members[0].interface_name == "Interface0"
+    assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members[1].interface_name == "Interface1"