add unit test for router creation

...and fix a bug in router creation workflow

gso/workflows/router/create_router.py

@@ -10,16 +10,16 @@ from orchestrator.workflow import StepList, conditional, done, init, step, workf
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
-from services.crm import customer_selector
 
 from gso.products.product_blocks.router import RouterRole, RouterVendor, generate_fqdn
 from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site
 from gso.services import infoblox, provisioning_proxy, subscriptions
+from gso.services.crm import customer_selector
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
+from gso.utils.functions import iso_from_ipv4
 from gso.utils.types.ip_port import PortNumber
-from utils.functions import iso_from_ipv4
 
 
 def _site_selector() -> Choice:

test/conftest.py

@@ -47,6 +47,12 @@ class FakerProvider(BaseProvider):
 
         return ipaddress.IPv6Network(str(network) + "/64")
 
+    def tt_number(self) -> str:
+        random_date = self.generator.date(pattern="%Y%m%d")
+        random_int = self.generator.random_int(min=10000000, max=99999999)
+
+        return f"TT#{random_date}{random_int}"
+
 
 @pytest.fixture(scope="session")
 def faker() -> Faker:

test/fixtures.py

@@ -32,7 +32,7 @@ def site_subscription_factory(faker):
         site_ts_address=None,
     ) -> UUIDstr:
         description = description or "Site Subscription"
-        site_name = site_name or faker.name()
+        site_name = site_name or faker.domain_word()
         site_city = site_city or faker.city()
         site_country = site_country or faker.country()
         site_country_code = site_country_code or faker.country_code()

test/workflows/__init__.py

