Add IPV4 and IPV6 netmask to Service Binding Port.

gso/migrations/versions/2024-10-10_df108295d917_add_ipv4_ipv6_netmask_to_service_.py

+"""Add IPV4/IPV6 netmask to Service Binding Port model .
+
+Revision ID: df108295d917
+Revises: bf05800fe9fc
+Create Date: 2024-10-10 11:39:43.051211
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'df108295d917'
+down_revision = 'bf05800fe9fc'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('ipv4_mask', 'IPV4 subnet mask') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('ipv6_mask', 'IPV6 subnet mask') 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 ('ServiceBindingPort')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask')))
+    """))
+    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 ('ServiceBindingPort')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask')))
+    """))
+    conn.execute(sa.text("""
+
+                WITH subscription_instance_ids AS (
+                    SELECT subscription_instances.subscription_instance_id
+                    FROM   subscription_instances
+                    WHERE  subscription_instances.product_block_id IN (
+                        SELECT product_blocks.product_block_id
+                        FROM   product_blocks
+                        WHERE  product_blocks.name = 'ServiceBindingPort'
+                    )
+                )
+
+                INSERT INTO
+                    subscription_instance_values (subscription_instance_id, resource_type_id, value)
+                SELECT
+                    subscription_instance_ids.subscription_instance_id,
+                    resource_types.resource_type_id,
+                    'None'
+                FROM resource_types
+                CROSS JOIN subscription_instance_ids
+                WHERE resource_types.resource_type = 'ipv4_mask'
+        
+    """))
+    conn.execute(sa.text("""
+
+                WITH subscription_instance_ids AS (
+                    SELECT subscription_instances.subscription_instance_id
+                    FROM   subscription_instances
+                    WHERE  subscription_instances.product_block_id IN (
+                        SELECT product_blocks.product_block_id
+                        FROM   product_blocks
+                        WHERE  product_blocks.name = 'ServiceBindingPort'
+                    )
+                )
+
+                INSERT INTO
+                    subscription_instance_values (subscription_instance_id, resource_type_id, value)
+                SELECT
+                    subscription_instance_ids.subscription_instance_id,
+                    resource_types.resource_type_id,
+                    'None'
+                FROM resource_types
+                CROSS JOIN subscription_instance_ids
+                WHERE resource_types.resource_type = 'ipv6_mask'
+        
+    """))
+
+
+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 ('ServiceBindingPort')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask'))
+    """))
+    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 ('ServiceBindingPort'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask'))
+    """))
+    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 ('ServiceBindingPort')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask'))
+    """))
+    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 ('ServiceBindingPort'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv6_mask'))
+    """))
+    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 ('ipv4_mask', 'ipv6_mask'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('ipv4_mask', 'ipv6_mask')
+    """))

gso/products/product_blocks/service_binding_port.py

@@ -12,7 +12,7 @@ from pydantic import Field
 from gso.products.product_blocks.bgp_session import BGPSession, BGPSessionInactive, BGPSessionProvisioning
 from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
 from gso.utils.shared_enums import SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 
 VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
 
