Finished release 1.5.

Changelog.md

@@ -2,6 +2,17 @@
 
 All notable changes to this project will be documented in this file.
 
+## [1.5] - 2024-04-22
+- Added deploy TWAMP functionality.
+- Made some changes on IP Trunk creation/modification workflow including:
+    - Replaced `minimum link` with `number of members` in the IP trunks workflows.
+    - Added a confirmation page for users to confirm the `minimum links` value.
+    - Made `interface description` optional.
+    - Made `GA-GID` and `GA-SID` optional.
+- Added a rule to OPA policies to give access to Inventory Provider.
+- Improved the codebase to include OIDC client ID in userinfo.
+- Added switch, LAN switch interconnect and Pop VLAN domain models. 
+
 ## [1.1] - 2024-04-04
 - Fixed the AttributeError in the migrate_iptrunk workflow.
 - Improved the delete device in Netbox functionality.

docs/source/glossary.rst

@@ -81,3 +81,6 @@ Glossary of terms
 
   WFO
     `Workflow Orchestrator <https://workfloworchestrator.org/>`_
+
+  LAN
+    Local Area Network

gso/api/v1/imports.py

@@ -62,7 +62,7 @@ class IptrunkImportModel(BaseModel):
     """Required fields for importing an existing :class:`gso.products.product_types.iptrunk`."""
 
     partner: str
-    geant_s_sid: str
+    geant_s_sid: str | None
     iptrunk_type: IptrunkType
     iptrunk_description: str
     iptrunk_speed: PhysicalPortCapacity
@@ -70,11 +70,11 @@ class IptrunkImportModel(BaseModel):
     iptrunk_isis_metric: int
     side_a_node_id: str
     side_a_ae_iface: str
-    side_a_ae_geant_a_sid: str
+    side_a_ae_geant_a_sid: str | None
     side_a_ae_members: list[LAGMember]
     side_b_node_id: str
     side_b_ae_iface: str
-    side_b_ae_geant_a_sid: str
+    side_b_ae_geant_a_sid: str | None
     side_b_ae_members: list[LAGMember]
 
     iptrunk_ipv4_network: ipaddress.IPv4Network

gso/auth/oidc_policy_helper.py

@@ -92,6 +92,11 @@ class OIDCUserModel(dict):
                 return self.get(key)
             raise error from None
 
+    @property
+    def client_id(self) -> str:
+        """Return the client id."""
+        return self.get("client_id") or ""
+
     @property
     def user_name(self) -> str:
         """Return the username of the user."""
@@ -236,6 +241,8 @@ class OIDCUser(HTTPBearer):
 
             user_info = await self.userinfo(async_request, token)
 
+            user_info["client_id"] = intercepted_token.get("client_id")
+
             logger.debug("OIDCUserModel object.", intercepted_token=intercepted_token)
             return user_info
 
@@ -418,7 +425,7 @@ def opa_decision(
                 **(opa_kwargs or {}),
                 **user_info,
                 "resource": request.url.path,
-                "method": request_method,
+                "method": request_method.upper(),
                 "arguments": {"path": request.path_params, "query": {**request.query_params}, "json": json},
             }
         }

gso/products/__init__.py

@@ -9,10 +9,13 @@ from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
 from pydantic_forms.types import strEnum
 
 from gso.products.product_types.iptrunk import Iptrunk
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
 from gso.products.product_types.office_router import OfficeRouter
+from gso.products.product_types.pop_vlan import PopVlan
 from gso.products.product_types.router import Router
 from gso.products.product_types.site import Site
 from gso.products.product_types.super_pop_switch import SuperPopSwitch
+from gso.products.product_types.switch import Switch
 
 
 class ProductName(strEnum):
@@ -23,6 +26,9 @@ class ProductName(strEnum):
     SITE = "Site"
     SUPER_POP_SWITCH = "Super PoP switch"
     OFFICE_ROUTER = "Office router"
+    SWITCH = "Switch"
+    LAN_SWITCH_INTERCONNECT = "LAN Switch Interconnect"
+    POP_VLAN = "Pop VLAN"
 
 
 class ProductType(strEnum):
@@ -33,6 +39,9 @@ class ProductType(strEnum):
     SITE = Site.__name__
     SUPER_POP_SWITCH = SuperPopSwitch.__name__
     OFFICE_ROUTER = OfficeRouter.__name__
