move types into their respective product blocks and types

docs/source/glossary.rst

@@ -3,6 +3,9 @@ Glossary of terms
 
 .. glossary::
 
+  API
+    Application Programming Interface
+
   BGP
     Border Gateway Protocol: a path vector routing protocol described in
     `RFC 4271 <https://datatracker.ietf.org/doc/html/rfc4271>`_.

docs/source/module/products/index.rst

 ``gso.products``
 ================
 
+.. automodule:: gso.products
+   :members:
+   :show-inheritance:
+
 Subpackages
 -----------
 

gso/api/v1/imports.py

+import ipaddress
 from typing import Any
 from uuid import UUID
 
 from fastapi import Depends, HTTPException, status
 from fastapi.routing import APIRouter
 from orchestrator.security import opa_security_default
-from orchestrator.services import processes, subscriptions
+from orchestrator.services import processes
+from orchestrator.services import subscriptions as wfo_subscriptions
+from pydantic import BaseModel, root_validator, validator
 from sqlalchemy.exc import MultipleResultsFound
 
-from gso.utils.types.imports import ImportResponseModel, IptrunkImportModel, RouterImportModel, SiteImportModel
+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.workflows.iptrunk.utils import LAGMember
 
 router = APIRouter(prefix="/imports", tags=["Imports"], dependencies=[Depends(opa_security_default)])
 
 
+class ImportResponseModel(BaseModel):
+    pid: UUID
+    detail: str
+
+
+class SiteImportModel(BaseModel):
+    site_name: str
+    site_city: str
+    site_country: str
+    site_country_code: str
+    site_latitude: float
+    site_longitude: float
+    site_bgp_community_id: int
+    site_internal_id: int
+    site_tier: SiteTier
+    site_ts_address: str
+    customer: str
+
+
+class RouterImportModel(BaseModel):
+    customer: str
+    router_site: str
+    hostname: str
+    ts_port: int
+    router_vendor: RouterVendor
+    router_role: RouterRole
+    is_ias_connected: bool
+    router_lo_ipv4_address: ipaddress.IPv4Address
+    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_iso_address: str
+    router_si_ipv4_network: ipaddress.IPv4Network | None = None
+    router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
+    router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
+
+
+class IptrunkImportModel(BaseModel):
+    customer: str
+    geant_s_sid: str
+    iptrunk_type: IptrunkType
+    iptrunk_description: str
+    iptrunk_speed: PhyPortCapacity
+    iptrunk_minimum_links: int
+    side_a_node_id: str
+    side_a_ae_iface: str
+    side_a_ae_geant_a_sid: str
+    side_a_ae_members: list[LAGMember]
+    side_b_node_id: str
+    side_b_ae_iface: str
+    side_b_ae_geant_a_sid: str
+    side_b_ae_members: list[LAGMember]
+
+    iptrunk_ipv4_network: ipaddress.IPv4Network
+    iptrunk_ipv6_network: ipaddress.IPv6Network
+
+    @classmethod
+    def _get_active_routers(cls) -> set[str]:
+        return {
+            str(router_id) for router_id in subscriptions.get_active_router_subscriptions(fields=["subscription_id"])
+        }
+
+    @validator("customer")
+    def check_if_customer_exists(cls, value: str) -> str:
+        try:
+            get_customer_by_name(value)
+        except CustomerNotFoundError:
+            raise ValueError(f"Customer {value} not found")
+
+        return value
+
+    @validator("side_a_node_id", "side_b_node_id")
+    def check_if_router_side_is_available(cls, value: str) -> str:
+        if value not in cls._get_active_routers():
+            raise ValueError(f"Router {value} not found")
+
+        return value
+
+    @validator("side_a_ae_members", "side_b_ae_members")
+    def check_side_uniqueness(cls, value: list[str]) -> list[str]:
+        if len(value) != len(set(value)):
+            raise ValueError("Items must be unique")
+
+        return value
+
+    @root_validator
+    def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
+        min_links = values["iptrunk_minimum_links"]
+        side_a_members = values.get("side_a_ae_members", [])
+        side_b_members = values.get("side_b_ae_members", [])
+
+        len_a = len(side_a_members)
+        len_b = len(side_b_members)
+
+        if len_a < min_links:
+            raise ValueError(f"Side A members should be at least {min_links} (iptrunk_minimum_links)")
+
+        if len_a != len_b:
+            raise ValueError("Mismatch between Side A and B members")
+
+        return values
+
+
 def _start_process(process_name: str, data: dict) -> UUID:
     """Start a process and handle common exceptions."""
 