@@ -26,7 +26,9 @@ class ServiceBindingPortInactive(
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType | None = None
     ipv4_address: IPv4AddressType | None = None
+    ipv4_mask: IPV4Netmask | None = None
     ipv6_address: IPv6AddressType | None = None
+    ipv6_mask: IPV6Netmask | None = None
     custom_firewall_filters: bool | None = None
     geant_sid: str | None = None
     sbp_bgp_session_list: list[BGPSessionInactive] = Field(default_factory=list)
@@ -40,7 +42,9 @@ class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[Subs
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType
     ipv4_address: IPv4AddressType | None = None
+    ipv4_mask: IPV4Netmask | None = None
     ipv6_address: IPv6AddressType | None = None
+    ipv6_mask: IPV6Netmask | None = None
     custom_firewall_filters: bool
     geant_sid: str
     sbp_bgp_session_list: list[BGPSessionProvisioning]  # type: ignore[assignment]
@@ -58,8 +62,12 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
     sbp_type: SBPType
     #: If layer 3, IPv4 resources.
     ipv4_address: IPv4AddressType | None = None
+    #: IPV4 subnet mask.
+    ipv4_mask: IPV4Netmask | None = None
     #: If layer 3, IPv6 resources.
     ipv6_address: IPv6AddressType | None = None
+    #: IPV6 subnet mask.
+    ipv6_mask: IPV6Netmask | None = None
     #: Any custom firewall filters that the partner may require.
     custom_firewall_filters: bool
     #: The GÉANT service ID of this binding port.

gso/settings.py

@@ -9,15 +9,13 @@ import json
 import logging
 import os
 from pathlib import Path
-from typing import Annotated
 
 from orchestrator.types import UUIDstr
-from pydantic import EmailStr, Field
+from pydantic import EmailStr
 from pydantic_forms.types import strEnum
 from pydantic_settings import BaseSettings
-from typing_extensions import Doc
 
-from gso.utils.types.ip_address import PortNumber
+from gso.utils.types.ip_address import IPV4Netmask, IPV6Netmask, PortNumber
 
 logger = logging.getLogger(__name__)
 
@@ -62,16 +60,12 @@ class InfoBloxParams(BaseSettings):
     password: str
 
 
-V4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
-V6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
-
-
 class V4NetworkParams(BaseSettings):
     """A set of parameters that describe an IPv4 network in InfoBlox."""
 
     containers: list[ipaddress.IPv4Network]
     networks: list[ipaddress.IPv4Network]
-    mask: V4Netmask
+    mask: IPV4Netmask
 
 
 class V6NetworkParams(BaseSettings):
@@ -79,7 +73,7 @@ class V6NetworkParams(BaseSettings):
 
     containers: list[ipaddress.IPv6Network]
     networks: list[ipaddress.IPv6Network]
-    mask: V6Netmask
+    mask: IPV6Netmask
 
 
 class ServiceNetworkParams(BaseSettings):

gso/utils/types/ip_address.py

@@ -39,6 +39,8 @@ IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(_str, return_
 IPv6NetworkType = Annotated[ipaddress.IPv6Network, PlainSerializer(_str, return_type=str, when_used="always")]
 IPAddress = Annotated[str, AfterValidator(validate_ipv4_or_ipv6)]
 IPNetwork = Annotated[str, AfterValidator(validate_ipv4_or_ipv6_network)]
+IPV4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
+IPV6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
 PortNumber = Annotated[
     int,
     Field(

gso/workflows/geant_ip/create_geant_ip.py

@@ -24,7 +24,7 @@ from gso.utils.helpers import (
     partner_choice,
 )
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 from gso.utils.types.tt_number import TTNumber
 
 
@@ -106,7 +106,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             is_tagged: bool = False
             vlan_id: VLAN_ID
             ipv4_address: IPv4AddressType
+            ipv4_mask: IPV4Netmask
             ipv6_address: IPv6AddressType
+            ipv6_mask: IPV6Netmask
             custom_firewall_filters: bool = False
             divider: Divider = Field(None, exclude=True)
             v4_bgp_peer: IPv4BGPPeer

gso/workflows/geant_ip/create_imported_geant_ip.py

@@ -20,7 +20,7 @@ from gso.products.product_types.geant_ip import ImportedGeantIPInactive
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.shared_enums import SBPType
-from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 
 
 def initial_input_form_generator() -> FormGenerator:
@@ -49,7 +49,9 @@ def initial_input_form_generator() -> FormGenerator:
         vlan_id: VLAN_ID
         custom_firewall_filters: bool = False
         ipv4_address: IPv4AddressType
+        ipv4_mask: IPV4Netmask
         ipv6_address: IPv6AddressType
+        ipv6_mask: IPV6Netmask
         rtbh_enabled: bool = True
         is_multi_hop: bool = True
         bgp_peers: list[BaseBGPPeer]

gso/workflows/geant_ip/modify_geant_ip.py

@@ -21,7 +21,7 @@ from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.geant_ip import GeantIP
 from gso.utils.helpers import active_edge_port_selector
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -137,7 +137,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             # it's a layer 3 service. The ignore statements are there to put our type checker at ease.
             vlan_id: VLAN_ID = current_sbp.vlan_id  # type: ignore[assignment]
             ipv4_address: IPv4AddressType = current_sbp.ipv4_address  # type: ignore[assignment]
+            ipv4_mask: IPV4Netmask = current_sbp.ipv4_mask  # type: ignore[assignment]
             ipv6_address: IPv6AddressType = current_sbp.ipv6_address  # type: ignore[assignment]
+            ipv6_mask: IPV6Netmask = current_sbp.ipv6_mask  # type: ignore[assignment]
             custom_firewall_filters: bool = current_sbp.custom_firewall_filters
             divider: Divider = Field(None, exclude=True)
             v4_bgp_peer: IPv4BGPPeer = IPv4BGPPeer(

test/conftest.py

@@ -108,6 +108,12 @@ class FakerProvider(BaseProvider):
 
         return site_name
 
+    def ipv4_netmask(self) -> int:
+        return self.generator.random_int(min=1, max=32)
+
+    def ipv6_netmask(self) -> int:
+        return self.generator.random_int(min=1, max=128)
+
     def network_interface(self) -> str:
         return self.generator.numerify("ge-@#/@#/@#")
 

test/fixtures/geant_ip_fixtures.py

@@ -59,7 +59,9 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_p
         geant_sid: str | None = None,
         sbp_type: SBPType = SBPType.L3,
         ipv4_address: str | None = None,
+        ipv4_mask: int | None = None,
         ipv6_address: str | None = None,
+        ipv6_mask: int | None = None,
         vlan_id: int | None = None,
         edge_port: EdgePort | None = None,
         *,
@@ -72,7 +74,9 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_p
             vlan_id=vlan_id or faker.pyint(min_value=1, max_value=4096),
             sbp_type=sbp_type,
             ipv4_address=ipv4_address or faker.ipv4(),
+            ipv4_mask=ipv4_mask or faker.ipv4_netmask(),
             ipv6_address=ipv6_address or faker.ipv6(),
+            ipv6_mask=ipv6_mask or faker.ipv6_netmask(),
             custom_firewall_filters=custom_firewall_filters,
             geant_sid=geant_sid or faker.geant_sid(),
             sbp_bgp_session_list=sbp_bgp_session_list

test/workflows/geant_ip/test_create_geant_ip.py

@@ -52,7 +52,9 @@ def test_create_geant_ip_success(
             "is_tagged": faker.boolean(),
             "vlan_id": faker.vlan_id(),
             "ipv4_address": faker.ipv4(),
+            "ipv4_mask": faker.ipv4_netmask(),
             "ipv6_address": faker.ipv6(),
+            "ipv6_mask": faker.ipv6_netmask(),
             "custom_firewall_filters": faker.boolean(),
             "v4_bgp_peer": base_bgp_peer_input() | {"add_v4_multicast": faker.boolean(), "peer_address": faker.ipv4()},
             "v6_bgp_peer": base_bgp_peer_input() | {"add_v6_multicast": faker.boolean(), "peer_address": faker.ipv6()},

test/workflows/geant_ip/test_create_imported_geant_ip.py

@@ -20,7 +20,9 @@ def imported_geant_ip_creation_input_form_data(edge_port_subscription_factory, p
                 "is_tagged": faker.boolean(),
                 "vlan_id": faker.vlan_id(),
                 "ipv4_address": faker.ipv4(),
+                "ipv4_mask": faker.ipv4_netmask(),
                 "ipv6_address": faker.ipv6(),
+                "ipv6_mask": faker.ipv6_netmask(),
                 "custom_firewall_filters": faker.boolean(),
                 "bgp_peers": [
                     {