Fix the issue with BFDSettings in GEANT IP creation.

gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py

@@ -13,7 +13,7 @@ from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import BaseModel, ConfigDict, Field, computed_field
 from pydantic_forms.validators import Divider
 
-from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
+from gso.products.product_blocks.bgp_session import BFDSettings, BGPSession, IPFamily
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPortInactive
 from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
 from gso.products.product_types.edge_port import EdgePort
@@ -61,7 +61,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         bfd_multiplier: int | None = None
 
     class BaseBGPPeer(BaseModel):
-        bfd_settings: BFDSettingsForm
+        bfd_settings: BFDSettingsForm = BFDSettingsForm(bfd_enabled=False)
         has_custom_policies: bool = False
         authentication_key: str | None = None
         multipath_enabled: bool = False
@@ -97,8 +97,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
         geant_sid: str
         is_tagged: bool = False
-        v4_bfd_settings: BFDSettingsForm
-        v6_bfd_settings: BFDSettingsForm
+        v4_bfd_settings: BFDSettingsForm = BFDSettingsForm(bfd_enabled=False)
+        v6_bfd_settings: BFDSettingsForm = BFDSettingsForm(bfd_enabled=False)
         vlan_id: VLAN_ID
         ipv4_address: IPv4AddressType
         ipv4_mask: IPV4Netmask
@@ -140,11 +140,19 @@ def initialize_subscription(
     edge_port_fqdn_list = []
     edge_port_subscription = EdgePort.from_subscription(edge_port["edge_port"])
     sbp_bgp_session_list = [
-        BGPSession.new(subscription_id=uuid4(), **session, rtbh_enabled=True, is_multi_hop=True)
+        BGPSession.new(
+            subscription_id=uuid4(),
+            bfd_settings=BFDSettings.new(subscription_id=uuid4(), **(session.pop("bfd_settings"))),
+            **session,
+            rtbh_enabled=True,
+            is_multi_hop=True,
+        )
         for session in binding_port_input["bgp_peers"]
     ]
     service_binding_port = ServiceBindingPortInactive.new(
         subscription_id=uuid4(),
+        v4_bfd_settings=BFDSettings.new(subscription_id=uuid4(), **(binding_port_input.pop("v4_bfd_settings"))),
+        v6_bfd_settings=BFDSettings.new(subscription_id=uuid4(), **(binding_port_input.pop("v6_bfd_settings"))),
         **binding_port_input,
         bgp_session_list=sbp_bgp_session_list,
         sbp_type=SBPType.L3,
@@ -161,15 +169,12 @@ def initialize_subscription(
 
     subscription.description = f"{product_name} service"
     partner_name = get_partner_by_id(subscription.customer_id).name
-
-
     return {"subscription": subscription, "edge_port_fqdn_list": edge_port_fqdn_list, "partner_name": partner_name}
 
 
 @step("[DRY RUN] Deploy service binding port")
 def provision_sbp_dry(
-    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, edge_port_fqdn_list: list[str],
-    partner_name: str
+    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, edge_port_fqdn_list: list[str], partner_name: str
 ) -> LSOState:
     """Perform a dry run of deploying Service Binding Ports."""
     extra_vars = {
@@ -191,8 +196,7 @@ def provision_sbp_dry(
 
 @step("[FOR REAL] Deploy service binding port")
 def provision_sbp_real(
-    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, edge_port_fqdn_list: list[str],
-    partner_name: str
+    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, edge_port_fqdn_list: list[str], partner_name: str
 ) -> LSOState:
     """Deploy Service Binding Ports."""
     extra_vars = {
@@ -226,8 +230,7 @@ def check_sbp_functionality(subscription: dict[str, Any], edge_port_fqdn_list: l
 
 @step("[DRY RUN] Deploy BGP peers")
 def deploy_bgp_peers_dry(
-    subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr,
-    partner_name: str
+    subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr, partner_name: str
 ) -> LSOState:
     """Perform a dry run of deploying :term:`BGP` peers."""
     extra_vars = {
@@ -249,8 +252,7 @@ def deploy_bgp_peers_dry(
 
 @step("[FOR REAL] Deploy BGP peers")
 def deploy_bgp_peers_real(
-    subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr,
-    partner_name: str
+    subscription: dict[str, Any], edge_port_fqdn_list: list[str], tt_number: str, process_id: UUIDstr, partner_name: str
 ) -> LSOState:
     """Deploy :term:`BGP` peers."""
     extra_vars = {