Add GÉANT IP product

gso/__init__.py

 """The main entrypoint for :term:`GSO`, and the different ways in which it can be run."""
 
 import os
+from typing import Annotated
 
 import sentry_sdk
+import strawberry
 import typer
 from celery import Celery
 from orchestrator import OrchestratorCore, app_settings
 from orchestrator.cli.main import app as cli_app
-from orchestrator.graphql import SCALAR_OVERRIDES
+from orchestrator.graphql import DEFAULT_GRAPHQL_MODELS, SCALAR_OVERRIDES
 from orchestrator.services.tasks import initialise_celery
 from orchestrator.settings import ExecutorType
 
@@ -19,6 +21,7 @@ from gso.auth.oidc import oidc_instance
 from gso.auth.opa import graphql_opa_instance, opa_instance
 from gso.graphql_api.types import GSO_SCALAR_OVERRIDES
 from gso.settings import load_oss_params
+from gso.utils.types.interfaces import LAGMember
 
 SCALAR_OVERRIDES.update(GSO_SCALAR_OVERRIDES)
 
@@ -35,7 +38,18 @@ def init_gso_app() -> OrchestratorCore:
     app.register_authentication(oidc_instance)
     app.register_authorization(opa_instance)
     app.register_graphql_authorization(graphql_opa_instance)
-    app.register_graphql()
+
+    @strawberry.experimental.pydantic.type(model=LAGMember)
+    class LAGMemberGraphql:
+        self_reference_block: Annotated["LAGMemberGraphql", strawberry.lazy("gso.utils.types.interfaces")] | None = None
+        interface_name: str
+        interface_description: str
+
+    updated_graphql_models = DEFAULT_GRAPHQL_MODELS | {
+        "LAGMemberGraphql": LAGMemberGraphql,
+    }
+
+    app.register_graphql(graphql_models=updated_graphql_models)
     app.include_router(api_router, prefix="/api")
 
     if app_settings.EXECUTOR == ExecutorType.WORKER:

gso/products/__init__.py

@@ -9,6 +9,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.geant_ip import GeantIP
 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
@@ -39,6 +40,7 @@ class ProductName(strEnum):
     OPENGEAR = "Opengear"
     IMPORTED_OPENGEAR = "Imported Opengear"
     EDGE_PORT = "Edge port"
+    GEANT_IP = "GEANT IP"
 
 
 class ProductType(strEnum):
@@ -60,6 +62,7 @@ class ProductType(strEnum):
     OPENGEAR = Opengear.__name__
     IMPORTED_OPENGEAR = Opengear.__name__
     EDGE_PORT = EdgePort.__name__
+    GEANT_IP = GeantIP.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -80,5 +83,6 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.OPENGEAR.value: Opengear,
         ProductName.IMPORTED_OPENGEAR.value: ImportedOpengear,
         ProductName.EDGE_PORT.value: EdgePort,
+        ProductType.GEANT_IP.value: GeantIP,
     },
 )

gso/products/product_blocks/bgp_session.py

@@ -4,6 +4,7 @@ from ipaddress import IPv4Address, IPv6Address
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
+from pydantic import Field
 from pydantic_forms.types import strEnum
 
 
@@ -23,7 +24,7 @@ class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INI
     bfd_enabled: bool | None = None
     bfd_interval: int | None = None
     bfd_multiplier: int | None = None
-    families: list[IPFamily] | None = None
+    families: list[IPFamily] = Field(default_factory=list)
     has_custom_policies: bool | None = None
     authentication_key: str | None = None
     multipath_enabled: bool | None = None

gso/products/product_blocks/geant_ip.py