@@ -42,7 +151,7 @@ def import_site(site: SiteImportModel) -> dict[str, Any]:
     :raises HTTPException: If the site already exists or if there's an error in the process.
     """
     try:
-        subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
+        subscription = wfo_subscriptions.retrieve_subscription_by_subscription_instance_value(
             resource_type="site_name", value=site.site_name, sub_status=("provisioning", "active")
         )
         if subscription:

gso/products/__init__.py

-"""Module that updates the domain model of :term:`GSO`. Should contain all types of subscriptions."""
+"""Module that updates the domain model of :term:`GSO`. Should contain all types of subscriptions.
+
+.. warning::
+   Whenever a new product type is added, this should be reflected in the :py:class:`gso.products.ProductType`
+   enumerator.
+"""
 from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
 from pydantic_forms.types import strEnum
 
@@ -15,8 +20,8 @@ class ProductType(strEnum):
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
     {
-        ProductType.SITE.value: Site,
-        ProductType.ROUTER.value: Router,
-        ProductType.IP_TRUNK.value: Iptrunk,
+        "Site": Site,
+        "Router": Router,
+        "IP trunk": Iptrunk,
     }
 )

gso/products/product_blocks/iptrunk.py

@@ -8,7 +8,18 @@ from orchestrator.forms.validators import UniqueConstrainedList
 from orchestrator.types import SubscriptionLifecycle, strEnum
 
 from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
-from gso.utils.types.phy_port import PhyPortCapacity
+
+
+class PhyPortCapacity(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"
 
 
 class IptrunkType(strEnum):

gso/products/product_blocks/router.py

@@ -3,9 +3,9 @@ import ipaddress
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import ConstrainedInt
 
 from gso.products.product_blocks.site import SiteBlock, SiteBlockInactive, SiteBlockProvisioning
-from gso.utils.types.ip_port import PortNumber
 
 
 class RouterVendor(strEnum):
@@ -23,6 +23,16 @@ class RouterRole(strEnum):
     AMT = "amt"
 
 
+class PortNumber(ConstrainedInt):
+    """Constrained integer for valid port numbers.
+
+    The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
+    """
+
+    gt = 0
+    le = 49151
+
+
 class RouterBlockInactive(
     ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="RouterBlock"
 ):

gso/products/product_blocks/site.py

 """The product block that describes a site subscription."""
+import re
+from typing import Union
+
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
-
-from gso.utils.types.snmp import LatitudeCoordinate, LongitudeCoordinate
+from pydantic import ConstrainedStr
 
 
 class SiteTier(strEnum):
@@ -18,6 +20,42 @@ class SiteTier(strEnum):
     TIER4 = 4
 
 
+class LatitudeCoordinate(ConstrainedStr):
+    """A latitude coordinate, modeled as a constrained 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
+    """
+
+    regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
+
+    @classmethod
+    def validate(cls, value: Union[str]) -> Union[str]:
+        if not cls.regex.match(value):
+            raise ValueError("Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'.")
+
+        return value
+
+
+class LongitudeCoordinate(ConstrainedStr):
+    """A longitude coordinate, modeled as a constrained 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
+    """
+
+    regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
+
+    @classmethod
+    def validate(cls, value: Union[str]) -> Union[str]:
+        if not cls.regex.match(value):
+            raise ValueError("Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180'")
+
+        return value
+
+
 class SiteBlockInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="SiteBlock"):
     """A site that's currently inactive, see :class:`SiteBlock`."""
 

