Allow for creation, modification and migration of L3 services onto Edge Ports...

Allow for creation, modification and migration of L3 services onto Edge Ports belonging to different partners

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
 

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:

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

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(),

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,

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