diff --git a/test/conftest.py b/test/conftest.py
index 34c6a3dd088d125882ef5cf6097da8113ae5cc08..f46642a13a3babe429e5edcb3667659d65871223 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -59,6 +59,9 @@ class FakerProvider(BaseProvider):
def geant_sid(self) -> str:
return self.generator.numerify("SID-#####")
+ def network_interface(self) -> str:
+ return self.generator.numerify("ge-@#/@#/@#")
+
@pytest.fixture(scope="session")
def faker() -> Faker:
diff --git a/test/fixtures.py b/test/fixtures.py
index 49fc410913723a01e6bac9d185c4cf3ec76def51..3b7df108803af7fd4cc55d365ce8827e9c14f827 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -140,7 +140,7 @@ def iptrunk_side_subscription_factory(router_subscription_factory, faker):
iptrunk_side_node = Router.from_subscription(iptrunk_side_node_id).router
iptrunk_side_ae_iface = iptrunk_side_ae_iface or faker.pystr()
iptrunk_side_ae_geant_a_sid = iptrunk_side_ae_geant_a_sid or faker.geant_sid()
- iptrunk_side_ae_members = iptrunk_side_ae_members or ["ge-0/0/0", "ge-0/0/1"]
+ iptrunk_side_ae_members = iptrunk_side_ae_members or [faker.network_interface(), faker.network_interface()]
iptrunk_side_ae_members_description = iptrunk_side_ae_members_description or [
faker.sentence(),
faker.sentence(),
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
new file mode 100644
index 0000000000000000000000000000000000000000..452e7971fff198386780a55a8ca6e0f39bab4d73
--- /dev/null
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -0,0 +1,128 @@
+from unittest.mock import patch
+
+import pytest
+
+from gso.products import Iptrunk
+from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.utils.types.phy_port import PhyPortCapacity
+from test.workflows import (
+ assert_complete,
+ assert_suspended,
+ extract_state,
+ resume_workflow,
+ run_workflow,
+ user_accept_and_assert_suspended,
+)
+
+
+@pytest.mark.workflow
+@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk")
+def test_iptrunk_modify_trunk_interface_success(
+ mock_provision_ip_trunk,
+ iptrunk_subscription_factory,
+ faker,
+):
+ # Set up mock return values
+ product_id = iptrunk_subscription_factory()
+ new_sid = faker.geant_sid()
+ new_description = faker.sentence()
+ new_type = IptrunkType.LEASED
+ new_speed = PhyPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND
+ new_link_count = 2
+
+ new_side_a_sid = faker.geant_sid()
+ new_side_a_ae_members = [
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ ]
+ new_side_a_ae_descriptions = [
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ ]
+
+ new_side_b_sid = faker.geant_sid()
+ new_side_b_ae_members = [
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ faker.network_interface(),
+ ]
+ new_side_b_ae_descriptions = [
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ faker.sentence(),
+ ]
+
+ # Run workflow
+ initial_iptrunk_data = [
+ {"subscription_id": product_id},
+ {
+ "tt_number": faker.tt_number(),
+ "geant_s_sid": new_sid,
+ "iptrunk_description": new_description,
+ "iptrunk_type": new_type,
+ "iptrunk_speed": new_speed,
+ "iptrunk_minimum_links": new_link_count,
+ },
+ {
+ "iptrunk_sideA_ae_geant_a_sid": new_side_a_sid,
+ "iptrunk_sideA_ae_members": new_side_a_ae_members,
+ "iptrunk_sideA_ae_members_descriptions": new_side_a_ae_descriptions,
+ },
+ {
+ "iptrunk_sideB_ae_geant_a_sid": new_side_b_sid,
+ "iptrunk_sideB_ae_members": new_side_b_ae_members,
+ "iptrunk_sideB_ae_members_descriptions": new_side_b_ae_descriptions,
+ },
+ ]
+
+ result, process_stat, step_log = run_workflow("modify_trunk_interface", initial_iptrunk_data)
+ assert_suspended(result)
+
+ lso_return = {
+ "pp_run_results": {
+ "status": "ok",
+ "job_id": faker.uuid4(),
+ "output": "parsed_output",
+ "return_code": 0,
+ },
+ "confirm": "ACCEPTED",
+ }
+
+ 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 all subscription properties have been updated correctly
+ assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}"
+ assert subscription.iptrunk.geant_s_sid == new_sid
+ assert subscription.iptrunk.iptrunk_description == new_description
+ assert subscription.iptrunk.iptrunk_type == new_type
+ assert subscription.iptrunk.iptrunk_speed == new_speed
+ assert subscription.iptrunk.iptrunk_minimum_links == new_link_count
+ # FIXME: update assertions below when IPtrunk model is updated
+ # assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid == new_side_a_sid
+ # assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members == new_side_a_ae_members
+ # assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members_description == new_side_a_ae_descriptions
+ # assert subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_geant_a_sid == new_side_b_sid
+ # assert subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members == new_side_b_ae_members
+ # assert subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members_description == new_side_b_ae_descriptions