@@ -201,7 +201,7 @@ def run_workflow(workflow_key: str, input_data: State | list[State]) -> tuple[WF
 
 
 def resume_workflow(
-    process: ProcessStat, step_log: list[tuple[Step, WFProcess]], input_data: State
+    process: ProcessStat, step_log: list[tuple[Step, WFProcess]], input_data: State | list[State]
 ) -> tuple[WFProcess, list]:
     # ATTENTION!! This code needs to be as similar as possible to `server.services.processes.resume_process`
     # The main differences are: we use a different step log function, and we don't run in a separate thread

test/workflows/router/test_create_router.py

+from unittest.mock import patch
+
+import pytest
+from infoblox_client import objects
+
+from gso.products import ProductType, Site
+from gso.products.product_blocks.router import RouterRole, RouterVendor
+from gso.products.product_types.router import Router
+from gso.services.crm import customer_selector, get_customer_by_name
+from gso.services.subscriptions import get_product_id_by_name
+from test.workflows import (
+    assert_aborted,
+    assert_complete,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+)
+
+
+@pytest.fixture
+def router_input_form_data(site_subscription_factory, faker):
+    router_site = site_subscription_factory()
+
+    return {
+        "tt_number": faker.tt_number(),
+        "customer": getattr(customer_selector(), get_customer_by_name("GÉANT")["id"]),
+        "router_site": router_site,
+        "hostname": faker.pystr(),
+        "ts_port": faker.pyint(),
+        "router_vendor": faker.random_choices(elements=(RouterVendor.NOKIA, RouterVendor.JUNIPER), length=1)[0],
+        "router_role": faker.random_choices(elements=(RouterRole.P, RouterRole.PE, RouterRole.AMT), length=1)[0],
+        "is_ias_connected": True,
+    }
+
+
+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")
+@patch("gso.workflows.router.create_router.infoblox.hostname_available")
+@patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
+@patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
+@patch("gso.workflows.router.create_router.infoblox.allocate_v6_network")
+@patch("gso.workflows.router.create_router.infoblox.allocate_v4_network")
+@patch("gso.workflows.router.create_router.infoblox.allocate_host")
+def test_create_router_success(
+    mock_allocate_host,
+    mock_allocate_v4_network,
+    mock_allocate_v6_network,
+    mock_find_host_by_fqdn,
+    mock_find_network_by_cidr,
+    mock_hostname_available,
+    mock_netbox_create_device,
+    mock_provision_router,
+    router_input_form_data,
+    faker,
+):
+    #  Set up mock return values
+    mock_site = Site.from_subscription(router_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"{mock_site.site_country_code.lower()}.geant.net"
+    )
+    mock_hostname_available.return_value = True
+    mock_allocate_host.return_value = str(mock_v4), str(mock_v6)
+    mock_allocate_v4_network.return_value = mock_v4_net
+    mock_allocate_v6_network.return_value = faker.ipv6_network()
+    mock_find_host_by_fqdn.return_value = objects.HostRecord(
+        connector=None,
+        aliases=[mock_fqdn],
+        comment=faker.sentence(),
+        ipv4addrs=[
+            objects.IPv4(
+                ipv4addr=str(mock_v4),
+                configure_for_dhcp=False,
+                mac="00:00:00:00:00:00",
+                ip=str(mock_v4),
+                host=f"lo0.{mock_fqdn}",
+            )
+        ],
+        name=mock_fqdn,
+    )
+    mock_find_network_by_cidr.return_value = objects.NetworkV4(
+        connector=None,
+        comment=faker.sentence(),
+        network=str(mock_v4_net),
+        network_view="default",
+        cidr=str(mock_v4_net),
+    )
+
+    #  Run workflow
+    product_id = get_product_id_by_name(ProductType.ROUTER)
+    initial_router_data = [{"product": product_id}, router_input_form_data]
+    result, process_stat, step_log = run_workflow("create_router", initial_router_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 = Router.from_subscription(subscription_id)
+
+    assert "active" == subscription.status
+    assert subscription.description == f"Router {mock_fqdn}"
+
+    assert mock_provision_router.call_count == 2
+    assert mock_netbox_create_device.call_count == 1
+    assert mock_find_host_by_fqdn.call_count == 1
+    assert mock_find_network_by_cidr.call_count == 3
+
+
+@pytest.mark.workflow
+@patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
+@patch("gso.workflows.router.create_router.NetBoxClient.create_device")
+@patch("gso.workflows.router.create_router.infoblox.hostname_available")
+@patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
+@patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
+@patch("gso.workflows.router.create_router.infoblox.allocate_v6_network")
+@patch("gso.workflows.router.create_router.infoblox.allocate_v4_network")
+@patch("gso.workflows.router.create_router.infoblox.allocate_host")
+def test_create_router_lso_failure(
+    mock_allocate_host,
+    mock_allocate_v4_network,
+    mock_allocate_v6_network,
+    mock_find_host_by_fqdn,
+    mock_find_network_by_cidr,
+    mock_hostname_available,
+    mock_netbox_create_device,
+    mock_provision_router,
+    router_input_form_data,
+    faker,
+):
+    #  Set up mock return values
+    mock_site = Site.from_subscription(router_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"{mock_site.site_country_code.lower()}.geant.net"
+    )
+    mock_hostname_available.return_value = True
+    mock_allocate_host.return_value = str(mock_v4), str(mock_v6)
+    mock_allocate_v4_network.return_value = mock_v4_net
+    mock_allocate_v6_network.return_value = faker.ipv6_network()
+    mock_find_host_by_fqdn.return_value = objects.HostRecord(
+        connector=None,
+        aliases=[mock_fqdn],
+        comment=faker.sentence(),
+        ipv4addrs=[
+            objects.IPv4(
+                ipv4addr=str(mock_v4),
+                configure_for_dhcp=False,
+                mac="00:00:00:00:00:00",
+                ip=str(mock_v4),
+                host=f"lo0.{mock_fqdn}",
+            )
+        ],
+        name=mock_fqdn,
+    )
+    mock_find_network_by_cidr.return_value = objects.NetworkV4(
+        connector=None,
+        comment=faker.sentence(),
+        network=str(mock_v4_net),
+        network_view="default",
+        cidr=str(mock_v4_net),
+    )
+
+    #  Run workflow
+    product_id = get_product_id_by_name(ProductType.ROUTER)
+    initial_router_data = [{"product": product_id}, router_input_form_data]
+    result, process_stat, step_log = run_workflow("create_router", initial_router_data)
+    assert_suspended(result)
+
+    lso_return = {
+        "pp_run_results": {
+            "status": "failure",
+            "job_id": faker.uuid4(),
+            "output": "parsed_output",
+            "return_code": 1,
+        },
+        "confirm": "ACCEPTED",
+    }
+
+    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 = resume_workflow(process_stat, step_log, [{}, {}])
+
+    assert_aborted(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Router.from_subscription(subscription_id)
+
+    assert "provisioning" == subscription.status
+    assert subscription.description == f"Router {mock_fqdn}"
+
+    assert mock_provision_router.call_count == attempts
+    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