gso/utils/types/imports.py

-import ipaddress
-from typing import Any
-from uuid import UUID
-
-from pydantic import BaseModel, root_validator, validator
-
-from gso.products.product_blocks.iptrunk import IptrunkType
-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.types.phy_port import PhyPortCapacity
-from gso.workflows.iptrunk.utils import LAGMember
-
-
-class ImportResponseModel(BaseModel):
-    pid: UUID
-    detail: str
-
-
-class SiteImportModel(BaseModel):
-    site_name: str
-    site_city: str
-    site_country: str
-    site_country_code: str
-    site_latitude: float
-    site_longitude: float
-    site_bgp_community_id: int
-    site_internal_id: int
-    site_tier: SiteTier
-    site_ts_address: str
-    customer: str
-
-
-class RouterImportModel(BaseModel):
-    customer: str
-    router_site: str
-    hostname: str
-    ts_port: int
-    router_vendor: RouterVendor
-    router_role: RouterRole
-    is_ias_connected: bool
-    router_lo_ipv4_address: ipaddress.IPv4Address
-    router_lo_ipv6_address: ipaddress.IPv6Address
-    router_lo_iso_address: str
-    router_si_ipv4_network: ipaddress.IPv4Network | None = None
-    router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
-    router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
-
-
-class IptrunkImportModel(BaseModel):
-    customer: str
-    geant_s_sid: str
-    iptrunk_type: IptrunkType
-    iptrunk_description: str
-    iptrunk_speed: PhyPortCapacity
-    iptrunk_minimum_links: int
-    side_a_node_id: str
-    side_a_ae_iface: str
-    side_a_ae_geant_a_sid: str
-    side_a_ae_members: list[LAGMember]
-    side_b_node_id: str
-    side_b_ae_iface: str
-    side_b_ae_geant_a_sid: str
-    side_b_ae_members: list[LAGMember]
-
-    iptrunk_ipv4_network: ipaddress.IPv4Network
-    iptrunk_ipv6_network: ipaddress.IPv6Network
-
-    @classmethod
-    def _get_active_routers(cls) -> set[str]:
-        return {
-            str(router_id) for router_id in subscriptions.get_active_router_subscriptions(fields=["subscription_id"])
-        }
-
-    @validator("customer")
-    def check_if_customer_exists(cls, value: str) -> str:
-        try:
-            get_customer_by_name(value)
-        except CustomerNotFoundError:
-            raise ValueError(f"Customer {value} not found")
-
-        return value
-
-    @validator("side_a_node_id", "side_b_node_id")
-    def check_if_router_side_is_available(cls, value: str) -> str:
-        if value not in cls._get_active_routers():
-            raise ValueError("Router not found")
-
-        return value
-
-    @validator("side_a_ae_members", "side_b_ae_members")
-    def check_side_uniqueness(cls, value: list[str]) -> list[str]:
-        if len(value) != len(set(value)):
-            raise ValueError("Items must be unique")
-
-        return value
-
-    @root_validator
-    def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
-        min_links = values["iptrunk_minimum_links"]
-        side_a_members = values.get("side_a_ae_members", {})
-        side_b_members = values.get("side_b_ae_members", {})
-
-        len_a = len(side_a_members)
-        len_b = len(side_b_members)
-
-        if len_a < min_links:
-            raise ValueError(f"Side A members should be at least {min_links} (iptrunk_minimum_links)")
-
-        if len_a != len_b:
-            raise ValueError("Mismatch between Side A and B members")
-
-        return values

gso/utils/types/ip_port.py

-from pydantic import ConstrainedInt
-
-
-class PortNumber(ConstrainedInt):
-    """Constrained integer for valid port numbers.
-
-    The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
-    """
-
-    gt = 0
-    le = 49151

gso/utils/types/phy_port.py

