Update input form of GÉANT IP creation, update model of BGP peer

gso/migrations/versions/2024-09-25_c3e98d657b36_add_passive_boolean_to_bgp_session_model.py → gso/migrations/versions/2024-09-25_f4239c9361b4_update_bgp_session_model.py

-"""Add passive boolean to BGP session model.
+"""Update BGP session model.
 
-Revision ID: c3e98d657b36
+Revision ID: f4239c9361b4
 Revises: f900cbaa47d7
-Create Date: 2024-09-25 14:05:57.072597
+Create Date: 2024-09-25 14:20:55.540546
 
 """
 import sqlalchemy as sa
 from alembic import op
 
 # revision identifiers, used by Alembic.
-revision = 'c3e98d657b36'
+revision = 'f4239c9361b4'
 down_revision = 'f900cbaa47d7'
 branch_labels = None
 depends_on = None
@@ -18,7 +18,13 @@ depends_on = None
 def upgrade() -> None:
     conn = op.get_bind()
     conn.execute(sa.text("""
-INSERT INTO resource_types (resource_type, description) VALUES ('is_passive', 'This BGP session is passive') RETURNING resource_types.resource_type_id
+INSERT INTO resource_types (resource_type, description) VALUES ('rtbh_enabled', 'Remote Triggered Blackhole is enabled') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('is_passive', 'BGP session is passive') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('rtbh_enabled')))
     """))
     conn.execute(sa.text("""
 INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('is_passive')))
@@ -28,14 +34,20 @@ INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VA
 def downgrade() -> None:
     conn = op.get_bind()
     conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('rtbh_enabled'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('rtbh_enabled'))
+    """))
+    conn.execute(sa.text("""
 DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('is_passive'))
     """))
     conn.execute(sa.text("""
 DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('BGPSession'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('is_passive'))
     """))
     conn.execute(sa.text("""
-DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('is_passive'))
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('rtbh_enabled', 'is_passive'))
     """))
     conn.execute(sa.text("""