+    SWITCH = Switch.__name__
+    LAN_SWITCH_INTERCONNECT = LanSwitchInterconnect.__name__
+    POP_VLAN = PopVlan.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -42,5 +51,8 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.SITE.value: Site,
         ProductName.SUPER_POP_SWITCH.value: SuperPopSwitch,
         ProductName.OFFICE_ROUTER.value: OfficeRouter,
+        ProductName.SWITCH.value: Switch,
+        ProductName.LAN_SWITCH_INTERCONNECT.value: LanSwitchInterconnect,
+        ProductName.POP_VLAN.value: PopVlan,
     },
 )

gso/products/product_blocks/iptrunk.py

@@ -55,14 +55,14 @@ class IptrunkInterfaceBlockProvisioning(IptrunkInterfaceBlockInactive, lifecycle
     """An IP trunk interface that is being provisioned."""
 
     interface_name: str
-    interface_description: str
+    interface_description: str | None = None
 
 
 class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     """An active IP trunk interface."""
 
     interface_name: str
-    interface_description: str
+    interface_description: str | None = None
 
 
 class IptrunkSides(UniqueConstrainedList[T_co]):  # type: ignore[type-var]
@@ -139,7 +139,7 @@ class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
     """A trunk that's currently deployed in the network."""
 
     #:  GÉANT service ID associated with this trunk.
-    geant_s_sid: str
+    geant_s_sid: str | None = None
     #:  A human-readable description of this trunk.
     iptrunk_description: str
     #:  The type of trunk, can be either dark fibre or leased capacity.

gso/products/product_blocks/lan_switch_interconnect.py

+"""LAN Switch Interconnect product block that has all parameters of a subscription throughout its lifecycle."""
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle, strEnum
+
+from gso.products.product_blocks.iptrunk import LAGMemberList
+from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
+from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
+
+
+class LanSwitchInterconnectAddressSpace(strEnum):
+    """Types of LAN Switch Interconnect. Can be private or public."""
+
+    PRIVATE = "Private"
+    PUBLIC = "Public"
+
+
+class LanSwitchInterconnectInterfaceBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="LanSwitchInterconnectInterfaceBlock",
+):
+    """An inactive LAN Switch Interconnect interface."""
+
+    interface_name: str | None = None
+    interface_description: str | None = None
+
+
+class LanSwitchInterconnectInterfaceBlockProvisioning(
+    LanSwitchInterconnectInterfaceBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """A LAN Switch Interconnect interface that is being provisioned."""
+
+    interface_name: str
+    interface_description: str
+
+
+class LanSwitchInterconnectInterfaceBlock(
+    LanSwitchInterconnectInterfaceBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
+):
+    """An active Switch Interconnect interface."""
+
+    interface_name: str
+    interface_description: str
+
+
+class LanSwitchInterconnectRouterSideBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="LanSwitchInterconnectRouterSideBlock",
+):
+    """An inactive LAN Switch Interconnect router side."""
+
+    node: RouterBlockInactive
+    ae_iface: str | None = None
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
+
+
+class LanSwitchInterconnectRouterSideBlockProvisioning(
+    LanSwitchInterconnectRouterSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """An LAN Switch Interconnect router side that is being provisioned."""
+
+    node: RouterBlockProvisioning
+    ae_iface: str | None = None
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+
+
+class LanSwitchInterconnectRouterSideBlock(
+    LanSwitchInterconnectRouterSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
+):
+    """An active LAN Switch Interconnect router side."""
+
+    node: RouterBlock
+    ae_iface: str
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+
+
+class LanSwitchInterconnectSwitchSideBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="LanSwitchInterconnectSwitchSideBlock",
+):
+    """An inactive LAN Switch Interconnect switch side."""
+
+    node: SwitchBlockInactive
+    ae_iface: str | None = None
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
+
+
+class LanSwitchInterconnectSwitchSideBlockProvisioning(
+    LanSwitchInterconnectSwitchSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """An LAN Switch Interconnect switch side that is being provisioned."""
+
+    node: SwitchBlockProvisioning
+    ae_iface: str | None = None
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+
+
+class LanSwitchInterconnectSwitchSideBlock(
+    LanSwitchInterconnectSwitchSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
+):
+    """An active LAN Switch Interconnect switch side."""
+
+    node: SwitchBlock
+    ae_iface: str
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+
+
+class LanSwitchInterconnectBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="LanSwitchInterconnectBlock",
+):
+    """A LAN Switch Interconnect that's currently inactive, see :class:`LanSwitchInterconnectBlock`."""
+
+    lan_switch_interconnect_description: str | None = None
+    address_space: LanSwitchInterconnectAddressSpace | None = None
+    minimum_links: int | None = None
+    router_side: LanSwitchInterconnectRouterSideBlockInactive
+    switch_side: LanSwitchInterconnectSwitchSideBlockInactive
+
+
+class LanSwitchInterconnectBlockProvisioning(
+    LanSwitchInterconnectBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """A LAN Switch Interconnect that's currently being provisioned, see :class:`LanSwitchInterconnectBlock`."""
+
+    lan_switch_interconnect_description: str | None = None
+    address_space: LanSwitchInterconnectAddressSpace | None = None
+    minimum_links: int | None = None
+    router_side: LanSwitchInterconnectRouterSideBlockProvisioning
+    switch_side: LanSwitchInterconnectSwitchSideBlockProvisioning
+
+
+class LanSwitchInterconnectBlock(LanSwitchInterconnectBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A LAN Switch Interconnect that's currently deployed in the network."""
+
+    #:  A human-readable description of this LAN Switch Interconnect.
+    lan_switch_interconnect_description: str
+    #:  The address space of the VLAN Switch Interconnect. It can be private or public.
+    address_space: LanSwitchInterconnectAddressSpace
+    #:  The minimum amount of links the LAN Switch Interconnect should consist of.
+    minimum_links: int
+    #:  The router side of the LAN Switch Interconnect.
+    router_side: LanSwitchInterconnectRouterSideBlock
+    #:  The switch side of the LAN Switch Interconnect.
+    switch_side: LanSwitchInterconnectSwitchSideBlock

gso/products/product_blocks/pop_vlan.py

+"""Pop VLAN product block that has all parameters of a subscription throughout its lifecycle."""
+
+from ipaddress import IPv4Network, IPv6Network
+from typing import TypeVar
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.forms.validators import UniqueConstrainedList
+from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
+
+from gso.products.product_blocks.lan_switch_interconnect import (
+    LanSwitchInterconnectBlock,
+    LanSwitchInterconnectBlockInactive,
+    LanSwitchInterconnectBlockProvisioning,
+)
+
+T_co = TypeVar("T_co", covariant=True)
+
+
+class LayerPreference(strEnum):
+    """Enumerator for the different types of layer preferences."""
+
+    L2 = "L2"
+    L3 = "L3"
+
+
+class PortList(UniqueConstrainedList[T_co]):  # type: ignore[type-var]
+    """A list of ports."""
+
+
+class PopVlanPortBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="PopVlanPortBlock"
+):
+    """An inactive Pop VLAN port."""
+
+    port_name: str | None = None
+    port_description: str | None = None
+    tagged: bool | None = None
+
+
+class PopVlanPortBlockProvisioning(PopVlanPortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Pop VLAN port that is being provisioned."""
+
+    port_name: str | None = None
+    port_description: str | None = None
+    tagged: bool | None = None
+
+
+class PopVlanPortBlock(PopVlanPortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active Pop VLAN port."""
+
+    port_name: str
+    port_description: str
+    tagged: bool
+
+
+class PopVlanBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="PopVlanBlock",
+):
+    """A Pop VLAN that's currently inactive, see :class:`PopVlanBlock`."""
+
+    vlan_id: int
+    pop_vlan_description: str | None
+    lan_switch_interconnect: LanSwitchInterconnectBlockInactive
+    ports: PortList[PopVlanPortBlockProvisioning]
+    layer_preference: LayerPreference
+    ipv4_network: IPv4Network | None = None
+    ipv6_network: IPv6Network | None = None
+
+
+class PopVlanBlockProvisioning(PopVlanBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Pop VLAN that's currently being provisioned, see :class:`PopVlanBlock`."""
+
+    vlan_id: int
+    pop_vlan_description: str | None
+    lan_switch_interconnect: LanSwitchInterconnectBlockProvisioning
+    ports: PortList[PopVlanPortBlockProvisioning]
+    layer_preference: LayerPreference
+    ipv4_network: IPv4Network | None = None
+    ipv6_network: IPv6Network | None = None
+
+
+class PopVlanBlock(PopVlanBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A Pop VLAN that's currently deployed in the network."""
+
+    #: The VLAN ID of the Pop VLAN.
+    vlan_id: int
+    #: The description of the Pop VLAN.
+    pop_vlan_description: str
+    #: The LAN Switch Interconnect that this Pop VLAN is connected to.
+    lan_switch_interconnect: LanSwitchInterconnectBlock
+    #: The ports of the Pop VLAN.
+    ports: PortList[PopVlanPortBlock]
+    #: The level of the layer preference for the Pop VLAN (L2 or L3).
+    layer_preference: LayerPreference
+    #: IPv4 network for the Pop VLAN if layer preference is L3.
+    ipv4_network: IPv4Network | None = None
+    #: IPv6 network for the Pop VLAN if layer preference is L3.
+    ipv6_network: IPv6Network | None = None

gso/products/product_blocks/switch.py

+"""Product block for :class:`Switch` products."""
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
+
+from gso.products.product_blocks.site import (
+    SiteBlock,
+    SiteBlockInactive,
+    SiteBlockProvisioning,
+)
+from gso.utils.shared_enums import PortNumber, Vendor
+
+
+class SwitchModel(strEnum):
+    """Enumerator for the different types of switches."""
+
+    EX3400 = "EX3400"
+
+
+class SwitchBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="SwitchBlock",
+):
+    """A switch that's being currently inactive. See :class:`SwitchBlock`."""
+
+    switch_hostname: str | None = None
+    switch_ts_port: PortNumber | None = None
+    switch_site: SiteBlockInactive | None
+    switch_vendor: Vendor | None
+    switch_model: SwitchModel | None = None
+
+
+class SwitchBlockProvisioning(SwitchBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A switch that's being provisioned. See :class:`SwitchBlock`."""
+
+    switch_hostname: str
+    switch_ts_port: PortNumber
+    switch_site: SiteBlockProvisioning
+    switch_vendor: Vendor
+    switch_model: SwitchModel | None = None
+
+
+class SwitchBlock(SwitchBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A switch that's currently deployed in the network."""
+
+    #: The hostname of the switch.
+    switch_hostname: str
+    #:  The port of the terminal server that this switch is connected to. Used to offer out of band access.
+    switch_ts_port: PortNumber
+    #:  The :class:`Site` that this switch resides in. Both physically and computationally.
+    switch_site: SiteBlock
+    #: The vendor of the switch.
+    switch_vendor: Vendor
+    #: The model of the switch.
+    switch_model: SwitchModel | None = None

gso/products/product_types/lan_switch_interconnect.py

+"""The product type for LAN Switch Interconnect."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.lan_switch_interconnect import (
+    LanSwitchInterconnectBlock,
+    LanSwitchInterconnectBlockInactive,
+    LanSwitchInterconnectBlockProvisioning,
+)
+
+
+class LanSwitchInterconnectInactive(SubscriptionModel, is_base=True):
+    """An LAN Switch Interconnect that is inactive."""
+
+    lan_switch_interconnect: LanSwitchInterconnectBlockInactive
+
+
+class LanSwitchInterconnectProvisioning(LanSwitchInterconnectInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """An LAN Switch Interconnect that is being provisioned."""
+
+    lan_switch_interconnect: LanSwitchInterconnectBlockProvisioning
+
+
+class LanSwitchInterconnect(LanSwitchInterconnectProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An LAN Switch Interconnect that is active."""
+
+    lan_switch_interconnect: LanSwitchInterconnectBlock

gso/products/product_types/pop_vlan.py

+"""The product type for Pop VLAN."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.pop_vlan import PopVlanBlock, PopVlanBlockInactive, PopVlanBlockProvisioning
+
+
+class PopVlanInactive(SubscriptionModel, is_base=True):
+    """A Pop VLAN that is inactive."""
+
+    pop_vlan: PopVlanBlockInactive
+
+
+class PopVlanProvisioning(PopVlanInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Pop VLAN that is being provisioned."""
+
+    pop_vlan: PopVlanBlockProvisioning
+
+
+class PopVlan(PopVlanProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A Pop VLAN that is active."""
+
+    pop_vlan: PopVlanBlock

gso/products/product_types/switch.py

+"""A switch product type."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
+
+
+class SwitchInactive(SubscriptionModel, is_base=True):
+    """An inactive switch."""
+
+    switch: SwitchBlockInactive
+
+
+class SwitchProvisioning(SwitchInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A switch that is being provisioned."""
+
+    switch: SwitchBlockProvisioning
+
+
+class Switch(SwitchProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """A switch that is currently active."""
+
+    switch: SwitchBlock

gso/services/infoblox.py

@@ -19,10 +19,6 @@ class AllocationError(Exception):
     """Raised when Infoblox failed to allocate a resource."""
 
 
-class DeletionError(Exception):
-    """Raised when Infoblox failed to delete a resource."""
-
-
 def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
     """Set up a new connection with an Infoblox instance.
 
@@ -158,7 +154,7 @@ def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) ->
         network.delete()
     else:
         msg = f"Could not find network {ip_network}, nothing has been deleted."
-        raise DeletionError(msg)
+        logger.warning(msg)
 
 
 def allocate_host(
@@ -253,8 +249,9 @@ def create_host_by_ip(
                         :term:`GSO`.
     """
     if not hostname_available(hostname):
-        msg = f"Cannot allocate new host, FQDN {hostname} already taken."
-        raise AllocationError(msg)
+        msg = f"FQDN {hostname} already taken, nothing to be done."
+        logger.warning(msg)
+        return
 
     conn, oss = _setup_connection()
     ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
@@ -331,7 +328,7 @@ def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) ->
         host.delete()
     else:
         msg = f"Could not find host at {ip_addr}, nothing has been deleted."
-        raise DeletionError(msg)
+        logger.warning(msg)
 
 
 def delete_host_by_fqdn(fqdn: str) -> None:
@@ -348,4 +345,4 @@ def delete_host_by_fqdn(fqdn: str) -> None:
         host.delete()
     else:
         msg = f"Could not find host at {fqdn}, nothing has been deleted."
-        raise DeletionError(msg)
+        logger.warning(msg)

gso/translations/en-GB.json

@@ -38,7 +38,6 @@
     "workflow": {
         "activate_iptrunk": "Activate IP Trunk",
         "activate_router": "Activate router",
-        "cancel_subscription": "Cancel subscription",
         "confirm_info": "Please verify this form looks correct.",
         "deploy_twamp": "Deploy TWAMP",
         "migrate_iptrunk": "Migrate IP Trunk",

gso/utils/helpers.py

@@ -25,7 +25,7 @@ class LAGMember(BaseModel):
     """A :term:`LAG` member interface that consists of a name and description."""
 
     interface_name: str
-    interface_description: str
+    interface_description: str | None
 
     def __hash__(self) -> int:
         """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""

gso/workflows/__init__.py

 """Initialisation class that imports all workflows into :term:`GSO`."""
 
 from orchestrator.services.subscriptions import WF_USABLE_MAP
+from orchestrator.types import SubscriptionLifecycle
 from orchestrator.workflows import LazyWorkflowInstance
 
+ALL_ALIVE_STATES: list[str] = [
+    SubscriptionLifecycle.INITIAL,
+    SubscriptionLifecycle.PROVISIONING,
+    SubscriptionLifecycle.ACTIVE,
+]
+
 WF_USABLE_MAP.update(
     {
-        "cancel_subscription": ["initial"],
-        "redeploy_base_config": ["provisioning", "active"],
-        "update_ibgp_mesh": ["provisioning", "active"],
-        "activate_router": ["provisioning"],
-        "deploy_twamp": ["provisioning", "active"],
-        "modify_trunk_interface": ["provisioning", "active"],
-        "activate_iptrunk": ["provisioning"],
+        "redeploy_base_config": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+        "update_ibgp_mesh": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+        "activate_router": [SubscriptionLifecycle.PROVISIONING],
+        "deploy_twamp": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+        "modify_trunk_interface": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+        "activate_iptrunk": [SubscriptionLifecycle.PROVISIONING],
+        "terminate_site": ALL_ALIVE_STATES,
+        "terminate_router": ALL_ALIVE_STATES,
+        "terminate_iptrunk": ALL_ALIVE_STATES,
     }
 )
 
@@ -28,7 +37,6 @@ LazyWorkflowInstance("gso.workflows.router.redeploy_base_config", "redeploy_base
 LazyWorkflowInstance("gso.workflows.router.terminate_router", "terminate_router")
 LazyWorkflowInstance("gso.workflows.router.update_ibgp_mesh", "update_ibgp_mesh")
 LazyWorkflowInstance("gso.workflows.router.modify_connection_strategy", "modify_connection_strategy")
-LazyWorkflowInstance("gso.workflows.shared.cancel_subscription", "cancel_subscription")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")
 LazyWorkflowInstance("gso.workflows.site.terminate_site", "terminate_site")

gso/workflows/iptrunk/create_iptrunk.py

@@ -57,11 +57,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
         tt_number: str
         partner: str = ReadOnlyField("GEANT")
-        geant_s_sid: str
+        geant_s_sid: str | None
         iptrunk_description: str
         iptrunk_type: IptrunkType
         iptrunk_speed: PhysicalPortCapacity
-        iptrunk_minimum_links: int
+        iptrunk_number_of_members: int
 
         @validator("tt_number", allow_reuse=True)
         def validate_tt_number(cls, tt_number: str) -> str:
@@ -69,6 +69,13 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     initial_user_input = yield CreateIptrunkForm
 
+    class VerifyMinimumLinksForm(FormPage):
+        info_label: Label = (
+            f"This is the calculated minimum-links for this LAG: " f"{initial_user_input.iptrunk_number_of_members - 1}"  # type: ignore[assignment]
+        )
+        info_label2: Label = "Please confirm or modify."  # type: ignore[assignment]
+
+    yield VerifyMinimumLinksForm
     router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True))  # type: ignore[arg-type]
 
     class SelectRouterSideA(FormPage):
@@ -86,7 +93,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     router_a_fqdn = Router.from_subscription(router_a).router.router_fqdn
 
     class JuniperAeMembers(UniqueConstrainedList[LAGMember]):
-        min_items = initial_user_input.iptrunk_minimum_links
+        min_items = initial_user_input.iptrunk_number_of_members
+        max_items = initial_user_input.iptrunk_number_of_members
 
     if get_router_vendor(router_a) == Vendor.NOKIA:
 
@@ -97,7 +105,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             )
 
         class NokiaAeMembersA(UniqueConstrainedList[NokiaLAGMemberA]):
-            min_items = initial_user_input.iptrunk_minimum_links
+            min_items = initial_user_input.iptrunk_number_of_members
+            max_items = initial_user_input.iptrunk_number_of_members
 
         ae_members_side_a = NokiaAeMembersA
     else:
@@ -108,7 +117,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             title = f"Provide subscription details for side A of the trunk.({router_a_fqdn})"
 
         side_a_ae_iface: available_lags_choices(router_a) or str  # type: ignore[valid-type]
-        side_a_ae_geant_a_sid: str
+        side_a_ae_geant_a_sid: str | None
         side_a_ae_members: ae_members_side_a  # type: ignore[valid-type]
 
         @validator("side_a_ae_members", allow_reuse=True)
@@ -159,7 +168,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             title = f"Provide subscription details for side B of the trunk.({router_b_fqdn})"
 
         side_b_ae_iface: available_lags_choices(router_b) or str  # type: ignore[valid-type]
-        side_b_ae_geant_a_sid: str
+        side_b_ae_geant_a_sid: str | None
         side_b_ae_members: ae_members_side_b  # type: ignore[valid-type]
 
         @validator("side_b_ae_members", allow_reuse=True)
@@ -209,18 +218,18 @@ def get_info_from_ipam(subscription: IptrunkInactive) -> State:
 @step("Initialize subscription")
 def initialize_subscription(
     subscription: IptrunkInactive,
-    geant_s_sid: str,
+    geant_s_sid: str | None,
     iptrunk_type: IptrunkType,
     iptrunk_description: str,
     iptrunk_speed: PhysicalPortCapacity,
-    iptrunk_minimum_links: int,
+    iptrunk_number_of_members: int,
     side_a_node_id: str,
     side_a_ae_iface: str,
-    side_a_ae_geant_a_sid: str,
+    side_a_ae_geant_a_sid: str | None,
     side_a_ae_members: list[dict],
     side_b_node_id: str,
     side_b_ae_iface: str,
-    side_b_ae_geant_a_sid: str,
+    side_b_ae_geant_a_sid: str | None,
     side_b_ae_members: list[dict],
 ) -> State:
     """Take all input from the user, and store it in the database."""
@@ -232,7 +241,7 @@ def initialize_subscription(
     subscription.iptrunk.iptrunk_type = iptrunk_type
     subscription.iptrunk.iptrunk_speed = iptrunk_speed
     subscription.iptrunk.iptrunk_isis_metric = oss_params.GENERAL.isis_high_metric
-    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
+    subscription.iptrunk.iptrunk_minimum_links = iptrunk_number_of_members - 1
 
     subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node = side_a
     subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface = side_a_ae_iface

gso/workflows/iptrunk/deploy_twamp.py

 """Workflow for adding TWAMP to an existing IP trunk."""
 
+import json
+
 from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Label
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
+from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, done, init, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
@@ -38,9 +41,10 @@ def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 def deploy_twamp_dry(subscription: Iptrunk, process_id: UUIDstr, callback_route: str, tt_number: str) -> State:
     """Perform a dry run of deploying the TWAMP session."""
     extra_vars = {
-        "subscription": subscription,
+        "subscription": json.loads(json_dumps(subscription)),
         "process_id": process_id,
         "dry_run": True,
+        "verb": "deploy",
         "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy TWAMP",
     }
 
@@ -58,9 +62,10 @@ def deploy_twamp_dry(subscription: Iptrunk, process_id: UUIDstr, callback_route:
 def deploy_twamp_real(subscription: Iptrunk, process_id: UUIDstr, callback_route: str, tt_number: str) -> State:
     """Deploy the TWAMP session."""
     extra_vars = {
-        "subscription": subscription,
+        "subscription": json.loads(json_dumps(subscription)),
         "process_id": process_id,
         "dry_run": False,
+        "verb": "deploy",
         "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy TWAMP",
     }
 
@@ -74,6 +79,24 @@ def deploy_twamp_real(subscription: Iptrunk, process_id: UUIDstr, callback_route
     return {"subscription": subscription}
 
 
+@step("Check TWAMP status on both sides")
+def check_twamp_status(subscription: Iptrunk, callback_route: str) -> State:
+    """Check TWAMP session."""
+    extra_vars = {
+        "subscription": json.loads(json_dumps(subscription)),
+        "verb": "check_twamp",
+    }
+
+    inventory = (
+        f"{subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}"
+        f"\n{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}"
+    )
+
+    execute_playbook("deploy_twamp.yaml", callback_route, inventory, extra_vars)
+
+    return {"subscription": subscription}
+
+
 @workflow(
     "Deploy TWAMP",
     initial_input_form=wrap_modify_initial_input_form(_initial_input_form_generator),
@@ -90,6 +113,7 @@ def deploy_twamp() -> StepList:
         >> unsync
         >> lso_interaction(deploy_twamp_dry)
         >> lso_interaction(deploy_twamp_real)
+        >> lso_interaction(check_twamp_status)
         >> resync
         >> done
     )

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -15,7 +15,6 @@ from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Choice, Label, UniqueConstrainedList
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
-from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, conditional, done, init, inputstep
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
@@ -28,7 +27,6 @@ from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.products.product_types.router import Router
 from gso.services import infoblox
-from gso.services.infoblox import DeletionError
 from gso.services.lso_client import execute_playbook, lso_interaction
 from gso.services.netbox_client import NetboxClient
 from gso.services.subscriptions import get_active_router_subscriptions
@@ -609,12 +607,7 @@ def update_ipam(subscription: Iptrunk, replace_index: int, new_node: Router, new
     v6_addr = subscription.iptrunk.iptrunk_ipv6_network[replace_index + 1]
 
     #  Out with the old
-    try:
-        infoblox.delete_host_by_ip(subscription.iptrunk.iptrunk_ipv4_network[replace_index])
-    except DeletionError as e:
-        msg = "Failed to delete record from Infoblox."
-        raise ProcessFailureError(msg) from e
-
+    infoblox.delete_host_by_ip(subscription.iptrunk.iptrunk_ipv4_network[replace_index])
     #  And in with the new
     new_fqdn = f"{new_lag_interface}-0.{new_node.router.router_fqdn}"
     comment = str(subscription.subscription_id)