Skip to content
Snippets Groups Projects
Commit 17a71b7b authored by Karel van Klink's avatar Karel van Klink :smiley_cat: Committed by Neda Moeini
Browse files

add unit test for router termination workflow

parent 8ce94e8b
Branches
Tags
1 merge request!83Clean up the repo a bit, and add some unit tests
...@@ -86,7 +86,7 @@ def router_subscription_factory(site_subscription_factory, faker): ...@@ -86,7 +86,7 @@ def router_subscription_factory(site_subscription_factory, faker):
router_is_ias_connected=True, router_is_ias_connected=True,
) -> UUIDstr: ) -> UUIDstr:
description = description or faker.text(max_nb_chars=30) description = description or faker.text(max_nb_chars=30)
router_fqdn = router_fqdn or faker.domain_name() router_fqdn = router_fqdn or faker.domain_name(levels=4)
router_ts_port = router_ts_port or faker.random_int(min=1, max=49151) router_ts_port = router_ts_port or faker.random_int(min=1, max=49151)
router_access_via_ts = router_access_via_ts or faker.boolean() router_access_via_ts = router_access_via_ts or faker.boolean()
router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4()) router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
......
...@@ -15,7 +15,7 @@ from test.workflows import ( ...@@ -15,7 +15,7 @@ from test.workflows import (
extract_state, extract_state,
resume_workflow, resume_workflow,
run_workflow, run_workflow,
user_accept_and_assert_suspended user_accept_and_assert_suspended,
) )
......
...@@ -15,7 +15,7 @@ from test.workflows import ( ...@@ -15,7 +15,7 @@ from test.workflows import (
extract_state, extract_state,
resume_workflow, resume_workflow,
run_workflow, run_workflow,
user_accept_and_assert_suspended user_accept_and_assert_suspended,
) )
...@@ -70,10 +70,18 @@ def test_create_router_success( ...@@ -70,10 +70,18 @@ def test_create_router_success(
mock_allocate_host.return_value = str(mock_v4), str(mock_v6) mock_allocate_host.return_value = str(mock_v4), str(mock_v6)
mock_allocate_v4_network.return_value = mock_v4_net mock_allocate_v4_network.return_value = mock_v4_net
mock_allocate_v6_network.return_value = faker.ipv6_network() mock_allocate_v6_network.return_value = faker.ipv6_network()
# Run workflow
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)
state = extract_state(result)
subscription_id = state["subscription_id"]
mock_find_host_by_fqdn.return_value = objects.HostRecord( mock_find_host_by_fqdn.return_value = objects.HostRecord(
connector=None, connector=None,
aliases=[mock_fqdn], aliases=[mock_fqdn],
comment=product_id, comment=subscription_id,
ipv4addrs=[ ipv4addrs=[
objects.IPv4( objects.IPv4(
ipv4addr=str(mock_v4), ipv4addr=str(mock_v4),
...@@ -87,17 +95,12 @@ def test_create_router_success( ...@@ -87,17 +95,12 @@ def test_create_router_success(
) )
mock_find_network_by_cidr.return_value = objects.NetworkV4( mock_find_network_by_cidr.return_value = objects.NetworkV4(
connector=None, connector=None,
comment=faker.sentence(), comment=subscription_id,
network=str(mock_v4_net), network=str(mock_v4_net),
network_view="default", network_view="default",
cidr=str(mock_v4_net), cidr=str(mock_v4_net),
) )
# Run workflow
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)
lso_return = { lso_return = {
"pp_run_results": { "pp_run_results": {
"status": "ok", "status": "ok",
...@@ -116,7 +119,6 @@ def test_create_router_success( ...@@ -116,7 +119,6 @@ def test_create_router_success(
assert_complete(result) assert_complete(result)
state = extract_state(result) state = extract_state(result)
subscription_id = state["subscription_id"]
subscription = Router.from_subscription(subscription_id) subscription = Router.from_subscription(subscription_id)
assert "active" == subscription.status assert "active" == subscription.status
...@@ -126,6 +128,8 @@ def test_create_router_success( ...@@ -126,6 +128,8 @@ def test_create_router_success(
assert mock_netbox_create_device.call_count == 1 assert mock_netbox_create_device.call_count == 1
assert mock_find_host_by_fqdn.call_count == 1 assert mock_find_host_by_fqdn.call_count == 1
assert mock_find_network_by_cidr.call_count == 3 assert mock_find_network_by_cidr.call_count == 3
for error in ["ipam_warning", "ipam_si_warning", "ipam_ias_lt_ipv4_warning", "ipam_ias_lt_ipv6_warning"]:
assert error not in state
@pytest.mark.workflow @pytest.mark.workflow
...@@ -146,7 +150,7 @@ def test_create_router_lso_failure( ...@@ -146,7 +150,7 @@ def test_create_router_lso_failure(
mock_hostname_available, mock_hostname_available,
mock_netbox_create_device, mock_netbox_create_device,
mock_provision_router, mock_provision_router,
router_creation_input_form_data, router_creation_input_form_data,
faker, faker,
): ):
# Set up mock return values # Set up mock return values
...@@ -222,5 +226,3 @@ def test_create_router_lso_failure( ...@@ -222,5 +226,3 @@ def test_create_router_lso_failure(
assert mock_netbox_create_device.call_count == 0 assert mock_netbox_create_device.call_count == 0
assert mock_find_host_by_fqdn.call_count == 0 assert mock_find_host_by_fqdn.call_count == 0
assert mock_find_network_by_cidr.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
from unittest.mock import patch
import pytest
from gso.products import Router
from test.workflows import assert_complete, extract_state, run_workflow
@pytest.fixture
def router_termination_input_form_data(site_subscription_factory, faker):
return {"tt_number": faker.tt_number(), "remove_configuration": True, "clean_up_ipam": True}
@pytest.mark.workflow
@patch("gso.workflows.router.terminate_router.NetBoxClient.delete_device")
@patch("gso.workflows.router.terminate_router.infoblox.delete_host_by_ip")
@patch("gso.workflows.router.terminate_router.infoblox.delete_network")
def test_terminate_router_success(
mock_delete_network,
mock_delete_host_by_ip,
mock_delete_device,
router_termination_input_form_data,
router_subscription_factory,
faker,
):
# Set up active subscription in database
product_id = router_subscription_factory()
# Run workflow
initial_router_data = [{"subscription_id": product_id}, router_termination_input_form_data]
result, process_stat, step_log = run_workflow("terminate_router", initial_router_data)
assert_complete(result)
state = extract_state(result)
subscription_id = state["subscription_id"]
subscription = Router.from_subscription(subscription_id)
assert "terminated" == subscription.status
assert mock_delete_network.call_count == 3
assert mock_delete_device.call_count == 1
assert mock_delete_host_by_ip.call_count == 1
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment