add unit test for iptrunk modify ISIS metric workflow

* add a fixture for an iptrunk subscription
* resolve some pytest warnings about an unregistered mark

test/conftest.py

@@ -53,6 +53,12 @@ class FakerProvider(BaseProvider):
 
         return f"TT#{random_date}{random_int}"
 
+    def geant_gid(self) -> str:
+        return self.generator.numerify("GID-#####")
+
+    def geant_sid(self) -> str:
+        return self.generator.numerify("SID-#####")
+
 
 @pytest.fixture(scope="session")
 def faker() -> Faker:

test/fixtures.py

@@ -6,11 +6,14 @@ from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
 from gso.products import ProductType
+from gso.products.product_blocks.iptrunk import IptrunkSideBlock, IptrunkType
 from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_blocks.site import SiteTier
-from gso.products.product_types.router import RouterInactive
+from gso.products.product_types.iptrunk import IptrunkInactive
+from gso.products.product_types.router import Router, RouterInactive
 from gso.products.product_types.site import Site, SiteInactive
 from gso.services import subscriptions
+from gso.utils.types.phy_port import PhyPortCapacity
 
 CUSTOMER_ID: UUIDstr = "2f47f65a-0911-e511-80d0-005056956c1a"
 
@@ -122,3 +125,85 @@ def router_subscription_factory(site_subscription_factory, faker):
         return str(router_subscription.subscription_id)
 
     return subscription_create
+
+
+@pytest.fixture
+def iptrunk_side_subscription_factory(router_subscription_factory, faker):
+    def subscription_create(
+        iptrunk_side_node=None,
+        iptrunk_side_ae_iface=None,
+        iptrunk_side_ae_geant_a_sid=None,
+        iptrunk_side_ae_members=None,
+        iptrunk_side_ae_members_description=None,
+    ) -> IptrunkSideBlock:
+        iptrunk_side_node_id = iptrunk_side_node or router_subscription_factory()
+        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_description = iptrunk_side_ae_members_description or [
+            faker.sentence(),
+            faker.sentence(),
+        ]
+
+        return IptrunkSideBlock.new(
+            faker.uuid4(),
+            iptrunk_side_node=iptrunk_side_node,
+            iptrunk_side_ae_iface=iptrunk_side_ae_iface,
+            iptrunk_side_ae_geant_a_sid=iptrunk_side_ae_geant_a_sid,
+            iptrunk_side_ae_members=iptrunk_side_ae_members,
+            iptrunk_side_ae_members_description=iptrunk_side_ae_members_description,
+        )
+
+    return subscription_create
+
+
+@pytest.fixture
+def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker):
+    def subscription_create(
+        description=None,
+        start_date="2023-05-24T00:00:00+00:00",
+        geant_s_sid=None,
+        iptrunk_description=None,
+        iptrunk_type=IptrunkType.DARK_FIBER,
+        iptrunk_speed=PhyPortCapacity.ONE_GIGABIT_PER_SECOND,
+        iptrunk_isis_metric=None,
+        iptrunk_ipv4_network=None,
+        iptrunk_ipv6_network=None,
+        iptrunk_sides=None,
+    ) -> UUIDstr:
+        product_id = subscriptions.get_product_id_by_name(ProductType.IP_TRUNK)
+        description = description or faker.sentence()
+
+        geant_s_sid = geant_s_sid or faker.geant_sid()
+        iptrunk_description = iptrunk_description or faker.sentence()
+        iptrunk_isis_metric = iptrunk_isis_metric or faker.pyint()
+        iptrunk_ipv4_network = iptrunk_ipv4_network or faker.ipv4_network()
+        iptrunk_ipv6_network = iptrunk_ipv6_network or faker.ipv6_network()
+        iptrunk_minimum_links = 1
+        iptrunk_side_a = iptrunk_side_subscription_factory()
+        iptrunk_side_b = iptrunk_side_subscription_factory()
+        iptrunk_sides = iptrunk_sides or [iptrunk_side_a, iptrunk_side_b]
+
+        iptrunk_subscription = IptrunkInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
+        iptrunk_subscription.iptrunk.geant_s_sid = geant_s_sid
+        iptrunk_subscription.iptrunk.iptrunk_description = iptrunk_description
+        iptrunk_subscription.iptrunk.iptrunk_type = iptrunk_type
+        iptrunk_subscription.iptrunk.iptrunk_speed = iptrunk_speed
+        iptrunk_subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
+        iptrunk_subscription.iptrunk.iptrunk_isis_metric = iptrunk_isis_metric
+        iptrunk_subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
+        iptrunk_subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network
+        iptrunk_subscription.iptrunk.iptrunk_sides = iptrunk_sides
+
+        iptrunk_subscription = SubscriptionModel.from_other_lifecycle(
+            iptrunk_subscription, SubscriptionLifecycle.ACTIVE
+        )
+        iptrunk_subscription.description = description
+        iptrunk_subscription.start_date = start_date
+        iptrunk_subscription.save()
+        db.session.commit()
+
+        return str(iptrunk_subscription.subscription_id)
+
+    return subscription_create

test/imports/conftest.py

-from test.fixtures import router_subscription_factory, site_subscription_factory  # noqa
+from test.fixtures import (  # noqa
+    iptrunk_side_subscription_factory,
+    iptrunk_subscription_factory,
+    router_subscription_factory,
+    site_subscription_factory,
+)

test/workflows/conftest.py

 import pytest
 from urllib3_mock import Responses
 
-from test.fixtures import router_subscription_factory, site_subscription_factory  # noqa
+from test.fixtures import (  # noqa
+    iptrunk_side_subscription_factory,
+    iptrunk_subscription_factory,
+    router_subscription_factory,
+    site_subscription_factory,
+)
 
 
 @pytest.fixture(autouse=True)

test/workflows/iptrunk/test_modify_isis_metric.py

+from unittest.mock import patch
+
+import pytest
+
+from gso.products import Iptrunk
+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_isis_metric.provisioning_proxy.provision_ip_trunk")
+def test_iptrunk_modify_isis_metric_success(
+    mock_provision_ip_trunk,
+    iptrunk_subscription_factory,
+    faker,
+):
+    #  Set up mock return values
+    product_id = iptrunk_subscription_factory()
+    new_isis_metric = faker.pyint()
+
+    #  Run workflow
+    initial_iptrunk_data = [
+        {"subscription_id": product_id},
+        {"tt_number": faker.tt_number(), "isis_metric": new_isis_metric},
+    ]
+    result, process_stat, step_log = run_workflow("modify_isis_metric", 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 subscription.iptrunk.iptrunk_isis_metric == new_isis_metric

tox.ini

@@ -7,6 +7,9 @@ select = B,C,D,E,F,G,I,N,S,T,W,B902,B903,R
 max-line-length = 120
 ban-relative-imports = true
 
+[pytest]
+markers = workflow
+
 [testenv]
 passenv = DATABASE_URI_TEST,SKIP_ALL_TESTS
 deps =