Modify commercial peer

Modification workflow for a Commercial Peer subscription.

Operation

Bases: strEnum

The three operations that can be performed to modify a Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

class Operation(strEnum):
    """The three operations that can be performed to modify a Commercial Peer subscription."""

    ADD = "Add a Peering Connection"
    REMOVE = "Remove a Peering Connection"
    EDIT = "Edit an existing Peering Connection"

make_peering_connection_modification_form(*, product_name, selected_placement_port, current_pc, commercial_peer, general_peering_connection)

Build a dynamic form that only exposes IPv4/IPv6 sections that actually exist.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

def make_peering_connection_modification_form(  # noqa: PLR0915
    *,
    product_name: str,
    selected_placement_port: SubscriptionModel,
    current_pc: PeeringConnection,
    commercial_peer: CommercialPeerBlock,
    general_peering_connection: FormPage,
) -> type[SubmitFormPage]:
    """Build a dynamic form that only exposes IPv4/IPv6 sections that actually exist."""
    v4 = current_pc.bgp_session_v4
    v6 = current_pc.bgp_session_v6

    annotations: dict[str, Any] = {}
    namespace: dict[str, Any] = {"__annotations__": annotations}

    namespace["model_config"] = ConfigDict(title=f"{product_name} - Modify BGP Sessions")

    # Always-present label
    annotations["placement_port"] = Label
    namespace["placement_port"] = f"Peering Connection Placement Port: {selected_placement_port.description}"

    # IPv4 section
    if v4 is not None:
        annotations["divider_a"] = Divider
        namespace["divider_a"] = Field(None, exclude=True)

        annotations["label_a"] = Label
        namespace["label_a"] = Field("IPv4 settings for BGP", exclude=True)

        annotations["v4_bgp_peer_address"] = IPv4AddressType
        namespace["v4_bgp_peer_address"] = Field(
            IPv4AddressType(v4.peer_address),
            exclude=True,
        )

        annotations["v4_bgp_authentication_key"] = str | None
        namespace["v4_bgp_authentication_key"] = Field(
            v4.authentication_key,
            exclude=True,
        )

        annotations["v4_bgp_has_custom_policies"] = bool
        namespace["v4_bgp_has_custom_policies"] = Field(
            v4.has_custom_policies,
            exclude=True,
        )

        annotations["v4_bgp_bfd_enabled"] = bool
        namespace["v4_bgp_bfd_enabled"] = Field(
            v4.bfd_enabled,
            exclude=True,
        )

        annotations["v4_bgp_multipath_enabled"] = bool
        namespace["v4_bgp_multipath_enabled"] = Field(
            v4.multipath_enabled,
            exclude=True,
        )

        annotations["v4_bgp_prefix_limit"] = read_only_field(commercial_peer.prefix_limit_v4)

        annotations["v4_bgp_ttl_security"] = TTL | None
        namespace["v4_bgp_ttl_security"] = Field(
            v4.ttl_security,
            exclude=True,
        )

        annotations["v4_bgp_is_passive"] = bool
        namespace["v4_bgp_is_passive"] = Field(
            general_peering_connection.session_state in PassiveSessionStates,  # type: ignore[attr-defined]
            exclude=True,
        )

        annotations["v4_bgp_send_default_route"] = bool
        namespace["v4_bgp_send_default_route"] = Field(
            v4.send_default_route,
            exclude=True,
        )

        annotations["v4_bgp_add_v4_multicast"] = bool
        namespace["v4_bgp_add_v4_multicast"] = Field(
            bool(IPFamily.V4MULTICAST in v4.families),
            exclude=True,
        )

        # Computed field
        def v4_bgp_peer(self: Any) -> IPv4BGPPeer:
            return IPv4BGPPeer(
                peer_address=self.v4_bgp_peer_address,
                authentication_key=self.v4_bgp_authentication_key,
                has_custom_policies=self.v4_bgp_has_custom_policies,
                bfd_enabled=self.v4_bgp_bfd_enabled,
                multipath_enabled=self.v4_bgp_multipath_enabled,
                prefix_limit=self.v4_bgp_prefix_limit,
                ttl_security=self.v4_bgp_ttl_security,
                is_passive=self.v4_bgp_is_passive,
                send_default_route=self.v4_bgp_send_default_route,
                add_v4_multicast=self.v4_bgp_add_v4_multicast,
            )

        namespace["v4_bgp_peer"] = computed_field(v4_bgp_peer)

    # IPv6 section
    if v6 is not None:
        annotations["divider_b"] = Divider
        namespace["divider_b"] = Field(None, exclude=True)

        annotations["label_b"] = Label
        namespace["label_b"] = Field("IPv6 settings for BGP", exclude=True)

        annotations["v6_bgp_peer_address"] = IPv6AddressType
        namespace["v6_bgp_peer_address"] = Field(
            IPv6AddressType(v6.peer_address),
            exclude=True,
        )

        annotations["v6_bgp_authentication_key"] = str | None
        namespace["v6_bgp_authentication_key"] = Field(
            v6.authentication_key,
            exclude=True,
        )

        annotations["v6_bgp_has_custom_policies"] = bool
        namespace["v6_bgp_has_custom_policies"] = Field(
            v6.has_custom_policies,
            exclude=True,
        )

        annotations["v6_bgp_bfd_enabled"] = bool
        namespace["v6_bgp_bfd_enabled"] = Field(
            v6.bfd_enabled,
            exclude=True,
        )

        annotations["v6_bgp_multipath_enabled"] = bool
        namespace["v6_bgp_multipath_enabled"] = Field(
            v6.multipath_enabled,
            exclude=True,
        )

        annotations["v6_bgp_prefix_limit"] = read_only_field(commercial_peer.prefix_limit_v6)

        annotations["v6_bgp_ttl_security"] = TTL | None
        namespace["v6_bgp_ttl_security"] = Field(
            v6.ttl_security,
            exclude=True,
        )

        annotations["v6_bgp_is_passive"] = bool
        namespace["v6_bgp_is_passive"] = Field(
            general_peering_connection.session_state in PassiveSessionStates,  # type: ignore[attr-defined]
            exclude=True,
        )

        annotations["v6_bgp_send_default_route"] = bool
        namespace["v6_bgp_send_default_route"] = Field(
            v6.send_default_route,
            exclude=True,
        )

        annotations["v6_bgp_add_v6_multicast"] = bool
        namespace["v6_bgp_add_v6_multicast"] = Field(
            bool(IPFamily.V6MULTICAST in v6.families),
            exclude=True,
        )

        def v6_bgp_peer(self: Any) -> IPv6BGPPeer:
            return IPv6BGPPeer(
                peer_address=self.v6_bgp_peer_address,
                authentication_key=self.v6_bgp_authentication_key,
                has_custom_policies=self.v6_bgp_has_custom_policies,
                bfd_enabled=self.v6_bgp_bfd_enabled,
                multipath_enabled=self.v6_bgp_multipath_enabled,
                prefix_limit=self.v6_bgp_prefix_limit,
                ttl_security=self.v6_bgp_ttl_security,
                is_passive=self.v6_bgp_is_passive,
                send_default_route=self.v6_bgp_send_default_route,
                add_v6_multicast=self.v6_bgp_add_v6_multicast,
            )

        namespace["v6_bgp_peer"] = computed_field(v6_bgp_peer)

    # Create the dynamic form class
    PeeringConnectionModificationForm = cast(  # noqa: N806
        type[SubmitFormPage],
        type(
            "PeeringConnectionModificationForm",
            (SubmitFormPage,),
            namespace,
        ),
    )
    PeeringConnectionModificationForm.model_rebuild()
    return PeeringConnectionModificationForm

initial_input_form_generator(subscription_id)

Get input about added, removed, and modified Placement Ports for a Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Get input about added, removed, and modified Placement Ports for a Commercial Peer subscription."""
    subscription = CommercialPeer.from_subscription(subscription_id)
    commercial_peer = subscription.commercial_peer
    product_name = subscription.product.name

    class OperationSelectionForm(FormPage):
        model_config = ConfigDict(title="Modify Commercial Peer")

        tt_number: TTNumber
        operation: Operation
        run_ansible_steps: bool = True

    def existing_peering_connection_selector() -> type[Choice]:
        """Generate a dropdown selector for choosing an existing Peering Connection in an input form."""

        def _format_peering_connection_label(pc: PeeringConnection) -> str:
            description = SubscriptionModel.from_subscription(pc.placement_port.owner_subscription_id).description

            parts: list[str] = [description]

            if pc.bgp_session_v4 is not None:
                parts.append(str(pc.bgp_session_v4.peer_address))

            if pc.bgp_session_v6 is not None:
                parts.append(str(pc.bgp_session_v6.peer_address))

            return " - ".join(parts)

        options = {
            str(pc.subscription_instance_id): _format_peering_connection_label(pc)
            for pc in commercial_peer.peering_connection_list
            if pc.placement_port
        }

        return cast(
            type[Choice],
            Choice.__call__(
                "Select a Peering Connection",
                zip(options.keys(), options.items(), strict=True),
            ),
        )

    initial_input = yield OperationSelectionForm

    match initial_input.operation:
        case Operation.ADD:

            class IPv4BGPPeerReadOnlyPrefixLimit(IPv4BGPPeer):
                prefix_limit: read_only_field(commercial_peer.prefix_limit_v4)  # type: ignore[valid-type]

            class IPv6BGPPeerReadOnlyPrefixLimit(IPv6BGPPeer):
                prefix_limit: read_only_field(commercial_peer.prefix_limit_v6)  # type: ignore[valid-type]

            class PartnerSelectionForm(FormPage):
                model_config = ConfigDict(title=f"Add a Placement Port to a {product_name}")
                label: Label = Field(
                    "Please select the partner who owns the Placement Port which is to be added.", exclude=True
                )
                placement_port_partner: partner_choice() = subscription.customer_id  # type: ignore[valid-type]

            partner_input = yield PartnerSelectionForm

            class AddPeeringConnectionForm(SubmitFormPage):
                model_config = ConfigDict(title=f"Add a Peering Connection to a {product_name}")
                bgp_session_v4: IPv4BGPPeerReadOnlyPrefixLimit | None = None
                bgp_session_v6: IPv6BGPPeerReadOnlyPrefixLimit | None = None
                minimum_hold_timer: NonNegativeInt | None = None
                session_state: SessionState
                placement_port: active_placement_port_selector(partner_id=partner_input.placement_port_partner)  # type: ignore[valid-type]

                @model_validator(mode="after")
                def validate_bgp_sessions(self) -> "AddPeeringConnectionForm":
                    if self.bgp_session_v4 is None and self.bgp_session_v6 is None:
                        message = "Either an IPv4 or an IPv6 BGP session must be provided."
                        raise ValueError(message)
                    return self

            user_input = yield AddPeeringConnectionForm

            return {
                "operation": initial_input.operation,
                "verb": "deploy",
                "tt_number": initial_input.tt_number,
                "added_peering_connection": user_input.model_dump(),
                "run_ansible_steps": initial_input.run_ansible_steps,
            }

        case Operation.REMOVE:

            class RemovePeeringConnectionForm(SubmitFormPage):
                model_config = ConfigDict(title=f"Remove a Peering Connection from a {product_name}")
                label: Label = Field(
                    f"Please select one of the Peering connections associated with this {product_name} that "
                    f"should get removed.",
                    exclude=True,
                )
                peering_connection: existing_peering_connection_selector()  # type: ignore[valid-type]

            user_input = yield RemovePeeringConnectionForm
            selected_peering_connection = PeeringConnection.from_db(user_input.peering_connection)

            return {
                "operation": initial_input.operation,
                "verb": "delete",
                "tt_number": initial_input.tt_number,
                "removed_peering_connection": user_input.peering_connection,
                "selected_peering_connection": selected_peering_connection,
                "run_ansible_steps": initial_input.run_ansible_steps,
            }

        case Operation.EDIT:

            class ModifyPeeringConnectionForm(FormPage):
                model_config = ConfigDict(title=f"Modify {product_name}")
                label: Label = Field(
                    f"Please select one of the peering connections associated with this {product_name} that "
                    f"should get modified.",
                    exclude=True,
                )
                peering_connection: existing_peering_connection_selector()  # type: ignore[valid-type]

            user_input = yield ModifyPeeringConnectionForm
            current_pc = PeeringConnection.from_db(user_input.peering_connection)
            selected_placement_port = SubscriptionModel.from_subscription(
                current_pc.placement_port.owner_subscription_id
            )

            class GeneralPeeringConnectionModificationForm(FormPage):
                model_config = ConfigDict(title=f"{product_name} - Modify Peering Connection")

                placement_port: Label = f"Peering Connection Placement Port: {selected_placement_port.description}"
                # FIXME: type hint workaround
                session_state: SessionState | str = current_pc.session_state
                minimum_hold_timer: int | None = current_pc.minimum_hold_timer

            general_peering_connection = yield GeneralPeeringConnectionModificationForm

            PeeringConnectionModificationForm = make_peering_connection_modification_form(  # noqa: N806
                product_name=product_name,
                selected_placement_port=selected_placement_port,
                current_pc=current_pc,
                commercial_peer=commercial_peer,
                general_peering_connection=general_peering_connection,
            )

            modified_peeing_connection = yield PeeringConnectionModificationForm
            return {
                "operation": initial_input.operation,
                "verb": "deploy",
                "modified_peering_connection": general_peering_connection.model_dump()
                | modified_peeing_connection.model_dump(),
                "modified_peering_connection_id": user_input.peering_connection,
                "tt_number": initial_input.tt_number,
                "run_ansible_steps": initial_input.run_ansible_steps,
            }

        case _:
            msg = f"Invalid operation selected: {initial_input.operation}"
            raise ValueError(msg)

create_new_peering_connection(subscription, added_peering_connection)

Add new Peering Connection to the Peering connection list of the specific Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("Instantiate new Peering Connection")
def create_new_peering_connection(subscription: CommercialPeer, added_peering_connection: dict[str, Any]) -> State:
    """Add new Peering Connection to the Peering connection list of the specific Commercial Peer subscription."""
    selected_placement_port = SubscriptionModel.from_subscription(added_peering_connection["placement_port"])
    new_peering_connection = PeeringConnection.new(
        subscription_id=uuid4(),
        bgp_session_v4=BGPSession.new(
            subscription_id=uuid4(),
            rtbh_enabled=True,
            is_multi_hop=True,
            **added_peering_connection["bgp_session_v4"],
        )
        if added_peering_connection.get("bgp_session_v4")
        else None,
        bgp_session_v6=BGPSession.new(
            subscription_id=uuid4(),
            rtbh_enabled=True,
            is_multi_hop=True,
            **added_peering_connection["bgp_session_v6"],
        )
        if added_peering_connection.get("bgp_session_v6")
        else None,
        minimum_hold_timer=added_peering_connection.get("minimum_hold_timer"),
        session_state=added_peering_connection.get("session_state"),
        placement_port=selected_placement_port.product_block,  # type: ignore[attr-defined]
    )
    subscription.commercial_peer.peering_connection_list.append(new_peering_connection)

    return {"subscription": subscription, "selected_peering_connection": new_peering_connection}

remove_old_peering_connection(subscription, removed_peering_connection)

Remove old Peering Connection from the Peering Connection list of the specific Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("Remove selected Peering Connection from Commercial Peer's Peering Connection list")
def remove_old_peering_connection(subscription: CommercialPeer, removed_peering_connection: UUID) -> State:
    """Remove old Peering Connection from the Peering Connection list of the specific Commercial Peer subscription."""
    peering_connection_list = subscription.commercial_peer.peering_connection_list
    peering_connection_list.remove(
        next(pc for pc in peering_connection_list if pc.subscription_instance_id == removed_peering_connection)
    )

    return {"subscription": subscription}

modify_existing_peering_connection(subscription, modified_peering_connection, modified_peering_connection_id)

Update the subscription model and modify the existing Peering Connection.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("Modify existing Peering Connection")
def modify_existing_peering_connection(
    subscription: CommercialPeer,
    modified_peering_connection: dict[str, Any],
    modified_peering_connection_id: UUID,
) -> State:
    """Update the subscription model and modify the existing Peering Connection."""
    current_pc = next(
        pc
        for pc in subscription.commercial_peer.peering_connection_list
        if pc.subscription_instance_id == modified_peering_connection_id
    )

    current_pc.minimum_hold_timer = modified_peering_connection["minimum_hold_timer"]
    current_pc.session_state = modified_peering_connection["session_state"]
    for attribute in modified_peering_connection.get("v4_bgp_peer", []):
        setattr(current_pc.bgp_session_v4, attribute, modified_peering_connection["v4_bgp_peer"][attribute])
    for attribute in modified_peering_connection.get("v6_bgp_peer", []):
        setattr(current_pc.bgp_session_v6, attribute, modified_peering_connection["v6_bgp_peer"][attribute])

    return {"subscription": subscription, "selected_peering_connection": current_pc}

scope_subscription(subscription, selected_peering_connection)

Scope the subscription model, and set the FQDN of the host to deploy on.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("Scope subscription for LSO interactions")
def scope_subscription(subscription: CommercialPeer, selected_peering_connection: dict[str, Any]) -> State:
    """Scope the subscription model, and set the FQDN of the host to deploy on."""
    partner_name = get_partner_by_id(subscription.customer_id).name
    placement_port = SubscriptionModel.from_subscription(
        selected_peering_connection["placement_port"]["owner_subscription_id"]
    )
    placement_port_partner_name = get_partner_by_id(placement_port.customer_id).name

    scoped_subscription = json.loads(json_dumps(subscription))
    scoped_subscription["commercial_peer"]["peering_connection_list"] = [selected_peering_connection]

    return {
        "scoped_subscription": scoped_subscription,
        "partner_name": partner_name,
        "placement_port_partner_name": placement_port_partner_name,
        "placement_port_fqdn": placement_port.l3_interface.edge_port.node.router_fqdn,  # type: ignore[attr-defined]
    }

deploy_commercial_peer_dry(workflow_name, scoped_subscription, process_id, tt_number, partner_name, placement_port_partner_name, placement_port_fqdn, verb)

Perform a dry run of deploying Commercial Peer configuration on a placement port.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("[DRY RUN] Deploy Commercial Peer on a Placement Port")
def deploy_commercial_peer_dry(
    workflow_name: str,
    scoped_subscription: dict[str, Any],
    process_id: UUIDstr,
    tt_number: TTNumber,
    partner_name: str,
    placement_port_partner_name: str,
    placement_port_fqdn: str,
    verb: str,
) -> LSOState:
    """Perform a dry run of deploying Commercial Peer configuration on a placement port."""
    extra_vars = {
        "subscription": scoped_subscription,
        "placement_port_partner_name": placement_port_partner_name,
        "partner_name": partner_name,
        "dry_run": True,
        "verb": verb,
        "is_redeploy_workflow": workflow_name.startswith("redeploy_"),
        "object": "bgp_peer",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deploy config for {scoped_subscription['description']}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_bgp_peers.yaml",
        "inventory": {"all": {"hosts": {placement_port_fqdn: None}}},
        "extra_vars": extra_vars,
    }

deploy_commercial_peer_real(workflow_name, scoped_subscription, process_id, tt_number, partner_name, placement_port_partner_name, placement_port_fqdn, verb)

Perform a dry run of deploying Commercial Peer configuration on a placement port.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@step("[FOR REAL] Deploy Commercial Peer on a Placement Port")
def deploy_commercial_peer_real(
    workflow_name: str,
    scoped_subscription: dict[str, Any],
    process_id: UUIDstr,
    tt_number: TTNumber,
    partner_name: str,
    placement_port_partner_name: str,
    placement_port_fqdn: str,
    verb: str,
) -> LSOState:
    """Perform a dry run of deploying Commercial Peer configuration on a placement port."""
    extra_vars = {
        "subscription": scoped_subscription,
        "placement_port_partner_name": placement_port_partner_name,
        "partner_name": partner_name,
        "dry_run": False,
        "verb": verb,
        "is_redeploy_workflow": workflow_name.startswith("redeploy_"),
        "object": "bgp_peer",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deploy config for {scoped_subscription['description']}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_bgp_peers.yaml",
        "inventory": {"all": {"hosts": {placement_port_fqdn: None}}},
        "extra_vars": extra_vars,
    }

modify_commercial_peer()

Modify Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/modify_commercial_peer.py

@modify_workflow("Modify Commercial Peer", initial_input_form=initial_input_form_generator)
def modify_commercial_peer() -> StepList:
    """Modify Commercial Peer subscription."""
    placement_port_is_added = conditional(lambda state: state["operation"] == Operation.ADD)
    placement_port_is_removed = conditional(lambda state: state["operation"] == Operation.REMOVE)
    placement_port_is_modified = conditional(lambda state: state["operation"] == Operation.EDIT)

    return (
        begin
        >> placement_port_is_added(create_new_peering_connection)
        >> placement_port_is_modified(modify_existing_peering_connection)
        >> scope_subscription
        >> run_ansible_steps(lso_interaction(deploy_commercial_peer_dry))
        >> run_ansible_steps(lso_interaction(deploy_commercial_peer_real))
        >> placement_port_is_removed(remove_old_peering_connection)
    )