From 4e633b51d0e0ccd021f1c3eafe487165aa3e057d Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Thu, 28 Nov 2024 13:59:14 +0100
Subject: [PATCH] Add validation to create VRF WF to check if the vrf name is
unique in the entire DB
---
gso/services/subscriptions.py | 43 ++++++++++++++++++++++++++-------
gso/workflows/vrf/create_vrf.py | 10 +++++++-
2 files changed, 43 insertions(+), 10 deletions(-)
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index b2c2561d..b3b8a66d 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -289,6 +289,26 @@ def get_all_active_sites() -> list[dict[str, Any]]:
]
+def is_resource_type_value_unique(resource_type: str, value: str) -> bool:
+ """Check if the given value for the specified resource type is unique in the database.
+
+ This function verifies if the specified value for the given resource type is not already in the core database.
+
+ :param resource_type: The resource type to check.
+ :type resource_type: str
+ :param value: The value to check.
+ :type value: str
+ :return: True if the value is unique (not found), False if it exists.
+ :rtype: bool
+ """
+ exists = (
+ ResourceTypeTable.query.join(SubscriptionInstanceValueTable)
+ .filter(ResourceTypeTable.resource_type == resource_type, SubscriptionInstanceValueTable.value == value)
+ .scalar()
+ )
+ return exists is None
+
+
def is_virtual_circuit_id_available(virtual_circuit_id: str) -> bool:
"""Check if the given virtual circuit ID is unique in the database.
@@ -300,12 +320,17 @@ def is_virtual_circuit_id_available(virtual_circuit_id: str) -> bool:
:return: True if the virtual circuit ID is unique (not found), False if it exists.
:rtype: bool
"""
- exists = (
- ResourceTypeTable.query.join(SubscriptionInstanceValueTable)
- .filter(
- ResourceTypeTable.resource_type == "virtual_circuit_id",
- SubscriptionInstanceValueTable.value == virtual_circuit_id,
- )
- .scalar()
- )
- return exists is None
+ return is_resource_type_value_unique("virtual_circuit_id", virtual_circuit_id)
+
+
+def is_vrf_name_unique(vrf_name: str) -> bool:
+ """Check if the given VRF name is unique in the database.
+
+ This function verifies if the specified VRF name is not already present in the core database.
+
+ :param vrf_name: The VRF name to check.
+ :type vrf_name: str
+ :return: True if the VRF name is unique (not found), False if it exists.
+ :rtype: bool
+ """
+ return is_resource_type_value_unique("vrf_name", vrf_name)
diff --git a/gso/workflows/vrf/create_vrf.py b/gso/workflows/vrf/create_vrf.py
index 89b39487..a555457e 100644
--- a/gso/workflows/vrf/create_vrf.py
+++ b/gso/workflows/vrf/create_vrf.py
@@ -6,11 +6,12 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
from orchestrator.workflow import StepList, begin, done, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
-from pydantic import ConfigDict
+from pydantic import ConfigDict, field_validator
from pydantic_forms.validators import ReadOnlyField
from gso.products.product_types.vrf import VRFInactive
from gso.services.partners import get_partner_by_name
+from gso.services.subscriptions import is_vrf_name_unique
from gso.utils.types.tt_number import TTNumber
from gso.workflows.shared import create_summary_form
@@ -28,6 +29,13 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
route_target: str
vrf_as_number: int
+ @field_validator("vrf_name")
+ def vrf_name_must_be_unique(cls, vrf_name: str) -> str:
+ if not is_vrf_name_unique(vrf_name):
+ msg = "VRF name must be unique."
+ raise ValueError(msg)
+ return vrf_name
+
user_input = yield CreateVRFForm
user_input = user_input.model_dump()
summary_fields = ["vrf_name", "route_distinguisher", "route_target", "vrf_as_number"]
--
GitLab