Update modify NrenL3CoreServic WF after BFD Settings change.

gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py

@@ -14,7 +14,7 @@ from pydantic import AfterValidator, BaseModel, ConfigDict, Field, computed_fiel
 from pydantic_forms.types import State
 from pydantic_forms.validators import Divider, Label
 
-from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
+from gso.products.product_blocks.bgp_session import BGPSession, IPFamily, IPTypes
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
 from gso.products.product_blocks.service_binding_port import BFDSettings, ServiceBindingPort
 from gso.products.product_types.edge_port import EdgePort
@@ -61,11 +61,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
     class BFDInputModel(BaseModel):
         bfd_enabled: bool = False
-        bfd_interval: int | None = None
+        bfd_interval_rx: int | None = None
+        bfd_interval_tx: int | None = None
         bfd_multiplier: int | None = None
 
     class BaseBGPPeer(BaseModel):
-        bfd_settings: BFDInputModel
+        bfd_enabled: bool = False
         has_custom_policies: bool = False
         authentication_key: str | None
         multipath_enabled: bool = False
@@ -75,6 +76,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     class IPv4BGPPeer(BaseBGPPeer):
         peer_address: IPv4AddressType
         add_v4_multicast: bool = Field(default=False, exclude=True)
+        ip_type: IPTypes = IPTypes.IPV4
 
         @computed_field  # type: ignore[misc]
         @property
@@ -84,6 +86,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     class IPv6BGPPeer(BaseBGPPeer):
         peer_address: IPv6AddressType
         add_v6_multicast: bool = Field(default=False, exclude=True)
+        ip_type: IPTypes = IPTypes.IPV6
 
         @computed_field  # type: ignore[misc]
         @property
