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"

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."""
        options = {
            str(pc.subscription_instance_id): (
                f"{SubscriptionModel.from_subscription(pc.placement_port.owner_subscription_id).description} - "
                f"{pc.bgp_session_v4.peer_address} - "
                f"{pc.bgp_session_v6.peer_address}"
            )
            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
                bgp_session_v6: IPv6BGPPeerReadOnlyPrefixLimit
                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]

            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

            class PeeringConnectionModificationForm(SubmitFormPage):
                model_config = ConfigDict(title=f"{product_name} - Modify BGP Sessions")

                placement_port: Label = f"Peering Connection Placement Port: {selected_placement_port.description}"

                divider_a: Divider = Field(None, exclude=True)
                label_a: Label = Field("IPv4 settings for BGP", exclude=True)
                v4_bgp_peer_address: IPv4AddressType = Field(
                    IPv4AddressType(current_pc.bgp_session_v4.peer_address), exclude=True
                )
                v4_bgp_authentication_key: str | None = Field(
                    current_pc.bgp_session_v4.authentication_key, exclude=True
                )
                v4_bgp_has_custom_policies: bool = Field(current_pc.bgp_session_v4.has_custom_policies, exclude=True)
                v4_bgp_bfd_enabled: bool = Field(current_pc.bgp_session_v4.bfd_enabled, exclude=True)
                v4_bgp_multipath_enabled: bool = Field(current_pc.bgp_session_v4.multipath_enabled, exclude=True)
                v4_bgp_prefix_limit: read_only_field(commercial_peer.prefix_limit_v4)  # type: ignore[valid-type]
                v4_bgp_ttl_security: TTL | None = Field(current_pc.bgp_session_v4.ttl_security, exclude=True)
                v4_bgp_is_passive: bool = Field(
                    general_peering_connection.session_state in PassiveSessionStates, exclude=True
                )
                v4_bgp_send_default_route: bool = Field(current_pc.bgp_session_v4.send_default_route, exclude=True)
                v4_bgp_add_v4_multicast: bool = Field(
                    bool(IPFamily.V4MULTICAST in current_pc.bgp_session_v4.families), exclude=True
                )

                divider_b: Divider = Field(None, exclude=True)
                label_b: Label = Field("IPv6 settings for BGP", exclude=True)
                v6_bgp_peer_address: IPv6AddressType = Field(
                    IPv6AddressType(current_pc.bgp_session_v6.peer_address), exclude=True
                )
                v6_bgp_authentication_key: str | None = Field(
                    current_pc.bgp_session_v6.authentication_key, exclude=True
                )
                v6_bgp_has_custom_policies: bool = Field(current_pc.bgp_session_v6.has_custom_policies, exclude=True)
                v6_bgp_bfd_enabled: bool = Field(current_pc.bgp_session_v6.bfd_enabled, exclude=True)
                v6_bgp_multipath_enabled: bool = Field(current_pc.bgp_session_v6.multipath_enabled, exclude=True)
                v6_bgp_prefix_limit: read_only_field(commercial_peer.prefix_limit_v6)  # type: ignore[valid-type]
                v6_bgp_ttl_security: TTL | None = Field(current_pc.bgp_session_v6.ttl_security, exclude=True)
                v6_bgp_is_passive: bool = Field(
                    general_peering_connection.session_state in PassiveSessionStates, exclude=True
                )
                v6_bgp_send_default_route: bool = Field(current_pc.bgp_session_v6.send_default_route, exclude=True)
                v6_bgp_add_v6_multicast: bool = Field(
                    bool(IPFamily.V6MULTICAST in current_pc.bgp_session_v6.families), exclude=True
                )

                @computed_field  # type: ignore[prop-decorator]
                @property
                def v4_bgp_peer(self) -> 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,
                    )

                @computed_field  # type: ignore[prop-decorator]
                @property
                def v6_bgp_peer(self) -> 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,
                    )

            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"],
        ),
        bgp_session_v6=BGPSession.new(
            subscription_id=uuid4(),
            rtbh_enabled=True,
            is_multi_hop=True,
            **added_peering_connection["bgp_session_v6"],
        ),
        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["v4_bgp_peer"]:
        setattr(current_pc.bgp_session_v4, attribute, modified_peering_connection["v4_bgp_peer"][attribute])
    for attribute in modified_peering_connection["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(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(
    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,
        "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(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(
    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,
        "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)
    )