Skip to content
Snippets Groups Projects
Commit ac56a875 authored by Mohammad Torkashvand's avatar Mohammad Torkashvand
Browse files

refactre validation and uniuqeness

parent 341701a9
No related branches found
No related tags found
1 merge request!188upgrade to orchestrato-core v2
Pipeline #86668 failed
...@@ -8,6 +8,7 @@ from uuid import UUID ...@@ -8,6 +8,7 @@ from uuid import UUID
import pycountry import pycountry
from orchestrator.types import UUIDstr from orchestrator.types import UUIDstr
from pydantic import BaseModel, field_validator from pydantic import BaseModel, field_validator
from pydantic_core.core_schema import ValidationInfo
from pydantic_forms.validators import Choice from pydantic_forms.validators import Choice
from gso import settings from gso import settings
...@@ -225,29 +226,15 @@ class BaseSiteValidatorModel(BaseModel): ...@@ -225,29 +226,15 @@ class BaseSiteValidatorModel(BaseModel):
validate_country_code(country_code) validate_country_code(country_code)
return country_code return country_code
@field_validator("site_ts_address") @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name")
def site_ts_address_must_be_unique(cls, site_ts_address: str) -> str: def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
"""Validate that the internal and :term:`BGP` community IDs are unique.""" """Validate that a field is unique."""
validate_site_fields_is_unique("site_ts_address", site_ts_address) if not info.field_name:
return site_ts_address raise ValueError("Field name is required")
@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_bgp_community_id") validate_site_fields_is_unique(info.field_name, value)
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
@field_validator("site_name") return value
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
@field_validator("site_name") @field_validator("site_name")
def site_name_must_be_valid(cls, site_name: str) -> str: def site_name_must_be_valid(cls, site_name: str) -> str:
......
...@@ -12,6 +12,7 @@ from orchestrator.workflows.steps import ( ...@@ -12,6 +12,7 @@ from orchestrator.workflows.steps import (
) )
from orchestrator.workflows.utils import wrap_modify_initial_input_form from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic import ConfigDict, field_validator from pydantic import ConfigDict, field_validator
from pydantic_core.core_schema import ValidationInfo
from pydantic_forms.validators import ReadOnlyField from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
...@@ -37,30 +38,26 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: ...@@ -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_tier: ReadOnlyField(subscription.site.site_tier, default_type=SiteTier) # type: ignore[valid-type]
site_ts_address: str | None = subscription.site.site_ts_address 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") @field_validator("site_ts_address")
def validate_ts_address(cls, site_ts_address: str) -> str: def validate_ts_address(cls, site_ts_address: str) -> str:
if site_ts_address and site_ts_address != subscription.site.site_ts_address: 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) validate_ipv4_or_ipv6(site_ts_address)
return 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 user_input = yield ModifySiteForm
return user_input.dict() return user_input.dict()
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment