gso/api/v1/imports.py

@@ -9,20 +9,13 @@ from fastapi.routing import APIRouter
 from orchestrator.security import opa_security_default
 from orchestrator.services import processes
 from pydantic import BaseModel, root_validator, validator
-from pydantic.fields import ModelField
 
 from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_blocks.site import SiteTier
 from gso.services import subscriptions
 from gso.services.crm import CustomerNotFoundError, get_customer_by_name
-from gso.utils.helpers import (
-    LAGMember,
-    validate_country_code,
-    validate_ipv4_or_ipv6,
-    validate_site_fields_is_unique,
-    validate_site_name,
-)
+from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
 
 router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
 
@@ -34,7 +27,7 @@ class ImportResponseModel(BaseModel):
     detail: str
 
 
-class SiteImportModel(BaseModel):
+class SiteImportModel(BaseSiteValidatorModel):
     """The required input for importing an existing :class:`gso.products.product_types.site`."""
 
     site_name: str
@@ -49,34 +42,6 @@ class SiteImportModel(BaseModel):
     site_ts_address: str
     customer: str
 
-    @validator("site_ts_address", allow_reuse=True)
-    def validate_ts_address(cls, site_ts_address: str) -> str:
-        """Validate the terminal server address."""
-        validate_site_fields_is_unique("site_ts_address", site_ts_address)
-        validate_ipv4_or_ipv6(site_ts_address)
-        return site_ts_address
-
-    @validator("site_country_code", allow_reuse=True)
-    def country_code_must_exist(cls, country_code: str) -> str:
-        """Validate the country code such that it exists."""
-        validate_country_code(country_code)
-        return country_code
-
-    @validator("site_internal_id", "site_bgp_community_id", allow_reuse=True)
-    def validate_unique_fields(cls, value: str, field: ModelField) -> str | int:
-        """Validate that the internal side ID and :term:`BGP` community IDs are unique."""
-        return validate_site_fields_is_unique(field.name, value)
-
-    @validator("site_name", allow_reuse=True)
-    def site_name_must_be_valid(cls, site_name: str) -> str:
-        """Validate the site name.
-
-        The site name must consist of three uppercase letters, optionally followed by a single digit.
-        """
-        validate_site_fields_is_unique("site_name", site_name)
-        validate_site_name(site_name)
-        return site_name
-
 
 class RouterImportModel(BaseModel):
     """Required fields for importing an existing :class:`gso.product.product_types.router`."""

gso/services/netbox_client.py

@@ -162,7 +162,7 @@ class NetboxClient:
     def create_device(self, device_name: str, site_tier: str) -> Devices:
         """Create a new device in Netbox."""
         # Get device type id
-        tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
+        tier_info = TierInfo().get_module_by_name(f"tier{site_tier}")
         device_type = self.netbox.dcim.device_types.get(model=tier_info.device_type)
 
         # Get device role id

gso/utils/device_info.py

@@ -18,14 +18,14 @@ class TierInfo:
 
     def __init__(self) -> None:
         """Initialise the different tiers of sites that exist."""
