diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index b2eb35454c9eac5c65bdfdbd02201000f7c4a309..5a5e3d7d90f516caad9e1a6a32d681d330246e14 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -62,7 +62,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
For Nokia routers, return a list of available interfaces.
For Juniper routers, return a string.
"""
- if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
+ if get_router_vendor(router_id) != RouterVendor.NOKIA:
return None
interfaces = {
interface["name"]: f"{interface['name']} - {interface['module']['display']} - {interface['description']}"
@@ -71,8 +71,9 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
return Choice("ae member", zip(interfaces.keys(), interfaces.items(), strict=True)) # type: ignore[arg-type]
+
def available_interfaces_choices_including_current_members(
- router_id: UUID | UUIDstr,
+ router_id: UUID,
speed: str,
interfaces: list[IptrunkInterfaceBlock],
) -> Choice | None:
@@ -81,7 +82,7 @@ def available_interfaces_choices_including_current_members(
For Nokia routers, return a list of available interfaces.
For Juniper routers, return a string.
"""
- if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
+ if get_router_vendor(router_id) != RouterVendor.NOKIA:
return None
available_interfaces = list(NetboxClient().get_available_interfaces(router_id, speed))
@@ -101,13 +102,14 @@ def available_interfaces_choices_including_current_members(
return Choice("ae member", zip(options.keys(), options.items(), strict=True)) # type: ignore[arg-type]
+
def available_lags_choices(router_id: UUID) -> Choice | None:
"""Return a list of available lags for a given router.
For Nokia routers, return a list of available lags.
- For Juniper routers, return a string.
+ For Juniper routers, return ``None``.
"""
- if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA:
+ if get_router_vendor(router_id) != RouterVendor.NOKIA:
return None
side_a_ae_iface_list = NetboxClient().get_available_lags(router_id)
return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)) # type: ignore[arg-type]
@@ -120,9 +122,9 @@ def get_router_vendor(router_id: UUID) -> str:
:type router_id: :class:`uuid.UUID`
:return: The vendor of the router.
- :rtype: str:
+ :rtype: RouterVendor:
"""
- return Router.from_subscription(router_id).router.router_vendor
+ return Router.from_subscription(router_id).vendor
def iso_from_ipv4(ipv4_address: IPv4Address) -> str:
@@ -148,9 +150,9 @@ def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
:return: The :term:`UUID` of the router subscription.
:rtype: :class:`UUIDstr`
"""
- router = Router.from_subscription(subscription_id).router
- if router.router_vendor == RouterVendor.NOKIA:
- device = NetboxClient().get_device_by_name(router.router_fqdn)
+ router_type = Router.from_subscription(subscription_id)
+ if router_type.vendor == RouterVendor.NOKIA:
+ device = NetboxClient().get_device_by_name(router_type.router.router_fqdn)
if not device:
msg = "The selected router does not exist in Netbox."
raise ValueError(msg)
diff --git a/gso/utils/validators.py b/gso/utils/validators.py
new file mode 100644
index 0000000000000000000000000000000000000000..3677003a62a07bf55272104c49658d9ba44c83e8
--- /dev/null
+++ b/gso/utils/validators.py
@@ -0,0 +1,36 @@
+from pydantic import BaseModel, validator, ModelField
+
+from gso.utils.helpers import (
+ validate_country_code,
+ validate_ipv4_or_ipv6,
+ validate_site_fields_is_unique,
+ validate_site_name,
+)
+
+def common_ts_address_validator(cls, site_ts_address: str) -> str:
+ """Validate that a terminal server address is valid."""
+ validate_ipv4_or_ipv6(site_ts_address)
+ return site_ts_address
+
+def common_country_code_validator(cls, country_code: str) -> str:
+ """Validate that the country code exists."""
+ validate_country_code(country_code)
+ return country_code
+
+def common_unique_fields_validator(cls, value: str, field: ModelField) -> str | int:
+ """Validate that the internal and :term:`BGP` community IDs are unique."""
+ return validate_site_fields_is_unique(field.name, value)
+
+def common_site_name_validator(cls, site_name: str) -> str:
+ """Validate the site name.
+
+ The site name must consist of three uppercase letters, followed by an optional single digit.
+ """
+ validate_site_name(site_name)
+ return site_name
+
+def validator_decorator(func):
+ def wrapper(cls, *args, **kwargs):
+ return validator(func.__name__, pre=True, allow_reuse=True)(*args, **kwargs)
+
+ return wrapper
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index d76aa5c8c9aae65e54af988d207ceab269034fd5..0d63cd325f789ea44bedea86ceb6c52d66cb81ce 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -43,22 +43,21 @@ def _netbox_client_mock():
@pytest.fixture()
-def input_form_wizard_data(request, router_subscription_factory, faker):
+def input_form_wizard_data(request, nokia_router_subscription_factory, faker):
# only side b is set to JUNIPER
vendor = getattr(request, "param", RouterVendor.NOKIA)
- router_side_a = router_subscription_factory()
- router_side_b = router_subscription_factory(router_vendor=vendor)
+ router_side_a = nokia_router_subscription_factory()
+ router_side_b = nokia_router_subscription_factory()
side_a_members = [
LAGMember(interface_name=f"Interface{interface}", interface_description=faker.sentence())
- for interface in range(5)]
+ for interface in range(2)]
side_b_members = None
lag_name_side_b = "LAG4"
if vendor == RouterVendor.JUNIPER:
side_b_members = [
- {"interface_name": "et-1/0/0", "interface_description": faker.sentence()},
- {"interface_name": "xe-1/0/0", "interface_description": faker.sentence()}] # type: ignore[assignment]
- lag_name_side_b = "ae4"
+ LAGMember(interface_name="et-1/0/0", interface_description=faker.sentence()),
+ LAGMember(interface_name="xe-1/0/0", interface_description=faker.sentence())] # type: ignore[assignment]
else:
side_b_members = side_a_members