Add product blocks for SBP and BGP session. Small edit to edge port product model.

gso/api/v1/network.py

@@ -10,8 +10,6 @@ from orchestrator.security import authorize
 from orchestrator.services.subscriptions import build_extended_domain_model
 from starlette import status
 
-from gso.annotations.coordinates import LatitudeCoordinate, LongitudeCoordinate
-from gso.annotations.interfaces import PhysicalPortCapacity
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
 from gso.services.subscriptions import get_active_iptrunk_subscriptions

gso/products/product_blocks/bgp_session.py

+""":term:`BGP` session product block."""
+
+from ipaddress import IPv4Address, IPv6Address
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
+
+
+class IPFamily(strEnum):
+    """Possible :term:`IP` families of a :term:`BGP` peering."""
+
+    V4UNICAST = "v4unicast"
+    V6UNICAST = "v6unicast"
+    V4MULTICAST = "v4multicast"
+    V6MULTICAST = "v6multicast"
+
+
+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
+    bfd_enabled: bool | None = None
+    bfd_interval: int | None = None
+    bfd_multiplier: int | None = None
+    families: list[IPFamily] | None = None
+    has_custom_policies: bool | None = None
+    authentication_key: str | None = None
+    multipath_enabled: bool | None = None
+    send_default_route: bool | None = None
+    is_multi_hop: bool = False
+
+
+class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A :term:`BGP` session that is currently being provisioned. See :class:`BGPSession`."""
+
+    peer_address: IPv4Address | IPv6Address
+    bfd_enabled: bool
+    bfd_interval: int | None = None
+    bfd_multiplier: int | None = None
+    families: list[IPFamily]
+    has_custom_policies: bool
+    authentication_key: str
+    multipath_enabled: bool
+    send_default_route: bool
+    is_multi_hop: bool
+
+
+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
+    #: Whether :term:`BFD` is enabled.
+    bfd_enabled: bool
+    #: The :term:`BFD` interval, if enabled.
+    bfd_interval: int | None = None
+    #: The :term:`BFD` multiplier, if enabled.
+    bfd_multiplier: int | None = None
+    #: The list of :term:`IP` families enabled for this session.
+    families: list[IPFamily]
+    #: Whether any custom policies exist for this session.
+    has_custom_policies: bool
+    #: The authentication key of the :term:`BGP` session.
+    authentication_key: str
+    #: Whether multipath is enabled.
+    multipath_enabled: bool
+    #: Whether we send a last resort route.
+    send_default_route: bool
+    #: Whether this session is multi-hop or not. Defaults to no.
+    is_multi_hop: bool

gso/products/product_blocks/edge_port.py

@@ -4,48 +4,11 @@ Edge port sets the boundary between Geant network and an external entity that co
 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.domain.base import ProductBlockModel
 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 RouterBlock, RouterBlockInactive, RouterBlockProvisioning
-
-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
+from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
 
 
 class EncapsulationType(strEnum):
@@ -87,7 +50,7 @@ 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[EdgePortInterfaceBlockInactive]
+    edge_port_ae_members: LAGMemberList[LAGMember]
 
 
 class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -104,7 +67,7 @@ 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[EdgePortInterfaceBlockProvisioning]  # type: ignore[assignment]
+    edge_port_ae_members: LAGMemberList[LAGMember]
 
 
 class EdgePortBlock(EdgePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -133,4 +96,4 @@ 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[EdgePortInterfaceBlock]  # type: ignore[assignment]
+    edge_port_ae_members: LAGMemberList[LAGMember]

gso/products/product_blocks/service_binding_port.py

+"""Service Binding Port.
+
+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 ProductModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic import Field
+from pydantic_forms.types import strEnum
+
+VLANID = Annotated[int, Field(gt=0, lt=4096)]
+
+
+class SBPType(strEnum):
+    """Enumerator for the two allowed types of service binding port: layer 2 or layer 3."""
+
+    L2 = "l2"
+    L3 = "l3"
+
+
+class ServiceBindingPortInactive(
+    ProductModel, 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
+    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
+
+
+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
+    sbp_type: SBPType
+    ipv4_address: IPv4Address | None = None
+    ipv6_address: IPv6Address | None = None
+    custom_firewall_filters: bool
+    geant_sid: str
+
+
+class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A service binding port that is actively used in the network."""
+
+    #: Whether this :term:`VLAN` is tagged or not.
+    is_tagged: bool
+    #: The :term:`VLAN` ID.
+    vlan_id: VLANID | None = None
+    #: Is this service binding port layer 2 or 3?
+    sbp_type: SBPType
+    #: If layer 3, IPv4 resources.
+    ipv4_address: IPv4Address | None = None
+    #: If layer 3, IPv6 resources.
+    ipv6_address: IPv6Address | 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

gso/services/infoblox.py

@@ -37,7 +37,7 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
     return connector.Connector(options), oss
 
 
-def _allocate_network(
+def _allocate_network(  # noqa: PLR0917
     conn: connector.Connector,
     dns_view: str,
     network_view: str,

gso/utils/helpers.py

@@ -11,10 +11,10 @@ from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
 from gso.services import subscriptions
 from gso.services.netbox_client import NetboxClient
-from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import IPv4AddressType
 from gso.services.partners import get_all_partners
 from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType
 
 if TYPE_CHECKING:
     from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock

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
@@ -9,6 +8,7 @@ from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Choice
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import StepList, begin, done
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
@@ -16,8 +16,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 EdgePortInterfaceBlockInactive, EdgePortType, EncapsulationType
-from gso.products.product_blocks.iptrunk import PhysicalPortCapacity
+from gso.products.product_blocks.edge_port import 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
@@ -25,13 +24,13 @@ from gso.services import lso_client, subscriptions
 from gso.services.lso_client import lso_interaction
 from gso.services.netbox_client import NetboxClient
 from gso.utils.helpers import (
-    LAGMember,
     available_interfaces_choices,
     available_service_lags_choices,
     partner_choice,
     validate_edge_port_number_of_members_based_on_lacp,
 )
-from gso.utils.types import TTNumber
+from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
+from gso.utils.types.tt_number import TTNumber
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -134,9 +133,7 @@ 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(
-            EdgePortInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
-        )
+        subscription.edge_port.edge_port_ae_members.append(LAGMember(**member))
     subscription = EdgePortProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
 
     return {"subscription": subscription}
@@ -181,8 +178,8 @@ def allocate_interfaces_in_netbox(subscription: EdgePortProvisioning) -> None:
         fqdn = subscription.edge_port.edge_port_node.router_fqdn
         iface_name = interface.interface_name
         if not fqdn or not iface_name:
-            msg = f"FQDN and/or interface name missing in subscription {interface.owner_subscription_id}"
-            raise ValueError(msg)
+            msg = "FQDN and/or interface name missing in subscription"
+            raise ProcessFailureError(msg, details=subscription.subscription_id)
 
         NetboxClient().allocate_interface(device_name=fqdn, iface_name=iface_name)
 

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
@@ -14,22 +13,18 @@ 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 (
-    EdgePortInterfaceBlock,
-    EncapsulationType,
-)
-from gso.products.product_blocks.iptrunk import PhysicalPortCapacity
+from gso.products.product_blocks.edge_port import 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
 from gso.services.partners import get_partner_by_id
 from gso.utils.helpers import (
-    LAGMember,
     available_interfaces_choices,
     available_interfaces_choices_including_current_members,
     validate_edge_port_number_of_members_based_on_lacp,
 )
-from gso.utils.types import TTNumber
+from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
+from gso.utils.types.tt_number import TTNumber
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -151,9 +146,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(
-            EdgePortInterfaceBlock.new(subscription_id=uuid4(), **member),
-        )
+        subscription.edge_port.edge_port_ae_members.append(LAGMember(**member))
     subscription.save()
 
     return {
@@ -166,19 +159,19 @@ def modify_edge_port_subscription(
 @step("Update interfaces in NetBox")
 def update_interfaces_in_netbox(
     subscription: EdgePort,
-    removed_ae_members: list[dict],
-    previous_ae_members: list[dict],
+    removed_ae_members: list[LAGMember],
+    previous_ae_members: list[LAGMember],
 ) -> 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"])
+        nbclient.free_interface(subscription.edge_port.edge_port_node.router_fqdn, 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:  # type: ignore[assignment]
-        if any(prev_member["interface_name"] == member.interface_name for prev_member in previous_ae_members):
+    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):
             continue
         nbclient.attach_interface_to_lag(
             device_name=subscription.edge_port.edge_port_node.router_fqdn,

gso/workflows/edge_port/terminate_edge_port.py

@@ -14,7 +14,7 @@ from pydantic_forms.types import FormGenerator, UUIDstr
 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
-from gso.utils.types import TTNumber
+from gso.utils.types.tt_number import TTNumber
 
 
 def initial_input_form_generator() -> FormGenerator:

gso/workflows/iptrunk/create_imported_iptrunk.py

@@ -13,7 +13,6 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
 from pydantic import AfterValidator, ConfigDict
 from pydantic_forms.validators import validate_unique_list
 
-from gso.annotations.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
 from gso.products import ProductName
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType
 from gso.products.product_types.iptrunk import ImportedIptrunkInactive

gso/workflows/iptrunk/create_iptrunk.py

@@ -20,9 +20,6 @@ from pydantic import ConfigDict
 from pydantic_forms.validators import ReadOnlyField
 from pynetbox.models.dcim import Interfaces
 
-from gso.annotations.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
-from gso.annotations.netbox_router import NetboxEnabledRouter
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlockInactive,
     IptrunkSideBlockInactive,

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -24,8 +24,6 @@ from pydantic import ConfigDict
 from pydantic_forms.validators import ReadOnlyField
 from pynetbox.models.dcim import Interfaces
 
-from gso.annotations.interfaces import JuniperAEInterface, JuniperLAGMember, LAGMember, LAGMemberList
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, IptrunkType
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -15,8 +15,6 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import ConfigDict
 from pydantic_forms.validators import Label, ReadOnlyField
 
-from gso.annotations.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlock,
     IptrunkSideBlock,

gso/workflows/iptrunk/terminate_iptrunk.py

@@ -17,7 +17,6 @@ from orchestrator.workflows.steps import (
 )
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.iptrunk import IptrunkSideBlock
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import infoblox

gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@@ -13,15 +13,6 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import AfterValidator, ConfigDict
 from pydantic_forms.validators import Divider, ReadOnlyField
 
-from gso.annotations.interfaces import (
-    JuniperAEInterface,
-    JuniperLAGMember,
-    JuniperPhyInterface,
-    LAGMember,
-    PhysicalPortCapacity,
-    validate_interface_names_are_unique,
-)
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectAddressSpace,
     LanSwitchInterconnectInterfaceBlockInactive,

gso/workflows/router/create_router.py

@@ -14,7 +14,6 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import ConfigDict, model_validator
 from pydantic_forms.validators import ReadOnlyField
 
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site

gso/workflows/router/promote_p_to_pe.py

@@ -15,7 +15,6 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import ConfigDict, model_validator
 
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
 from gso.services.kentik_client import KentikClient, NewKentikDevice

gso/workflows/router/terminate_router.py

@@ -20,7 +20,6 @@ from orchestrator.workflows.steps import (
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from requests import HTTPError
 
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
 from gso.services import infoblox

gso/workflows/router/update_ibgp_mesh.py

@@ -12,8 +12,6 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import ConfigDict, model_validator
 
-from gso.annotations.snmp import SNMPVersion
-from gso.annotations.tt_number import TTNumber
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
 from gso.services import librenms_client

test/cli/test_imports.py

@@ -4,7 +4,6 @@ from unittest.mock import patch
 
 import pytest
 
-from gso.annotations.interfaces import PhysicalPortCapacity
 from gso.cli.imports import (
     import_iptrunks,
     import_office_routers,