@@ -150,12 +153,14 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             v4_bfd_settings: BFDInputModel = BFDInputModel(
                 bfd_enabled=current_sbp.v4_bfd_settings.bfd_enabled,
                 bfd_multiplier=current_sbp.v4_bfd_settings.bfd_multiplier,
-                bfd_interval=current_sbp.v4_bfd_settings.bfd_interval,
+                bfd_interval_rx=current_sbp.v4_bfd_settings.bfd_interval_rx,
+                bfd_interval_tx=current_sbp.v4_bfd_settings.bfd_interval_tx,
             )
             v6_bfd_settings: BFDInputModel = BFDInputModel(
                 bfd_enabled=current_sbp.v6_bfd_settings.bfd_enabled,
                 bfd_multiplier=current_sbp.v6_bfd_settings.bfd_multiplier,
-                bfd_interval=current_sbp.v6_bfd_settings.bfd_interval,
+                bfd_interval_rx=current_sbp.v6_bfd_settings.bfd_interval_rx,
+                bfd_interval_tx=current_sbp.v6_bfd_settings.bfd_interval_tx,
             )
             divider: Divider = Field(None, exclude=True)
             v4_bgp_peer: IPv4BGPPeer = IPv4BGPPeer(
@@ -249,18 +254,12 @@ def modify_existing_sbp_blocks(subscription: NRENL3CoreService, modified_sbp_lis
         )
 
         v4_peer = next(peer for peer in current_sbp.bgp_session_list if IPFamily.V4UNICAST in peer.families)
-        v4_peer_bfd_settings = modified_sbp_data["v4_bgp_peer"].pop("bfd_settings")
-        for attribute in v4_peer_bfd_settings:
-            setattr(v4_peer.bfd_settings, attribute, v4_peer_bfd_settings[attribute])
         for attribute in modified_sbp_data["v4_bgp_peer"]:
             setattr(v4_peer, attribute, modified_sbp_data["v4_bgp_peer"][attribute])
         for attribute in modified_sbp_data["v4_bfd_settings"]:
             setattr(current_sbp.v4_bfd_settings, attribute, modified_sbp_data["v4_bfd_settings"][attribute])
 
         v6_peer = next(peer for peer in current_sbp.bgp_session_list if IPFamily.V6UNICAST in peer.families)
-        v6_peer_bfd_settings = modified_sbp_data["v6_bgp_peer"].pop("bfd_settings")
-        for attribute in v6_peer_bfd_settings:
-            setattr(v6_peer.bfd_settings, attribute, v6_peer_bfd_settings[attribute])
         for attribute in modified_sbp_data["v6_bgp_peer"]:
             setattr(v6_peer, attribute, modified_sbp_data["v6_bgp_peer"][attribute])
         for attribute in modified_sbp_data["v6_bfd_settings"]:
@@ -283,14 +282,10 @@ def create_new_sbp_blocks(subscription: NRENL3CoreService, added_service_binding
     """Add new two :term:`SBP` to the :term:`NREN` L3 Core Service subscription."""
     for sbp_input in added_service_binding_ports:
         edge_port = EdgePort.from_subscription(sbp_input["edge_port_id"])
-        bgp_session_list = []
-        for session in sbp_input["bgp_peers"]:
-            bfd_settings = BFDSettings.new(subscription_id=uuid4(), **session.pop("bfd_settings"))
-            bgp_session_list.append(
-                BGPSession.new(
-                    subscription_id=uuid4(), **session, bfd_settings=bfd_settings, rtbh_enabled=True, is_multi_hop=True
-                )
-            )
+        bgp_session_list = [
+            BGPSession.new(subscription_id=uuid4(), **session, rtbh_enabled=True, is_multi_hop=True)
+            for session in sbp_input["bgp_peers"]
+        ]
         v4_bfd_settings = BFDSettings.new(subscription_id=uuid4(), **sbp_input.pop("v4_bfd_settings"))
         v6_bfd_settings = BFDSettings.new(subscription_id=uuid4(), **sbp_input.pop("v6_bfd_settings"))
         service_binding_port = ServiceBindingPort.new(

test/fixtures/nren_l3_core_service_fixtures.py

@@ -7,7 +7,7 @@ from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
 
 from gso.products import ProductName
-from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
+from gso.products.product_blocks.bgp_session import BGPSession, IPFamily, IPTypes
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
 from gso.products.product_blocks.service_binding_port import BFDSettings, ServiceBindingPort
 from gso.products.product_types.edge_port import EdgePort
@@ -55,6 +55,7 @@ def bgp_session_subscription_factory(faker):
         is_passive: bool | None = False,
         rtbh_enabled: bool | None = False,
         bfd_enabled: bool | None = False,
+        ip_type: IPTypes | None = None,
     ):
         return BGPSession.new(
             subscription_id=uuid4(),
@@ -68,6 +69,7 @@ def bgp_session_subscription_factory(faker):
             is_multi_hop=is_multi_hop,
             rtbh_enabled=rtbh_enabled,
             is_passive=is_passive,
+            ip_type=ip_type if ip_type else faker.random_element(elements=[IPTypes.IPV4, IPTypes.IPV6]),
         )
 
     return create_bgp_session

test/workflows/nren_l3_core_service/test_modify_nren_l3_core_service.py

@@ -77,12 +77,12 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success(
             "v4_bgp_peer": {
                 "authentication_key": faker.password(),
                 "peer_address": faker.ipv4(),
-                "bfd_settings": {"bfd_enabled": False},
+                "bfd_enabled": False,
             },
             "v6_bgp_peer": {
                 "authentication_key": faker.password(),
                 "peer_address": faker.ipv6(),
-                "bfd_settings": {"bfd_enabled": False},
+                "bfd_enabled": False,
             },
             "v4_bfd_settings": {"bfd_enabled": False},
             "v6_bfd_settings": {"bfd_enabled": False},
@@ -108,20 +108,18 @@ def sbp_input_form_data(faker):
             "custom_firewall_filters": True,
             "v4_bfd_settings": {
                 "bfd_enabled": True,
-                "bfd_interval": faker.pyint(),
+                "bfd_interval_rx": faker.pyint(),
+                "bfd_interval_tx": faker.pyint(),
                 "bfd_multiplier": faker.pyint(),
             },
             "v6_bfd_settings": {
                 "bfd_enabled": True,
-                "bfd_interval": faker.pyint(),
+                "bfd_interval_rx": faker.pyint(),
+                "bfd_interval_tx": faker.pyint(),
                 "bfd_multiplier": faker.pyint(),
             },
             "v4_bgp_peer": {
-                "bfd_settings": {
-                    "bfd_enabled": True,
-                    "bfd_interval": faker.pyint(),
-                    "bfd_multiplier": faker.pyint(),
-                },
+                "bfd_enabled": True,
                 "has_custom_policies": True,
                 "authentication_key": faker.password(),
                 "multipath_enabled": True,
@@ -131,11 +129,7 @@ def sbp_input_form_data(faker):
                 "add_v4_multicast": True,
             },
             "v6_bgp_peer": {
-                "bfd_settings": {
-                    "bfd_enabled": True,
-                    "bfd_interval": faker.pyint(),
-                    "bfd_multiplier": faker.pyint(),
-                },
+                "bfd_enabled": True,
                 "has_custom_policies": True,
                 "authentication_key": faker.password(),
                 "multipath_enabled": True,
@@ -194,16 +188,8 @@ def test_modify_nren_l3_core_service_modify_edge_port_success(
         )
 
         assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].bfd_settings.bfd_enabled
-            == new_sbp_data[i]["v4_bgp_peer"]["bfd_settings"]["bfd_enabled"]
-        )
-        assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].bfd_settings.bfd_interval
-            == new_sbp_data[i]["v4_bgp_peer"]["bfd_settings"]["bfd_interval"]
-        )
-        assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].bfd_settings.bfd_multiplier
-            == new_sbp_data[i]["v4_bgp_peer"]["bfd_settings"]["bfd_multiplier"]
+            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].bfd_enabled
+            == new_sbp_data[i]["v4_bgp_peer"]["bfd_enabled"]
         )
         assert (
             subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].has_custom_policies
