From ac56a875bf89737b179812e8414a8b7c6210c95b Mon Sep 17 00:00:00 2001
From: Mohammad Torkashvand <mohammad.torkashvand@geant.org>
Date: Wed, 1 May 2024 09:55:17 +0200
Subject: [PATCH] refactre validation and uniuqeness
---
gso/utils/helpers.py | 29 ++++++++---------------------
gso/workflows/site/modify_site.py | 31 ++++++++++++++-----------------
2 files changed, 22 insertions(+), 38 deletions(-)
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 1daa2d81..7e12d8f3 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -8,6 +8,7 @@ from uuid import UUID
import pycountry
from orchestrator.types import UUIDstr
from pydantic import BaseModel, field_validator
+from pydantic_core.core_schema import ValidationInfo
from pydantic_forms.validators import Choice
from gso import settings
@@ -225,29 +226,15 @@ class BaseSiteValidatorModel(BaseModel):
validate_country_code(country_code)
return country_code
- @field_validator("site_ts_address")
- def site_ts_address_must_be_unique(cls, site_ts_address: str) -> str:
- """Validate that the internal and :term:`BGP` community IDs are unique."""
- validate_site_fields_is_unique("site_ts_address", site_ts_address)
- return site_ts_address
-
- @field_validator("site_internal_id")
- def site_internal_id_must_be_unique(cls, site_internal_id: int) -> int:
- """Validate that the internal and :term:`BGP` community IDs are unique."""
- validate_site_fields_is_unique("site_internal_id", site_internal_id)
- return site_internal_id
+ @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name")
+ def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
+ """Validate that a field is unique."""
+ if not info.field_name:
+ raise ValueError("Field name is required")
- @field_validator("site_bgp_community_id")
- def site_bgp_community_id_must_be_unique(cls, site_bgp_community_id: int) -> int:
- """Validate that the internal and :term:`BGP` community IDs are unique."""
- validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id)
- return site_bgp_community_id
+ validate_site_fields_is_unique(info.field_name, value)
- @field_validator("site_name")
- def site_name_must_be_unique(cls, site_name: str) -> str:
- """Validate that the internal and :term:`BGP` community IDs are unique."""
- validate_site_fields_is_unique("site_name", site_name)
- return site_name
+ return value
@field_validator("site_name")
def site_name_must_be_valid(cls, site_name: str) -> str:
diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py
index 93ac6f51..ee48784e 100644
--- a/gso/workflows/site/modify_site.py
+++ b/gso/workflows/site/modify_site.py
@@ -12,6 +12,7 @@ from orchestrator.workflows.steps import (
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic import ConfigDict, field_validator
+from pydantic_core.core_schema import ValidationInfo
from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
@@ -37,30 +38,26 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
site_tier: ReadOnlyField(subscription.site.site_tier, default_type=SiteTier) # type: ignore[valid-type]
site_ts_address: str | None = subscription.site.site_ts_address
+ @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id")
+ def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
+ if not info.field_name:
+ msg = "Field name must be provided."
+ raise ValueError(msg)
+
+ if value and value == getattr(subscription.site, info.field_name):
+ return value
+
+ validate_site_fields_is_unique(info.field_name, value)
+
+ return value
+
@field_validator("site_ts_address")
def validate_ts_address(cls, site_ts_address: str) -> str:
if site_ts_address and site_ts_address != subscription.site.site_ts_address:
- validate_site_fields_is_unique("site_ts_address", site_ts_address)
validate_ipv4_or_ipv6(site_ts_address)
return site_ts_address
- @field_validator("site_internal_id")
- def validate_site_internal_id(cls, site_internal_id: int) -> int:
- if site_internal_id == subscription.site.site_internal_id:
- return site_internal_id
-
- validate_site_fields_is_unique("site_internal_id", site_internal_id)
- return site_internal_id
-
- @field_validator("site_bgp_community_id")
- def validate_site_bgp_community_id(cls, site_bgp_community_id: int) -> int:
- if site_bgp_community_id == subscription.site.site_bgp_community_id:
- return site_bgp_community_id
-
- validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id)
- return site_bgp_community_id
-
user_input = yield ModifySiteForm
return user_input.dict()
--
GitLab