-DELETE FROM resource_types WHERE resource_types.resource_type IN ('is_passive')
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('rtbh_enabled', 'is_passive')
     """))

gso/products/product_blocks/bgp_session.py

@@ -33,6 +33,7 @@ class BGPSessionInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INI
     send_default_route: bool | None = None
     is_multi_hop: bool = False
     is_passive: bool = False
+    rtbh_enabled: bool = False
 
 
 class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -49,6 +50,7 @@ class BGPSessionProvisioning(BGPSessionInactive, lifecycle=[SubscriptionLifecycl
     send_default_route: bool
     is_multi_hop: bool
     is_passive: bool
+    rtbh_enabled: bool
 
 
 class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -76,3 +78,5 @@ class BGPSession(BGPSessionProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE
     is_multi_hop: bool
     #: Whether this is a passive session.
     is_passive: bool
+    #: Whether Remote Triggered Blackhole is enabled
+    rtbh_enabled: bool

gso/translations/en-GB.json

@@ -19,6 +19,7 @@
             "router_role": "Router role",
 
             "geant_s_sid": "GÉANT S-SID",
+            "geant_sid": "GÉANT S-SID",
             "iptrunk_description": "IPtrunk description",
             "iptrunk_type": "IPtrunk type",
             "iptrunk_speed": "Capacity per port (in Gbits/s)",
@@ -32,13 +33,7 @@
             "migrate_to_different_site": "Migrating to a different Site",
             "remove_configuration": "Remove configuration from the router",
             "clean_up_ipam": "Clean up related entries in IPAM",
-            "restore_isis_metric": "Restore ISIS metric to original value",
-
-            "bgp_peers": {
-                "bfd_enabled": "BFD enabled",
-                "bfd_interval": "BFD interval",
-                "bfd_multiplier": "BFD multiplier"
-            }
+            "restore_isis_metric": "Restore ISIS metric to original value"
         }
     },
     "workflow": {

gso/workflows/geant_ip/create_geant_ip.py

@@ -12,7 +12,6 @@ from orchestrator.workflow import StepList, begin, done, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import AfterValidator, BaseModel, ConfigDict, Field, computed_field
-from pydantic_forms.types import strEnum
 from pydantic_forms.validators import Divider
 
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
@@ -26,7 +25,7 @@ from gso.utils.helpers import (
     partner_choice,
 )
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 from gso.utils.types.tt_number import TTNumber
 
 
@@ -41,15 +40,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     initial_user_input = yield CreateGeantIPForm
 
-    class BGPPeerFamily(strEnum):
-        V4 = "IPv4 only"
-        V6 = "IPv6 only"
-        DUAL_STACK = "Dual stack"
-
     class EdgePortSelection(BaseModel):
         edge_port: active_edge_port_selector(partner_id=initial_user_input.partner)  # type: ignore[valid-type]
         ap_type: APType
-        ip_family: BGPPeerFamily
 
     def validate_edge_ports_are_unique(edge_ports: list[EdgePortSelection]) -> list[EdgePortSelection]:
         """Verify if interfaces are unique."""
@@ -76,13 +69,15 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         bfd_enabled: bool = False
         bfd_interval: int | None = None
         bfd_multiplier: int | None = None
+        rtbh_enabled: bool = False
         has_custom_policies: bool = False
         authentication_key: str
         multipath_enabled: bool = False
         send_default_route: bool = False
         is_multi_hop: bool = False
+        is_passive: bool = False
 
-    class V4OnlyBGPPeer(BaseBGPPeer):
+    class IPv4BGPPeer(BaseBGPPeer):
         peer_address: IPv4AddressType
         add_v4_multicast: bool = Field(default=False, exclude=True)
 
@@ -91,7 +86,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         def families(self) -> list[IPFamily]:
             return [IPFamily.V4UNICAST, IPFamily.V4MULTICAST] if self.add_v4_multicast else [IPFamily.V4UNICAST]
 
-    class V6OnlyBGPPeer(BaseBGPPeer):
+    class IPv6BGPPeer(BaseBGPPeer):
         peer_address: IPv6AddressType
         add_v6_multicast: bool = Field(default=False, exclude=True)
 
@@ -100,33 +95,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         def families(self) -> list[IPFamily]:
             return [IPFamily.V6UNICAST, IPFamily.V6MULTICAST] if self.add_v6_multicast else [IPFamily.V6UNICAST]
 
-    class DualStackBGPPeer(V4OnlyBGPPeer, V6OnlyBGPPeer):
-        peer_address: IPAddress  # type: ignore[assignment]
-
-        @computed_field  # type: ignore[misc]
-        @property
-        def families(self) -> list[IPFamily]:
-            result = [IPFamily.V4UNICAST, IPFamily.V6UNICAST]
-            if self.add_v4_multicast:
-                result.append(IPFamily.V4MULTICAST)
-            if self.add_v6_multicast:
-                result.append(IPFamily.V6MULTICAST)
-            return result
-
     binding_port_inputs = []
     while current_ep_index < total_ep_count:
         current_edge_port = EdgePort.from_subscription(ep_list[current_ep_index].edge_port)
-        bgp_peer_type: type[object]
-        match ep_list[current_ep_index].ip_family:
-            case BGPPeerFamily.V4:
-                bgp_peer_type = V4OnlyBGPPeer
-            case BGPPeerFamily.V6:
-                bgp_peer_type = V6OnlyBGPPeer
-            case BGPPeerFamily.DUAL_STACK:
-                bgp_peer_type = DualStackBGPPeer
-            case _:
-                msg = f"Invalid IP family selected: {ep_list[current_ep_index].ip_family}"
-                raise ValueError(msg)
 
         class BindingPortsInputForm(FormPage):
             model_config = ConfigDict(
@@ -146,12 +117,17 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             ipv6_address: IPv6AddressType
             custom_firewall_filters: bool = False
             divider: Divider = Field(None, exclude=True)
-            bgp_peers: list[bgp_peer_type]  # type: ignore[valid-type]
+            v4_bgp_peer: IPv4BGPPeer
+            v6_bgp_peer: IPv6BGPPeer
 
         binding_port_input_form = yield BindingPortsInputForm
         binding_port_input = binding_port_input_form.model_dump()
 
         binding_port_input["sbp_type"] = SBPType.L3
+        binding_port_input["bgp_peers"] = [
+            binding_port_input_form.v4_bgp_peer.model_dump(),
+            binding_port_input_form.v6_bgp_peer.model_dump(),
+        ]
         binding_port_inputs.append(binding_port_input)
         current_ep_index += 1