@@ -238,16 +224,8 @@ def test_modify_nren_l3_core_service_modify_edge_port_success(
         )
 
         assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].bfd_settings.bfd_enabled
-            == new_sbp_data[i]["v6_bgp_peer"]["bfd_settings"]["bfd_enabled"]
-        )
-        assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].bfd_settings.bfd_interval
-            == new_sbp_data[i]["v6_bgp_peer"]["bfd_settings"]["bfd_interval"]
-        )
-        assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].bfd_settings.bfd_multiplier
-            == new_sbp_data[i]["v6_bgp_peer"]["bfd_settings"]["bfd_multiplier"]
+            subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].bfd_enabled
+            == new_sbp_data[i]["v6_bgp_peer"]["bfd_enabled"]
         )
         assert (
             subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].has_custom_policies
@@ -285,8 +263,12 @@ def test_modify_nren_l3_core_service_modify_edge_port_success(
             == new_sbp_data[i]["v4_bfd_settings"]["bfd_enabled"]
         )
         assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_interval
-            == new_sbp_data[i]["v4_bfd_settings"]["bfd_interval"]
+            subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_interval_rx
+            == new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_rx"]
+        )
+        assert (
+                subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_interval_tx
+                == new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_tx"]
         )
         assert (
             subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_multiplier
@@ -297,8 +279,12 @@ def test_modify_nren_l3_core_service_modify_edge_port_success(
             == new_sbp_data[i]["v6_bfd_settings"]["bfd_enabled"]
         )
         assert (
-            subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_interval
-            == new_sbp_data[i]["v6_bfd_settings"]["bfd_interval"]
+            subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_interval_rx
+            == new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_rx"]
+        )
+        assert (
+                subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_interval_tx
+                == new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_tx"]
         )
         assert (
             subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_multiplier