gso/utils/helpers.py

@@ -75,9 +75,9 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
 
 
 def available_interfaces_choices_including_current_members(
-    router_id: UUID,
-    speed: str,
-    interfaces: list[IptrunkInterfaceBlock],
+        router_id: UUID,
+        speed: str,
+        interfaces: list[IptrunkInterfaceBlock],
 ) -> Choice | None:
     """Return a list of available interfaces for a given router and speed including the current members.
 
@@ -223,6 +223,7 @@ def validate_site_name(site_name: str) -> str:
     return site_name
 
 
+<<<<<<< HEAD
 class BaseSiteValidatorModel(BaseModel):
     """A base site validator model extended by create site and by import site."""
 
@@ -263,3 +264,32 @@ class BaseSiteValidatorModel(BaseModel):
         """
         validate_site_name(site_name)
         return site_name
+
+
+def validate_interface_name_list(interface_name_list: list, vendor: str) -> list:
+    """Validate that the provided interface name matches the expected pattern.
+
+    The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
+    then a digit between 0 and 9, a forward slash '/', another digit between 0 and 9,
+    another forward slash '/', and ends with a digit between 0 and 9.
+    For example: 'xe-1/0/0'.
+
+    Args:
+    ----
+    interface_name_list: List of interface names to validate.
+    vendor: Router vendor to check interface names
+
+    Returns:
+    -------
+    list: The list of interface names if all match was successful, otherwise it will throw a ValueError exception.
+    """
+    # For Nokia nothing to do
+    if vendor == RouterVendor.NOKIA:
+        return interface_name_list
+    pattern = re.compile(r"^(ge|et|xe)-[0-9]/[0-9]/[0-9]$")
+    for interface in interface_name_list:
+        if not bool(pattern.match(interface.interface_name)):
+            raise ValueError(
+                f"Invalid interface name. The interface name should be of format: "
+                f"xe-1/0/0. Got: [{interface.interface_name}]")
+    return interface_name_list

gso/workflows/iptrunk/create_iptrunk.py

@@ -29,6 +29,7 @@ from gso.utils.helpers import (
     available_interfaces_choices,
     available_lags_choices,
     get_router_vendor,
+    validate_interface_name_list,
     validate_iptrunk_unique_interface,
     validate_router_in_netbox,
 )
@@ -103,6 +104,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
             return validate_iptrunk_unique_interface(side_a_ae_members)
 
+        @validator("side_a_ae_members", allow_reuse=True)
+        def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
+            vendor = get_router_vendor(router_a)
+            return validate_interface_name_list(side_a_ae_members, vendor)
+
     user_input_side_a = yield CreateIptrunkSideAForm
     # Remove the selected router for side A, to prevent any loops
     routers.pop(str(router_a))
@@ -150,6 +156,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
             return validate_iptrunk_unique_interface(side_b_ae_members)
 
+        @validator("side_b_ae_members", allow_reuse=True)
+        def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
+            vendor = get_router_vendor(router_b)
+            return validate_interface_name_list(side_b_ae_members, vendor)
+
     user_input_side_b = yield CreateIptrunkSideBForm
 
     return (

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -28,6 +28,7 @@ from gso.utils.helpers import (
     available_interfaces_choices,
     available_interfaces_choices_including_current_members,
     get_router_vendor,
+    validate_interface_name_list,
     validate_iptrunk_unique_interface,
 )
 
@@ -108,6 +109,11 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
             return validate_iptrunk_unique_interface(side_a_ae_members)
 
+        @validator("side_a_ae_members", allow_reuse=True)
+        def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
+            vendor = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.vendor
+            return validate_interface_name_list(side_a_ae_members, vendor)
+
     user_input_side_a = yield ModifyIptrunkSideAForm
     ae_members_side_b = initialize_ae_members(subscription, initial_user_input.dict(), 1)
 
@@ -128,6 +134,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
             return validate_iptrunk_unique_interface(side_b_ae_members)
 
+        @validator("side_b_ae_members", allow_reuse=True)
+        def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
+            vendor = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.vendor
+            return validate_interface_name_list(side_b_ae_members, vendor)
     user_input_side_b = yield ModifyIptrunkSideBForm
 
     return initial_user_input.dict() | user_input_side_a.dict() | user_input_side_b.dict()

test/conftest.py

@@ -17,12 +17,14 @@ from oauth2_lib.settings import oauth2lib_settings
 from orchestrator import app_settings
 from orchestrator.db import Database, db
 from orchestrator.db.database import ENGINE_ARGUMENTS, SESSION_ARGUMENTS, BaseModel
+from orchestrator.types import strEnum
 from sqlalchemy import create_engine, text
 from sqlalchemy.engine import make_url
 from sqlalchemy.orm import scoped_session, sessionmaker
 from starlette.testclient import TestClient
 
 from gso.main import init_gso_app
+from gso.utils.helpers import LAGMember
 
 logging.getLogger("faker.factory").setLevel(logging.WARNING)
 
@@ -33,6 +35,14 @@ def pytest_collection_modifyitems(config, items):
             item.add_marker(pytest.mark.skip(reason="Skipped due to SKIP_ALL_TESTS env variable"))
 
 
+class UseJuniperSide(strEnum):
+    """Define on tests on which side to use Juniper router"""
+    NONE = "none"
+    SIDE_A = "side_a"
+    SIDE_B = "side_b"
+    SIDE_BOTH = "side_both"
+
+
 class FakerProvider(BaseProvider):
     def ipv4_network(self):
         ipv4 = self.generator.ipv4()
@@ -72,6 +82,14 @@ class FakerProvider(BaseProvider):
     def network_interface(self) -> str:
         return self.generator.numerify("ge-@#/@#/@#")
 
+    def generate_junniper_members_list(self) -> list[LAGMember]:
+        iface_amount = self.generator.random_int(min=1, max=5)
+        interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]
+        return [
+            LAGMember(interface_name=interface_name, interface_description=self.generator.sentence())
+            for interface_name in interface_names]
+
+
 
 @pytest.fixture(scope="session")
 def faker() -> Faker:

