diff --git a/gso/workflows/l3_core_service/base_create_l3_core_service.py b/gso/workflows/l3_core_service/base_create_l3_core_service.py index 90221ec0ed482de386762c097c88eb7a9e8298e0..0fb2c84357aa58e891f6578ae0c99b0de91838cc 100644 --- a/gso/workflows/l3_core_service/base_create_l3_core_service.py +++ b/gso/workflows/l3_core_service/base_create_l3_core_service.py @@ -38,12 +38,17 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: model_config = ConfigDict(title=f"{product_name} - Select partner") tt_number: TTNumber + label_a: Label = Field(f"Please select the partner for this {product_name}.", exclude=True) partner: partner_choice() # type: ignore[valid-type] + label_b: Label = Field( + f"Please select the partner who owns the Edge Port this {product_name} will be deployed on.", exclude=True + ) + edge_port_partner: partner_choice() # type: ignore[valid-type] initial_user_input = yield CreateL3CoreServiceForm class EdgePortSelection(BaseModel): - edge_port: active_edge_port_selector(partner_id=initial_user_input.partner) # type: ignore[valid-type] + edge_port: active_edge_port_selector(partner_id=initial_user_input.edge_port_partner) # type: ignore[valid-type] ap_type: APType custom_service_name: str | None = None diff --git a/gso/workflows/l3_core_service/base_migrate_l3_core_service.py b/gso/workflows/l3_core_service/base_migrate_l3_core_service.py index de2cae8950883b8c43affa8631339902cf09bc0e..8a18543b36302693aa0a2e9bfb1078abde5cfdb1 100644 --- a/gso/workflows/l3_core_service/base_migrate_l3_core_service.py +++ b/gso/workflows/l3_core_service/base_migrate_l3_core_service.py @@ -25,6 +25,7 @@ from gso.products.product_types.edge_port import EdgePort from gso.services.lso_client import LSOState from gso.services.partners import get_partner_by_id from gso.services.subscriptions import get_active_edge_port_subscriptions +from gso.utils.helpers import partner_choice from gso.utils.types.tt_number import TTNumber from gso.utils.workflow_steps import IS_HUMAN_INITIATED_WF_KEY, MOODI_EXTRA_KWARGS_KEY, SKIP_MOODI_KEY from gso.workflows.shared import create_summary_form @@ -33,9 +34,17 @@ from gso.workflows.shared import create_summary_form def initial_input_form(subscription_id: UUIDstr) -> FormGenerator: """Gather input from the operator on what destination Edge Ports this L3 Core Service should be migrated to.""" subscription = SubscriptionModel.from_subscription(subscription_id) - partner_id = subscription.customer_id ap_list = subscription.l3_core.ap_list # type: ignore[attr-defined] + class PartnerSelectionForm(FormPage): + model_config = ConfigDict(title=f"Migrating a(n) {subscription.product.name} AP to a new Edge Port") + label: Label = Field( + "Please select the partner who owns the Edge Port which we are migrating to.", exclude=True + ) + edge_port_partner: partner_choice() = subscription.customer_id # type: ignore[valid-type] + + partner_input = yield PartnerSelectionForm + current_ep_list = { str( ap.sbp.edge_port.owner_subscription_id @@ -51,14 +60,16 @@ def initial_input_form(subscription_id: UUIDstr) -> FormGenerator: model_config = ConfigDict(title=f"Migrating a(n) {subscription.product.name} AP to a new Edge Port") tt_number: TTNumber - divider: Divider = Field(None, exclude=True) + divider_a: Divider = Field(None, exclude=True) skip_moodi: bool = False is_human_initiated_wf: bool = True - source_edge_port: source_edge_port_selector | str # type: ignore[valid-type] expected_number_of_ipv4_received_routes: int | None = None expected_number_of_ipv4_advertised_routes: int | None = None expected_number_of_ipv6_received_routes: int | None = None expected_number_of_ipv6_advertised_routes: int | None = None + divider_b: Divider = Field(None, exclude=True) + label: Label = Field("Source Edge Port", exclude=True) + source_edge_port: source_edge_port_selector | str # type: ignore[valid-type] source_ep_input = yield L3CoreServiceSourceEdgePortSelectionForm @@ -78,7 +89,8 @@ def initial_input_form(subscription_id: UUIDstr) -> FormGenerator: ) class L3CoreServiceEdgePortSelectionForm(FormPage): - destination_edge_port: _destination_edge_port_selector(partner_id) | str # type: ignore[valid-type] + label: Label = Field("Destination Edge Port", exclude=True) + destination_edge_port: _destination_edge_port_selector(partner_input.edge_port_partner) | str # type: ignore[valid-type] destination_ep_user_input = yield L3CoreServiceEdgePortSelectionForm if source_ep_input.is_human_initiated_wf: diff --git a/gso/workflows/l3_core_service/base_modify_l3_core_service.py b/gso/workflows/l3_core_service/base_modify_l3_core_service.py index e16364ad1095734f4904c7288d54f46f9dbd808c..4b04a4c6a2bb0c19fb8189ac1dc382c4af4483ea 100644 --- a/gso/workflows/l3_core_service/base_modify_l3_core_service.py +++ b/gso/workflows/l3_core_service/base_modify_l3_core_service.py @@ -15,6 +15,7 @@ from gso.products.product_blocks.l3_core_service import AccessPort from gso.products.product_blocks.service_binding_port import BFDSettings, ServiceBindingPort from gso.products.product_types.edge_port import EdgePort from gso.services.subscriptions import generate_unique_id, get_active_edge_port_subscriptions +from gso.utils.helpers import partner_choice from gso.utils.shared_enums import APType, SBPType from gso.utils.types.geant_ids import IMPORTED_GS_ID from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask @@ -115,6 +116,15 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: 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 @@ -122,7 +132,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: 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=subscription.customer_id) + edge_ports = get_active_edge_port_subscriptions(partner_id=partner_input.edge_port_partner) options = { str(edge_port.subscription_id): edge_port.description diff --git a/test/workflows/l3_core_service/test_create_l3_core_service.py b/test/workflows/l3_core_service/test_create_l3_core_service.py index a8ee5bc6a27b090acfaf936010999fc8ab0be7fe..b5865a74e417d2002aa666d030be5f956e456742 100644 --- a/test/workflows/l3_core_service/test_create_l3_core_service.py +++ b/test/workflows/l3_core_service/test_create_l3_core_service.py @@ -55,7 +55,7 @@ def test_create_l3_core_service_success( form_input_data = [ {"product": product_id}, - {"tt_number": faker.tt_number(), "partner": partner["partner_id"]}, + {"tt_number": faker.tt_number(), "partner": partner["partner_id"], "edge_port_partner": partner["partner_id"]}, {"edge_port": {"edge_port": edge_port_a, "ap_type": APType.PRIMARY, "custom_service_name": faker.sentence()}}, { "is_tagged": faker.boolean(), diff --git a/test/workflows/l3_core_service/test_migrate_l3_core_service.py b/test/workflows/l3_core_service/test_migrate_l3_core_service.py index eb1b292a90b769ffd091e1d677a91df16518803f..f9960a053f988d455898164fa0f195927719417d 100644 --- a/test/workflows/l3_core_service/test_migrate_l3_core_service.py +++ b/test/workflows/l3_core_service/test_migrate_l3_core_service.py @@ -35,6 +35,7 @@ def test_migrate_l3_core_service_success( subscription = SubscriptionModel.from_subscription(subscription_id) form_input_data = [ {"subscription_id": subscription_id}, + {"edge_port_partner": partner["partner_id"]}, { "tt_number": faker.tt_number(), "source_edge_port": subscription.l3_core.ap_list[0].sbp.edge_port.owner_subscription_id, @@ -86,6 +87,7 @@ def test_migrate_l3_core_service_scoped_emission( form_input_data = [ {"subscription_id": str(subscription.subscription_id)}, + {"edge_port_partner": partner["partner_id"]}, { "tt_number": faker.tt_number(), "source_edge_port": source_edge_port, diff --git a/test/workflows/l3_core_service/test_modify_l3_core_service.py b/test/workflows/l3_core_service/test_modify_l3_core_service.py index 93f8680b68dc20273030535499364f8a5a0e121f..b3421c79e0089cb7a05744bcc95053973a31a5a3 100644 --- a/test/workflows/l3_core_service/test_modify_l3_core_service.py +++ b/test/workflows/l3_core_service/test_modify_l3_core_service.py @@ -53,6 +53,7 @@ def test_modify_l3_core_service_add_new_edge_port_success( input_form_data = [ {"subscription_id": str(subscription.subscription_id)}, {"tt_number": faker.tt_number(), "operation": Operation.ADD}, + {"edge_port_partner": partner["partner_id"]}, { # Adding configuration for the new SBP "edge_port": str(new_edge_port), "ap_type": APType.BACKUP, @@ -94,12 +95,12 @@ def test_modify_l3_core_service_add_new_edge_port_success( ap_list = subscription.l3_core.ap_list new_ap = ap_list[-1] assert new_ap.ap_type == APType.BACKUP - assert new_ap.sbp.gs_id == input_form_data[2]["gs_id"] - assert new_ap.sbp.vlan_id == input_form_data[2]["vlan_id"] - assert str(new_ap.sbp.ipv4_address) == input_form_data[2]["ipv4_address"] - assert new_ap.sbp.ipv4_mask == input_form_data[2]["ipv4_mask"] - assert str(new_ap.sbp.ipv6_address) == input_form_data[2]["ipv6_address"] - assert new_ap.sbp.ipv6_mask == input_form_data[2]["ipv6_mask"] + assert new_ap.sbp.gs_id == input_form_data[3]["gs_id"] + assert new_ap.sbp.vlan_id == input_form_data[3]["vlan_id"] + assert str(new_ap.sbp.ipv4_address) == input_form_data[3]["ipv4_address"] + assert new_ap.sbp.ipv4_mask == input_form_data[3]["ipv4_mask"] + assert str(new_ap.sbp.ipv6_address) == input_form_data[3]["ipv6_address"] + assert new_ap.sbp.ipv6_mask == input_form_data[3]["ipv6_mask"] assert len(ap_list) == 3 if product_name == ProductName.IAS: assert subscription.ias.ias_flavor == IASFlavor.IASGWS