Add EdgePort product block and product type.

gso/products/__init__.py

@@ -8,6 +8,7 @@
 from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
 from pydantic_forms.types import strEnum
 
+from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.iptrunk import ImportedIptrunk, Iptrunk
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.products.product_types.office_router import ImportedOfficeRouter, OfficeRouter
@@ -37,6 +38,7 @@ class ProductName(strEnum):
     IMPORTED_OFFICE_ROUTER = "Imported office router"
     OPENGEAR = "Opengear"
     IMPORTED_OPENGEAR = "Imported Opengear"
+    EDGE_PORT = "Edge port"
 
 
 class ProductType(strEnum):
@@ -57,6 +59,7 @@ class ProductType(strEnum):
     IMPORTED_OFFICE_ROUTER = ImportedOfficeRouter.__name__
     OPENGEAR = Opengear.__name__
     IMPORTED_OPENGEAR = Opengear.__name__
+    EDGE_PORT = EdgePort.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -76,5 +79,6 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.IMPORTED_OFFICE_ROUTER.value: ImportedOfficeRouter,
         ProductName.OPENGEAR.value: Opengear,
         ProductName.IMPORTED_OPENGEAR.value: ImportedOpengear,
+        ProductName.EDGE_PORT.value: EdgePort,
     },
 )

gso/products/product_blocks/edge_port.py

+"""Edge port product block.
+
+Edge port sets the boundary between Geant network and an external entity that could also be a different technological
+domain still managed by GEANT. In other words, an Edge port determines where the network ends.
+"""
+
+from typing import Annotated
+
+from annotated_types import Len
+from orchestrator.domain.base import ProductBlockModel, T
+from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import AfterValidator
+from pydantic_forms.validators import validate_unique_list
+from typing_extensions import Doc
+
+from gso.products.product_blocks.iptrunk import PhysicalPortCapacity
+from gso.products.product_blocks.router import RouterBlockInactive
+
+LAGMemberList = Annotated[
+    list[T], AfterValidator(validate_unique_list), Len(min_length=0), Doc("A list of :term:`LAG` member interfaces.")
+]
+
+
+class EdgePortInterfaceBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="EdgePortInterfaceBlock",
+):
+    """An inactive edge port interface that's currently inactive."""
+
+    interface_name: str | None = None
+    interface_description: str | None = None
+
+
+class EdgePortInterfaceBlockProvisioning(
+    EdgePortInterfaceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """An IP trunk interface that is being provisioned."""
+
+    interface_name: str
+    interface_description: str | None = None
+
+
+class EdgePortInterfaceBlock(EdgePortInterfaceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active edge port interface."""
+
+    interface_name: str
+    interface_description: str | None = None
+
+
+class EncapsulationType(strEnum):
+    """Enum representing different Ethernet encapsulation service options.
+
+    Null supports a single service on the port.
+    Dot1Q supports multiple services for one customer or services for multiple customers.
+    QinQ expands VLAN space by double-tagging frames.
+    """
+
+    DOT1Q = "dot1q"
+    QINQ = "qinq"
+    NULL = "null"
+
+
+class EdgePortType(strEnum):
+    """Types of edge ports."""
+
+    CUSTOMER = "CUSTOMER"
+    INFRASTRUCTURE = "INFRASTRUCTURE"
+    PRIVATE = "PRIVATE"
+    PUBLIC = "PUBLIC"
+    RE_INTERCONNECT = "RE_INTERCONNECT"
+
+
+class EdgePortBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="EdgePortBlock"
+):
+    """An edge port that's currently inactive. See :class:`EdgePortBlock`."""
+
+    node: RouterBlockInactive
+    edge_port_name: str
+    enable_lacp: bool
+    edge_port_encapsulation: EncapsulationType = EncapsulationType.DOT1Q
+    edge_port_mac_address: str | None = None
+    edge_port_member_speed: PhysicalPortCapacity
+    edge_port_minimum_links: int | None = None
+    edge_port_type: EdgePortType
+    edge_port_ignore_if_down: bool = False
+    edge_port_geant_ga_id: str | None = None
+    edge_port_ae_iface: str | None = None
+    edge_port_ae_members: LAGMemberList[EdgePortInterfaceBlockInactive]
+
+
+class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An edge port that's being provisioned. See :class:`EdgePortBlock`."""
+
+    node: RouterBlockInactive
+    edge_port_name: str
+    edge_port_enable_lacp: bool
+    edge_port_encapsulation: EncapsulationType = EncapsulationType.DOT1Q
+    edge_port_mac_address: str | None = None
+    edge_port_member_speed: PhysicalPortCapacity
+    edge_port_minimum_links: int | None = None
+    edge_port_type: EdgePortType
+    edge_port_ignore_if_down: bool = False
+    edge_port_geant_ga_id: str | None = None
+    edge_port_ae_iface: str | None = None
+    edge_port_ae_members: LAGMemberList[EdgePortInterfaceBlockProvisioning]  # type: ignore[assignment]
+
+
+class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An edge port that's currently deployed in the network."""
+
+    #: The router that this edge port is connected to.
+    node: RouterBlockInactive
+    #: The name of the edge port.
+    edge_port_name: str
+    #: Indicates whether LACP (Link Aggregation Control Protocol) is enabled for this edge port.
+    edge_port_enable_lacp: bool
+    #: The type of encapsulation used on this edge port, by default DOT1Q.
+    edge_port_encapsulation: EncapsulationType = EncapsulationType.DOT1Q
+    #: The MAC address assigned to this edge port, if applicable.
+    edge_port_mac_address: str | None = None
+    #: The speed capacity of each member in the physical port.
+    edge_port_member_speed: PhysicalPortCapacity
+    #: The minimum number of links required for this edge port.
+    edge_port_minimum_links: int | None = None
+    #: The type of edge port (e.g., access, trunk).
+    edge_port_type: EdgePortType
+    #: If set to True, the edge port will be ignored if it is down.
+    edge_port_ignore_if_down: bool = False
+    #: The GEANT GA ID associated with this edge port, if any.
+    edge_port_geant_ga_id: str | None = None
+    #: The interface name for the aggregated Ethernet (AE) interface, if applicable.
+    edge_port_ae_iface: str | None = None
+    #: A list of LAG members associated with this edge port.
+    edge_port_ae_members: LAGMemberList[EdgePortInterfaceBlock]  # type: ignore[assignment]

gso/products/product_types/edge_port.py

+"""Product types for Edge Port."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.edge_port import (
+    EdgePortBlock,
+    EdgePortBlockInactive,
+    EdgePortBlockProvisioning,
+)
+
+
+class EdgePortInactive(SubscriptionModel, is_base=True):
+    """An Edge Port that is inactive."""
+
+    edge_port: EdgePortBlockInactive
+
+
+class EdgePortProvisioning(EdgePortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An Edge Port that is being provisioned."""
+
+    edge_port: EdgePortBlockProvisioning
+
+
+class EdgePort(EdgePortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An Edge Port that is active."""
+
+    edge_port: EdgePortBlock