test/fixtures.py

@@ -124,6 +124,57 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
     return subscription_create
 
 
+@pytest.fixture()
+def juniper_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_role=RouterRole.PE,
+        router_site=None,
+        status: SubscriptionLifecycle | None = None,
+    ) -> UUIDstr:
+        description = description or faker.text(max_nb_chars=30)
+        router_fqdn = router_fqdn or faker.domain_name(levels=4)
+        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_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_role = router_role
+        router_subscription.router.router_site = Site.from_subscription(router_site).site
+        router_subscription.router.vendor = RouterVendor.JUNIPER
+
+        router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
+        router_subscription.description = description
+        router_subscription.start_date = start_date
+
+        if status:
+            router_subscription.status = status
+
+        router_subscription.save()
+        db.session.commit()
+
+        return str(router_subscription.subscription_id)
+
+    return subscription_create
+
+
 @pytest.fixture()
 def iptrunk_side_subscription_factory(nokia_router_subscription_factory, faker):
     def subscription_create(

test/workflows/conftest.py

@@ -4,6 +4,7 @@ from urllib3_mock import Responses
 from test.fixtures import (  # noqa: F401
     iptrunk_side_subscription_factory,
     iptrunk_subscription_factory,
+    juniper_router_subscription_factory,
     nokia_router_subscription_factory,
     site_subscription_factory,
 )

test/workflows/iptrunk/test_create_iptrunk.py

@@ -5,6 +5,7 @@ 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
@@ -42,9 +43,22 @@ def _netbox_client_mock():
 
 
 @pytest.fixture()
-def input_form_wizard_data(nokia_router_subscription_factory, faker):
+def input_form_wizard_data(request, juniper_router_subscription_factory, nokia_router_subscription_factory, faker):
+    vendor = getattr(request, "param", RouterVendor.NOKIA)
     router_side_a = nokia_router_subscription_factory()
-    router_side_b = nokia_router_subscription_factory()
+    side_b_members = None
+
+    # Set side b router to Juniper
+    if vendor == RouterVendor.JUNIPER:
+        router_side_b = juniper_router_subscription_factory()
+        side_b_members = faker.generate_junniper_members_list()
+    else:
+        router_side_b = nokia_router_subscription_factory()
+        side_b_members = [
+            LAGMember(
+                interface_name=f"Interface{interface}",
+                interface_description=faker.sentence())
+            for interface in range(5)]
 
     create_ip_trunk_step = {
         "tt_number": faker.tt_number(),
@@ -71,13 +85,7 @@ def input_form_wizard_data(nokia_router_subscription_factory, faker):
     create_ip_trunk_side_b_step = {
         "side_b_ae_iface": "LAG4",
         "side_b_ae_geant_a_sid": faker.geant_sid(),
-        "side_b_ae_members": [
-            LAGMember(
-                interface_name=f"Interface{interface}",
-                interface_description=faker.sentence(),
-            )
-            for interface in range(5)
-        ],
+        "side_b_ae_members": side_b_members,
     }
 
     return [
@@ -157,3 +165,33 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
 
     assert mock_check_ip_trunk.call_count == 0
     assert mock_provision_ip_trunk.call_count == 2
+
+
+@pytest.mark.parametrize("input_form_wizard_data", [RouterVendor.JUNIPER], indirect=True)
+@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_juniper_interface_names(
+    mock_allocate_v4_network,
+    mock_allocate_v6_network,
+    mock_provision_ip_trunk,
+    mock_check_ip_trunk,
+    responses,
+    input_form_wizard_data,
+    faker,
+    data_config_filename: PathLike,
+    _netbox_client_mock,  # noqa: PT019
+    test_client,
+):
+    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)
+
+    for _ in range(6):
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    assert_complete(result)

test/workflows/iptrunk/test_modify_trunk_interface.py

@@ -13,36 +13,11 @@ from test.workflows import (
 from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
 
 
-@pytest.mark.workflow()
-@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk")
-@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
-@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag")
-@patch("gso.services.netbox_client.NetboxClient.reserve_interface")
-@patch("gso.services.netbox_client.NetboxClient.allocate_interface")
-@patch("gso.services.netbox_client.NetboxClient.free_interface")
-@patch("gso.services.netbox_client.NetboxClient.detach_interfaces_from_lag")
-def test_iptrunk_modify_trunk_interface_success(
-    mocked_detach_interfaces_from_lag,
-    mocked_free_interface,
-    mocked_allocate_interface,
-    mocked_reserve_interface,
-    mocked_attach_interface_to_lag,
-    mocked_get_available_interfaces,
-    mock_provision_ip_trunk,
-    iptrunk_subscription_factory,
-    faker,
-    data_config_filename,
-):
-    #  Set up mock return values
-    mocked_netbox = MockedNetboxClient()
-    mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
-    mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
-    mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
-    mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
-    mocked_free_interface.return_value = mocked_netbox.free_interface()
-    mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
+@pytest.fixture()
+def input_form_iptrunk_data(faker, iptrunk_subscription_factory):
 
     product_id = iptrunk_subscription_factory()
+
     new_sid = faker.geant_sid()
     new_description = faker.sentence()
     new_type = IptrunkType.LEASED
@@ -59,8 +34,7 @@ def test_iptrunk_modify_trunk_interface_success(
         {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5)
     ]
 
-    #  Run workflow
-    initial_iptrunk_data = [
+    return [
         {"subscription_id": product_id},
         {
             "tt_number": faker.tt_number(),
@@ -80,7 +54,38 @@ def test_iptrunk_modify_trunk_interface_success(
         },
     ]
 
-    result, process_stat, step_log = run_workflow("modify_trunk_interface", initial_iptrunk_data)
+
+@pytest.mark.workflow()
+@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk")
+@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces")
+@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag")
+@patch("gso.services.netbox_client.NetboxClient.reserve_interface")
+@patch("gso.services.netbox_client.NetboxClient.allocate_interface")
+@patch("gso.services.netbox_client.NetboxClient.free_interface")
+@patch("gso.services.netbox_client.NetboxClient.detach_interfaces_from_lag")
+def test_iptrunk_modify_trunk_interface_success(
+    mocked_detach_interfaces_from_lag,
+    mocked_free_interface,
+    mocked_allocate_interface,
+    mocked_reserve_interface,
+    mocked_attach_interface_to_lag,
+    mocked_get_available_interfaces,
+    mock_provision_ip_trunk,
+    input_form_iptrunk_data,
+    faker,
+    data_config_filename,
+):
+    #  Set up mock return values
+    mocked_netbox = MockedNetboxClient()
+    mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces()
+    mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag()
+    mocked_reserve_interface.return_value = mocked_netbox.reserve_interface()
+    mocked_allocate_interface.return_value = mocked_netbox.allocate_interface()
+    mocked_free_interface.return_value = mocked_netbox.free_interface()
+    mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag()
+
+    #  Run workflow
+    result, process_stat, step_log = run_workflow("modify_trunk_interface", input_form_iptrunk_data)
 
     for _ in range(2):
         result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
@@ -100,12 +105,17 @@ def test_iptrunk_modify_trunk_interface_success(
     assert mocked_detach_interfaces_from_lag.call_count == 2  # 1 time per side
 
     # Assert all subscription properties have been updated correctly
+    new_sid = input_form_iptrunk_data[1]["geant_s_sid"]
+    new_side_a_sid = input_form_iptrunk_data[2]["side_a_ae_geant_a_sid"]
+    new_side_a_ae_members = input_form_iptrunk_data[2]["side_a_ae_members"]
+    new_side_b_sid = input_form_iptrunk_data[3]["side_b_ae_geant_a_sid"]
+    new_side_b_ae_members = input_form_iptrunk_data[3]["side_b_ae_members"]
     assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}"
-    assert subscription.iptrunk.geant_s_sid == new_sid
-    assert subscription.iptrunk.iptrunk_description == new_description
-    assert subscription.iptrunk.iptrunk_type == new_type
-    assert subscription.iptrunk.iptrunk_speed == new_speed
-    assert subscription.iptrunk.iptrunk_minimum_links == new_link_count
+    assert subscription.iptrunk.geant_s_sid == input_form_iptrunk_data[1]["geant_s_sid"]
+    assert subscription.iptrunk.iptrunk_description == input_form_iptrunk_data[1]["iptrunk_description"]
+    assert subscription.iptrunk.iptrunk_type == input_form_iptrunk_data[1]["iptrunk_type"]
+    assert subscription.iptrunk.iptrunk_speed == input_form_iptrunk_data[1]["iptrunk_speed"]
+    assert subscription.iptrunk.iptrunk_minimum_links == input_form_iptrunk_data[1]["iptrunk_minimum_links"]
     assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid == new_side_a_sid
 
     def _find_interface_by_name(interfaces: list[dict[str, str]], name: str):
@@ -126,5 +136,5 @@ def test_iptrunk_modify_trunk_interface_success(
     for member in subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members:
         assert (
             member.interface_description
-            == _find_interface_by_name(new_side_b_ae_members, member.interface_name)["interface_description"]
+            == _find_interface_by_name(new_side_b_ae_members , member.interface_name)["interface_description"]
         )