diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 385191637d32de97c028d5953e25afbe1e13980b..3d19684eb2b83044a40a23a69c72a6a2b65351ec 100644
--- a/gso/workflows/router/create_router.py
+++ b/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:
diff --git a/test/conftest.py b/test/conftest.py
index 2bb17b7a4baf57e34ca19bfadcb70f0790cc5786..c9859e8c5fed66a2fd9e2256d2f8bf720c79b639 100644
--- a/test/conftest.py
+++ b/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:
diff --git a/test/fixtures.py b/test/fixtures.py
index 24864c5a96b9f4b26b2e2e648c63f12a26f1f1f4..1dcd2d790918be9bfed466ebc8edd17c19c4e692 100644
--- a/test/fixtures.py
+++ b/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()
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 9704dceec41bfc7644dfcfe5d81e6918bfd0ad55..899f7aa2e9f41688f63cbac52d3903eec6d85076 100644
--- a/test/workflows/__init__.py
+++ b/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
diff --git a/test/workflows/router/__init__.py b/test/workflows/router/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
new file mode 100644
index 0000000000000000000000000000000000000000..8c818b4e9cf041e43ae226880a5a95ee28c4a6dc
--- /dev/null
+++ b/test/workflows/router/test_create_router.py
@@ -0,0 +1,231 @@
+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