diff --git a/gso/api/v1/network.py b/gso/api/v1/network.py
index b92a135e63ff1a4fcee89e987b3c568fccc28d7f..62983f77f0b2287cb366ba2cb062babc2e972d05 100644
--- a/gso/api/v1/network.py
+++ b/gso/api/v1/network.py
@@ -10,10 +10,11 @@ from orchestrator.security import authorize
from orchestrator.services.subscriptions import build_extended_domain_model
from starlette import status
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
from gso.services.subscriptions import get_active_iptrunk_subscriptions
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.types.interfaces import PhysicalPortCapacity
from gso.utils.shared_enums import Vendor
router = APIRouter(prefix="/networks", tags=["Network"], dependencies=[Depends(authorize)])
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index 406fa85d1dbc0e11d58187909a63dc5f6285285c..eddb8dd9c6949aa5828e602f49e8db34bf0d224d 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -18,7 +18,7 @@ from sqlalchemy.exc import SQLAlchemyError
from gso.db.models import PartnerTable
from gso.products import ProductType
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.services.partners import (
PartnerNotFoundError,
@@ -31,7 +31,8 @@ from gso.services.subscriptions import (
get_active_subscriptions_by_field_and_value,
get_subscriptions,
)
-from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
+from gso.types.base_site import BaseSiteValidatorModel
+from gso.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
app: typer.Typer = typer.Typer()
@@ -115,11 +116,11 @@ class IptrunkImportModel(BaseModel):
side_a_node_id: str
side_a_ae_iface: str
side_a_ae_geant_a_sid: str | None
- side_a_ae_members: list[LAGMember]
+ side_a_ae_members: LAGMemberList[LAGMember]
side_b_node_id: str
side_b_ae_iface: str
side_b_ae_geant_a_sid: str | None
- side_b_ae_members: list[LAGMember]
+ side_b_ae_members: LAGMemberList[LAGMember]
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
@@ -150,15 +151,6 @@ class IptrunkImportModel(BaseModel):
return value
- @field_validator("side_a_ae_members", "side_b_ae_members")
- def check_side_uniqueness(cls, value: list[str]) -> list[str]:
- """:term:`LAG` members must be unique."""
- if len(value) != len(set(value)):
- msg = "Items must be unique"
- raise ValueError(msg)
-
- return value
-
@model_validator(mode="after")
def check_members(self) -> Self:
"""Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
diff --git a/gso/products/product_blocks/iptrunk.py b/gso/products/product_blocks/iptrunk.py
index 88ff75cb2afad1960c4bc4f03953903c23040745..1ff4fe4652d6fea6f1a80491976ff3063978f82e 100644
--- a/gso/products/product_blocks/iptrunk.py
+++ b/gso/products/product_blocks/iptrunk.py
@@ -15,18 +15,7 @@ from gso.products.product_blocks.router import (
RouterBlockInactive,
RouterBlockProvisioning,
)
-
-
-class PhysicalPortCapacity(strEnum):
- """Physical port capacity enumerator.
-
- An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
- """
-
- ONE_GIGABIT_PER_SECOND = "1G"
- TEN_GIGABIT_PER_SECOND = "10G"
- HUNDRED_GIGABIT_PER_SECOND = "100G"
- FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
+from gso.types.interfaces import LAGMemberList, PhysicalPortCapacity
class IptrunkType(strEnum):
@@ -36,9 +25,6 @@ class IptrunkType(strEnum):
LEASED = "Leased"
-LAGMemberList = Annotated[
- list[T], AfterValidator(validate_unique_list), Len(min_length=0), Doc("A list of :term:`LAG` member interfaces.")
-]
IptrunkSides = Annotated[
list[T],
AfterValidator(validate_unique_list),
diff --git a/gso/products/product_blocks/lan_switch_interconnect.py b/gso/products/product_blocks/lan_switch_interconnect.py
index a9b1c77366662e24499f936f9762d515b882b934..544da1e757c279ca3c16f2e52489dba2624e68aa 100644
--- a/gso/products/product_blocks/lan_switch_interconnect.py
+++ b/gso/products/product_blocks/lan_switch_interconnect.py
@@ -3,9 +3,9 @@
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
-from gso.products.product_blocks.iptrunk import LAGMemberList
from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
+from gso.types.interfaces import LAGMemberList
class LanSwitchInterconnectAddressSpace(strEnum):
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index be7d086adc021992ede2048b6ef4a843c1793755..a0e21a584ddc6afed3377ee6d52bde2e6b665bd7 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -1,18 +1,9 @@
"""The product block that describes a site subscription."""
-import re
-from typing import Annotated
-
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
-from pydantic import AfterValidator
-from typing_extensions import Doc
-
-MAX_LONGITUDE = 180
-MIN_LONGITUDE = -180
-MAX_LATITUDE = 90
-MIN_LATITUDE = -90
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
class SiteTier(strEnum):
@@ -28,56 +19,6 @@ class SiteTier(strEnum):
TIER4 = 4
-def validate_latitude(v: str) -> str:
- """Validate a latitude coordinate."""
- msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
- regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
- if not regex.match(str(v)):
- raise ValueError(msg)
-
- float_v = float(v)
- if float_v > MAX_LATITUDE or float_v < MIN_LATITUDE:
- raise ValueError(msg)
-
- return v
-
-
-def validate_longitude(v: str) -> str:
- """Validate a longitude coordinate."""
- regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
- msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
- if not regex.match(v):
- raise ValueError(msg)
-
- float_v = float(v)
- if float_v > MAX_LONGITUDE or float_v < MIN_LONGITUDE:
- raise ValueError(msg)
-
- return v
-
-
-LatitudeCoordinate = Annotated[
- str,
- AfterValidator(validate_latitude),
- Doc(
- "A latitude coordinate, modeled as a string. "
- "The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. "
- "It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0."
- ),
-]
-
-LongitudeCoordinate = Annotated[
- str,
- AfterValidator(validate_longitude),
- Doc(
- "A longitude coordinate, modeled as a string. "
- "The coordinate must match the format conforming to the longitude "
- "range of -180 to +180 degrees. It can be a floating-point number or an integer. "
- "Valid examples: 40.7128, -74.0060, 180, -180, 0."
- ),
-]
-
-
class SiteBlockInactive(
ProductBlockModel,
lifecycle=[SubscriptionLifecycle.INITIAL],
diff --git a/gso/services/librenms_client.py b/gso/services/librenms_client.py
index 2e04a866e3f85b539f22031b3f878b2149823d0f..4d8e4b197f647aadf829648a01e056cf8b199068 100644
--- a/gso/services/librenms_client.py
+++ b/gso/services/librenms_client.py
@@ -10,7 +10,7 @@ from requests import HTTPError, Response
from requests.adapters import HTTPAdapter
from gso.settings import load_oss_params
-from gso.utils.helpers import SNMPVersion
+from gso.types.snmp import SNMPVersion
logger = logging.getLogger(__name__)
diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py
index 25ed84e0593b2963656167737f17b4697ef2b0df..eed969a729c2da5ef943ec72c79dc94b65d60674 100644
--- a/gso/services/subscriptions.py
+++ b/gso/services/subscriptions.py
@@ -69,20 +69,6 @@ def get_subscriptions(
return [dict(zip(includes, result, strict=True)) for result in results]
-def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
- """Retrieve active subscriptions specifically for sites.
-
- :param includes: The fields to be included in the returned Subscription objects.
- :type includes: list[str]
-
- :return: A list of Subscription objects for sites.
- :rtype: list[Subscription]
- """
- return get_subscriptions(
- product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
- )
-
-
def get_router_subscriptions(
includes: list[str] | None = None, lifecycles: list[SubscriptionLifecycle] | None = None
) -> list[SubscriptionType]:
@@ -122,6 +108,20 @@ def get_provisioning_router_subscriptions(includes: list[str] | None = None) ->
)
+def get_active_switch_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+ """Retrieve active subscriptions specifically for switches.
+
+ :param includes: The fields to be included in the returned Subscription objects.
+ :type includes: list[str]
+
+ :return: A list of Subscription objects for switches.
+ :rtype: list[Subscription]
+ """
+ return get_subscriptions(
+ product_types=[ProductType.SWITCH], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+ )
+
+
def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active subscriptions specifically for IP trunks.
@@ -233,6 +233,18 @@ def get_active_insync_subscriptions() -> list[SubscriptionTable]:
)
+def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+ """Retrieve active subscriptions specifically for sites.
+
+ :param includes: The fields to be included in the returned Subscription objects.
+ :type includes: list[str]
+
+ :return: A list of Subscription objects for sites.
+ :rtype: list[Subscription]
+ """
+ return get_subscriptions(product_types=["Site"], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes) # type: ignore[list-item]
+
+
def get_site_by_name(site_name: str) -> Site:
"""Get a site by its name.
diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index 108a28522a74747461352dbd5133ba5ccf63f3ec..f1ede053f813aefef202d92ffd31b6d48c3908a5 100644
--- a/gso/translations/en-GB.json
+++ b/gso/translations/en-GB.json
@@ -32,16 +32,18 @@
"migrate_to_different_site": "Migrating to a different Site",
"remove_configuration": "Remove configuration from the router",
"clean_up_ipam": "Clean up related entries in IPAM",
- "restore_isis_metric": "Restore ISIS metric to original value"
+ "restore_isis_metric": "Restore ISIS metric to original value",
+ "confirm_info": "Please verify this form looks correct."
}
},
"workflow": {
"activate_iptrunk": "Activate IP Trunk",
- "activate_router": "Activate router",
- "confirm_info": "Please verify this form looks correct.",
+ "activate_router": "Activate Router",
+ "activate_switch": "Activate Switch",
"create_iptrunk": "Create IP Trunk",
"create_router": "Create Router",
"create_site": "Create Site",
+ "create_switch": "Create Switch",
"deploy_twamp": "Deploy TWAMP",
"migrate_iptrunk": "Migrate IP Trunk",
"modify_isis_metric": "Modify the ISIS metric",
@@ -52,6 +54,7 @@
"terminate_iptrunk": "Terminate IP Trunk",
"terminate_router": "Terminate Router",
"terminate_site": "Terminate Site",
+ "terminate_switch": "Terminate Switch",
"redeploy_base_config": "Redeploy base config",
"update_ibgp_mesh": "Update iBGP mesh",
"create_imported_site": "NOT FOR HUMANS -- Import existing site",
@@ -67,7 +70,8 @@
"import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
"import_opengear": "NOT FOR HUMANS -- Finalize import into an OpenGear",
"validate_iptrunk": "Validate IP Trunk configuration",
- "validate_router": "Validate router configuration",
+ "validate_router": "Validate Router configuration",
+ "validate_switch": "Validate Switch configuration",
"task_validate_geant_products": "Validation task for GEANT products",
"task_send_email_notifications": "Send email notifications for failed tasks",
"task_create_partners": "Create partner task",
diff --git a/gso/types/__init__.py b/gso/types/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..c3d1994f0c0c918fc35c2aabd757c67488065aa9
--- /dev/null
+++ b/gso/types/__init__.py
@@ -0,0 +1 @@
+"""Define custom types for use across the application."""
diff --git a/gso/types/base_site.py b/gso/types/base_site.py
new file mode 100644
index 0000000000000000000000000000000000000000..1c29c1417aaeb2797c2f4007fd03e64cfa5e81b2
--- /dev/null
+++ b/gso/types/base_site.py
@@ -0,0 +1,59 @@
+"""A base site type for validation purposes that can be extended elsewhere."""
+
+from pydantic import BaseModel, field_validator
+from pydantic_core.core_schema import ValidationInfo
+
+from gso.products.product_blocks.site import SiteTier
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.types.country_code import validate_country_code
+from gso.types.ip_address import validate_ipv4_or_ipv6
+from gso.types.site_name import validate_site_name
+from gso.types.unique_field import validate_field_is_unique
+
+
+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
+ site_country_code: str
+ site_name: str
+ site_city: str
+ site_country: str
+ site_latitude: LatitudeCoordinate
+ site_longitude: LongitudeCoordinate
+ partner: str
+
+ @field_validator("site_ts_address")
+ 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
+
+ @field_validator("site_country_code")
+ def country_code_must_exist(cls, country_code: str) -> str:
+ """Validate that the country code exists."""
+ validate_country_code(country_code)
+ return country_code
+
+ @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:
+ msg = "Field name must be provided."
+ raise ValueError(msg)
+
+ validate_field_is_unique(info.field_name, value)
+
+ return value
+
+ @field_validator("site_name")
+ 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
diff --git a/gso/types/coordinates.py b/gso/types/coordinates.py
new file mode 100644
index 0000000000000000000000000000000000000000..91f1188fbe9196a564af4ab9f34a8d2aedeafb16
--- /dev/null
+++ b/gso/types/coordinates.py
@@ -0,0 +1,61 @@
+"""Custom coordinate types for latitude and longitude."""
+
+import re
+from typing import Annotated
+
+from pydantic import AfterValidator
+from typing_extensions import Doc
+
+MAX_LONGITUDE = 180
+MIN_LONGITUDE = -180
+MAX_LATITUDE = 90
+MIN_LATITUDE = -90
+
+
+def validate_latitude(v: str) -> str:
+ """Validate a latitude coordinate."""
+ msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
+ regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
+ if not regex.match(str(v)):
+ raise ValueError(msg)
+
+ float_v = float(v)
+ if float_v > MAX_LATITUDE or float_v < MIN_LATITUDE:
+ raise ValueError(msg)
+
+ return v
+
+
+def validate_longitude(v: str) -> str:
+ """Validate a longitude coordinate."""
+ regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
+ msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
+ if not regex.match(v):
+ raise ValueError(msg)
+
+ float_v = float(v)
+ if float_v > MAX_LONGITUDE or float_v < MIN_LONGITUDE:
+ raise ValueError(msg)
+
+ return v
+
+
+LatitudeCoordinate = Annotated[
+ str,
+ AfterValidator(validate_latitude),
+ Doc(
+ "A latitude coordinate, modeled as a string. "
+ "The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. "
+ "It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0."
+ ),
+]
+LongitudeCoordinate = Annotated[
+ str,
+ AfterValidator(validate_longitude),
+ Doc(
+ "A longitude coordinate, modeled as a string. "
+ "The coordinate must match the format conforming to the longitude "
+ "range of -180 to +180 degrees. It can be a floating-point number or an integer. "
+ "Valid examples: 40.7128, -74.0060, 180, -180, 0."
+ ),
+]
diff --git a/gso/types/country_code.py b/gso/types/country_code.py
new file mode 100644
index 0000000000000000000000000000000000000000..4f03392845e34dde7299cda8a3135ded67041def
--- /dev/null
+++ b/gso/types/country_code.py
@@ -0,0 +1,15 @@
+"""Country codes."""
+
+import pycountry
+
+
+def validate_country_code(country_code: str) -> str:
+ """Validate that a country code is valid."""
+ # Check for the UK code before attempting to look it up since it's known as "GB" in the pycountry database.
+ if country_code != "UK":
+ try:
+ pycountry.countries.lookup(country_code)
+ except LookupError as e:
+ msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
+ raise ValueError(msg) from e
+ return country_code
diff --git a/gso/types/interfaces.py b/gso/types/interfaces.py
new file mode 100644
index 0000000000000000000000000000000000000000..15c91167de822fee116600b71bf0b05dedfd87ae
--- /dev/null
+++ b/gso/types/interfaces.py
@@ -0,0 +1,102 @@
+"""Custom types for interfaces, both physical and logical."""
+
+import re
+from typing import Annotated
+
+from annotated_types import Len
+from orchestrator.domain.base import T
+from pydantic import AfterValidator, BaseModel
+from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
+from typing_extensions import Doc
+
+
+class LAGMember(BaseModel):
+ """A :term:`LAG` member interface that consists of a name and description."""
+
+ interface_name: str
+ interface_description: str | None = None
+
+ def __hash__(self) -> int:
+ """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
+ return hash(self.interface_name)
+
+
+def validate_interface_names_are_unique(interfaces: list[LAGMember]) -> list[LAGMember]:
+ """Verify if interfaces are unique.
+
+ Raises a :class:`ValueError` if the interfaces are not unique.
+
+ :param interfaces: The list of interfaces.
+ :type interfaces: list[:class:`utils.types.LAGMember`]
+
+ :return: The list of interfaces
+ :rtype: list[:class:`utils.types.LAGMember`]
+ """
+ interface_names = [member.interface_name for member in interfaces]
+ if len(interface_names) != len(set(interface_names)):
+ msg = "Interfaces must be unique."
+ raise ValueError(msg)
+ return interfaces
+
+
+def validate_juniper_phy_interface_name(interface_name: str) -> str:
+ """Validate that the provided interface name matches the expected pattern.
+
+ The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
+ then a number between 0 and 19, a forward slash '/', another number between 0 and 99,
+ another forward slash '/', and ends with a number between 0 and 99.
+ For example: 'xe-1/0/0'. This only applies to Juniper-brand hardware.
+
+ :param str interface_name: Interface name to validate.
+
+ :return str: The interface name if match was successful, otherwise it will throw a ValueError exception.
+ """
+ pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
+ if not bool(pattern.match(interface_name)):
+ error_msg = (
+ f"Invalid interface name. The interface name should be of format: xe-1/0/0. " f"Got: {interface_name}"
+ )
+ raise ValueError(error_msg)
+ return interface_name
+
+
+def validate_juniper_ae_interface_name(interface_name: str) -> str:
+ """Validate that the provided interface name matches the expected pattern for a :term:`LAG` interface.
+
+ Interface names must match the pattern 'ae' followed by a one- or two-digit number.
+ """
+ juniper_lag_re = re.compile("^ae\\d{1,2}$")
+ if not juniper_lag_re.match(interface_name):
+ msg = "Invalid LAG name, please try again."
+ raise ValueError(msg)
+ return interface_name
+
+
+JuniperPhyInterface = Annotated[str, AfterValidator(validate_juniper_phy_interface_name)]
+JuniperAEInterface = Annotated[str, AfterValidator(validate_juniper_ae_interface_name)]
+LAGMemberList = Annotated[
+ list[T],
+ AfterValidator(validate_unique_list),
+ AfterValidator(validate_interface_names_are_unique),
+ Len(min_length=0),
+ Doc("A list of :term:`LAG` member interfaces."),
+]
+
+
+class JuniperLAGMember(LAGMember):
+ """A Juniper-specific :term:`LAG` member interface."""
+
+ interface_name: JuniperPhyInterface
+
+
+class PhysicalPortCapacity(strEnum):
+ """Physical port capacity enumerator.
+
+ An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
+ """
+
+ ONE_GIGABIT_PER_SECOND = "1G"
+ TEN_GIGABIT_PER_SECOND = "10G"
+ HUNDRED_GIGABIT_PER_SECOND = "100G"
+ FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
diff --git a/gso/types/ip_address.py b/gso/types/ip_address.py
new file mode 100644
index 0000000000000000000000000000000000000000..793ebb4f1a7aefd66c85d4f307de983255edee31
--- /dev/null
+++ b/gso/types/ip_address.py
@@ -0,0 +1,14 @@
+"""IP addresses."""
+
+import ipaddress
+
+
+def validate_ipv4_or_ipv6(value: str) -> str:
+ """Validate that a value is a valid IPv4 or IPv6 address."""
+ try:
+ ipaddress.ip_address(value)
+ except ValueError as e:
+ msg = "Enter a valid IPv4 or IPv6 address."
+ raise ValueError(msg) from e
+ else:
+ return value
diff --git a/gso/types/netbox_router.py b/gso/types/netbox_router.py
new file mode 100644
index 0000000000000000000000000000000000000000..a5845a479b605f98ba1faaa6f4a3c6e461bbf8d3
--- /dev/null
+++ b/gso/types/netbox_router.py
@@ -0,0 +1,27 @@
+"""A router that must be present in Netbox."""
+
+from pydantic_forms.types import UUIDstr
+
+from gso.products.product_types.router import Router
+from gso.services.netbox_client import NetboxClient
+from gso.utils.shared_enums import Vendor
+
+
+def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
+ """Verify if a device exists in Netbox.
+
+ Raises a :class:`ValueError` if the device is not found.
+
+ :param subscription_id: The :term:`UUID` of the router subscription.
+ :type subscription_id: :class:`UUIDstr`
+
+ :return: The :term:`UUID` of the router subscription.
+ :rtype: :class:`UUIDstr`
+ """
+ router_type = Router.from_subscription(subscription_id)
+ if router_type.router.vendor == Vendor.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)
+ return subscription_id
diff --git a/gso/types/site_name.py b/gso/types/site_name.py
new file mode 100644
index 0000000000000000000000000000000000000000..69a46b1a43646bb50389f47a4bb5dc0ace1f170b
--- /dev/null
+++ b/gso/types/site_name.py
@@ -0,0 +1,17 @@
+"""Type for the name of a site."""
+
+import re
+
+
+def validate_site_name(site_name: str) -> None:
+ """Validate the site name.
+
+ The site name must consist of three uppercase letters, optionally followed by a single digit.
+ """
+ pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
+ if not pattern.match(site_name):
+ msg = (
+ "Enter a valid site name. It must consist of three uppercase letters (A-Z), followed by an optional single "
+ f"digit (0-9). Received: {site_name}"
+ )
+ raise ValueError(msg)
diff --git a/gso/types/snmp.py b/gso/types/snmp.py
new file mode 100644
index 0000000000000000000000000000000000000000..03581cf970c036db37ea901d8b159d25fc480136
--- /dev/null
+++ b/gso/types/snmp.py
@@ -0,0 +1,10 @@
+"""An enumerator of SNMP version numbers."""
+
+from enum import StrEnum
+
+
+class SNMPVersion(StrEnum):
+ """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
+
+ V2C = "v2c"
+ V3 = "v3"
diff --git a/gso/types/tt_number.py b/gso/types/tt_number.py
new file mode 100644
index 0000000000000000000000000000000000000000..a9e1d05c6ad9563047764d56461afe846b99be88
--- /dev/null
+++ b/gso/types/tt_number.py
@@ -0,0 +1,29 @@
+"""A Trouble Ticket number."""
+
+import re
+from typing import Annotated
+
+from pydantic import AfterValidator
+
+
+def validate_tt_number(tt_number: str) -> str:
+ """Validate a string to match a specific pattern.
+
+ This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
+
+ :param str tt_number: The TT number as string to validate
+
+ :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
+ """
+ pattern = r"^TT#\d{16}$"
+ if not bool(re.match(pattern, tt_number)):
+ err_msg = (
+ f"The given TT number: {tt_number} is not valid. "
+ f" A valid TT number starts with 'TT#' followed by 16 digits."
+ )
+ raise ValueError(err_msg)
+
+ return tt_number
+
+
+TTNumber = Annotated[str, AfterValidator(validate_tt_number)]
diff --git a/gso/types/unique_field.py b/gso/types/unique_field.py
new file mode 100644
index 0000000000000000000000000000000000000000..1b47dcb0db386d09dc58e2c48119854c6fb4adb3
--- /dev/null
+++ b/gso/types/unique_field.py
@@ -0,0 +1,10 @@
+"""An input field that must be unique in the database."""
+
+from gso.services import subscriptions
+
+
+def validate_field_is_unique(field_name: str, value: str | int) -> None:
+ """Validate that a site field is unique."""
+ if len(subscriptions.get_active_subscriptions_by_field_and_value(field_name, str(value))) > 0:
+ msg = f"{field_name} must be unique"
+ raise ValueError(msg)
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 13016c76f13028db17e8397ebff7ce23455a997a..f1063cecd97209247ecd950a49de95bf46f5c0e0 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -1,42 +1,21 @@
"""Helper methods that are used across :term:`GSO`."""
-import ipaddress
import re
-from enum import StrEnum
+from typing import TYPE_CHECKING
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
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, PhysicalPortCapacity
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
from gso.products.product_types.router import Router
+from gso.services import subscriptions
from gso.services.netbox_client import NetboxClient
-from gso.services.subscriptions import get_active_router_subscriptions, get_active_subscriptions_by_field_and_value
+from gso.types.interfaces import PhysicalPortCapacity
from gso.utils.shared_enums import IPv4AddressType, Vendor
-
-class LAGMember(BaseModel):
- """A :term:`LAG` member interface that consists of a name and description."""
-
- interface_name: str
- interface_description: str | None = None
-
- def __hash__(self) -> int:
- """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
- return hash((self.interface_name, self.interface_description))
-
-
-class SNMPVersion(StrEnum):
- """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
-
- V2C = "v2c"
- V3 = "v3"
+if TYPE_CHECKING:
+ from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
@@ -57,7 +36,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
def available_interfaces_choices_including_current_members(
router_id: UUID,
speed: str,
- interfaces: list[IptrunkInterfaceBlock],
+ interfaces: list["IptrunkInterfaceBlock"],
) -> Choice | None:
"""Return a list of available interfaces for a given router and speed including the current members.
@@ -119,184 +98,6 @@ def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
return f"49.51e5.0001.{re_split}.00"
-def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
- """Verify if a device exists in Netbox.
-
- Raises a :class:`ValueError` if the device is not found.
-
- :param subscription_id: The :term:`UUID` of the router subscription.
- :type subscription_id: :class:`UUIDstr`
-
- :return: The :term:`UUID` of the router subscription.
- :rtype: :class:`UUIDstr`
- """
- router_type = Router.from_subscription(subscription_id)
- if router_type.router.vendor == Vendor.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)
- return subscription_id
-
-
-def validate_iptrunk_unique_interface(interfaces: list[LAGMember]) -> list[LAGMember]:
- """Verify if the interfaces are unique.
-
- Raises a :class:`ValueError` if the interfaces are not unique.
-
- :param interfaces: The list of interfaces.
- :type interfaces: list[:class:`LAGMember`]
-
- :return: The list of interfaces
- :rtype: list[:class:`LAGMember`]
- """
- interface_names = [member.interface_name for member in interfaces]
- if len(interface_names) != len(set(interface_names)):
- msg = "Interfaces must be unique."
- raise ValueError(msg)
- return interfaces
-
-
-def validate_site_fields_is_unique(field_name: str, value: str | int) -> None:
- """Validate that a site field is unique."""
- if len(get_active_subscriptions_by_field_and_value(field_name, str(value))) > 0:
- msg = f"{field_name} must be unique"
- raise ValueError(msg)
-
-
-def validate_ipv4_or_ipv6(value: str) -> str:
- """Validate that a value is a valid IPv4 or IPv6 address."""
- try:
- ipaddress.ip_address(value)
- except ValueError as e:
- msg = "Enter a valid IPv4 or IPv6 address."
- raise ValueError(msg) from e
- else:
- return value
-
-
-def validate_country_code(country_code: str) -> str:
- """Validate that a country code is valid."""
- # Check for the UK code before attempting to look it up since it's known as "GB" in the pycountry database.
- if country_code != "UK":
- try:
- pycountry.countries.lookup(country_code)
- except LookupError as e:
- msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
- raise ValueError(msg) from e
- return country_code
-
-
-def validate_site_name(site_name: str) -> None:
- """Validate the site name.
-
- The site name must consist of three uppercase letters, optionally followed by a single digit.
- """
- pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
- if not pattern.match(site_name):
- msg = (
- "Enter a valid site name. It must consist of three uppercase letters (A-Z), followed by an optional single "
- f"digit (0-9). Received: {site_name}"
- )
- raise ValueError(msg)
-
-
-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
- site_country_code: str
- site_name: str
- site_city: str
- site_country: str
- site_latitude: LatitudeCoordinate
- site_longitude: LongitudeCoordinate
- partner: str
-
- @field_validator("site_ts_address")
- 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
-
- @field_validator("site_country_code")
- def country_code_must_exist(cls, country_code: str) -> str:
- """Validate that the country code exists."""
- validate_country_code(country_code)
- return country_code
-
- @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:
- msg = "Field name must be provided."
- raise ValueError(msg)
-
- validate_site_fields_is_unique(info.field_name, value)
-
- return value
-
- @field_validator("site_name")
- 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
-
-
-def validate_interface_name_list(interface_name_list: list, vendor: str) -> list:
- """Validate that the provided interface name matches the expected pattern.
-
- The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
- then a number between 0 and 19, a forward slash '/', another number between 0 and 99,
- another forward slash '/', and ends with a number between 0 and 99.
- For example: 'xe-1/0/0'.
-
- :param list interface_name_list: List of interface names to validate.
- :param str vendor: Router vendor to check interface names
-
- :return list: The list of interface names if all match was successful, otherwise it will throw a ValueError
- exception.
- """
- # For Nokia nothing to do
- if vendor == Vendor.NOKIA:
- return interface_name_list
- pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
- for interface in interface_name_list:
- if not bool(pattern.match(interface.interface_name)):
- error_msg = (
- f"Invalid interface name. The interface name should be of format: xe-1/0/0. "
- f"Got: [{interface.interface_name}]"
- )
- raise ValueError(error_msg)
- return interface_name_list
-
-
-def validate_tt_number(tt_number: str) -> str:
- """Validate a string to match a specific pattern.
-
- This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
-
- :param str tt_number: The TT number as string to validate
-
- :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
- """
- pattern = r"^TT#\d{16}$"
- if not bool(re.match(pattern, tt_number)):
- err_msg = (
- f"The given TT number: {tt_number} is not valid. "
- f" A valid TT number starts with 'TT#' followed by 16 digits."
- )
- raise ValueError(err_msg)
-
- return tt_number
-
-
def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
"""Generate an :term:`FQDN` from a hostname, site name, and a country code."""
oss = settings.load_oss_params()
@@ -318,7 +119,9 @@ def generate_inventory_for_active_routers(
:return: A dictionary representing the inventory of active routers.
:rtype: dict[str, Any]
"""
- all_routers = [Router.from_subscription(r["subscription_id"]) for r in get_active_router_subscriptions()]
+ all_routers = [
+ Router.from_subscription(r["subscription_id"]) for r in subscriptions.get_active_router_subscriptions()
+ ]
exclude_routers = exclude_routers or []
return {
@@ -351,3 +154,33 @@ def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_
if iptrunk_speed == PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND:
return iptrunk_number_of_members - 1
return iptrunk_number_of_members
+
+
+def active_site_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active site in an input form."""
+ site_subscriptions = {
+ str(site["subscription_id"]): site["description"]
+ for site in subscriptions.get_active_site_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True)) # type: ignore[arg-type]
+
+
+def active_router_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active Router in an input form."""
+ router_subscriptions = {
+ str(router["subscription_id"]): router["description"]
+ for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)) # type: ignore[arg-type]
+
+
+def active_switch_selector() -> Choice:
+ """Generate a dropdown selector for choosing an active Switch in an input form."""
+ switch_subscriptions = {
+ str(switch["subscription_id"]): switch["description"]
+ for switch in subscriptions.get_active_switch_subscriptions(includes=["subscription_id", "description"])
+ }
+
+ return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)) # type: ignore[arg-type]
diff --git a/gso/utils/types.py b/gso/utils/types.py
deleted file mode 100644
index 3e1b4091b127d9a572c12c7fad462dc4887de9f7..0000000000000000000000000000000000000000
--- a/gso/utils/types.py
+++ /dev/null
@@ -1,9 +0,0 @@
-"""Define custom types for use across the application."""
-
-from typing import Annotated
-
-from pydantic import AfterValidator
-
-from gso.utils.helpers import validate_tt_number
-
-TTNumber = Annotated[str, AfterValidator(validate_tt_number)]
diff --git a/gso/workflows/iptrunk/create_imported_iptrunk.py b/gso/workflows/iptrunk/create_imported_iptrunk.py
index 9b0e6b87a8a095073875721e79d72c9e03baa66d..18f76a7f93d30e7bc082bb2696ac9daf8448ce86 100644
--- a/gso/workflows/iptrunk/create_imported_iptrunk.py
+++ b/gso/workflows/iptrunk/create_imported_iptrunk.py
@@ -6,7 +6,6 @@ from uuid import uuid4
from orchestrator import workflow
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, begin, done, step
@@ -15,30 +14,17 @@ from pydantic import AfterValidator, ConfigDict
from pydantic_forms.validators import validate_unique_list
from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType
from gso.products.product_types.iptrunk import ImportedIptrunkInactive
from gso.products.product_types.router import Router
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
-from gso.utils.helpers import LAGMember
-
-
-def _generate_routers() -> dict[str, str]:
- """Generate a dictionary of router IDs and descriptions."""
- routers = {}
- for subscription in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"]):
- routers[str(subscription["subscription_id"])] = subscription["description"]
-
- return routers
-
-
-LAGMemberList = Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
+from gso.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.utils.helpers import active_router_selector
def initial_input_form_generator() -> FormGenerator:
"""Take all information passed to this workflow by the :term:`API` endpoint that was called."""
- routers = _generate_routers()
- router_enum = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title="Import Iptrunk")
@@ -51,15 +37,15 @@ def initial_input_form_generator() -> FormGenerator:
iptrunk_minimum_links: int
iptrunk_isis_metric: int
- side_a_node_id: router_enum # type: ignore[valid-type]
+ side_a_node_id: active_router_selector() # type: ignore[valid-type]
side_a_ae_iface: str
side_a_ae_geant_a_sid: str | None = None
- side_a_ae_members: LAGMemberList
+ side_a_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
- side_b_node_id: router_enum # type: ignore[valid-type]
+ side_b_node_id: active_router_selector() # type: ignore[valid-type]
side_b_ae_iface: str
side_b_ae_geant_a_sid: str | None = None
- side_b_ae_members: LAGMemberList
+ side_b_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
@@ -94,11 +80,11 @@ def initialize_subscription(
side_a_node_id: str,
side_a_ae_iface: str,
side_a_ae_geant_a_sid: str | None,
- side_a_ae_members: list[dict],
+ side_a_ae_members: LAGMemberList,
side_b_node_id: str,
side_b_ae_iface: str,
side_b_ae_geant_a_sid: str | None,
- side_b_ae_members: list[dict],
+ side_b_ae_members: LAGMemberList,
) -> State:
"""Take all input from the user, and store it in the database."""
subscription.iptrunk.geant_s_sid = geant_s_sid
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 246d0639974fd90a32b08f3600d90231c0707400..aba759808ff59cd8886a2b01cff715fe95421f38 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -16,15 +16,14 @@ from orchestrator.workflow import StepList, begin, conditional, done, step, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from ping3 import ping
-from pydantic import AfterValidator, ConfigDict, field_validator
-from pydantic_forms.validators import ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict, field_validator
+from pydantic_forms.validators import ReadOnlyField
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlockInactive,
IptrunkSideBlockInactive,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_types.iptrunk import Iptrunk, IptrunkInactive, IptrunkProvisioning
from gso.products.product_types.router import Router
@@ -35,29 +34,27 @@ from gso.services.partners import get_partner_by_name
from gso.services.sharepoint import SharePointClient
from gso.services.subscriptions import get_non_terminated_iptrunk_subscriptions
from gso.settings import load_oss_params
+from gso.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.types.netbox_router import validate_router_in_netbox
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_lags_choices,
calculate_recommended_minimum_links,
get_router_vendor,
- validate_interface_name_list,
- validate_iptrunk_unique_interface,
- validate_router_in_netbox,
)
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
from gso.utils.workflow_steps import prompt_sharepoint_checklist_url
def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Gather input from the user in three steps. General information, and information on both sides of the trunk."""
- routers = {}
- for router in subscriptions.get_active_router_subscriptions(
- includes=["subscription_id", "description"]
- ) + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"]):
- # Add both provisioning and active routers, since trunks are required for promoting a router to active.
- routers[str(router["subscription_id"])] = router["description"]
+ # Add both provisioning and active routers, since trunks are required for promoting a router to active.
+ routers = {
+ str(router["subscription_id"]): router["description"]
+ for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"])
+ + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"])
+ }
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title=product_name)
@@ -97,8 +94,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
router_a_fqdn = Router.from_subscription(router_a).router.router_fqdn
juniper_ae_members = Annotated[
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[JuniperLAGMember],
Len(
min_length=initial_user_input.iptrunk_number_of_members,
max_length=initial_user_input.iptrunk_number_of_members,
@@ -114,8 +110,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
)
ae_members_side_a_type = Annotated[
- list[NokiaLAGMemberA],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMemberA],
Len(
min_length=initial_user_input.iptrunk_number_of_members,
max_length=initial_user_input.iptrunk_number_of_members,
@@ -125,19 +120,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
ae_members_side_a_type = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideAForm(FormPage):
- model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk.({router_a_fqdn})")
+ model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk. ({router_a_fqdn})")
side_a_ae_iface: available_lags_choices(router_a) or str # type: ignore[valid-type]
side_a_ae_geant_a_sid: str | None
side_a_ae_members: ae_members_side_a_type
- @field_validator("side_a_ae_members")
- def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- validate_iptrunk_unique_interface(side_a_ae_members)
- vendor = get_router_vendor(router_a)
- validate_interface_name_list(side_a_ae_members, vendor)
- return side_a_ae_members
-
user_input_side_a = yield CreateIptrunkSideAForm
# Remove the selected router for side A, to prevent any loops
routers.pop(str(router_a))
@@ -165,8 +153,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
)
ae_members_side_b = Annotated[
- list[NokiaLAGMemberB],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMemberB],
Len(
min_length=len(user_input_side_a.side_a_ae_members), max_length=len(user_input_side_a.side_a_ae_members)
),
@@ -175,19 +162,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
ae_members_side_b = juniper_ae_members # type: ignore[assignment, misc]
class CreateIptrunkSideBForm(FormPage):
- model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk.({router_b_fqdn})")
+ model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk. ({router_b_fqdn})")
side_b_ae_iface: available_lags_choices(router_b) or str # type: ignore[valid-type]
side_b_ae_geant_a_sid: str | None
side_b_ae_members: ae_members_side_b
- @field_validator("side_b_ae_members")
- def validate_side_b_ae_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- validate_iptrunk_unique_interface(side_b_ae_members)
- vendor = get_router_vendor(router_b)
- validate_interface_name_list(side_b_ae_members, vendor)
- return side_b_ae_members
-
user_input_side_b = yield CreateIptrunkSideBForm
return (
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index a45b5eca61144577c5dbf58a251b0b1ca7c76f6d..feb7100a46360f5c9fb14ef28f98b6bb4cb165ac 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -13,7 +13,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.utils.types import TTNumber
+from gso.types.tt_number import TTNumber
def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 73fb344e169a0bb26d3ef1eba18b84f0e0367531..0ed107a6fb9aab243887a6af3fdf88a355e7b887 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -6,7 +6,6 @@ configured to run from A to C. B is then no longer associated with this IP trunk
import copy
import json
-import re
from typing import Annotated
from uuid import uuid4
@@ -21,8 +20,8 @@ from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, begin, conditional, done, inputstep
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import AfterValidator, ConfigDict, field_validator
-from pydantic_forms.validators import ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict
+from pydantic_forms.validators import ReadOnlyField
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkType
@@ -34,15 +33,14 @@ from gso.services.netbox_client import NetboxClient
from gso.services.sharepoint import SharePointClient
from gso.services.subscriptions import get_active_router_subscriptions
from gso.settings import load_oss_params
+from gso.types.interfaces import JuniperAEInterface, JuniperLAGMember, LAGMember, LAGMemberList
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_lags_choices,
get_router_vendor,
- validate_interface_name_list,
)
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
from gso.utils.workflow_steps import prompt_sharepoint_checklist_url, set_isis_to_max
@@ -103,7 +101,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
new_side_iptrunk_router_input = yield NewSideIPTrunkRouterForm
new_router = new_side_iptrunk_router_input.new_node
- side_a_ae_iface = available_lags_choices(new_router) or str
+ side_a_ae_iface = available_lags_choices(new_router) or JuniperAEInterface
new_side_is_nokia = get_router_vendor(new_router) == Vendor.NOKIA
if new_side_is_nokia:
@@ -115,8 +113,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
ae_members = Annotated[
- list[NokiaLAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[NokiaLAGMember],
Len(
min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
@@ -124,8 +121,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
]
else:
ae_members = Annotated[ # type: ignore[assignment, misc]
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ LAGMemberList[JuniperLAGMember],
Len(
min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
@@ -150,23 +146,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
model_config = ConfigDict(title=form_title)
new_lag_interface: side_a_ae_iface # type: ignore[valid-type]
- existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=list[LAGMember]) # type: ignore[valid-type]
+ existing_lag_interface: ReadOnlyField(existing_lag_ae_members, default_type=LAGMemberList[LAGMember]) # type: ignore[valid-type]
new_lag_member_interfaces: ae_members
- @field_validator("new_lag_interface")
- def lag_interface_proper_name(cls, new_lag_interface: str) -> str:
- if get_router_vendor(new_router) == Vendor.JUNIPER:
- juniper_lag_re = re.compile("^ae\\d{1,2}$")
- if not juniper_lag_re.match(new_lag_interface):
- msg = "Invalid LAG name, please try again."
- raise ValueError(msg)
- return new_lag_interface
-
- @field_validator("new_lag_member_interfaces")
- def is_interface_names_valid_juniper(cls, new_lag_member_interfaces: list[LAGMember]) -> list[LAGMember]:
- vendor = get_router_vendor(new_router)
- return validate_interface_name_list(new_lag_member_interfaces, vendor)
-
new_side_input = yield NewSideIPTrunkForm
return (
migrate_form_input.model_dump()
diff --git a/gso/workflows/iptrunk/modify_isis_metric.py b/gso/workflows/iptrunk/modify_isis_metric.py
index 285907b45508249794bb8c5fd486b62ed0b4dac6..493f0eb98c5db0b5edc1d3b20269ef50019d4432 100644
--- a/gso/workflows/iptrunk/modify_isis_metric.py
+++ b/gso/workflows/iptrunk/modify_isis_metric.py
@@ -12,7 +12,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.utils.types import TTNumber
+from gso.types.tt_number import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 394e369a88d1f64481750080410ffcb8f4335064..6ec7d59a83ef272dc63837d8d01b59c81fac07b2 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -1,7 +1,7 @@
"""A modification workflow that updates the :term:`LAG` interfaces that are part of an existing IP trunk."""
import json
-from typing import Annotated, TypeVar
+from typing import Annotated
from uuid import UUID, uuid4
from annotated_types import Len
@@ -12,38 +12,33 @@ from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, begin, conditional, done, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import AfterValidator, ConfigDict, field_validator
-from pydantic_forms.validators import Label, ReadOnlyField, validate_unique_list
+from pydantic import ConfigDict
+from pydantic_forms.validators import Label, ReadOnlyField
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
+from gso.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import (
- LAGMember,
available_interfaces_choices,
available_interfaces_choices_including_current_members,
calculate_recommended_minimum_links,
get_router_vendor,
- validate_interface_name_list,
- validate_iptrunk_unique_interface,
)
from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, Vendor
-from gso.utils.types import TTNumber
from gso.workflows.iptrunk.migrate_iptrunk import check_ip_trunk_optical_levels_pre
from gso.workflows.iptrunk.validate_iptrunk import check_ip_trunk_isis
-T = TypeVar("T", bound=LAGMember)
-
def initialize_ae_members(
subscription: Iptrunk, initial_user_input: dict, side_index: int
-) -> Annotated[list[LAGMember], ""]:
+) -> type[LAGMemberList[LAGMember]]:
"""Initialize the list of AE members."""
router = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_node
router_vendor = get_router_vendor(router.owner_subscription_id)
@@ -67,17 +62,15 @@ def initialize_ae_members(
)
)
- return Annotated[
- list[NokiaLAGMember],
- AfterValidator(validate_unique_list),
+ return Annotated[ # type: ignore[return-value]
+ LAGMemberList[NokiaLAGMember],
Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
- ] # type: ignore[return-value]
+ ]
- return Annotated[
- list[LAGMember],
- AfterValidator(validate_unique_list),
+ return Annotated[ # type: ignore[return-value]
+ LAGMemberList[JuniperLAGMember],
Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
- ] # type: ignore[return-value]
+ ]
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -132,15 +125,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @field_validator("side_a_ae_members")
- def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_a_ae_members)
-
- @field_validator("side_a_ae_members")
- def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- vendor = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.vendor
- return validate_interface_name_list(side_a_ae_members, vendor)
-
user_input_side_a = yield ModifyIptrunkSideAForm
ae_members_side_b = initialize_ae_members(subscription, initial_user_input.model_dump(), 1)
@@ -158,15 +142,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @field_validator("side_b_ae_members")
- def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_b_ae_members)
-
- @field_validator("side_b_ae_members")
- def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- vendor = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.vendor
- return validate_interface_name_list(side_b_ae_members, vendor)
-
user_input_side_b = yield ModifyIptrunkSideBForm
return (
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index bb1a6fd90b3d9a5e3b9aa9bb633db58cc2eb1cd4..fec1c48bb59142d51b6a65f5a1bcfada4d69ee28 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -22,9 +22,9 @@ from gso.products.product_types.iptrunk import Iptrunk
from gso.services import infoblox
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import get_router_vendor
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
from gso.utils.workflow_steps import set_isis_to_max
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 59dd6210578d9556fdf79ecacf7df0b5518fdbd5..68c55cf0909e9781864b14348cc5e47ccd6a3bf0 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -23,9 +23,9 @@ from gso.services.netbox_client import NetboxClient
from gso.services.partners import get_partner_by_name
from gso.services.sharepoint import SharePointClient
from gso.settings import load_oss_params
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import generate_fqdn, iso_from_ipv4
from gso.utils.shared_enums import PortNumber, Vendor
-from gso.utils.types import TTNumber
from gso.utils.workflow_steps import (
deploy_base_config_dry,
deploy_base_config_real,
diff --git a/gso/workflows/router/promote_p_to_pe.py b/gso/workflows/router/promote_p_to_pe.py
index 29a110c63589701f239715a030f87567fde6d17f..f294bd99468942094016687e93937c093788b074 100644
--- a/gso/workflows/router/promote_p_to_pe.py
+++ b/gso/workflows/router/promote_p_to_pe.py
@@ -21,9 +21,9 @@ from gso.services import lso_client
from gso.services.kentik_client import KentikClient, NewKentikDevice
from gso.services.lso_client import lso_interaction
from gso.services.subscriptions import get_all_active_sites
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/router/redeploy_base_config.py b/gso/workflows/router/redeploy_base_config.py
index b30d02f16b6bb197a20257f9016eabb6dad0d8fa..eef8b6e74fe3db1c5d5d292fc498a1872955c622 100644
--- a/gso/workflows/router/redeploy_base_config.py
+++ b/gso/workflows/router/redeploy_base_config.py
@@ -10,7 +10,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.router import Router
from gso.services.lso_client import lso_interaction
-from gso.utils.types import TTNumber
+from gso.types.tt_number import TTNumber
from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 5b0c6ef546c9cbb08d93c5cd277f8e31e71f57c1..60895ab33d7fc0c40a5d8df58db7d6a543c3361d 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -28,9 +28,9 @@ from gso.services.librenms_client import LibreNMSClient
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.settings import load_oss_params
+from gso.types.tt_number import TTNumber
from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
-from gso.utils.types import TTNumber
logger = logging.getLogger(__name__)
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 8fbb2813c94cff8443647a033c146e93f34b7ca4..f8f17a4829060c6654782a342e0380415e01336c 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -17,8 +17,9 @@ from gso.products.product_types.router import Router
from gso.services import librenms_client, lso_client
from gso.services.lso_client import lso_interaction
from gso.services.subscriptions import get_trunks_that_terminate_on_router
-from gso.utils.helpers import SNMPVersion, generate_inventory_for_active_routers
-from gso.utils.types import TTNumber
+from gso.types.snmp import SNMPVersion
+from gso.types.tt_number import TTNumber
+from gso.utils.helpers import generate_inventory_for_active_routers
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
diff --git a/gso/workflows/site/create_imported_site.py b/gso/workflows/site/create_imported_site.py
index 1fd9d59d7aeddbb04faac6131a0cc68533a7b1d4..1e149ab89733ad093f894cb76556949c677cc10d 100644
--- a/gso/workflows/site/create_imported_site.py
+++ b/gso/workflows/site/create_imported_site.py
@@ -10,11 +10,12 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
from pydantic import ConfigDict
from gso.products import ProductName
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
+from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import ImportedSiteInactive
from gso.services import subscriptions
from gso.services.partners import get_partner_by_name
-from gso.utils.helpers import BaseSiteValidatorModel
+from gso.types.base_site import BaseSiteValidatorModel
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
@step("Create subscription")
diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py
index d2e99d510e678a7f7e654de4cbee1a162868ed2f..0858b372843828aa0944e712673b2cf11cbe3a10 100644
--- a/gso/workflows/site/create_site.py
+++ b/gso/workflows/site/create_site.py
@@ -10,10 +10,10 @@ from pydantic import ConfigDict
from pydantic_forms.validators import ReadOnlyField
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.partners import get_partner_by_name
-from gso.utils.helpers import BaseSiteValidatorModel
+from gso.types.base_site import BaseSiteValidatorModel
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
def initial_input_form_generator(product_name: str) -> FormGenerator:
diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py
index 0fb2b50d3b54432138b2c15addfffddc39de4e66..aac144356899f123c387db2db0694831ca9dea63 100644
--- a/gso/workflows/site/modify_site.py
+++ b/gso/workflows/site/modify_site.py
@@ -15,9 +15,11 @@ 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
+from gso.products.product_blocks.site import SiteTier
from gso.products.product_types.site import Site
-from gso.utils.helpers import validate_ipv4_or_ipv6, validate_site_fields_is_unique
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.types.ip_address import validate_ipv4_or_ipv6
+from gso.types.unique_field import validate_field_is_unique
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -47,7 +49,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
if value and value == getattr(subscription.site, info.field_name):
return value
- validate_site_fields_is_unique(info.field_name, value)
+ validate_field_is_unique(info.field_name, value)
return value
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 8933627b82b06a35ca17e5a2a4417dc9cd9859c6..89933fc8de3f9b4860f60d387867dedc2ed99371 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -13,9 +13,10 @@ from gso.cli.imports import (
import_super_pop_switches,
)
from gso.products import Router, Site
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
+from gso.types.interfaces import PhysicalPortCapacity
from gso.utils.helpers import iso_from_ipv4
from gso.utils.shared_enums import Vendor
@@ -327,11 +328,11 @@ def test_import_iptrunk_non_unique_members_side_a_and_b(mock_start_process, mock
assert "Validation error: 2 validation errors for IptrunkImportModel" in captured_output
assert (
"""side_a_ae_members
- Value error, Items must be unique [type=value_error, input_value=[{'interface_name':"""
+ List must be unique [type=unique_list, input_value=[{'interface_name':"""
) in captured_output
assert (
"""side_b_ae_members
- Value error, Items must be unique [type=value_error, input_value=[{'interface_name':"""
+ List must be unique [type=unique_list, input_value=[{'interface_name':"""
) in captured_output
assert mock_start_process.call_count == 0
diff --git a/test/conftest.py b/test/conftest.py
index 0c4e36e4f52d60d8270493b9dd234ffb080ba507..8cfea9d28eacfd24c6397c514245024cb15b5f6d 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -34,7 +34,7 @@ from urllib3_mock import Responses
from gso.main import init_gso_app
from gso.services.partners import PartnerSchema, create_partner
-from gso.utils.helpers import LAGMember
+from gso.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import ( # noqa: F401
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
@@ -107,7 +107,7 @@ class FakerProvider(BaseProvider):
def network_interface(self) -> str:
return self.generator.numerify("ge-@#/@#/@#")
- def link_members_juniper(self) -> list[LAGMember]:
+ def link_members_juniper(self) -> LAGMemberList[LAGMember]:
iface_amount = self.generator.random_int(min=2, max=5)
interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]
return [
@@ -115,7 +115,7 @@ class FakerProvider(BaseProvider):
for interface_name in interface_names
]
- def link_members_nokia(self) -> list[LAGMember]:
+ def link_members_nokia(self) -> LAGMemberList[LAGMember]:
iface_amount = self.generator.random_int(min=2, max=5)
return [
LAGMember(interface_name=f"Interface{i}", interface_description=self.generator.sentence())
diff --git a/test/fixtures.py b/test/fixtures.py
index 84481eda9c9d816e438414636cd335ecb2ef1b63..e642c4128113669cad29693e61c7506d3eafc8fc 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -26,7 +26,6 @@ from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
IptrunkSideBlock,
IptrunkType,
- PhysicalPortCapacity,
)
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
@@ -37,6 +36,7 @@ from gso.products.product_types.router import ImportedRouterInactive, Router, Ro
from gso.products.product_types.site import ImportedSiteInactive, Site, SiteInactive
from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitchInactive, SuperPopSwitchInactive
from gso.services import subscriptions
+from gso.types.interfaces import PhysicalPortCapacity
from gso.utils.shared_enums import Vendor
from test.workflows import WorkflowInstanceForTests
diff --git a/test/schemas/test_types.py b/test/schemas/test_types.py
index a968084f06e2674828b6a59df58484e1fd965851..33c3f20937232b75b92c9db5813462b5f0a5c57c 100644
--- a/test/schemas/test_types.py
+++ b/test/schemas/test_types.py
@@ -1,7 +1,7 @@
import pytest
from pydantic import BaseModel, ValidationError
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
+from gso.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
class LatitudeModel(BaseModel):
diff --git a/test/services/test_librenms_client.py b/test/services/test_librenms_client.py
index e28eaef74742d59d8b443ba0fad1d323f5b26254..1bd14b9b65c06dcc8d1ad3785b575b11de3bfc54 100644
--- a/test/services/test_librenms_client.py
+++ b/test/services/test_librenms_client.py
@@ -5,7 +5,7 @@ import pytest
from requests import HTTPError
from gso.services.librenms_client import LibreNMSClient
-from gso.utils.helpers import SNMPVersion
+from gso.types.snmp import SNMPVersion
@pytest.fixture()
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index dc7854eaa3287b4b98132a8243bf2068636a684a..5f9476096ea070d3c8856e623e3dc1301f001941 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -6,10 +6,10 @@ from orchestrator.types import SubscriptionLifecycle
from gso.products import Router
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
from gso.products.product_blocks.router import RouterRole
+from gso.types.tt_number import validate_tt_number
from gso.utils.helpers import (
available_interfaces_choices_including_current_members,
generate_inventory_for_active_routers,
- validate_tt_number,
)
from gso.utils.shared_enums import Vendor
diff --git a/test/workflows/iptrunk/test_create_imported_iptrunk.py b/test/workflows/iptrunk/test_create_imported_iptrunk.py
index c08ddbe643a122e0ed710cadffdddea70d78add3..263c4f161d99985b48da258d1660925e58c24157 100644
--- a/test/workflows/iptrunk/test_create_imported_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_imported_iptrunk.py
@@ -2,8 +2,9 @@ import pytest
from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_types.iptrunk import ImportedIptrunk
+from gso.types.interfaces import PhysicalPortCapacity
from test.workflows import (
assert_complete,
extract_state,
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 117444b7a33b5725df7558c2a3f7cd67b1cd00f1..cd33943ce75887824d73ac2a810cef2a75edfa04 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -5,8 +5,9 @@ import pytest
from infoblox_client.objects import HostRecord
from gso.products import Iptrunk, ProductName
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
from gso.services.subscriptions import get_product_id_by_name
+from gso.types.interfaces import PhysicalPortCapacity
from gso.utils.shared_enums import Vendor
from test import USER_CONFIRM_EMPTY_FORM
from test.services.conftest import MockedNetboxClient, MockedSharePointClient
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 1383ea2edc1380613900ae62b0024b4622ac6d57..b784d262a75533e5e46945763c76c06e391cd269 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -3,7 +3,8 @@ from unittest.mock import patch
import pytest
from gso.products import Iptrunk
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.types.interfaces import LAGMemberList, PhysicalPortCapacity
from gso.utils.shared_enums import Vendor
from test.conftest import UseJuniperSide
from test.workflows import (
@@ -172,7 +173,7 @@ def test_iptrunk_modify_trunk_interface_success(
assert subscription.iptrunk.iptrunk_minimum_links == input_form_iptrunk_data[1]["iptrunk_number_of_members"] - 1
assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid == new_side_a_sid
- def _find_interface_by_name(interfaces: list[dict[str, str]], name: str):
+ def _find_interface_by_name(interfaces: LAGMemberList, name: str):
for interface in interfaces:
if interface.interface_name == name:
return interface