From cb1438e6386e35e4806b5ed75f6a8da6e8e76a2c Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Thu, 10 Oct 2024 13:45:22 +0200
Subject: [PATCH] Add IPV4 and IPV6 netmask to Service Binding Port.
---
...5d917_add_ipv4_ipv6_netmask_to_service_.py | 99 +++++++++++++++++++
.../product_blocks/service_binding_port.py | 10 +-
gso/settings.py | 14 +--
gso/utils/types/ip_address.py | 2 +
gso/workflows/geant_ip/create_geant_ip.py | 4 +-
.../geant_ip/create_imported_geant_ip.py | 4 +-
gso/workflows/geant_ip/modify_geant_ip.py | 4 +-
test/conftest.py | 6 ++
test/fixtures/geant_ip_fixtures.py | 4 +
.../geant_ip/test_create_geant_ip.py | 2 +
.../geant_ip/test_create_imported_geant_ip.py | 2 +
11 files changed, 137 insertions(+), 14 deletions(-)
create mode 100644 gso/migrations/versions/2024-10-10_df108295d917_add_ipv4_ipv6_netmask_to_service_.py
diff --git a/gso/migrations/versions/2024-10-10_df108295d917_add_ipv4_ipv6_netmask_to_service_.py b/gso/migrations/versions/2024-10-10_df108295d917_add_ipv4_ipv6_netmask_to_service_.py
new file mode 100644
index 00000000..98a1bb47
--- /dev/null
+++ b/gso/migrations/versions/2024-10-10_df108295d917_add_ipv4_ipv6_netmask_to_service_.py
@@ -0,0 +1,99 @@
+"""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')
+ """))
diff --git a/gso/products/product_blocks/service_binding_port.py b/gso/products/product_blocks/service_binding_port.py
index 1aaf0862..f88a2e2c 100644
--- a/gso/products/product_blocks/service_binding_port.py
+++ b/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.
diff --git a/gso/settings.py b/gso/settings.py
index 0979dcb3..3596eb1c 100644
--- a/gso/settings.py
+++ b/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):
diff --git a/gso/utils/types/ip_address.py b/gso/utils/types/ip_address.py
index 820377b2..6186a1f8 100644
--- a/gso/utils/types/ip_address.py
+++ b/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(
diff --git a/gso/workflows/geant_ip/create_geant_ip.py b/gso/workflows/geant_ip/create_geant_ip.py
index 7c95268e..85b114a4 100644
--- a/gso/workflows/geant_ip/create_geant_ip.py
+++ b/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
diff --git a/gso/workflows/geant_ip/create_imported_geant_ip.py b/gso/workflows/geant_ip/create_imported_geant_ip.py
index ef620e73..3c5b7a03 100644
--- a/gso/workflows/geant_ip/create_imported_geant_ip.py
+++ b/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]
diff --git a/gso/workflows/geant_ip/modify_geant_ip.py b/gso/workflows/geant_ip/modify_geant_ip.py
index d568147b..fadf6aed 100644
--- a/gso/workflows/geant_ip/modify_geant_ip.py
+++ b/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(
diff --git a/test/conftest.py b/test/conftest.py
index 1083a001..e8d0a283 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -101,6 +101,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-@#/@#/@#")
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/geant_ip_fixtures.py
index 5e35a7c8..49b501e9 100644
--- a/test/fixtures/geant_ip_fixtures.py
+++ b/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
diff --git a/test/workflows/geant_ip/test_create_geant_ip.py b/test/workflows/geant_ip/test_create_geant_ip.py
index a93698ed..1b5f68d8 100644
--- a/test/workflows/geant_ip/test_create_geant_ip.py
+++ b/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()},
diff --git a/test/workflows/geant_ip/test_create_imported_geant_ip.py b/test/workflows/geant_ip/test_create_imported_geant_ip.py
index ffd975f2..87487bc7 100644
--- a/test/workflows/geant_ip/test_create_imported_geant_ip.py
+++ b/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": [
{
--
GitLab