-        self.Tier1 = ModuleInfo(
+        self.tier1 = ModuleInfo(
             device_type="7750 SR-7s",
             module_bays_slots=[1, 2],
             module_type="XMA2-s-36p-400g",
             breakout_interfaces_per_slot=[36, 35, 34, 33],
             total_10g_interfaces=80,
         )
-        self.Tier2 = ModuleInfo(
+        self.tier2 = ModuleInfo(
             device_type="7750-SR7s",
             module_bays_slots=[1, 2],
             module_type="XMA2-s-36p-400g",

gso/utils/helpers.py

@@ -8,11 +8,13 @@ from uuid import UUID
 import pycountry
 from orchestrator import step
 from orchestrator.types import State, UUIDstr
-from pydantic import BaseModel
+from pydantic import BaseModel, validator
+from pydantic.fields import ModelField
 from pydantic_forms.validators import Choice
 
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
 from gso.products.product_blocks.router import RouterVendor
+from gso.products.product_blocks.site import SiteTier
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
 from gso.services import provisioning_proxy
@@ -219,3 +221,45 @@ def validate_site_name(site_name: str) -> str:
         )
         raise ValueError(msg)
     return site_name
+
+
+class BaseSiteValidatorModel(BaseModel):
+    """A base site validator model extended by create site and by import site."""
+
+    site_bgp_community_id: int
+    site_internal_id: int
+    site_tier: SiteTier
+    site_ts_address: str
+
+    @validator("site_ts_address", check_fields=False, allow_reuse=True)
+    def validate_ts_address(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
+
+    @validator("site_country_code", check_fields=False, allow_reuse=True)
+    def country_code_must_exist(cls, country_code: str) -> str:
+        """Validate that the country code exists."""
+        validate_country_code(country_code)
+        return country_code
+
+    @validator(
+        "site_ts_address",
+        "site_internal_id",
+        "site_bgp_community_id",
+        "site_name",
+        check_fields=False,
+        allow_reuse=True,
+    )
+    def validate_unique_fields(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)
+
+    @validator("site_name", check_fields=False, allow_reuse=True)
+    def site_name_must_be_valid(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

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -41,6 +41,8 @@ from gso.utils.helpers import (
 
 logger = getLogger(__name__)
 
+PLAYBOOK_VERB_NOT_YET_PROPERLY_SET = "Playbook verb is not yet properly set."
+
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     """Gather input from the operator on the new router that the IP trunk should connect to."""
@@ -251,7 +253,7 @@ def deploy_new_config_dry(
         "trunk_interface",
     )
 
-    logger.warning("Playbook verb is not yet properly set.")
+    logger.warning(PLAYBOOK_VERB_NOT_YET_PROPERLY_SET)
 
     return {
         "subscription": subscription,
@@ -287,7 +289,7 @@ def deploy_new_config_real(
         dry_run=False,
     )
 
-    logger.warning("Playbook verb is not yet properly set.")
+    logger.warning(PLAYBOOK_VERB_NOT_YET_PROPERLY_SET)
 
     return {
         "subscription": subscription,
@@ -338,7 +340,7 @@ def deploy_new_isis(
         dry_run=False,
     )
 
-    logger.warning("Playbook verb is not yet properly set.")
+    logger.warning(PLAYBOOK_VERB_NOT_YET_PROPERLY_SET)
 
     return {
         "subscription": subscription,
@@ -410,7 +412,7 @@ def delete_old_config_dry(
         "delete",
     )
 
-    logger.warning("Playbook verb is not yet properly set.")
+    logger.warning(PLAYBOOK_VERB_NOT_YET_PROPERLY_SET)
 
     return {"subscription": subscription}
 
@@ -444,7 +446,7 @@ def delete_old_config_real(
         dry_run=False,
     )
 
-    logger.warning("Playbook verb is not yet properly set.")
+    logger.warning(PLAYBOOK_VERB_NOT_YET_PROPERLY_SET)
 
     return {"subscription": subscription}
 

gso/workflows/site/create_site.py

@@ -6,25 +6,18 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
 from orchestrator.workflow import StepList, done, init, 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 validator
-from pydantic.fields import ModelField
 
 from gso.products.product_blocks import site as site_pb
 from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
 from gso.products.product_types import site
 from gso.services.crm import customer_selector
-from gso.utils.helpers import (
-    validate_country_code,
-    validate_ipv4_or_ipv6,
-    validate_site_fields_is_unique,
-    validate_site_name,
-)
+from gso.utils.helpers import BaseSiteValidatorModel
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
     """Get input from the operator about the new site subscription."""
 
-    class CreateSiteForm(FormPage):
+    class CreateSiteForm(FormPage, BaseSiteValidatorModel):
         class Config:
             title = product_name
 
@@ -40,34 +33,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         site_tier: site_pb.SiteTier
         site_ts_address: str
 
-        @validator("site_ts_address", allow_reuse=True)
-        def validate_ts_address(cls, site_ts_address: str) -> str:
-            """Validate that a terminal server address is valid."""
-            validate_site_fields_is_unique("site_ts_address", site_ts_address)
-            validate_ipv4_or_ipv6(site_ts_address)
-            return site_ts_address
-
-        @validator("site_country_code", allow_reuse=True)
-        def country_code_must_exist(cls, country_code: str) -> str:
-            """Validate that the country code exists."""
-            validate_country_code(country_code)
-            return country_code
-
-        @validator("site_internal_id", "site_bgp_community_id", allow_reuse=True)
-        def validate_unique_fields(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)
-
-        @validator("site_name", allow_reuse=True)
-        def site_name_must_be_valid(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_fields_is_unique("site_name", site_name)
-            validate_site_name(site_name)
-            return site_name
-
     user_input = yield CreateSiteForm
 
     return user_input.dict()