Update edge port models, add translations, add domain model migration for GÉANT IP

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 DEFAULT_GRAPHQL_MODELS, SCALAR_OVERRIDES
+from orchestrator.graphql import SCALAR_OVERRIDES
 from orchestrator.services.tasks import initialise_celery
 from orchestrator.settings import ExecutorType
 
@@ -21,7 +19,6 @@ 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)
 
@@ -38,18 +35,7 @@ def init_gso_app() -> OrchestratorCore:
     app.register_authentication(oidc_instance)
     app.register_authorization(opa_instance)
     app.register_graphql_authorization(graphql_opa_instance)
-
-    @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.register_graphql()
     app.include_router(api_router, prefix="/api")
 
     if app_settings.EXECUTOR == ExecutorType.WORKER:

gso/products/product_blocks/bgp_session.py

 """:term:`BGP` session product block."""
 
-from ipaddress import IPv4Address, IPv6Address
-
+import strawberry
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import Field
 from pydantic_forms.types import strEnum
 
+from gso.utils.types.ip_address import IPAddress
+
 
+@strawberry.enum
 class IPFamily(strEnum):
     """Possible :term:`IP` families of a :term:`BGP` peering."""
 
@@ -20,7 +22,7 @@ class IPFamily(strEnum):
 class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="BGPSession"):
     """A :term:`BGP` session that is currently inactive. See :class:`BGPSession`."""
 
-    peer_address: IPv4Address | IPv6Address | None = None
+    peer_address: IPAddress | None = None
     bfd_enabled: bool | None = None
     bfd_interval: int | None = None
     bfd_multiplier: int | None = None
@@ -35,7 +37,7 @@ class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INI
 class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """A :term:`BGP` session that is currently being provisioned. See :class:`BGPSession`."""
 
-    peer_address: IPv4Address | IPv6Address
+    peer_address: IPAddress
     bfd_enabled: bool
     bfd_interval: int | None = None
     bfd_multiplier: int | None = None
@@ -51,7 +53,7 @@ class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE
     """A :term:`BGP` session that is currently deployed in the network."""
 
     #: The peering address of the session.
-    peer_address: IPv4Address | IPv6Address
+    peer_address: IPAddress
     #: Whether :term:`BFD` is enabled.
     bfd_enabled: bool
     #: The :term:`BFD` interval, if enabled.

gso/products/product_blocks/edge_port.py

@@ -6,9 +6,15 @@ domain still managed by GEANT. In other words, an Edge port determines where the
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import Field
 
 from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
-from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.products.product_blocks.service_binding_port import (
+    ServiceBindingPort,
+    ServiceBindingPortInactive,
+    ServiceBindingPortProvisioning,
+)
+from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
 
 
 class EncapsulationType(strEnum):
@@ -34,6 +40,29 @@ class EdgePortType(strEnum):
     RE_INTERCONNECT = "RE_INTERCONNECT"
 
 
+class EdgePortAEMemberBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="EdgePortAEMemberBlock"
+):
+    """An inactive Edge Port AE interface."""
+
+    interface_name: str | None = None
+    interface_description: str | None = None
+
+
+class EdgePortAEMemberBlockProvisioning(EdgePortAEMemberBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A provisional Edge Port AE interface."""
+
+    interface_name: str
+    interface_description: str | None = None
+
+
+class EdgePortAEMemberBlock(EdgePortAEMemberBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An Edge Port AE interface."""
+
+    interface_name: str
+    interface_description: str | None = None
+
+
 class EdgePortBlockInactive(
     ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="EdgePortBlock"
 ):
@@ -50,7 +79,8 @@ class EdgePortBlockInactive(
     edge_port_type: EdgePortType | None = None
     edge_port_ignore_if_down: bool = False
     edge_port_geant_ga_id: str | None = None
-    edge_port_ae_members: LAGMemberList[LAGMember]
+    edge_port_ae_members: LAGMemberList[EdgePortAEMemberBlockInactive]
+    edge_port_sbp_list: list[ServiceBindingPortInactive] = Field(default_factory=list)
 
 
 class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -67,7 +97,8 @@ class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLi
     edge_port_type: EdgePortType
     edge_port_ignore_if_down: bool = False
     edge_port_geant_ga_id: str | None = None
-    edge_port_ae_members: LAGMemberList[LAGMember]
+    edge_port_ae_members: LAGMemberList[EdgePortAEMemberBlockProvisioning]  # type: ignore[assignment]
+    edge_port_sbp_list: list[ServiceBindingPortProvisioning]  # type: ignore[assignment]
 
 
 class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -96,4 +127,6 @@ class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.
     #: The GEANT GA ID associated with this edge port, if any.
     edge_port_geant_ga_id: str | None = None
     #: A list of LAG members associated with this edge port.
-    edge_port_ae_members: LAGMemberList[LAGMember]
+    edge_port_ae_members: LAGMemberList[EdgePortAEMemberBlock]  # type: ignore[assignment]
+    #: A list of Service Binding Ports associated with this Edge Port
+    edge_port_sbp_list: list[ServiceBindingPort]  # type: ignore[assignment]

gso/products/product_blocks/geant_ip.py

@@ -5,11 +5,6 @@ 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
 
 
@@ -18,22 +13,24 @@ class NRENAccessPortInactive(
 ):
     """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
+    geant_ip_ep_list: list[EdgePortBlockInactive] = Field(default_factory=list)
 
 
 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
+    geant_ip_ep_list: list[EdgePortBlockProvisioning]  # type: ignore[assignment]
 
 
 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]
+    #: The type of Access Port
     nren_ap_type: APType
+    #: The list of Edge Ports where this service terminates.
+    geant_ip_ep_list: list[EdgePortBlock]  # type: ignore[assignment]
 
 
 class GeantIPBlockInactive(
@@ -42,14 +39,12 @@ class GeantIPBlockInactive(
     """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]):
