Product block and type - 10GGBS - WIP

gso/products/__init__.py

@@ -21,6 +21,7 @@ from gso.products.product_types.site import ImportedSite, Site
 from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitch, SuperPopSwitch
 from gso.products.product_types.switch import ImportedSwitch, Switch
 from gso.products.product_types.vrf import VRF
+from gso.products.product_types.tenggbs import TenGGBS, ImportedTenGGBS
 
 
 class ProductName(strEnum):
@@ -78,6 +79,9 @@ class ProductName(strEnum):
     IMPORTED_EXPRESSROUTE = Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE
     VRF = "VRF"
     """VRFs."""
+    TenGGBS = "TenGGBS"
+    """10GGBS."""
+    IMPORTED_TenGGBS = "Imported TenGGBS"
 
 
 L3_CORE_SERVICE_PRODUCT_TYPE = L3CoreService.__name__
@@ -121,6 +125,8 @@ class ProductType(strEnum):
     EXPRESSROUTE = L2_CIRCUIT_PRODUCT_TYPE
     IMPORTED_EXPRESSROUTE = ImportedLayer2Circuit.__name__
     VRF = VRF.__name__
+    TenGGBS = TenGGBS.__name__
+    IMPORTED_TenGGBS = ImportedTenGGBS.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -159,6 +165,8 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.EXPRESSROUTE.value: Layer2Circuit,
         ProductName.IMPORTED_EXPRESSROUTE.value: ImportedLayer2Circuit,
         ProductName.VRF.value: VRF,
+        ProductName.TenGGBS.value: TenGGBS,
+        ProductName.IMPORTED_TenGGBS.value: ImportedTenGGBS,
     },
 )
 

gso/products/product_blocks/tenggbs.py

+"""10GGBS product block models."""
+from typing import Annotated, TypeVar
+
+from annotated_types import Len
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from orchestrator.types import 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 (
+    IptrunkSideBlock,
+    IptrunkSideBlockInactive,
+    IptrunkSideBlockProvisioning,
+)
+from gso.products.product_blocks.layer_2_circuit import (
+    Layer2CircuitBlock,
+    Layer2CircuitBlockInactive,
+    Layer2CircuitBlockProvisioning,
+)
+
+T = TypeVar("T")
+
+
+class PathRole(strEnum):
+    """Defines path role types."""
+    PRIMARY = "Primary"
+    """Primary path."""
+    SECONDARY = "Secondary"
+    """Secondary path."""
+
+
+Paths = Annotated[
+    list[T],
+    AfterValidator(validate_unique_list),
+    Len(min_length=0),
+    Doc("A list of TrunkSideBlocks that make up a path."),
+]
+
+
+class PathBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="PathBlock"
+):
+    """A path representing an ordered list of adjacent trunk sides."""
+
+    role: PathRole | None = None
+    path: Paths[IptrunkSideBlockInactive]
+
+
+class PathBlockProvisioning(PathBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A path that is being provisioned."""
+
+    role: PathRole | None = None
+    path: Paths[IptrunkSideBlockProvisioning]
+
+
+class PathBlock(PathBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active path."""
+
+    role: PathRole
+    path: Paths[IptrunkSideBlock]
+
+
+PathsList = Annotated[
+    list[T],
+    AfterValidator(validate_unique_list),
+    Doc("A list of paths (1 primary, multiple secondary)."),
+]
+
+
+class TenGGBSBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="TenGGBSBlock"
+):
+    """Inactive 10GGBS service model."""
+
+    l2_circuit: Layer2CircuitBlockInactive
+    paths: PathsList[PathBlockInactive] | None = None
+
+
+class TenGGBSBlockProvisioning(TenGGBSBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """Provisioning state of 10GGBS."""
+
+    l2_circuit: Layer2CircuitBlockProvisioning
+    paths: PathsList[PathBlockProvisioning] | None = None
+
+
+class TenGGBSBlock(TenGGBSBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """Active 10GGBS service model."""
+
+    l2_circuit: Layer2CircuitBlock
+    paths: PathsList[PathBlock] | None = None

gso/products/product_types/tenggbs.py

+"""The product type for 10GGBS services."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.tenggbs import (
+    TenGGBSBlock,
+    TenGGBSBlockInactive,
+    TenGGBSBlockProvisioning,
+)
+
+
+class TenGGBSInactive(SubscriptionModel, is_base=True):
+    """A 10GGBS service that is inactive."""
+
+    tenggbs: TenGGBSBlockInactive
+
+
+class TenGGBSProvisioning(TenGGBSInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A 10GGBS service that is being provisioned."""
+
+    tenggbs: TenGGBSBlockProvisioning
+
+
+class TenGGBS(TenGGBSProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A 10GGBS service that is active."""
+
+    tenggbs: TenGGBSBlock
+
+
+class ImportedTenGGBSInactive(SubscriptionModel, is_base=True):
+    """An imported 10GGBS service that is inactive."""
+
+    tenggbs: TenGGBSBlockInactive
+
+
+class ImportedTenGGBS(
+    ImportedTenGGBSInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
+):
+    """An imported 10GGBS service that is active."""
+
+    tenggbs: TenGGBSBlock