split off reused method to __init__ module, update router creation test

gso/workflows/router/terminate_router.py

@@ -57,6 +57,8 @@ def deprovision_lt_ips(subscription: Router) -> dict:
 
 @step("Remove configuration from router")
 def remove_config_from_router() -> None:
+    #  FIXME: Add actual content
+    #  TODO: update unit test accordingly
     pass
 
 

test/workflows/__init__.py

@@ -294,3 +294,11 @@ def run_form_generator(
         result = stop.value
 
     return forms, result
+
+
+def user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
+    extra_data = extra_data or {}
+    result, step_log = resume_workflow(process_stat, step_log, extra_data)
+    assert_suspended(result)
+
+    return result, step_log

test/workflows/iptrunk/test_create_iptrunk.py

@@ -15,6 +15,7 @@ from test.workflows import (
     extract_state,
     resume_workflow,
     run_workflow,
+    user_accept_and_assert_suspended
 )
 
 
@@ -116,14 +117,6 @@ def input_form_wizard_data(router_subscription_factory, faker):
     ]
 
 
-def _user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
-    extra_data = extra_data or {}
-    result, step_log = resume_workflow(process_stat, step_log, extra_data)
-    assert_suspended(result)
-
-    return result, step_log
-
-
 @pytest.mark.workflow
 @patch("gso.workflows.iptrunk.create_iptrunk.provisioning_proxy.check_ip_trunk")
 @patch("gso.workflows.iptrunk.create_iptrunk.provisioning_proxy.provision_ip_trunk")
@@ -157,10 +150,10 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
         "confirm": "ACCEPTED",
     }
     for _ in range(5):
-        result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
-        result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
 
-    result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
+    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
     result, step_log = resume_workflow(process_stat, step_log, [{}, {}])
     assert_complete(result)
 
@@ -211,10 +204,10 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
 
     attempts = 3
     for _ in range(0, attempts - 1):
-        result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
-        result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
+        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
 
-    result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
+    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, standard_lso_result)
     result, step_log = resume_workflow(process_stat, step_log, [{}, {}])
     assert_aborted(result)
 

test/workflows/router/test_create_router.py

@@ -15,11 +15,12 @@ from test.workflows import (
     extract_state,
     resume_workflow,
     run_workflow,
+    user_accept_and_assert_suspended
 )
 
 
 @pytest.fixture
-def router_input_form_data(site_subscription_factory, faker):
+def router_creation_input_form_data(site_subscription_factory, faker):
     router_site = site_subscription_factory()
 
     return {
@@ -34,14 +35,6 @@ def router_input_form_data(site_subscription_factory, faker):
     }
 
 
-def _user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
-    extra_data = extra_data or {}
-    result, step_log = resume_workflow(process_stat, step_log, extra_data)
-    assert_suspended(result)
-
-    return result, step_log
-
-
 @pytest.mark.workflow
 @patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
 @patch("gso.workflows.router.create_router.NetBoxClient.create_device")
@@ -60,16 +53,17 @@ def test_create_router_success(
     mock_hostname_available,
     mock_netbox_create_device,
     mock_provision_router,
-    router_input_form_data,
+    router_creation_input_form_data,
     faker,
 ):
     #  Set up mock return values
-    mock_site = Site.from_subscription(router_input_form_data["router_site"]).site
+    product_id = get_product_id_by_name(ProductType.ROUTER)
+    mock_site = Site.from_subscription(router_creation_input_form_data["router_site"]).site
     mock_v4 = faker.ipv4()
     mock_v4_net = faker.ipv4_network()
     mock_v6 = faker.ipv6()
     mock_fqdn = (
-        f"{router_input_form_data['hostname']}.{mock_site.site_name.lower()}."
+        f"{router_creation_input_form_data['hostname']}.{mock_site.site_name.lower()}."
         f"{mock_site.site_country_code.lower()}.geant.net"
     )
     mock_hostname_available.return_value = True
@@ -79,7 +73,7 @@ def test_create_router_success(
     mock_find_host_by_fqdn.return_value = objects.HostRecord(
         connector=None,
         aliases=[mock_fqdn],
-        comment=faker.sentence(),
+        comment=product_id,
         ipv4addrs=[
             objects.IPv4(
                 ipv4addr=str(mock_v4),
@@ -100,8 +94,7 @@ def test_create_router_success(
     )
 
     #  Run workflow
-    product_id = get_product_id_by_name(ProductType.ROUTER)
-    initial_router_data = [{"product": product_id}, router_input_form_data]
+    initial_router_data = [{"product": product_id}, router_creation_input_form_data]
     result, process_stat, step_log = run_workflow("create_router", initial_router_data)
     assert_suspended(result)
 
@@ -115,9 +108,9 @@ def test_create_router_success(
         "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 = 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)
@@ -153,16 +146,16 @@ def test_create_router_lso_failure(
     mock_hostname_available,
     mock_netbox_create_device,
     mock_provision_router,
-    router_input_form_data,
+        router_creation_input_form_data,
     faker,
 ):
     #  Set up mock return values
-    mock_site = Site.from_subscription(router_input_form_data["router_site"]).site
+    mock_site = Site.from_subscription(router_creation_input_form_data["router_site"]).site
     mock_v4 = faker.ipv4()
     mock_v4_net = faker.ipv4_network()
     mock_v6 = faker.ipv6()
     mock_fqdn = (
-        f"{router_input_form_data['hostname']}.{mock_site.site_name.lower()}."
+        f"{router_creation_input_form_data['hostname']}.{mock_site.site_name.lower()}."
         f"{mock_site.site_country_code.lower()}.geant.net"
     )
     mock_hostname_available.return_value = True
@@ -194,7 +187,7 @@ def test_create_router_lso_failure(
 
     #  Run workflow
     product_id = get_product_id_by_name(ProductType.ROUTER)
-    initial_router_data = [{"product": product_id}, router_input_form_data]
+    initial_router_data = [{"product": product_id}, router_creation_input_form_data]
     result, process_stat, step_log = run_workflow("create_router", initial_router_data)
     assert_suspended(result)
 
@@ -210,10 +203,10 @@ def test_create_router_lso_failure(
 
     attempts = 3
     for _ in range(attempts - 1):
-        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 = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
     result, step_log = resume_workflow(process_stat, step_log, [{}, {}])
 
     assert_aborted(result)
@@ -229,3 +222,5 @@ def test_create_router_lso_failure(
     assert mock_netbox_create_device.call_count == 0
     assert mock_find_host_by_fqdn.call_count == 0
     assert mock_find_network_by_cidr.call_count == 0
+    for error in ["ipam_warning", "ipam_si_warning", "ipam_ias_lt_ipv4_warning", "ipam_ias_lt_ipv6_warning"]:
+        assert error not in state