Skip to content
Snippets Groups Projects

Allow iBGP update on both provisioning and active trunks

Merged Karel van Klink requested to merge feature/update-ibgp-on-provisioning-trunk into develop
All threads resolved!
1 file
+ 2
2
Compare changes
  • Side-by-side
  • Inline
from unittest.mock import patch
import pytest
from orchestrator.types import SubscriptionLifecycle
from orchestrator.workflow import StepStatus
from pydantic_forms.exceptions import FormValidationError
@@ -16,23 +17,22 @@ from test.workflows import (
)
@pytest.fixture()
def ibgp_mesh_input_form_data(iptrunk_subscription_factory, faker):
ip_trunk = Iptrunk.from_subscription(iptrunk_subscription_factory())
return {"subscription_id": ip_trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id}
@pytest.mark.parametrize("trunk_status", [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE])
@pytest.mark.workflow()
@patch("gso.workflows.router.update_ibgp_mesh.lso_client.execute_playbook")
@patch("gso.workflows.router.update_ibgp_mesh.librenms_client.LibreNMSClient.add_device")
def test_update_ibgp_mesh_success(
mock_librenms_add_device,
mock_execute_playbook,
ibgp_mesh_input_form_data,
trunk_status,
iptrunk_subscription_factory,
data_config_filename,
faker,
):
ip_trunk = Iptrunk.from_subscription(iptrunk_subscription_factory(status=trunk_status))
ibgp_mesh_input_form_data = {
"subscription_id": ip_trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id
}
result, process_stat, step_log = run_workflow(
"update_ibgp_mesh", [ibgp_mesh_input_form_data, {"tt_number": faker.tt_number()}]
)
@@ -53,10 +53,23 @@ def test_update_ibgp_mesh_success(
assert state["subscription"]["router"]["router_access_via_ts"] is False
@pytest.mark.parametrize("trunk_status", [SubscriptionLifecycle.INITIAL, SubscriptionLifecycle.TERMINATED])
@pytest.mark.workflow()
def test_update_ibgp_mesh_failure(iptrunk_subscription_factory, data_config_filename, trunk_status):
ip_trunk = Iptrunk.from_subscription(iptrunk_subscription_factory(status=trunk_status))
ibgp_mesh_input_form_data = {
"subscription_id": ip_trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id
}
exception_message = "Selected router does not terminate any available IP trunks."
with pytest.raises(FormValidationError, match=exception_message):
run_workflow("update_ibgp_mesh", [ibgp_mesh_input_form_data, {}])
@pytest.mark.workflow()
def test_update_ibgp_mesh_isolated_router(nokia_router_subscription_factory, data_config_filename):
router_id = nokia_router_subscription_factory(router_role=RouterRole.P)
exception_message = "Selected router does not terminate any active IP trunks."
exception_message = "Selected router does not terminate any available IP trunks."
with pytest.raises(FormValidationError, match=exception_message):
run_workflow("update_ibgp_mesh", [{"subscription_id": router_id}, {}])
Loading