Reflect changes in model for GÉANT IP modification workflow. Fix modification of existing SBPs

gso/workflows/edge_port/modify_edge_port.py

@@ -148,7 +148,6 @@ def modify_edge_port_subscription(
     subscription.edge_port.edge_port_ae_members.clear()
     for member in ae_members:
         subscription.edge_port.edge_port_ae_members.append(EdgePortAEMemberBlock.new(subscription_id=uuid4(), **member))
-    subscription.save()
 
     return {
         "subscription": subscription,

gso/workflows/geant_ip/create_geant_ip.py

@@ -151,19 +151,20 @@ def initialize_subscription(
             for session in sbp_input["bgp_peers"]
         ]
         service_binding_port = ServiceBindingPort.new(
-            subscription_id=uuid4(), **sbp_input, sbp_bgp_session_list=sbp_bgp_session_list, sbp_type=SBPType.L3
+            subscription_id=uuid4(),
+            **sbp_input,
+            sbp_bgp_session_list=sbp_bgp_session_list,
+            sbp_type=SBPType.L3,
+            edge_port=edge_port_subscription.edge_port,
         )
         subscription.geant_ip.geant_ip_ap_list.append(
             NRENAccessPortInactive.new(
                 subscription_id=uuid4(),
                 nren_ap_type=edge_port_input["ap_type"],
-                geant_ip_ep=edge_port_subscription.edge_port,
                 geant_ip_sbp=service_binding_port,
             )
         )
-        edge_port_subscription.edge_port.edge_port_sbp_list.append(service_binding_port)
         edge_port_fqdn_list.append(edge_port_subscription.edge_port.edge_port_node.router_fqdn)
-        edge_port_subscription.save()
 
     subscription.description = "GEANT IP service"
 

gso/workflows/geant_ip/modify_geant_ip.py

 """A modification workflow for a GÉANT IP subscription."""
 
-from typing import Annotated, Any, ClassVar
-from uuid import UUID, uuid4
+from typing import Annotated, Any
+from uuid import uuid4
 
 from orchestrator import begin, conditional, done, step, workflow
 from orchestrator.forms import FormPage
@@ -40,9 +40,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
     class ModifyGeantIPAccessPortsForm(FormPage):
         model_config = ConfigDict(title="Modify GÉANT IP")
-        access_ports: ClassVar[Annotated[list[AccessPortSelection], AfterValidator(validate_edge_ports_are_unique)]] = [
+        access_ports: Annotated[list[AccessPortSelection], AfterValidator(validate_edge_ports_are_unique)] = [  # noqa: RUF012
             AccessPortSelection(
-                geant_ip_ep=str(access_port.geant_ip_ep.owner_subscription_id), nren_ap_type=access_port.nren_ap_type
+                geant_ip_ep=str(access_port.geant_ip_sbp.edge_port.owner_subscription_id),
+                nren_ap_type=access_port.nren_ap_type,
             )
             for access_port in subscription.geant_ip.geant_ip_ap_list
         ]
@@ -50,7 +51,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     access_port_input = yield ModifyGeantIPAccessPortsForm
     input_ap_list = access_port_input.access_ports
     input_ep_list = [str(ap.geant_ip_ep) for ap in input_ap_list]
-    existing_ep_list = [str(ap.geant_ip_ep.owner_subscription_id) for ap in subscription.geant_ip.geant_ip_ap_list]
+    existing_ep_list = [
+        str(ap.geant_ip_sbp.edge_port.owner_subscription_id) for ap in subscription.geant_ip.geant_ip_ap_list
+    ]
 
     class BaseBGPPeer(BaseModel):
         bfd_enabled: bool = False
@@ -81,11 +84,11 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             return [IPFamily.V6UNICAST, IPFamily.V6MULTICAST] if self.add_v6_multicast else [IPFamily.V6UNICAST]
 
     #  There are three possible scenarios for Edge Ports. They can be added, removed, or their relevant SBP can be
-    #  modified. SBPs need to be removed and added accordingly to keep the Edge Port subscriptions up to date.
+    #  modified.
     removed_ap_list = [
         access_port.subscription_instance_id
         for access_port in subscription.geant_ip.geant_ip_ap_list
-        if str(access_port.geant_ip_ep.owner_subscription_id) not in input_ep_list
+        if str(access_port.geant_ip_sbp.edge_port.owner_subscription_id) not in input_ep_list
     ]
     modified_ap_list = [
         (
@@ -94,13 +97,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
                 (
                     ap.nren_ap_type
                     for ap in input_ap_list
-                    if str(ap.geant_ip_ep) == str(access_port.geant_ip_ep.owner_subscription_id)
+                    if str(ap.geant_ip_ep) == str(access_port.geant_ip_sbp.edge_port.owner_subscription_id)
                 ),
                 None,
             ),
         )
         for access_port in subscription.geant_ip.geant_ip_ap_list
-        if str(access_port.geant_ip_ep.owner_subscription_id) in input_ep_list
+        if str(access_port.geant_ip_sbp.edge_port.owner_subscription_id) in input_ep_list
     ]
     added_ap_list = [
         (ep, next((ap.nren_ap_type for ap in input_ap_list if str(ap.geant_ip_ep) == ep), None))
@@ -121,7 +124,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
                 title=f"GÉANT IP - Modify Edge Port configuration ({access_port_index + 1}/{len(input_ap_list)})"
             )
             current_ep_label: Label = Field(
-                f"Currently configuring on {access_port.geant_ip_ep.description} "
+                f"Currently configuring on {access_port.geant_ip_sbp.edge_port.description} "
                 f"(Access Port type: {access_port.nren_ap_type})",
                 exclude=True,
             )
@@ -209,43 +212,6 @@ def remove_old_sbp_blocks(subscription: GeantIP, removed_access_ports: list[UUID
         if str(ap.subscription_instance_id) not in removed_access_ports
     ]
 
-    for ap in removed_access_ports:
-        access_port = NRENAccessPort.from_db(UUID(ap))
-        #  Also remove the :term:`SBP` from the related Edge Port subscription.
-        edge_port = EdgePort.from_subscription(access_port.geant_ip_ep.owner_subscription_id)
-        edge_port.edge_port.edge_port_sbp_list = [
-            sbp
-            for sbp in edge_port.edge_port.edge_port_sbp_list
-            if str(sbp.subscription_instance_id) != access_port.geant_ip_sbp.subscription_instance_id
-        ]
-        edge_port.save()
-
-    return {"subscription": subscription}
-
-
-@step("Instantiate new Service Binding Ports")
-def create_new_sbp_blocks(subscription: GeantIP, added_service_binding_ports: list[dict[str, Any]]):
-    """Add new :term:`SBP`s to the GÉANT IP subscription."""
-    for sbp_input in added_service_binding_ports:
-        edge_port = EdgePort.from_subscription(sbp_input["edge_port_id"])
-        sbp_bgp_session_list = [
-            BGPSession.new(subscription_id=uuid4(), **session, rtbh_enabled=True, is_multi_hop=True)
-            for session in sbp_input["bgp_peers"]
-        ]
-        service_binding_port = ServiceBindingPort.new(
-            subscription_id=uuid4(), **sbp_input, sbp_bgp_session_list=sbp_bgp_session_list, sbp_type=SBPType.L3
-        )
-        subscription.geant_ip.geant_ip_ap_list.append(
-            NRENAccessPort.new(
-                subscription_id=uuid4(),
-                nren_ap_type=sbp_input["ap_type"],
-                geant_ip_ep=edge_port.edge_port,
-                geant_ip_sbp=service_binding_port,
-            )
-        )
-        edge_port.edge_port.edge_port_sbp_list.append(service_binding_port)
-        edge_port.save()
-
     return {"subscription": subscription}
 
 
@@ -258,22 +224,50 @@ def modify_existing_sbp_blocks(subscription: GeantIP, modified_sbp_list: list[di
             (sbp for sbp in modified_sbp_list if sbp["current_sbp_id"] == str(current_sbp.subscription_instance_id)),
             None,
         )
-        modified_sbp_data.pop("current_sbp_id", None)
 
         v4_peer = next((peer for peer in current_sbp.sbp_bgp_session_list if IPFamily.V4UNICAST in peer.families), None)
         for attribute in modified_sbp_data["v4_bgp_peer"]:
             setattr(v4_peer, attribute, modified_sbp_data["v4_bgp_peer"][attribute])
-        modified_sbp_data.pop("v4_bgp_peer")
 
         v6_peer = next((peer for peer in current_sbp.sbp_bgp_session_list if IPFamily.V6UNICAST in peer.families), None)
         for attribute in modified_sbp_data["v6_bgp_peer"]:
             setattr(v6_peer, attribute, modified_sbp_data["v6_bgp_peer"][attribute])
-        modified_sbp_data.pop("v6_bgp_peer")
 
         current_sbp.sbp_bgp_session_list = [v4_peer, v6_peer]
-        access_port.nren_ap_type = modified_sbp_data.pop("new_ap_type")
-        for attribute in modified_sbp_data:
-            setattr(current_sbp, attribute, modified_sbp_data[attribute])
+        current_sbp.vlan_id = modified_sbp_data["vlan_id"]
+        current_sbp.geant_sid = modified_sbp_data["geant_sid"]
+        current_sbp.is_tagged = modified_sbp_data["is_tagged"]
+        current_sbp.ipv4_address = modified_sbp_data["ipv4_address"]
+        current_sbp.ipv6_address = modified_sbp_data["ipv6_address"]
+        current_sbp.custom_firewall_filters = modified_sbp_data["custom_firewall_filters"]
+        access_port.nren_ap_type = modified_sbp_data["new_ap_type"]
+
+    return {"subscription": subscription}
+
+
+@step("Instantiate new Service Binding Ports")
+def create_new_sbp_blocks(subscription: GeantIP, added_service_binding_ports: list[dict[str, Any]]):
+    """Add new :term:`SBP`s to the GÉANT IP subscription."""
+    for sbp_input in added_service_binding_ports:
+        edge_port = EdgePort.from_subscription(sbp_input["edge_port_id"])
+        sbp_bgp_session_list = [
+            BGPSession.new(subscription_id=uuid4(), **session, rtbh_enabled=True, is_multi_hop=True)
+            for session in sbp_input["bgp_peers"]
+        ]
+        service_binding_port = ServiceBindingPort.new(
+            subscription_id=uuid4(),
+            **sbp_input,
+            sbp_bgp_session_list=sbp_bgp_session_list,
+            sbp_type=SBPType.L3,
+            edge_port=edge_port.edge_port,
+        )
+        subscription.geant_ip.geant_ip_ap_list.append(
+            NRENAccessPort.new(
+                subscription_id=uuid4(),
+                nren_ap_type=sbp_input["ap_type"],
+                geant_ip_sbp=service_binding_port,
+            )
+        )
 
     return {"subscription": subscription}
 
@@ -286,16 +280,16 @@ def modify_existing_sbp_blocks(subscription: GeantIP, modified_sbp_list: list[di
 def modify_geant_ip():
     """Modify a GÉANT IP subscription."""
     access_ports_are_removed = conditional(lambda state: bool(len(state["removed_access_ports"]) > 0))
-    access_ports_are_added = conditional(lambda state: bool(len(state["added_service_binding_ports"]) > 0))
     access_ports_are_modified = conditional(lambda state: bool(len(state["modified_sbp_list"]) > 0))
+    access_ports_are_added = conditional(lambda state: bool(len(state["added_service_binding_ports"]) > 0))
 
     return (
         begin
         >> store_process_subscription(Target.MODIFY)
         >> unsync
         >> access_ports_are_removed(remove_old_sbp_blocks)
-        >> access_ports_are_added(create_new_sbp_blocks)
         >> access_ports_are_modified(modify_existing_sbp_blocks)
+        >> access_ports_are_added(create_new_sbp_blocks)
         >> resync
         >> done
     )