Skip to content
Snippets Groups Projects
Verified Commit 67fa386e authored by Neda Moeini's avatar Neda Moeini Committed by Karel van Klink
Browse files

Update NREN L3 Service migrate workflows and fix ruff errors.

parent 3cfaeb00
No related branches found
No related tags found
1 merge request!286Add Edge Port, GÉANT IP and IAS products
......@@ -40,4 +40,3 @@ class ImportedEdgePort(
"""An imported Edge Port that is currently active."""
edge_port: EdgePortBlock
......@@ -7,7 +7,7 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products import ProductName
from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, ImportedNRENL3CoreService
from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreService
from gso.services.subscriptions import get_product_id_by_name
......
"""A modification workflow that migrates a GÉANT IP subscription to a different set of Edge Ports."""
"""A modification workflow that migrates a L3 Core Service to a new set of Edge Ports."""
from typing import Annotated
......@@ -20,14 +20,14 @@ from gso.utils.types.tt_number import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
"""Gather input from the operator on what new Edge Ports this GÉANT IP should migrate to."""
"""Gather input from the operator on what new Edge Ports this L3 Core Service should be migrated to."""
subscription = NRENL3CoreService.from_subscription(subscription_id)
partner_id = subscription.customer_id
edge_port_count = len(subscription.geant_ip.geant_ip_ap_list)
edge_port_count = len(subscription.nren_l3_core_service.nren_ap_list)
def _new_edge_port_selector(pid: UUIDstr) -> Choice:
existing_ep_name_list = [
ap.geant_ip_sbp.edge_port.owner_subscription_id for ap in subscription.geant_ip.geant_ip_ap_list
ap.sbp.edge_port.owner_subscription_id for ap in subscription.nren_l3_core_service.nren_ap_list
]
edge_port_subscriptions = list(
filter(
......@@ -55,7 +55,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
return edge_ports
class NRENL3CoreServiceEdgePortSelectionForm(FormPage):
model_config = ConfigDict(title="Migrating GÉANT IP to a new set of Edge Ports")
model_config = ConfigDict(title=f"Migrating {subscription.product.name} to a new set of Edge Ports")
tt_number: TTNumber
divider: Divider = Field(None, exclude=True)
......@@ -65,12 +65,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
Len(min_length=edge_port_count, max_length=edge_port_count),
] = [ # noqa: RUF012
NewEdgePortSelection(
old_edge_port=f"{
EdgePort.from_subscription(ap.geant_ip_sbp.edge_port.owner_subscription_id).description
} ({ap.nren_ap_type})",
old_edge_port=f"{EdgePort.from_subscription(ap.sbp.edge_port.owner_subscription_id).description} ({
ap.ap_type
})",
new_edge_port="",
)
for ap in subscription.geant_ip.geant_ip_ap_list
for ap in subscription.nren_l3_core_service.nren_ap_list
]
ep_user_input = yield NRENL3CoreServiceEdgePortSelectionForm
......@@ -82,7 +82,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
def update_subscription_model(subscription: NRENL3CoreService, edge_port_selection: list[dict]) -> State:
"""Update the subscription model with the new list of Access Ports."""
for index, selected_port in enumerate(edge_port_selection):
subscription.geant_ip.geant_ip_ap_list[index].geant_ip_sbp.edge_port = EdgePort.from_subscription(
subscription.nren_l3_core_service.nren_ap_list[index].sbp.edge_port = EdgePort.from_subscription(
selected_port["new_edge_port"]
).edge_port
......@@ -90,10 +90,10 @@ def update_subscription_model(subscription: NRENL3CoreService, edge_port_selecti
@workflow(
"Migrate GÉANT IP",
"Migrate NREN L3 Core Service",
initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
target=Target.MODIFY,
)
def migrate_nren_l3_core_service() -> StepList:
"""Migrate a GÉANT IP to a new set of Edge Ports."""
"""Migrate a :term:`NREN` L3 Core Service to a new set of Edge Ports."""
return begin >> store_process_subscription(Target.MODIFY) >> unsync >> update_subscription_model >> resync >> done
......@@ -125,7 +125,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class BindingPortModificationForm(FormPage):
model_config = ConfigDict(
title=f"{product_name} - Modify Edge Port configuration "
f"({access_port_index + 1}/{len(input_ap_list)})"
f"({access_port_index + 1}/{len(input_ap_list)})"
)
current_ep_label: Label = Field(
f"Currently configuring on {access_port.sbp.edge_port.edge_port_description} "
......
......@@ -10,7 +10,7 @@ from gso.products import EdgePort, ProductName
from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
from gso.products.product_blocks.service_binding_port import ServiceBindingPort
from gso.products.product_types.nren_l3_core_service import NRENL3CoreServiceInactive, ImportedNRENL3CoreService
from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreServiceInactive
from gso.services import subscriptions
from gso.utils.shared_enums import APType, SBPType
from gso.utils.types.ip_address import IPAddress
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment