From 8ce94e8b37988342ef2d2534c3f004ae7b07d131 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Tue, 10 Oct 2023 11:41:07 +0200
Subject: [PATCH] split off reused method to __init__ module, update router
creation test
---
gso/workflows/router/terminate_router.py | 2 +
test/workflows/__init__.py | 8 ++++
test/workflows/iptrunk/test_create_iptrunk.py | 21 +++------
test/workflows/router/test_create_router.py | 45 +++++++++----------
4 files changed, 37 insertions(+), 39 deletions(-)
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 4989823f..47d09b41 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/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
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 899f7aa2..876b0e3e 100644
--- a/test/workflows/__init__.py
+++ b/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
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index c72329d7..dfded4f7 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/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)
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
index 8c818b4e..2c7a8a47 100644
--- a/test/workflows/router/test_create_router.py
+++ b/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
--
GitLab