From a87f48fec0a7bca39bbc0cf51e4f8e206a4d765e Mon Sep 17 00:00:00 2001
From: Jorge Sasiain <jorge.sasiain@ehu.eus>
Date: Thu, 23 Nov 2023 12:34:39 +0000
Subject: [PATCH] NAT-329: fix mismatches with develop
---
gso/utils/helpers.py | 22 ++++++------
gso/utils/validators.py | 36 +++++++++++++++++++
test/workflows/iptrunk/test_create_iptrunk.py | 13 ++++---
3 files changed, 54 insertions(+), 17 deletions(-)
create mode 100644 gso/utils/validators.py
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index b2eb35454..5a5e3d7d9 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 000000000..3677003a6
--- /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 d76aa5c8c..0d63cd325 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
--
GitLab