Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision

Target

Select target project
  • goat/gap/geant-service-orchestrator
1 result
Select Git revision
Show changes
Showing
with 814 additions and 21 deletions
"""Add Edge Port and GÉANT IP workflows.
Revision ID: bf05800fe9fc
Revises: a08bf228f112
Create Date: 2024-10-08 11:22:00.038925
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = 'bf05800fe9fc'
down_revision = 'a08bf228f112'
branch_labels = None
depends_on = None
from orchestrator.migrations.helpers import create_workflow, delete_workflow
new_workflows = [
{
"name": "create_edge_port",
"target": "CREATE",
"description": "Create Edge Port",
"product_type": "EdgePort"
},
{
"name": "modify_edge_port",
"target": "MODIFY",
"description": "Modify Edge Port",
"product_type": "EdgePort"
},
{
"name": "terminate_edge_port",
"target": "TERMINATE",
"description": "Terminate Edge Port",
"product_type": "EdgePort"
},
{
"name": "validate_edge_port",
"target": "SYSTEM",
"description": "Validate Edge Port Configuration",
"product_type": "EdgePort"
},
{
"name": "create_imported_edge_port",
"target": "CREATE",
"description": "Import Edge Port",
"product_type": "ImportedEdgePort"
},
{
"name": "import_edge_port",
"target": "MODIFY",
"description": "Import Edge Port",
"product_type": "ImportedEdgePort"
},
{
"name": "create_geant_ip",
"target": "CREATE",
"description": "Create G\u00c9ANT IP",
"product_type": "GeantIP"
},
{
"name": "modify_geant_ip",
"target": "MODIFY",
"description": "Modify G\u00c9ANT IP",
"product_type": "GeantIP"
},
{
"name": "migrate_geant_ip",
"target": "MODIFY",
"description": "Migrate G\u00c9ANT IP",
"product_type": "GeantIP"
},
{
"name": "create_imported_geant_ip",
"target": "CREATE",
"description": "Import G\u00c9ANT IP",
"product_type": "ImportedGeantIP"
},
{
"name": "import_geant_ip",
"target": "MODIFY",
"description": "Import G\u00c9ANT IP",
"product_type": "ImportedGeantIP"
}
]
def upgrade() -> None:
conn = op.get_bind()
for workflow in new_workflows:
create_workflow(conn, workflow)
def downgrade() -> None:
conn = op.get_bind()
for workflow in new_workflows:
delete_workflow(conn, workflow["name"])
"""Add IPV4/IPV6 netmask to Service Binding Port model .
Revision ID: df108295d917
Revises: bf05800fe9fc
Create Date: 2024-10-10 11:39:43.051211
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = 'df108295d917'
down_revision = 'bf05800fe9fc'
branch_labels = None
depends_on = None
def upgrade() -> None:
conn = op.get_bind()
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('ipv4_mask', 'IPV4 subnet mask') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO resource_types (resource_type, description) VALUES ('ipv6_mask', 'IPV6 subnet mask') RETURNING resource_types.resource_type_id
"""))
conn.execute(sa.text("""
INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask')))
"""))
conn.execute(sa.text("""
INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask')))
"""))
conn.execute(sa.text("""
WITH subscription_instance_ids AS (
SELECT subscription_instances.subscription_instance_id
FROM subscription_instances
WHERE subscription_instances.product_block_id IN (
SELECT product_blocks.product_block_id
FROM product_blocks
WHERE product_blocks.name = 'ServiceBindingPort'
)
)
INSERT INTO
subscription_instance_values (subscription_instance_id, resource_type_id, value)
SELECT
subscription_instance_ids.subscription_instance_id,
resource_types.resource_type_id,
'None'
FROM resource_types
CROSS JOIN subscription_instance_ids
WHERE resource_types.resource_type = 'ipv4_mask'
"""))
conn.execute(sa.text("""
WITH subscription_instance_ids AS (
SELECT subscription_instances.subscription_instance_id
FROM subscription_instances
WHERE subscription_instances.product_block_id IN (
SELECT product_blocks.product_block_id
FROM product_blocks
WHERE product_blocks.name = 'ServiceBindingPort'
)
)
INSERT INTO
subscription_instance_values (subscription_instance_id, resource_type_id, value)
SELECT
subscription_instance_ids.subscription_instance_id,
resource_types.resource_type_id,
'None'
FROM resource_types
CROSS JOIN subscription_instance_ids
WHERE resource_types.resource_type = 'ipv6_mask'
"""))
def downgrade() -> None:
conn = op.get_bind()
conn.execute(sa.text("""
DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask'))
"""))
conn.execute(sa.text("""
DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask'))
"""))
conn.execute(sa.text("""
DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask'))
"""))
conn.execute(sa.text("""
DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask'))
"""))
conn.execute(sa.text("""
DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask', 'ipv6_mask'))
"""))
conn.execute(sa.text("""
DELETE FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask', 'ipv6_mask')
"""))
......@@ -8,8 +8,10 @@
from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
from pydantic_forms.types import strEnum
from gso.products.product_types.edge_port import EdgePort, ImportedEdgePort
from gso.products.product_types.iptrunk import ImportedIptrunk, Iptrunk
from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreService
from gso.products.product_types.office_router import ImportedOfficeRouter, OfficeRouter
from gso.products.product_types.opengear import ImportedOpengear, Opengear
from gso.products.product_types.pop_vlan import PopVlan
......@@ -37,6 +39,12 @@ class ProductName(strEnum):
IMPORTED_OFFICE_ROUTER = "Imported office router"
OPENGEAR = "Opengear"
IMPORTED_OPENGEAR = "Imported Opengear"
EDGE_PORT = "Edge Port"
IMPORTED_EDGE_PORT = "Imported Edge Port"
GEANT_IP = "GÉANT IP"
IMPORTED_GEANT_IP = "Imported GÉANT IP"
IAS = "IAS"
IMPORTED_IAS = "Imported IAS"
class ProductType(strEnum):
......@@ -57,6 +65,12 @@ class ProductType(strEnum):
IMPORTED_OFFICE_ROUTER = ImportedOfficeRouter.__name__
OPENGEAR = Opengear.__name__
IMPORTED_OPENGEAR = Opengear.__name__
EDGE_PORT = EdgePort.__name__
IMPORTED_EDGE_PORT = ImportedEdgePort.__name__
GEANT_IP = NRENL3CoreService.__name__
IMPORTED_GEANT_IP = ImportedNRENL3CoreService.__name__
IAS = NRENL3CoreService.__name__
IMPORTED_IAS = ImportedNRENL3CoreService.__name__
SUBSCRIPTION_MODEL_REGISTRY.update(
......@@ -76,5 +90,11 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
ProductName.IMPORTED_OFFICE_ROUTER.value: ImportedOfficeRouter,
ProductName.OPENGEAR.value: Opengear,
ProductName.IMPORTED_OPENGEAR.value: ImportedOpengear,
ProductName.EDGE_PORT.value: EdgePort,
ProductName.IMPORTED_EDGE_PORT.value: ImportedEdgePort,
ProductName.GEANT_IP.value: NRENL3CoreService,
ProductName.IMPORTED_GEANT_IP.value: ImportedNRENL3CoreService,
ProductName.IAS.value: NRENL3CoreService,
ProductName.IMPORTED_IAS.value: ImportedNRENL3CoreService,
},
)
""":term:`BGP` session product block."""
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 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: IPAddress | None = None
bfd_enabled: bool | None = None
bfd_interval: int | None = None
bfd_multiplier: int | 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
send_default_route: bool | None = None
is_multi_hop: bool = False
is_passive: bool = False
rtbh_enabled: bool = False
class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""A :term:`BGP` session that is currently being provisioned. See :class:`BGPSession`."""
peer_address: IPAddress
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
is_passive: bool
rtbh_enabled: 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: IPAddress
#: 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 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 multi-path 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
#: Whether this is a passive session.
is_passive: bool
#: Whether Remote Triggered Blackhole is enabled
rtbh_enabled: bool
"""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 orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
class EncapsulationType(strEnum):
"""Types of encapsulation for edge ports.
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 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"
):
"""An edge port that's currently inactive. See :class:`EdgePortBlock`."""
edge_port_node: RouterBlockInactive | None = None
edge_port_name: str | None = None
edge_port_description: str | None = None
edge_port_enable_lacp: bool | None = None
edge_port_encapsulation: EncapsulationType = EncapsulationType.DOT1Q
edge_port_mac_address: str | None = None
edge_port_member_speed: PhysicalPortCapacity | None = None
edge_port_minimum_links: int | None = None
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[EdgePortAEMemberBlockInactive]
class EdgePortBlockProvisioning(EdgePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""An edge port that's being provisioned. See :class:`EdgePortBlock`."""
edge_port_node: RouterBlockProvisioning
edge_port_name: str
edge_port_description: str | None = None
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_members: LAGMemberList[EdgePortAEMemberBlockProvisioning] # 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.
edge_port_node: RouterBlock
#: The name of the edge port, in our case, corresponds to the name of the :term:`LAG` interface.
edge_port_name: str
#: A description of the edge port.
edge_port_description: str | None = None
#: Indicates whether :term:`LACP` 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., customer, private, public).
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
#: A list of :term:`LAG` members associated with this edge port.
edge_port_ae_members: LAGMemberList[EdgePortAEMemberBlock] # type: ignore[assignment]
"""Product blocks for :class:`NREN` Layer 3 Core Service products."""
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
from pydantic import Field
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."""
ap_type: APType | None = None
sbp: ServiceBindingPortInactive
class NRENAccessPortProvisioning(NRENAccessPortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""An access port for an R&E :term:`NREN` service that is being provisioned."""
ap_type: APType
sbp: ServiceBindingPortProvisioning
class NRENAccessPort(NRENAccessPortProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An access port for an R&E :term:`NREN` service."""
#: The type of Access Port
ap_type: APType
#: The corresponding :term:`SBP` of this Access Port.
sbp: ServiceBindingPort
class NRENL3CoreServiceBlockInactive(
ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="NRENL3CoreServiceBlock"
):
"""An inactive :term:`NREN` L3 Core service subscription. See :class:`NRENL3CoreServiceBlock`."""
nren_ap_list: list[NRENAccessPortInactive] = Field(default_factory=list)
class NRENL3CoreServiceBlockProvisioning(
NRENL3CoreServiceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
"""A provisioning :term:`NREN` L3 Core Service subscription. See :class:`NRENL3CoreServiceBlock`."""
nren_ap_list: list[NRENAccessPortProvisioning] # type: ignore[assignment]
class NRENL3CoreServiceBlock(NRENL3CoreServiceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An active :term:`NREN` L3 Core Service subscription block."""
#: The list of Access Points where this service is present.
nren_ap_list: list[NRENAccessPort] # type: ignore[assignment]
......@@ -43,9 +43,9 @@ class OpengearBlock(OpengearBlockProvisioning, lifecycle=[SubscriptionLifecycle.
opengear_hostname: str
#: The site where the Opengear device is located.
opengear_site: SiteBlock
#: The WAN address of the Opengear device.
#: The :term:`WAN` address of the Opengear device.
opengear_wan_address: ipaddress.IPv4Address
#: The WAN netmask of the Opengear device.
#: The :term:`WAN` netmask of the Opengear device.
opengear_wan_netmask: ipaddress.IPv4Address
#: The WAN gateway of the Opengear device.
#: The :term:`WAN` gateway of the Opengear device.
opengear_wan_gateway: ipaddress.IPv4Address
"""Service Binding Port.
A service binding port is used to logically attach an edge port to a customer service using a :term:`VLAN`.
"""
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.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
from gso.utils.shared_enums import SBPType
from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
class ServiceBindingPortInactive(
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: VLAN_ID | None = None
sbp_type: SBPType | None = None
ipv4_address: IPv4AddressType | None = None
ipv4_mask: IPV4Netmask | None = None
ipv6_address: IPv6AddressType | None = None
ipv6_mask: IPV6Netmask | None = None
custom_firewall_filters: bool | None = None
geant_sid: str | None = None
sbp_bgp_session_list: list[BGPSessionInactive] = Field(default_factory=list)
edge_port: EdgePortBlockInactive | 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: VLAN_ID | None = None
sbp_type: SBPType
ipv4_address: IPv4AddressType | None = None
ipv4_mask: IPV4Netmask | None = None
ipv6_address: IPv6AddressType | None = None
ipv6_mask: IPV6Netmask | None = None
custom_firewall_filters: bool
geant_sid: str
sbp_bgp_session_list: list[BGPSessionProvisioning] # type: ignore[assignment]
edge_port: EdgePortBlockProvisioning
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: VLAN_ID | None = None
#: Is this service binding port layer 2 or 3?
sbp_type: SBPType
#: If layer 3, IPv4 resources.
ipv4_address: IPv4AddressType | None = None
#: IPV4 subnet mask.
ipv4_mask: IPV4Netmask | None = None
#: If layer 3, IPv6 resources.
ipv6_address: IPv6AddressType | None = None
#: IPV6 subnet mask.
ipv6_mask: IPV6Netmask | 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: list[BGPSession] # type: ignore[assignment]
#: The Edge Port on which this :term:`SBP` resides.
edge_port: EdgePortBlock
"""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
class ImportedEdgePortInactive(SubscriptionModel, is_base=True):
"""An imported, inactive Edge Port."""
edge_port: EdgePortBlockInactive
class ImportedEdgePort(
ImportedEdgePortInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
):
"""An imported Edge Port that is currently active."""
edge_port: EdgePortBlock
""":term:`NREN` L3 Core Service product type."""
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle
from pydantic_forms.types import strEnum
from gso.products.product_blocks.nren_l3_core_service import (
NRENL3CoreServiceBlock,
NRENL3CoreServiceBlockInactive,
NRENL3CoreServiceBlockProvisioning,
)
class NRENL3CoreServiceType(strEnum):
"""Available types of :term:`NREN` Layer 3 Core Services.
The core services offered include GÉANT IP for R&E access, and the Internet Access Service.
"""
GEANT_IP = "GÉANT IP"
IAS = "IAS"
class NRENL3CoreServiceInactive(SubscriptionModel, is_base=True):
"""An inactive :term:`NREN` L3 Core Service subscription."""
nren_l3_core_service_type: NRENL3CoreServiceType
nren_l3_core_service: NRENL3CoreServiceBlockInactive
class NRENL3CoreServiceProvisioning(NRENL3CoreServiceInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""A :term:`NREN` L3 Core Service subscription that's being provisioned."""
nren_l3_core_service_type: NRENL3CoreServiceType
nren_l3_core_service: NRENL3CoreServiceBlockProvisioning
class NRENL3CoreService(NRENL3CoreServiceProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
"""An active :term:`NREN` L3 Core Service subscription."""
nren_l3_core_service_type: NRENL3CoreServiceType
nren_l3_core_service: NRENL3CoreServiceBlock
class ImportedNRENL3CoreServiceInactive(SubscriptionModel, is_base=True):
"""An imported, inactive :term:`NREN` L3 Core Service subscription."""
nren_l3_core_service_type: NRENL3CoreServiceType
nren_l3_core_service: NRENL3CoreServiceBlockInactive
class ImportedNRENL3CoreService(
ImportedNRENL3CoreServiceInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
):
"""An imported :term:`NREN` L3 Core Service subscription."""
nren_l3_core_service_type: NRENL3CoreServiceType
nren_l3_core_service: NRENL3CoreServiceBlock
......@@ -99,6 +99,8 @@ class KentikClient:
If the site is not found, return an empty dict.
.. vale off
:param str site_slug: The name of the site, should be a three-letter slug like COR or POZ.
"""
sites = self.get_sites()
......
......@@ -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__)
......
......@@ -13,6 +13,7 @@ from gso.settings import load_oss_params
from gso.utils.device_info import (
DEFAULT_SITE,
FEASIBLE_IP_TRUNK_LAG_RANGE,
FEASIBLE_SERVICES_LAG_RANGE,
ROUTER_ROLE,
TierInfo,
)
......@@ -289,8 +290,8 @@ class NetboxClient:
interface.lag = None
interface.save()
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Return all available :term:`LAG` not assigned to a device."""
def get_available_lags_in_range(self, router_id: UUID, lag_range: range) -> list[str]:
"""Return all available LAGs within a given range not assigned to a device."""
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
......@@ -299,12 +300,20 @@ class NetboxClient:
interface["name"] for interface in self.netbox.dcim.interfaces.filter(device=device.name, type="lag")
]
# Generate all feasible LAGs
all_feasible_lags = [f"lag-{i}" for i in FEASIBLE_IP_TRUNK_LAG_RANGE]
# Generate all feasible LAGs in the specified range
all_feasible_lags = [f"lag-{i}" for i in lag_range]
# Return available LAGs not assigned to the device
return [lag for lag in all_feasible_lags if lag not in lag_interface_names]
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Return all available :term:`LAG` not assigned to a device."""
return self.get_available_lags_in_range(router_id, FEASIBLE_IP_TRUNK_LAG_RANGE)
def get_available_services_lags(self, router_id: UUID) -> list[str]:
"""Return all available Edge port LAGs not assigned to a device."""
return self.get_available_lags_in_range(router_id, FEASIBLE_SERVICES_LAG_RANGE)
@staticmethod
def calculate_speed_bits_per_sec(speed: str) -> int:
"""Extract the numeric part from the speed."""
......
......@@ -246,6 +246,20 @@ def get_active_site_subscriptions(includes: list[str] | None = None) -> list[Sub
)
def get_active_edge_port_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
"""Retrieve active Edge Port subscriptions.
:param includes: The fields to be included in the returned Subscription objects.
:type includes: list[str]
:return: A list of Subscription objects for Edge Ports.
:rtype: list[Subscription]
"""
return get_subscriptions(
product_types=[ProductType.EDGE_PORT], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
)
def get_site_by_name(site_name: str) -> Site:
"""Get a site by its name.
......
......@@ -9,15 +9,13 @@ import json
import logging
import os
from pathlib import Path
from typing import Annotated
from orchestrator.types import UUIDstr
from pydantic import EmailStr, Field
from pydantic import EmailStr
from pydantic_forms.types import strEnum
from pydantic_settings import BaseSettings
from typing_extensions import Doc
from gso.utils.types.ip_address import PortNumber
from gso.utils.types.ip_address import IPV4Netmask, IPV6Netmask, PortNumber
logger = logging.getLogger(__name__)
......@@ -62,16 +60,12 @@ class InfoBloxParams(BaseSettings):
password: str
V4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
V6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
class V4NetworkParams(BaseSettings):
"""A set of parameters that describe an IPv4 network in InfoBlox."""
containers: list[ipaddress.IPv4Network]
networks: list[ipaddress.IPv4Network]
mask: V4Netmask
mask: IPV4Netmask
class V6NetworkParams(BaseSettings):
......@@ -79,7 +73,7 @@ class V6NetworkParams(BaseSettings):
containers: list[ipaddress.IPv6Network]
networks: list[ipaddress.IPv6Network]
mask: V6Netmask
mask: IPV6Netmask
class ServiceNetworkParams(BaseSettings):
......
......@@ -19,6 +19,7 @@
"router_role": "Router role",
"geant_s_sid": "GÉANT S-SID",
"geant_sid": "GÉANT S-SID",
"iptrunk_description": "IPtrunk description",
"iptrunk_type": "IPtrunk type",
"iptrunk_speed": "Capacity per port (in Gbits/s)",
......@@ -32,8 +33,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,17 +44,23 @@
"create_router": "Create Router",
"create_site": "Create Site",
"create_switch": "Create Switch",
"create_edge_port": "Create Edge Port",
"create_nren_l3_core_service": "Create NREN L3 Core Service",
"deploy_twamp": "Deploy TWAMP",
"migrate_iptrunk": "Migrate IP Trunk",
"migrate_nren_l3_core_service": "Migrate NREN L3 Core Service",
"modify_isis_metric": "Modify the ISIS metric",
"modify_site": "Modify Site",
"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",
"modify_nren_l3_core_service": "Modify NREN L3 Core Service",
"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",
......@@ -63,15 +69,20 @@
"create_imported_super_pop_switch": "NOT FOR HUMANS -- Import existing super PoP switch",
"create_imported_office_router": "NOT FOR HUMANS -- Import existing office router",
"create_imported_opengear": "NOT FOR HUMANS -- Import existing OpenGear",
"create_imported_edge_port": "NOT FOR HUMANS -- Import existing Edge Port",
"create_imported_nren_l3_core_service": "NOT FOR HUMANS -- Import existing NREN L3 Core Service",
"import_site": "NOT FOR HUMANS -- Finalize import into a Site product",
"import_router": "NOT FOR HUMANS -- Finalize import into a Router product",
"import_iptrunk": "NOT FOR HUMANS -- Finalize import into an IP trunk product",
"import_office_router": "NOT FOR HUMANS -- Finalize import into an Office router product",
"import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
"import_opengear": "NOT FOR HUMANS -- Finalize import into an OpenGear",
"import_edge_port": "NOT FOR HUMANS -- Finalize import into an Edge Port",
"import_nren_l3_core_service": "NOT FOR HUMANS -- Finalize import into a NREN L3 Core Service",
"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",
......
......@@ -45,8 +45,8 @@ class TierInfo:
return getattr(self, name)
# The range includes values from 1 to 10 (11 is not included)
FEASIBLE_IP_TRUNK_LAG_RANGE = range(1, 11)
FEASIBLE_SERVICES_LAG_RANGE = range(20, 51)
# Define default values
ROUTER_ROLE = {"name": "router", "slug": "router"}
......
......@@ -4,6 +4,7 @@ import re
from typing import TYPE_CHECKING
from uuid import UUID
from pydantic_forms.types import UUIDstr
from pydantic_forms.validators import Choice
from gso import settings
......@@ -11,6 +12,7 @@ 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.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
......@@ -75,6 +77,18 @@ def available_lags_choices(router_id: UUID) -> Choice | None:
return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)) # type: ignore[arg-type]
def available_service_lags_choices(router_id: UUID) -> Choice | None:
"""Return a list of available lags for a given router for services.
For Nokia routers, return a list of available lags.
For Juniper routers, return ``None``.
"""
if get_router_vendor(router_id) != Vendor.NOKIA:
return None
side_a_ae_iface_list = NetboxClient().get_available_services_lags(router_id)
return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)) # type: ignore[arg-type]
def get_router_vendor(router_id: UUID) -> Vendor:
"""Retrieve the vendor of a router.
......@@ -177,6 +191,16 @@ def active_router_selector() -> Choice:
return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)) # type: ignore[arg-type]
def active_pe_router_selector() -> Choice:
"""Generate a dropdown selector for choosing an active PE Router in an input form."""
routers = {
str(router.subscription_id): router.description
for router in subscriptions.get_active_subscriptions_by_field_and_value("router_role", RouterRole.PE)
}
return Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
def active_switch_selector() -> Choice:
"""Generate a dropdown selector for choosing an active Switch in an input form."""
switch_subscriptions = {
......@@ -185,3 +209,42 @@ def active_switch_selector() -> Choice:
}
return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)) # type: ignore[arg-type]
def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> Choice:
"""Generate a dropdown selector for choosing an active Edge Port in an input form."""
edge_port_subscriptions = subscriptions.get_active_edge_port_subscriptions(
includes=["subscription_id", "description", "customer_id"]
)
if partner_id:
# ``partner_id`` is set, so we will filter accordingly.
edge_port_subscriptions = list(
filter(lambda subscription: bool(subscription["customer_id"] == partner_id), edge_port_subscriptions)
)
edge_ports = {str(port["subscription_id"]): port["description"] for port in edge_port_subscriptions}
return Choice(
"Select an Edge Port",
zip(edge_ports.keys(), edge_ports.items(), strict=True), # type: ignore[arg-type]
)
def partner_choice() -> Choice:
"""Return a Choice object containing a list of available partners."""
partners = {partner["partner_id"]: partner["name"] for partner in get_all_partners()}
return Choice("Select a partner", zip(partners.values(), partners.items(), strict=True)) # type: ignore[arg-type]
def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int, enable_lacp: bool) -> None:
"""Validate the number of edge port members based on the :term:`LACP` setting.
:param number_of_members: The number of members to validate.
:param enable_lacp: Whether :term:`LACP` is enabled or not.
:raises ValueError: If the number of members is greater than 1 and :term:`LACP` is disabled.
"""
if number_of_members > 1 and not enable_lacp:
err_msg = "Number of members must be 1 if LACP is disabled."
raise ValueError(err_msg)
"""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"
......@@ -18,13 +18,29 @@ def validate_ipv4_or_ipv6(value: str) -> str:
return value
def validate_ipv4_or_ipv6_network(value: str) -> str:
"""Validate that a value is a valid IPv4 or IPv6 network."""
try:
ipaddress.ip_network(value)
except ValueError as e:
msg = "Enter a valid IPv4 or IPv6 network."
raise ValueError(msg) from e
else:
return value
def _str(value: Any) -> str:
return str(value)
IPv4AddressType = Annotated[ipaddress.IPv4Address, PlainSerializer(_str, return_type=str, when_used="always")]
IPv4NetworkType = Annotated[ipaddress.IPv4Network, PlainSerializer(_str, return_type=str, when_used="always")]
IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(_str, return_type=str, when_used="always")]
IPv6NetworkType = Annotated[ipaddress.IPv6Network, PlainSerializer(_str, return_type=str, when_used="always")]
IPAddress = Annotated[str, AfterValidator(validate_ipv4_or_ipv6)]
IPNetwork = Annotated[str, AfterValidator(validate_ipv4_or_ipv6_network)]
IPV4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
IPV6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
PortNumber = Annotated[
int,
Field(
......