+"""Product blocks for :class:`GeantIP` products."""
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic import Field
+
+from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
+from gso.products.product_blocks.service_binding_port import (
+    ServiceBindingPort,
+    ServiceBindingPortInactive,
+    ServiceBindingPortProvisioning,
+)
+from gso.utils.shared_enums import APType
+
+
+class NRENAccessPortInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="NRENAccessPort"
+):
+    """An access port for an R&E :term:`NREN` service that is inactive."""
+
+    nren_ap_sbp_list: list[ServiceBindingPortInactive] = Field(default_factory=list)
+    nren_ap_type: APType | None = None
+
+
+class NRENAccessPortProvisioning(NRENAccessPortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An access port for an R&E :term:`NREN` service that is being provisioned."""
+
+    nren_ap_sbp_list: list[ServiceBindingPortProvisioning]  # type: ignore[assignment]
+    nren_ap_type: APType
+
+
+class NRENAccessPort(NRENAccessPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An access port for an R&E :term:`NREN` service."""
+
+    nren_ap_sbp_list: list[ServiceBindingPort]  # type: ignore[assignment]
+    nren_ap_type: APType
+
+
+class GeantIPBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="GeantIPBlock"
+):
+    """A GÉANT IP subscription that is currently inactive. See :class:`GeantIPBlock`."""
+
+    geant_ip_ap_list: list[NRENAccessPortInactive] = Field(default_factory=list)
+    geant_ip_ep_list: list[EdgePortBlockInactive] = Field(default_factory=list)
+
+
+class GeantIPBlockProvisioning(GeantIPBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A GÉANT IP subscription that is currently being provisioned. See :class:`GeantIPBlock`."""
+
+    geant_ip_ap_list: list[NRENAccessPortProvisioning]  # type: ignore[assignment]
+    geant_ip_ep_list: list[EdgePortBlockProvisioning]  # type: ignore[assignment]
+
+
+class GeantIPBlock(GeantIPBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A GÉANT IP subscription block."""
+
+    #: The list of Access Points where this service is present.
+    geant_ip_ap_list: list[NRENAccessPort]  # type: ignore[assignment]
+    #: The list of Edge Ports where this service terminates.
+    geant_ip_ep_list: list[EdgePortBlock]  # type: ignore[assignment]

gso/products/product_blocks/service_binding_port.py

@@ -6,45 +6,42 @@ A service binding port is used to logically attach an edge port to a customer se
 from ipaddress import IPv4Address, IPv6Address
 from typing import Annotated
 
-from orchestrator.domain.base import ProductModel
+from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import Field
-from pydantic_forms.types import strEnum
 
-VLANID = Annotated[int, Field(gt=0, lt=4096)]
+from gso.utils.shared_enums import SBPType
+from gso.utils.types.bgp_session import BGPSessionSet
 
-
-class SBPType(strEnum):
-    """Enumerator for the two allowed types of service binding port: layer 2 or layer 3."""
-
-    L2 = "l2"
-    L3 = "l3"
+VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
 
 
 class ServiceBindingPortInactive(
-    ProductModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="ServiceBindingPort"
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="ServiceBindingPort"
 ):
     """A Service Binding Port that's currently inactive. See :class:`ServiceBindingPort`."""
 
     is_tagged: bool | None = None
-    vlan_id: VLANID | None = None
+    vlan_id: VLAN_ID | None = None
     sbp_type: SBPType | None = None
     ipv4_address: IPv4Address | None = None
     ipv6_address: IPv6Address | None = None
     custom_firewall_filters: bool | None = None
     geant_sid: str | None = None
+    sbp_bgp_session_list: BGPSessionSet | None = None
 
 
 class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """A Service Binding Port that's currently being provisioned. See :class:`ServiceBindingPort`."""
 
     is_tagged: bool
-    vlan_id: VLANID | None = None
+    vlan_id: VLAN_ID | None = None
     sbp_type: SBPType
     ipv4_address: IPv4Address | None = None
     ipv6_address: IPv6Address | None = None
     custom_firewall_filters: bool
     geant_sid: str
+    sbp_bgp_session_list: BGPSessionSet
 
 
 class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -53,7 +50,7 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
     #: Whether this :term:`VLAN` is tagged or not.
     is_tagged: bool
     #: The :term:`VLAN` ID.
-    vlan_id: VLANID | None = None
+    vlan_id: VLAN_ID | None = None
     #: Is this service binding port layer 2 or 3?
     sbp_type: SBPType
     #: If layer 3, IPv4 resources.
@@ -64,3 +61,5 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
     custom_firewall_filters: bool
     #: The GÉANT service ID of this binding port.
     geant_sid: str
+    #: The :term:`BGP` sessions associated with this service binding port.
+    sbp_bgp_session_list: BGPSessionSet

gso/products/product_types/geant_ip.py

+"""GÉANT IP product type."""
+
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.geant_ip import GeantIPBlock, GeantIPBlockInactive, GeantIPBlockProvisioning
+
+
+class GeantIPInactive(SubscriptionModel, is_base=True):
+    """An inactive GÉANT IP subscription."""
+
+    geant_ip: GeantIPBlockInactive
+
+
+class GeantIPProvisioning(GeantIPInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A GÉANT IP subscription that's being provisioned."""
+
+    geant_ip: GeantIPBlockProvisioning
+
+
+class GeantIP(GeantIPProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active GÉANT IP subscription."""
+
+    geant_ip: GeantIPBlock
+
+
+class ImportedGeantIPInactive(SubscriptionModel, is_base=True):
+    """An imported, inactive GÉANT IP subscription."""
+
+    geant_ip: GeantIPBlockInactive
+
+
+class ImportedGeantIP(
+    ImportedGeantIPInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
+):
+    """An imported GÉANT IP subscription."""
+
+    geant_ip: GeantIPBlock

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.types.snmp import SNMPVersion
+from gso.utils.shared_enums import SNMPVersion
 
 logger = logging.getLogger(__name__)
 

gso/utils/shared_enums.py

 """Shared choices for the different models."""
 
+from enum import StrEnum
+
 from pydantic_forms.types import strEnum
 
 
@@ -15,3 +17,25 @@ class ConnectionStrategy(strEnum):
 
     IN_BAND = "IN BAND"
     OUT_OF_BAND = "OUT OF BAND"
+
+
+class SNMPVersion(StrEnum):
+    """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
+
+    V2C = "v2c"
+    V3 = "v3"
+
+
+class APType(strEnum):
+    """Enumerator of the types of Access Port."""
+
+    PRIMARY = "PRIMARY"
+    BACKUP = "BACKUP"
+    LOAD_BALANCED = "LOAD_BALANCED"
+
+
+class SBPType(strEnum):
+    """Enumerator for the two allowed types of service binding port: layer 2 or layer 3."""
+
+    L2 = "l2"
+    L3 = "l3"

gso/utils/types/bgp_session.py

+""":term:`BGP` session sets."""
+
+from typing import Annotated, TypeVar
+
+from annotated_types import Len
+from pydantic import AfterValidator
+
+from gso.products.product_blocks.bgp_session import BGPSession, BGPSessionInactive, BGPSessionProvisioning
+
+BGPSessionTypes = TypeVar("BGPSessionTypes", BGPSessionInactive, BGPSessionProvisioning, BGPSession)
+
+
+def validate_bgp_session_set(bgp_session_set: list[BGPSessionTypes]) -> list[BGPSessionTypes]:
+    """:term:`BGP` sessions grouped together.
+
+    It consists of either a single item, or a pair of IPv4 and v6 sessions. It is not allowed to have two IPv4 or IPv6
+    :term:`BGP` sessions as part of a set.
+    """
+    if any(bgp_session.peer_address is None for bgp_session in bgp_session_set):
+        msg = "BGP session is missing a peer address."
+        raise ValueError(msg)
+    if len(bgp_session_set) == 2 and bgp_session_set[0].peer_address.version == bgp_session_set[1].peer_address.version:  # type: ignore[union-attr] # noqa: PLR2004
+        msg = (
+            "When defining two separate BGP sessions, IP families must differ. "
+            f"Both are IPv{bgp_session_set[0].peer_address.version}."  # type: ignore[union-attr]
+        )
+        raise ValueError(msg)
+    return bgp_session_set
+
+
+BGPSessionSet = Annotated[
+    list[BGPSessionTypes], Len(min_length=1, max_length=2), AfterValidator(validate_bgp_session_set)
+]

gso/utils/types/snmp.py

-"""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"

gso/workflows/router/update_ibgp_mesh.py

@@ -18,7 +18,7 @@ from gso.services import librenms_client
 from gso.services.lso_client import LSOState, lso_interaction
 from gso.services.subscriptions import get_trunks_that_terminate_on_router
 from gso.utils.helpers import generate_inventory_for_active_routers
-from gso.utils.types.snmp import SNMPVersion
+from gso.utils.shared_enums import SNMPVersion
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.workflow_steps import (
     add_all_p_to_pe_dry,

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.types.snmp import SNMPVersion
+from gso.utils.shared_enums import SNMPVersion
 
 
 @pytest.fixture()