@@ -57,5 +52,3 @@ class GeantIPBlock(GeantIPBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
 
     #: 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

@@ -3,15 +3,15 @@
 A service binding port is used to logically attach an edge port to a customer service using a :term:`VLAN`.
 """
 
-from ipaddress import IPv4Address, IPv6Address
 from typing import Annotated
 
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import Field
 
+from gso.products.product_blocks.bgp_session import BGPSession, BGPSessionInactive, BGPSessionProvisioning
 from gso.utils.shared_enums import SBPType
-from gso.utils.types.bgp_session import BGPSessionSet
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 
 VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
 
@@ -24,11 +24,11 @@ class ServiceBindingPortInactive(
     is_tagged: bool | None = None
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType | None = None
-    ipv4_address: IPv4Address | None = None
-    ipv6_address: IPv6Address | None = None
+    ipv4_address: IPv4AddressType | None = None
+    ipv6_address: IPv6AddressType | None = None
     custom_firewall_filters: bool | None = None
     geant_sid: str | None = None
-    sbp_bgp_session_list: BGPSessionSet | None = None
+    sbp_bgp_session_list: list[BGPSessionInactive] = Field(default_factory=list)
 
 
 class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -37,11 +37,11 @@ class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[Subs
     is_tagged: bool
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType
-    ipv4_address: IPv4Address | None = None
-    ipv6_address: IPv6Address | None = None
+    ipv4_address: IPv4AddressType | None = None
+    ipv6_address: IPv6AddressType | None = None
     custom_firewall_filters: bool
     geant_sid: str
-    sbp_bgp_session_list: BGPSessionSet
+    sbp_bgp_session_list: list[BGPSessionProvisioning]  # type: ignore[assignment]
 
 
 class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -54,12 +54,12 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
     #: Is this service binding port layer 2 or 3?
     sbp_type: SBPType
     #: If layer 3, IPv4 resources.
-    ipv4_address: IPv4Address | None = None
+    ipv4_address: IPv4AddressType | None = None
     #: If layer 3, IPv6 resources.
-    ipv6_address: IPv6Address | None = None
+    ipv6_address: IPv6AddressType | None = None
     #: Any custom firewall filters that the partner may require.
     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
+    sbp_bgp_session_list: list[BGPSession]  # type: ignore[assignment]

gso/translations/en-GB.json

@@ -32,8 +32,7 @@
             "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",
-            "confirm_info": "Please verify this form looks correct."
+            "restore_isis_metric": "Restore ISIS metric to original value"
         }
     },
     "workflow": {
@@ -44,6 +43,7 @@
         "create_router": "Create Router",
         "create_site": "Create Site",
         "create_switch": "Create Switch",
+        "create_edge_port": "Create Edge Port",
         "deploy_twamp": "Deploy TWAMP",
         "migrate_iptrunk": "Migrate IP Trunk",
         "modify_isis_metric": "Modify the ISIS metric",
@@ -51,10 +51,12 @@
         "modify_trunk_interface": "Modify IP Trunk interface",
         "modify_connection_strategy": "Modify connection strategy",
         "modify_router_kentik_license": "Modify device license in Kentik",
+        "modify_edge_port": "Modify Edge Port",
         "terminate_iptrunk": "Terminate IP Trunk",
         "terminate_router": "Terminate Router",
         "terminate_site": "Terminate Site",
         "terminate_switch": "Terminate Switch",
+        "terminate_edge_port": "Terminate Edge Port",
         "redeploy_base_config": "Redeploy base config",
         "update_ibgp_mesh": "Update iBGP mesh",
         "create_imported_site": "NOT FOR HUMANS -- Import existing site",
@@ -72,6 +74,7 @@
         "validate_iptrunk": "Validate IP Trunk configuration",
         "validate_router": "Validate Router configuration",
         "validate_switch": "Validate Switch configuration",
+        "validate_edge_port": "Validate Edge Port",
         "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",

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/workflows/edge_port/create_edge_port.py

 """A creation workflow for adding a new edge port to the network."""
 
 from typing import Annotated, Any, Self
+from uuid import uuid4
 
 from annotated_types import Len
 from orchestrator import step, workflow
@@ -16,7 +17,7 @@ from pydantic import AfterValidator, ConfigDict, model_validator
 from pydantic_forms.validators import validate_unique_list
 from pynetbox.models.dcim import Interfaces
 
-from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
+from gso.products.product_blocks.edge_port import EdgePortAEMemberBlockInactive, EdgePortType, EncapsulationType
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.edge_port import EdgePortInactive, EdgePortProvisioning
 from gso.products.product_types.router import Router
@@ -133,7 +134,9 @@ def initialize_subscription(
     subscription.description = f"Edge Port {name} on {router.router_fqdn}, {partner}, {geant_ga_id or ""}"
     subscription.edge_port.edge_port_description = description
     for member in ae_members:
-        subscription.edge_port.edge_port_ae_members.append(LAGMember(**member))
+        subscription.edge_port.edge_port_ae_members.append(
+            EdgePortAEMemberBlockInactive.new(subscription_id=uuid4(), **member)
+        )
     subscription = EdgePortProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
 
     return {"subscription": subscription}

gso/workflows/edge_port/modify_edge_port.py

 """Modify an existing edge port subscription."""
 
 from typing import Annotated, Any, Self
+from uuid import uuid4
 
 from annotated_types import Len
 from orchestrator import workflow
@@ -13,7 +14,7 @@ from pydantic import AfterValidator, ConfigDict, model_validator
 from pydantic_forms.types import FormGenerator, UUIDstr
 from pydantic_forms.validators import ReadOnlyField, validate_unique_list
 
-from gso.products.product_blocks.edge_port import EncapsulationType
+from gso.products.product_blocks.edge_port import EdgePortAEMemberBlock, EncapsulationType
 from gso.products.product_types.edge_port import EdgePort
 from gso.services.lso_client import execute_playbook, lso_interaction
 from gso.services.netbox_client import NetboxClient
@@ -146,7 +147,7 @@ def modify_edge_port_subscription(
     )
     subscription.edge_port.edge_port_ae_members.clear()
     for member in ae_members:
-        subscription.edge_port.edge_port_ae_members.append(LAGMember(**member))
+        subscription.edge_port.edge_port_ae_members.append(EdgePortAEMemberBlock.new(subscription_id=uuid4(), **member))
     subscription.save()
 
     return {
@@ -159,19 +160,19 @@ def modify_edge_port_subscription(
 @step("Update interfaces in NetBox")
 def update_interfaces_in_netbox(
     subscription: EdgePort,
-    removed_ae_members: list[LAGMember],
-    previous_ae_members: list[LAGMember],
+    removed_ae_members: list[dict],
+    previous_ae_members: list[dict],
 ) -> dict[str, Any]:
     """Update the interfaces in NetBox."""
     nbclient = NetboxClient()
     # Free removed interfaces
-    for member in removed_ae_members:
-        nbclient.free_interface(subscription.edge_port.edge_port_node.router_fqdn, member.interface_name)
+    for removed_member in removed_ae_members:
+        nbclient.free_interface(subscription.edge_port.edge_port_node.router_fqdn, removed_member["interface_name"])
     # Attach physical interfaces to :term:`LAG`
     # Update interface description to subscription ID
     # Reserve interfaces
     for member in subscription.edge_port.edge_port_ae_members:
-        if any(prev_member.interface_name == member.interface_name for prev_member in previous_ae_members):
+        if any(prev_member["interface_name"] == member.interface_name for prev_member in previous_ae_members):
             continue
         nbclient.attach_interface_to_lag(
             device_name=subscription.edge_port.edge_port_node.router_fqdn,

gso/workflows/geant_ip/__init__.py

+"""GÉANT IP workflows."""

gso/workflows/geant_ip/create_geant_ip.py

+"""Create a new GÉANT IP subscription."""