def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
"""Get input about added, removed, and modified Access Ports."""
subscription = SubscriptionModel.from_subscription(subscription_id)
product_name = subscription.product.name
class OperationSelectionForm(FormPage):
model_config = ConfigDict(title="Modify Edge Port")
tt_number: TTNumber
operation: Operation
include_ansible_playbook_sbp_config: bool = True
include_ansible_playbook_bgp_config: bool = True
@model_validator(mode="after")
def validate_no_duplicate_subscriptions(self) -> "OperationSelectionForm":
"""Warn if duplicate subscriptions exist for this partner + product.
For modify workflows, we allow the operation to proceed but warn the operator
that duplicate subscriptions exist and they should use the merge workflow.
"""
existing_subscriptions = get_active_l3_subscriptions_by_partner_and_product(
subscription.customer_id,
ProductName(product_name),
)
# Exclude current subscription from count
other_subscriptions = [
s for s in existing_subscriptions if s.subscription_id != subscription.subscription_id
]
if len(other_subscriptions) >= 1:
partner_name = get_partner_by_id(subscription.customer_id).name
subscription_ids = [str(s.subscription_id) for s in other_subscriptions]
msg = (
f"Warning: {len(other_subscriptions)} additional active {product_name} subscription(s) found "
f"for partner {partner_name} ({subscription.customer_id}): {', '.join(subscription_ids)}. "
"\n\nThe target model requires ONE subscription per partner with multiple Access Ports. "
"Consider using the 'Merge L3 Core Subscriptions' workflow to consolidate these subscriptions."
)
raise ValueError(msg)
return self
def access_port_selector() -> TypeAlias:
"""Generate a dropdown selector for choosing an Access Port in an input form."""
access_ports = subscription.l3_core.ap_list # type: ignore[attr-defined]
options = {
str(access_port.subscription_instance_id): (
f"{access_port.sbp.gs_id} on "
f"{EdgePort.from_subscription(access_port.sbp.edge_port.owner_subscription_id).description} "
f"({access_port.ap_type})"
)
for access_port in access_ports
}
return cast(
type[Choice],
Choice.__call__(
"Select an Access Port",
zip(options.keys(), options.items(), strict=True),
),
)
class BFDInputModel(BaseModel):
bfd_enabled: bool = False
bfd_interval_rx: int | None = None
bfd_interval_tx: int | None = None
bfd_multiplier: int | None = None
initial_input = yield OperationSelectionForm
match initial_input.operation:
case Operation.ADD:
class PartnerSelectionForm(FormPage):
model_config = ConfigDict(title=f"Add an Edge Port to a {product_name}")
label: Label = Field(
"Please select the partner who owns the Edge Port which is to be added.", exclude=True
)
edge_port_partner: partner_choice() = subscription.customer_id # type: ignore[valid-type]
partner_input = yield PartnerSelectionForm
class AccessPortListItem(BaseModel):
edge_port: str
ap_type: str
custom_service_name: str
def available_new_edge_port_selector() -> TypeAlias:
"""Generate a dropdown selector for choosing an active Edge Port in an input form."""
edge_ports = get_active_edge_port_subscriptions(partner_id=partner_input.edge_port_partner)
options = {
str(edge_port.subscription_id): edge_port.description
for edge_port in edge_ports
if edge_port.subscription_id
not in [ap.sbp.edge_port.owner_subscription_id for ap in subscription.l3_core.ap_list] # type: ignore[attr-defined]
}
return cast(
type[Choice],
Choice.__call__(
"Select an Edge Port",
zip(options.keys(), options.items(), strict=True),
),
)
def existing_ap_list() -> type[list]:
return cast(
type[list],
read_only_list([
AccessPortListItem(
edge_port=EdgePort.from_subscription(
access_port.sbp.edge_port.owner_subscription_id
).description,
ap_type=access_port.ap_type.value,
custom_service_name=access_port.custom_service_name or "",
)
for access_port in subscription.l3_core.ap_list # type: ignore[attr-defined]
]),
)
class AddAccessPortForm(FormPage):
model_config = ConfigDict(title=f"Add an Edge Port to a {product_name}")
existing_access_ports: existing_ap_list() # type: ignore[valid-type]
divider_a: Divider = Field(exclude=True)
label_a: Label = Field(
"Please use the fields below to configure a new Access Port, in addition to the existing ones "
"listed above.",
exclude=True,
)
edge_port: available_new_edge_port_selector() # type: ignore[valid-type]
ap_type: APType
generate_gs_id: bool = True
gs_id: IMPORTED_GS_ID | None = None
custom_service_name: str | None = None
is_tagged: bool = False
vlan_id: VLAN_ID
ipv4_address: IPv4AddressType
ipv4_mask: IPv4Netmask
ipv6_address: IPv6AddressType
ipv6_mask: IPv6Netmask
custom_firewall_filters: bool = False
divider_b: Divider = Field(None, exclude=True)
label_b: Label = Field("IPv4 settings for BFD and BGP", exclude=True)
v4_bfd_settings: BFDInputModel
v4_bgp_peer: IPv4BGPPeer
divider_c: Divider = Field(None, exclude=True)
label_c: Label = Field("IPv6 settings for BFD and BGP", exclude=True)
v6_bfd_settings: BFDInputModel
v6_bgp_peer: IPv6BGPPeer
@model_validator(mode="before")
def validate_gs_id(cls, input_data: dict[str, Any]) -> dict[str, Any]:
gs_id = input_data.get("gs_id")
generate_gs_id = input_data.get("generate_gs_id", True)
if generate_gs_id and gs_id:
error_message = (
"You cannot provide a GS ID manually while the 'Auto-generate GS ID' option is enabled."
"Please either uncheck 'Auto-generate GS ID' or remove the manual GS ID."
)
raise ValueError(error_message)
return input_data
@field_validator("edge_port")
def selected_edge_port_is_new(cls, value: UUIDstr) -> UUIDstr:
if value in [str(ap.sbp.edge_port.owner_subscription_id) for ap in subscription.l3_core.ap_list]: # type: ignore[attr-defined]
error_message = (
f"This {product_name} service is already deployed on "
f"{EdgePort.from_subscription(value).description}."
)
raise ValueError(error_message)
return value
user_input = yield AddAccessPortForm
return {
"operation": initial_input.operation,
"verb": "deploy",
"tt_number": initial_input.tt_number,
"added_access_port": user_input.model_dump(),
"include_ansible_playbook_sbp_config": initial_input.include_ansible_playbook_sbp_config,
"include_ansible_playbook_bgp_config": initial_input.include_ansible_playbook_bgp_config,
}
case Operation.REMOVE:
class RemoveAccessPortForm(FormPage):
model_config = ConfigDict(title=f"Remove an Edge Port from a {product_name}")
label: Label = Field(
f"Please select one of the Access Ports associated with this {product_name} that should get "
f"removed.",
exclude=True,
)
access_port: access_port_selector() # type: ignore[valid-type]
user_input = yield RemoveAccessPortForm
return {
"operation": initial_input.operation,
"verb": "delete",
"tt_number": initial_input.tt_number,
"active_ap": user_input.access_port,
"edge_port_fqdn_list": [AccessPort.from_db(user_input.access_port).sbp.edge_port.node.router_fqdn],
"include_ansible_playbook_sbp_config": initial_input.include_ansible_playbook_sbp_config,
"include_ansible_playbook_bgp_config": initial_input.include_ansible_playbook_bgp_config,
}
case Operation.EDIT:
class SelectModifyAccessPortForm(FormPage):
model_config = ConfigDict(title=f"Modify {product_name}")
label: Label = Field(
f"Please select one of the Access Ports associated with this {product_name} to be modified.",
exclude=True,
)
access_port: access_port_selector() # type: ignore[valid-type]
user_input = yield SelectModifyAccessPortForm
current_ap = AccessPort.from_db(user_input.access_port)
v4_peer = next(peer for peer in current_ap.sbp.bgp_session_list if IPFamily.V4UNICAST in peer.families)
v6_peer = next(peer for peer in current_ap.sbp.bgp_session_list if IPFamily.V6UNICAST in peer.families)
class BindingPortModificationForm(FormPage):
model_config = ConfigDict(title=f"{product_name} - Modify Edge Port configuration")
current_ep_label: Label = Field(
f'Currently configuring on Edge Port "{current_ap.sbp.edge_port.edge_port_description}"',
exclude=True,
)
gs_id: str = current_ap.sbp.gs_id
custom_service_name: str | None = current_ap.custom_service_name
is_tagged: bool = current_ap.sbp.is_tagged
ap_type: APType = current_ap.ap_type
# The SBP model does not require these five fields, but in the case of L3 Core Services this will never
# occur since it's a layer 3 service. The ignore statements are there to put our type checker at ease.
vlan_id: VLAN_ID = current_ap.sbp.vlan_id # type: ignore[assignment]
ipv4_address: IPv4AddressType = current_ap.sbp.ipv4_address # type: ignore[assignment]
ipv4_mask: IPv4Netmask = current_ap.sbp.ipv4_mask # type: ignore[assignment]
ipv6_address: IPv6AddressType = current_ap.sbp.ipv6_address # type: ignore[assignment]
ipv6_mask: IPv6Netmask = current_ap.sbp.ipv6_mask # type: ignore[assignment]
custom_firewall_filters: bool = current_ap.sbp.custom_firewall_filters
divider_a: Divider = Field(None, exclude=True)
label_a: Label = Field("IPv4 settings for BFD and BGP", exclude=True)
v4_bfd_enabled: bool = Field(current_ap.sbp.v4_bfd_settings.bfd_enabled, exclude=True)
v4_bfd_multiplier: int | None = Field(current_ap.sbp.v4_bfd_settings.bfd_multiplier, exclude=True)
v4_bfd_interval_rx: int | None = Field(current_ap.sbp.v4_bfd_settings.bfd_interval_rx, exclude=True)
v4_bfd_interval_tx: int | None = Field(current_ap.sbp.v4_bfd_settings.bfd_interval_tx, exclude=True)
v4_bgp_peer_address: IPv4AddressType = Field(IPv4AddressType(v4_peer.peer_address), exclude=True)
v4_bgp_authentication_key: str | None = Field(v4_peer.authentication_key, exclude=True)
v4_bgp_has_custom_policies: bool = Field(v4_peer.has_custom_policies, exclude=True)
v4_bgp_bfd_enabled: bool = Field(v4_peer.bfd_enabled, exclude=True)
v4_bgp_multipath_enabled: bool = Field(v4_peer.multipath_enabled, exclude=True)
v4_bgp_prefix_limit: NonNegativeInt | None = Field(v4_peer.prefix_limit, exclude=True)
v4_bgp_ttl_security: TTL | None = Field(v4_peer.ttl_security, exclude=True)
v4_bgp_is_passive: bool = Field(v4_peer.is_passive, exclude=True)
v4_bgp_send_default_route: bool = Field(v4_peer.send_default_route, exclude=True)
v4_bgp_add_v4_multicast: bool = Field(bool(IPFamily.V4MULTICAST in v4_peer.families), exclude=True)
divider_b: Divider = Field(None, exclude=True)
label_b: Label = Field("IPv6 settings for BFD and BGP", exclude=True)
v6_bfd_enabled: bool = Field(current_ap.sbp.v6_bfd_settings.bfd_enabled, exclude=True)
v6_bfd_multiplier: int | None = Field(current_ap.sbp.v6_bfd_settings.bfd_multiplier, exclude=True)
v6_bfd_interval_rx: int | None = Field(current_ap.sbp.v6_bfd_settings.bfd_interval_rx, exclude=True)
v6_bfd_interval_tx: int | None = Field(current_ap.sbp.v6_bfd_settings.bfd_interval_tx, exclude=True)
v6_bgp_peer_address: IPv6AddressType = Field(IPv6AddressType(v6_peer.peer_address), exclude=True)
v6_bgp_authentication_key: str | None = Field(v6_peer.authentication_key, exclude=True)
v6_bgp_has_custom_policies: bool = Field(v6_peer.has_custom_policies, exclude=True)
v6_bgp_bfd_enabled: bool = Field(v6_peer.bfd_enabled, exclude=True)
v6_bgp_multipath_enabled: bool = Field(v6_peer.multipath_enabled, exclude=True)
v6_bgp_prefix_limit: NonNegativeInt | None = Field(v6_peer.prefix_limit, exclude=True)
v6_bgp_ttl_security: TTL | None = Field(v6_peer.ttl_security, exclude=True)
v6_bgp_is_passive: bool = Field(v6_peer.is_passive, exclude=True)
v6_bgp_send_default_route: bool = Field(v6_peer.send_default_route, exclude=True)
v6_bgp_add_v6_multicast: bool = Field(bool(IPFamily.V6MULTICAST in v6_peer.families), exclude=True)
@computed_field # type: ignore[prop-decorator]
@property
def v4_bfd_settings(self) -> BFDInputModel:
return BFDInputModel(
bfd_enabled=self.v4_bfd_enabled,
bfd_multiplier=self.v4_bfd_multiplier,
bfd_interval_rx=self.v4_bfd_interval_rx,
bfd_interval_tx=self.v4_bfd_interval_tx,
)
@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_bfd_settings(self) -> BFDInputModel:
return BFDInputModel(
bfd_enabled=self.v6_bfd_enabled,
bfd_multiplier=self.v6_bfd_multiplier,
bfd_interval_rx=self.v6_bfd_interval_rx,
bfd_interval_tx=self.v6_bfd_interval_tx,
)
@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,
)
binding_port_input_form = yield BindingPortModificationForm
return {
"operation": initial_input.operation,
"verb": "deploy",
"active_ap": user_input.access_port,
"tt_number": initial_input.tt_number,
"modified_sbp": binding_port_input_form.model_dump(),
"include_ansible_playbook_sbp_config": initial_input.include_ansible_playbook_sbp_config,
"include_ansible_playbook_bgp_config": initial_input.include_ansible_playbook_bgp_config,
}
case _:
msg = f"Invalid operation selected: {initial_input.operation}"
raise ValueError(msg)