Finished release 0.3.

test/subscriptions/conftest.py

-from test.fixtures import nokia_router_subscription_factory, site_subscription_factory  # noqa: F401

test/workflows/iptrunk/test_create_iptrunk.py

@@ -6,7 +6,6 @@ import pytest
 from gso.products import Iptrunk, ProductType
 from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.products.product_blocks.router import RouterVendor
-from gso.services.crm import customer_selector, get_customer_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.helpers import LAGMember
 from test.services.conftest import MockedNetboxClient
@@ -60,7 +59,6 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
 
     create_ip_trunk_step = {
         "tt_number": faker.tt_number(),
-        "customer": getattr(customer_selector(), get_customer_by_name("GÉANT")["id"]),
         "geant_s_sid": faker.geant_sid(),
         "iptrunk_type": IptrunkType.DARK_FIBER,
         "iptrunk_description": faker.sentence(),
@@ -69,7 +67,7 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
     }
     create_ip_trunk_side_a_router_name = {"side_a_node_id": router_side_a}
     create_ip_trunk_side_a_step = {
-        "side_a_ae_iface": "LAG1",
+        "side_a_ae_iface": "lag-1",
         "side_a_ae_geant_a_sid": faker.geant_sid(),
         "side_a_ae_members": [
             LAGMember(
@@ -81,7 +79,7 @@ def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_r
     }
     create_ip_trunk_side_b_router_name = {"side_b_node_id": router_side_b}
     create_ip_trunk_side_b_step = {
-        "side_b_ae_iface": "LAG4",
+        "side_b_ae_iface": "lag-4",
         "side_b_ae_geant_a_sid": faker.geant_sid(),
         "side_b_ae_members": side_b_members,
     }
@@ -125,8 +123,16 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     subscription_id = state["subscription_id"]
     subscription = Iptrunk.from_subscription(subscription_id)
 
+    sorted_sides = sorted(
+        [
+            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
+            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
+        ]
+    )
     assert subscription.status == "active"
-    assert subscription.description == f"IP trunk, geant_s_sid:{input_form_wizard_data[0]['geant_s_sid']}"
+    assert subscription.description == (
+        f"IP trunk {sorted_sides[0]} {sorted_sides[1]}, geant_s_sid:{input_form_wizard_data[0]['geant_s_sid']}"
+    )
 
     assert mock_execute_playbook.call_count == 6
 

test/workflows/iptrunk/test_migrate_iptrunk.py

@@ -49,7 +49,7 @@ def migrate_form_input(
         new_router = nokia_router_subscription_factory()
         replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
         new_side_ae_members = faker.link_members_nokia()[0:2]
-        lag_name = "LAG1"
+        lag_name = "lag-1"
     elif use_juniper == UseJuniperSide.SIDE_BOTH:
         # Juniper -> Juniper
         old_side_a_node = juniper_router_subscription_factory()
@@ -69,7 +69,7 @@ def migrate_form_input(
         new_router = nokia_router_subscription_factory()
         replace_side = str(old_subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id)
         new_side_ae_members = faker.link_members_nokia()[0:2]
-        lag_name = "LAG1"
+        lag_name = "lag-1"
 
     return [
         {"subscription_id": product_id},

test/workflows/router/test_create_router.py

@@ -6,7 +6,6 @@ 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_complete,
@@ -23,7 +22,6 @@ def router_creation_input_form_data(site_subscription_factory, faker):
 
     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(),

test/workflows/router/test_update_ibgp_mesh.py

+from unittest.mock import patch
+
+import pytest
+from orchestrator.workflow import StepStatus
+from pydantic_forms.exceptions import FormValidationError
+
+from gso.products import Iptrunk
+from gso.products.product_blocks.router import RouterRole
+from test.workflows import assert_pp_interaction_success, extract_state, run_workflow
+
+
+@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.workflow()
+@patch("gso.workflows.router.update_ibgp_mesh.provisioning_proxy.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,
+    data_config_filename,
+    faker,
+):
+    result, process_stat, step_log = run_workflow(
+        "update_ibgp_mesh", [ibgp_mesh_input_form_data, {"tt_number": faker.tt_number()}]
+    )
+
+    for _ in range(5):
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    state = extract_state(result)
+
+    assert mock_execute_playbook.call_count == 5
+    assert mock_librenms_add_device.call_count == 1
+    assert result.status == StepStatus.COMPLETE
+    assert state["subscription"]["router"]["router_access_via_ts"] is False
+
+
+@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."
+    with pytest.raises(FormValidationError, match=exception_message):
+        run_workflow("update_ibgp_mesh", [{"subscription_id": router_id}, {}])

test/workflows/site/test_create_site.py

@@ -25,7 +25,6 @@ def test_create_site(responses, faker):
             "site_internal_id": faker.pyint(),
             "site_tier": SiteTier.TIER1,
             "site_ts_address": faker.ipv4(),
-            "customer": get_customer_by_name("GÉANT")["id"],
         },
     ]
     result, _, _ = run_workflow("create_site", initial_site_data)