diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index b34d7d2c915f8f0472be3feb67b93bc395ee252c..ef69a304e589ea26e3582074fc3890aaec06a2c5 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/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}
diff --git a/test/services/conftest.py b/test/services/conftest.py
index 4501963708fd8ebbbed1eceac3ed3fc2283c76ea..282ae9edc8521e77cc6b8688670ad5218fa676c2 100644
--- a/test/services/conftest.py
+++ b/test/services/conftest.py
@@ -38,3 +38,6 @@ class MockedNetboxClient:
def detach_interfaces_from_lag(self):
return None
+
+ def delete_interface(self):
+ return None
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
new file mode 100644
index 0000000000000000000000000000000000000000..89eb44f55cb9ea5776487e760d8335101b3980f4
--- /dev/null
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -0,0 +1,133 @@
+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"