-from orchestrator.types import strEnum
-
-
-class PhyPortCapacity(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"

gso/utils/types/snmp.py

-import re
-from typing import Union
-
-from pydantic import ConstrainedStr
-
-
-class LatitudeCoordinate(ConstrainedStr):
-    """A latitude coordinate, modeled as a constrained 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
-    """
-
-    regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
-
-    @classmethod
-    def validate(cls, value: Union[str]) -> Union[str]:
-        if not cls.regex.match(value):
-            raise ValueError("Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'.")
-
-        return value
-
-
-class LongitudeCoordinate(ConstrainedStr):
-    """A longitude coordinate, modeled as a constrained 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
-    """
-
-    regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
-
-    @classmethod
-    def validate(cls, value: Union[str]) -> Union[str]:
-        if not cls.regex.match(value):
-            raise ValueError("Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180'")
-
-        return value

gso/workflows/iptrunk/create_iptrunk.py

@@ -10,7 +10,7 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
 from pynetbox.models.dcim import Interfaces
 
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhyPortCapacity
 from gso.products.product_blocks.router import RouterVendor
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -9,11 +9,10 @@ from orchestrator.workflow import StepList, done, init, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkType
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkType, PhyPortCapacity
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import provisioning_proxy
 from gso.services.provisioning_proxy import pp_interaction
-from gso.utils.types.phy_port import PhyPortCapacity
 from gso.workflows.iptrunk.utils import LAGMember
 
 

gso/workflows/router/create_router.py

@@ -11,15 +11,14 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
 
-from gso.products.product_blocks.router import RouterRole, RouterVendor, generate_fqdn
+from gso.products.product_blocks.router import PortNumber, RouterRole, RouterVendor, generate_fqdn
 from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site
 from gso.services import infoblox, provisioning_proxy, subscriptions
 from gso.services.crm import customer_selector
 from gso.services.netbox_client import NetboxClient
 from gso.services.provisioning_proxy import pp_interaction
-from gso.utils.functions import iso_from_ipv4
-from gso.utils.types.ip_port import PortNumber
+from gso.utils.helpers import iso_from_ipv4
 
 
 def _site_selector() -> Choice:

gso/workflows/site/create_site.py

@@ -11,9 +11,9 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
 
 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.types.snmp import LatitudeCoordinate, LongitudeCoordinate
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:  # noqa: C901

gso/workflows/tasks/import_iptrunk.py

@@ -9,11 +9,10 @@ from orchestrator.workflow import StepList, done, init, step
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 
 from gso.products import ProductType
-from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.services import subscriptions
 from gso.services.crm import get_customer_by_name
-from gso.utils.types.phy_port import PhyPortCapacity
 from gso.workflows.iptrunk.create_iptrunk import initialize_subscription
 from gso.workflows.iptrunk.utils import LAGMember
 

gso/workflows/tasks/import_router.py

@@ -10,13 +10,12 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
 
 from gso.products import ProductType
 from gso.products.product_blocks import router as router_pb
-from gso.products.product_blocks.router import RouterRole, RouterVendor
+from gso.products.product_blocks.router import PortNumber, RouterRole, RouterVendor
 from gso.products.product_types import router
 from gso.products.product_types.router import RouterInactive
 from gso.products.product_types.site import Site
 from gso.services import subscriptions
 from gso.services.crm import get_customer_by_name
-from gso.utils.types.ip_port import PortNumber
 
 
 def _get_site_by_name(site_name: str) -> Site:

test/fixtures.py

@@ -6,14 +6,13 @@ from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
 from gso.products import ProductType
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkSideBlock, IptrunkType
+from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkSideBlock, IptrunkType, PhyPortCapacity
 from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_blocks.site import SiteTier
 from gso.products.product_types.iptrunk import IptrunkInactive
 from gso.products.product_types.router import Router, RouterInactive
 from gso.products.product_types.site import Site, SiteInactive
 from gso.services import subscriptions
-from gso.utils.types.phy_port import PhyPortCapacity
 
 CUSTOMER_ID: UUIDstr = "2f47f65a-0911-e511-80d0-005056956c1a"