add (unfinished) unit test for iptrunk interface modification workflow

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:

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(),

test/workflows/iptrunk/test_modify_trunk_interface.py

+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