diff --git a/gso/services/provisioning_proxy.py b/gso/services/provisioning_proxy.py
index c96ef37b1b3f3beb76418e8ecf46a4a2f28238cb..f3e18eb8503b15fe8e7e1bc561548811d0bd8f4c 100644
--- a/gso/services/provisioning_proxy.py
+++ b/gso/services/provisioning_proxy.py
@@ -4,7 +4,6 @@
"""
import json
import logging
-from typing import NoReturn
import requests
from orchestrator import conditional, inputstep, step
@@ -252,7 +251,7 @@ def _await_pp_results(subscription: SubscriptionModel, label_text: str = DEFAULT
confirm: Accept = Accept("INCOMPLETE")
@validator("pp_run_results", allow_reuse=True, pre=True, always=True)
- def run_results_must_be_given(cls, run_results: dict) -> dict | NoReturn:
+ def run_results_must_be_given(cls, run_results: dict) -> dict:
if run_results is None:
raise ValueError("Run results may not be empty. Wait for the provisioning proxy to finish.")
return run_results
diff --git a/requirements.txt b/requirements.txt
index 8b230dfd5001ec8ce78d940f8a7c01eb4a74bdd7..34aace52d4c3c20360ca4ed9d8d134335de51105 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -13,3 +13,4 @@ mypy==1.5.1
ruff==0.0.290
sphinx==7.2.6
sphinx-rtd-theme==1.3.0
+urllib3_mock
\ No newline at end of file
diff --git a/test/conftest.py b/test/conftest.py
index a811b7c34665c993f704b16ae0e2528f90e4ea82..39dff29c18ca71c53b50d29f8dee86624bace6b9 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -1,6 +1,7 @@
import contextlib
import ipaddress
import json
+import logging
import os
import socket
import tempfile
@@ -22,6 +23,8 @@ from starlette.testclient import TestClient
from gso.main import init_gso_app
+logging.getLogger("faker.factory").setLevel(logging.WARNING)
+
def pytest_collection_modifyitems(config, items):
if bool(os.environ.get("SKIP_ALL_TESTS")):
diff --git a/test/fixtures.py b/test/fixtures.py
new file mode 100644
index 0000000000000000000000000000000000000000..2f3ba8993da821982e2792a3b12b125e88958ce6
--- /dev/null
+++ b/test/fixtures.py
@@ -0,0 +1,124 @@
+import ipaddress
+
+import pytest
+from orchestrator.db import db
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+
+from gso.products.product_blocks.router import RouterRole, RouterVendor
+from gso.products.product_blocks.site import SiteTier
+from gso.products.product_types.router import RouterInactive
+from gso.products.product_types.site import Site, SiteInactive
+from gso.schemas.enums import ProductType
+from gso.services import subscriptions
+
+CUSTOMER_ID: UUIDstr = "2f47f65a-0911-e511-80d0-005056956c1a"
+
+
+@pytest.fixture
+def site_subscription_factory(faker):
+ def subscription_create(
+ description=None,
+ start_date="2023-05-24T00:00:00+00:00",
+ site_name=None,
+ site_city=None,
+ site_country=None,
+ site_country_code=None,
+ site_latitude=None,
+ site_longitude=None,
+ site_bgp_community_id=None,
+ site_internal_id=None,
+ site_tier=SiteTier.TIER1,
+ site_ts_address=None,
+ ) -> UUIDstr:
+ description = description or "Site Subscription"
+ site_name = site_name or faker.name()
+ site_city = site_city or faker.city()
+ site_country = site_country or faker.country()
+ site_country_code = site_country_code or faker.country_code()
+ site_latitude = site_latitude or float(faker.latitude())
+ site_longitude = site_longitude or float(faker.longitude())
+ site_bgp_community_id = site_bgp_community_id or faker.pyint()
+ site_internal_id = site_internal_id or faker.pyint()
+ site_ts_address = site_ts_address or faker.ipv4()
+
+ product_id = subscriptions.get_product_id_by_name(ProductType.SITE)
+ site_subscription = SiteInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
+ site_subscription.site.site_city = site_city
+ site_subscription.site.site_name = site_name
+ site_subscription.site.site_country = site_country
+ site_subscription.site.site_country_code = site_country_code
+ site_subscription.site.site_latitude = site_latitude
+ site_subscription.site.site_longitude = site_longitude
+ site_subscription.site.site_bgp_community_id = site_bgp_community_id
+ site_subscription.site.site_internal_id = site_internal_id
+ site_subscription.site.site_tier = site_tier
+ site_subscription.site.site_ts_address = site_ts_address
+
+ site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
+ site_subscription.description = description
+ site_subscription.start_date = start_date
+ site_subscription.save()
+ db.session.commit()
+
+ return str(site_subscription.subscription_id)
+
+ return subscription_create
+
+
+@pytest.fixture
+def router_subscription_factory(site_subscription_factory, faker):
+ def subscription_create(
+ description=None,
+ start_date="2023-05-24T00:00:00+00:00",
+ router_fqdn=None,
+ router_ts_port=None,
+ router_access_via_ts=None,
+ router_lo_ipv4_address=None,
+ router_lo_ipv6_address=None,
+ router_lo_iso_address=None,
+ router_si_ipv4_network=None,
+ router_ias_lt_ipv4_network=None,
+ router_ias_lt_ipv6_network=None,
+ router_vendor=RouterVendor.NOKIA,
+ router_role=RouterRole.PE,
+ router_site=None,
+ router_is_ias_connected=True,
+ ) -> UUIDstr:
+ description = description or faker.text(max_nb_chars=30)
+ router_fqdn = router_fqdn or faker.domain_name()
+ 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_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
+ router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
+ router_lo_iso_address = router_lo_iso_address or faker.word()
+ router_si_ipv4_network = router_si_ipv4_network or faker.ipv4_network()
+ router_ias_lt_ipv4_network = router_ias_lt_ipv4_network or faker.ipv4_network()
+ router_ias_lt_ipv6_network = router_ias_lt_ipv6_network or faker.ipv6_network()
+ router_site = router_site or site_subscription_factory()
+
+ product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
+ router_subscription = RouterInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
+ router_subscription.router.router_fqdn = router_fqdn
+ router_subscription.router.router_ts_port = router_ts_port
+ router_subscription.router.router_access_via_ts = router_access_via_ts
+ router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
+ router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
+ router_subscription.router.router_lo_iso_address = router_lo_iso_address
+ router_subscription.router.router_si_ipv4_network = router_si_ipv4_network
+ router_subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
+ router_subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
+ router_subscription.router.router_vendor = router_vendor
+ router_subscription.router.router_role = router_role
+ router_subscription.router.router_site = Site.from_subscription(router_site).site
+ router_subscription.router.router_is_ias_connected = router_is_ias_connected
+
+ router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
+ router_subscription.description = description
+ router_subscription.start_date = start_date
+ router_subscription.save()
+ db.session.commit()
+
+ return str(router_subscription.subscription_id)
+
+ return subscription_create
diff --git a/test/imports/conftest.py b/test/imports/conftest.py
index 320b0669726b1909f8a693ad528fc652bf917563..425a0e627a4592241e2c3f81cce910255dd34a5e 100644
--- a/test/imports/conftest.py
+++ b/test/imports/conftest.py
@@ -1,124 +1 @@
-import ipaddress
-
-import pytest
-from orchestrator.db import db
-from orchestrator.domain import SubscriptionModel
-from orchestrator.types import SubscriptionLifecycle, UUIDstr
-
-from gso.products.product_blocks.router import RouterRole, RouterVendor
-from gso.products.product_blocks.site import SiteTier
-from gso.products.product_types.router import RouterInactive
-from gso.products.product_types.site import Site, SiteInactive
-from gso.schemas.enums import ProductType
-from gso.services import subscriptions
-
-CUSTOMER_ID: UUIDstr = "2f47f65a-0911-e511-80d0-005056956c1a"
-
-
-@pytest.fixture
-def site_subscription_factory(faker):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- site_name=None,
- site_city=None,
- site_country=None,
- site_country_code=None,
- site_latitude=None,
- site_longitude=None,
- site_bgp_community_id=None,
- site_internal_id=None,
- site_tier=SiteTier.TIER1,
- site_ts_address=None,
- ) -> UUIDstr:
- description = description or "Site Subscription"
- site_name = site_name or faker.name()
- site_city = site_city or faker.city()
- site_country = site_country or faker.country()
- site_country_code = site_country_code or faker.country_code()
- site_latitude = site_latitude or float(faker.latitude())
- site_longitude = site_longitude or float(faker.longitude())
- site_bgp_community_id = site_bgp_community_id or faker.pyint()
- site_internal_id = site_internal_id or faker.pyint()
- site_ts_address = site_ts_address or faker.ipv4()
-
- product_id = subscriptions.get_product_id_by_name(ProductType.SITE)
- site_subscription = SiteInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
- site_subscription.site.site_city = site_city
- site_subscription.site.site_name = site_name
- site_subscription.site.site_country = site_country
- site_subscription.site.site_country_code = site_country_code
- site_subscription.site.site_latitude = site_latitude
- site_subscription.site.site_longitude = site_longitude
- site_subscription.site.site_bgp_community_id = site_bgp_community_id
- site_subscription.site.site_internal_id = site_internal_id
- site_subscription.site.site_tier = site_tier
- site_subscription.site.site_ts_address = site_ts_address
-
- site_subscription = SubscriptionModel.from_other_lifecycle(site_subscription, SubscriptionLifecycle.ACTIVE)
- site_subscription.description = description
- site_subscription.start_date = start_date
- site_subscription.save()
- db.session.commit()
-
- return str(site_subscription.subscription_id)
-
- return subscription_create
-
-
-@pytest.fixture
-def router_subscription_factory(site_subscription_factory, faker):
- def subscription_create(
- description=None,
- start_date="2023-05-24T00:00:00+00:00",
- router_fqdn=None,
- router_ts_port=None,
- router_access_via_ts=None,
- router_lo_ipv4_address=None,
- router_lo_ipv6_address=None,
- router_lo_iso_address=None,
- router_si_ipv4_network=None,
- router_ias_lt_ipv4_network=None,
- router_ias_lt_ipv6_network=None,
- router_vendor=RouterVendor.NOKIA,
- router_role=RouterRole.PE,
- router_site=None,
- router_is_ias_connected=True,
- ) -> UUIDstr:
- description = description or faker.text(max_nb_chars=30)
- router_fqdn = router_fqdn or faker.domain_name()
- 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_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
- router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
- router_lo_iso_address = router_lo_iso_address or faker.word()
- router_si_ipv4_network = router_si_ipv4_network or faker.ipv4_network()
- router_ias_lt_ipv4_network = router_ias_lt_ipv4_network or faker.ipv4_network()
- router_ias_lt_ipv6_network = router_ias_lt_ipv6_network or faker.ipv6_network()
- router_site = router_site or site_subscription_factory()
-
- product_id = subscriptions.get_product_id_by_name(ProductType.ROUTER)
- router_subscription = RouterInactive.from_product_id(product_id, customer_id=CUSTOMER_ID, insync=True)
- router_subscription.router.router_fqdn = router_fqdn
- router_subscription.router.router_ts_port = router_ts_port
- router_subscription.router.router_access_via_ts = router_access_via_ts
- router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
- router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
- router_subscription.router.router_lo_iso_address = router_lo_iso_address
- router_subscription.router.router_si_ipv4_network = router_si_ipv4_network
- router_subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
- router_subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
- router_subscription.router.router_vendor = router_vendor
- router_subscription.router.router_role = router_role
- router_subscription.router.router_site = Site.from_subscription(router_site).site
- router_subscription.router.router_is_ias_connected = router_is_ias_connected
-
- router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
- router_subscription.description = description
- router_subscription.start_date = start_date
- router_subscription.save()
- db.session.commit()
-
- return str(router_subscription.subscription_id)
-
- return subscription_create
+from test.fixtures import router_subscription_factory, site_subscription_factory # noqa
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 8239575e9c63ec888c7a5acdfc212b98b21d84b5..05c9de4f693d356f8d2ca614c7f9dba30896b912 100644
--- a/test/workflows/__init__.py
+++ b/test/workflows/__init__.py
@@ -6,16 +6,14 @@ from typing import Callable, Dict, List, Optional, Tuple, Union, cast
from uuid import uuid4
import structlog
-
from orchestrator.db import ProcessTable
from orchestrator.forms import post_process
from orchestrator.services.processes import StateMerger, _db_create_process
-from orchestrator.types import State
+from orchestrator.types import FormGenerator, InputForm, State
from orchestrator.utils.json import json_dumps, json_loads
from orchestrator.workflow import Process as WFProcess
from orchestrator.workflow import ProcessStat, Step, Success, Workflow, runwf
from orchestrator.workflows import ALL_WORKFLOWS, LazyWorkflowInstance, get_workflow
-from orchestrator.types import FormGenerator, InputForm
logger = structlog.get_logger(__name__)
@@ -133,9 +131,7 @@ class WorkflowInstanceForTests(LazyWorkflowInstance):
This can be as simple as merely importing a workflow function. However, if it concerns a workflow generating
function, that function will be called with or without arguments as specified.
- Returns:
- A workflow function.
-
+ Returns: A workflow function.
"""
self.workflow.name = self.name
return self.workflow
@@ -227,20 +223,23 @@ def run_form_generator(
) -> Tuple[List[dict], State]:
"""Run a form generator to get the resulting forms and result.
- Warning! This does not run the actual pydantic validation on purpose. However you should
+ Warning! This does not run the actual pydantic validation on purpose. However, you should
make sure that anything in extra_inputs matched the values and types as if the pydantic validation has
- been ran.
+ been run.
Args:
- form_generator: A form generator
- extra_inputs: Optional list of user input dicts for each page in the generator.
- If no input is given for a page an empty dict is used.
- The default value from the form is used as default value for a field.
+ ----
+ form_generator (FormGenerator): The form generator that will be run.
+ extra_inputs (Optional[List[State]]): List of user input dicts for each page in the generator.
+ If no input is given for a page, an empty dict is used.
+ The default value from the form is used as the default value for a field.
Returns:
- A list of generated forms and the result state for the whole generator.
+ -------
+ Tuple[List[dict], State]: A list of generated forms and the result state for the whole generator.
Example:
+ -------
Given the following form generator:
>>> from pydantic_forms.core import FormPage
@@ -253,7 +252,8 @@ def run_form_generator(
You can run this without extra_inputs
>>> forms, result = run_form_generator(form_generator({"state_field": 1}))
>>> forms
- [{'title': 'unknown', 'type': 'object', 'properties': {'field': {'title': 'Field', 'default': 'foo', 'type': 'string'}}, 'additionalProperties': False}]
+ [{'title': 'unknown', 'type': 'object', 'properties': {
+ 'field': {'title': 'Field', 'default': 'foo', 'type': 'string'}}, 'additionalProperties': False}]
>>> result
{'field': 'foo', 'bar': 42}
@@ -261,7 +261,8 @@ def run_form_generator(
Or with extra_inputs:
>>> forms, result = run_form_generator(form_generator({'state_field': 1}), [{'field':'baz'}])
>>> forms
- [{'title': 'unknown', 'type': 'object', 'properties': {'field': {'title': 'Field', 'default': 'foo', 'type': 'string'}}, 'additionalProperties': False}]
+ [{'title': 'unknown', 'type': 'object', 'properties': {
+ 'field': {'title': 'Field', 'default': 'foo', 'type': 'string'}}, 'additionalProperties': False}]
>>> result
{'field': 'baz', 'bar': 42}
diff --git a/test/workflows/conftest.py b/test/workflows/conftest.py
index c548aec6d49a4f07076389a0b4d665bf348b1c49..6e6630890466c6a1f3d1d63000eefff387927472 100644
--- a/test/workflows/conftest.py
+++ b/test/workflows/conftest.py
@@ -1,6 +1,8 @@
import pytest
from urllib3_mock import Responses
+from test.fixtures import router_subscription_factory, site_subscription_factory # noqa
+
@pytest.fixture(autouse=True)
def responses():
@@ -13,7 +15,7 @@ def responses():
return mock_url
def _to_tuple(url_mock):
- return (url_mock["url"], url_mock["method"], url_mock["match_querystring"])
+ return url_mock["url"], url_mock["method"], url_mock["match_querystring"]
with responses_mock:
yield responses_mock
diff --git a/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py b/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py
new file mode 100644
index 0000000000000000000000000000000000000000..dfb6fa4f98e7ebf0ec7afd1b5e09855d09add6f8
--- /dev/null
+++ b/test/workflows/iptrunks/iptrunks/test_create_iptrunks.py
@@ -0,0 +1,154 @@
+from unittest.mock import patch
+
+import pytest
+
+from gso.products import Iptrunk
+from gso.products.product_blocks import PhyPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.schemas.enums import ProductType
+from gso.services.crm import get_customer_by_name
+from gso.services.subscriptions import get_product_id_by_name
+from gso.workflows.utils import customer_selector
+from test.workflows import (
+ assert_aborted,
+ assert_complete,
+ assert_suspended,
+ extract_state,
+ resume_workflow,
+ run_workflow,
+)
+
+
+@pytest.fixture
+def input_form_wizard_data(router_subscription_factory, faker):
+ router_side_a = router_subscription_factory()
+ router_side_b = router_subscription_factory()
+
+ create_ip_trunk_step = {
+ "tt_number": faker.pystr(),
+ "customer": getattr(customer_selector(), get_customer_by_name("GÉANT")["id"]),
+ "geant_s_sid": faker.pystr(),
+ "iptrunk_type": IptrunkType.DARK_FIBER,
+ "iptrunk_description": faker.sentence(),
+ "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
+ "iptrunk_minimum_links": 5,
+ }
+ create_ip_trunk_side_a_step = {
+ "iptrunk_sideA_node_id": router_side_a,
+ "iptrunk_sideA_ae_iface": faker.pystr(),
+ "iptrunk_sideA_ae_geant_a_sid": faker.pystr(),
+ "iptrunk_sideA_ae_members": [faker.pystr() for _ in range(5)],
+ "iptrunk_sideA_ae_members_descriptions": [faker.sentence() for _ in range(5)],
+ }
+
+ create_ip_trunk_side_b_step = {
+ "iptrunk_sideB_node_id": router_side_b,
+ "iptrunk_sideB_ae_iface": faker.pystr(),
+ "iptrunk_sideB_ae_geant_a_sid": faker.pystr(),
+ "iptrunk_sideB_ae_members": [faker.pystr() for _ in range(5)],
+ "iptrunk_sideB_ae_members_descriptions": [faker.sentence() for _ in range(5)],
+ }
+
+ return [create_ip_trunk_step, create_ip_trunk_side_a_step, create_ip_trunk_side_b_step]
+
+
+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")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
+def test_successful_iptrunk_creation_with_standard_lso_result(
+ mock_allocate_v4_network,
+ mock_allocate_v6_network,
+ mock_provision_ip_trunk,
+ mock_check_ip_trunk,
+ responses,
+ input_form_wizard_data,
+ faker,
+):
+ mock_allocate_v4_network.return_value = faker.ipv4_network()
+ mock_allocate_v6_network.return_value = faker.ipv6_network()
+ product_id = get_product_id_by_name(ProductType.IP_TRUNK)
+ initial_site_data = [{"product": product_id}, *input_form_wizard_data]
+ result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
+ assert_suspended(result)
+
+ standard_lso_result = {
+ "pp_run_results": {
+ "status": "ok",
+ "job_id": "random_job_id",
+ "output": "parsed_output",
+ "return_code": 0,
+ },
+ "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 = resume_workflow(process_stat, step_log, {})
+ assert_complete(result)
+
+ state = extract_state(result)
+ subscription_id = state["subscription_id"]
+ subscription = Iptrunk.from_subscription(subscription_id)
+
+ assert "active" == subscription.status
+ assert subscription.description == f"IP trunk, geant_s_sid:{input_form_wizard_data[0]['geant_s_sid']}"
+
+ assert mock_provision_ip_trunk.call_count == 4
+ assert mock_check_ip_trunk.call_count == 2
+
+
+@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")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
+@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
+def test_iptrunk_creation_fails_when_lso_return_code_is_one(
+ mock_allocate_v4_network,
+ mock_allocate_v6_network,
+ mock_provision_ip_trunk,
+ mock_check_ip_trunk,
+ responses,
+ input_form_wizard_data,
+ faker,
+):
+ mock_allocate_v4_network.return_value = faker.ipv4_network()
+ mock_allocate_v6_network.return_value = faker.ipv6_network()
+ product_id = get_product_id_by_name(ProductType.IP_TRUNK)
+
+ initial_site_data = [{"product": product_id}, *input_form_wizard_data]
+ result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
+ assert_suspended(result)
+
+ standard_lso_result = {
+ "pp_run_results": {
+ "status": "ok",
+ "job_id": "random_job_id",
+ "output": "parsed_output",
+ "return_code": 1,
+ },
+ "confirm": "ACCEPTED",
+ }
+
+ 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 = resume_workflow(process_stat, step_log, {})
+ assert_aborted(result)
+
+ assert mock_provision_ip_trunk.call_count == attempts
+ assert mock_check_ip_trunk.call_count == 0
diff --git a/test/workflows/site/test_create_site.py b/test/workflows/site/test_create_site.py
index 826fcdacf52bbe17fad82309f01fe6700fc6af3e..da25a2b09f0e3183b3f85f69c75f4fe7c146f2ce 100644
--- a/test/workflows/site/test_create_site.py
+++ b/test/workflows/site/test_create_site.py
@@ -1,23 +1,16 @@
import pytest
-from orchestrator.db import ProductTable
from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import Site
+from gso.schemas.enums import ProductType
from gso.services.crm import get_customer_by_name
-from test.workflows import (
- assert_complete,
- assert_failed,
- extract_error,
- extract_state,
- run_workflow,
- assert_suspended,
- resume_workflow,
-)
+from gso.services.subscriptions import get_product_id_by_name
+from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.workflow
def test_create_site(responses, faker):
- product_id = ProductTable.query.filter(ProductTable.name == "Site").one().product_id
+ product_id = get_product_id_by_name(ProductType.SITE)
initial_site_data = [
{"product": product_id},
{
@@ -25,8 +18,8 @@ def test_create_site(responses, faker):
"site_city": faker.city(),
"site_country": faker.country(),
"site_country_code": faker.country_code(),
- "site_latitude": 1,
- "site_longitude": 2,
+ "site_latitude": "-74.0060",
+ "site_longitude": "40.7128",
"site_bgp_community_id": faker.pyint(),
"site_internal_id": faker.pyint(),
"site_tier": SiteTier.TIER1,