Add PVLAN commands for Neutron

Following the merge of the spec for PVLAN [1] support, this patch
introduces the parameter --pvlan in networks, as well as --pvlan-type
and --pvlan-community in ports. It handles the validation on the client
side of the different parameters:
For Network:
- PVLAN can only be enabled on if port security is enabled.
For Port:
- PVLAN attributes can only be set if associated network if PVLAN is
enabled and port security is enabled.
- pvlan-community can only be set if pvlan-type is community

There are now unit tests for these too.

[1] https://specs.openstack.org/openstack/neutron-specs/specs/2026.1/pvlan-semantics-for-provider-networks.html

Change-Id: I10033f59b52edea37350ca148dd2328bf2322cdb
Signed-off-by: Elvira Garcia <egarciar@redhat.com>

Author: elvgarrui

openstackclient/network/v2/network.py

@@ -146,6 +146,13 @@ def _get_attrs_network(
         attrs['qos_policy_id'] = _qos_policy.id
     if 'no_qos_policy' in parsed_args and parsed_args.no_qos_policy:
         attrs['qos_policy_id'] = None
+
+    # Set pvlan
+    if parsed_args.pvlan:
+        attrs['pvlan'] = True
+    if parsed_args.no_pvlan:
+        attrs['pvlan'] = False
+
     # Update DNS network options
     if parsed_args.dns_domain is not None:
         attrs['dns_domain'] = parsed_args.dns_domain
@@ -348,6 +355,24 @@ def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
             help=_("Disable VLAN QinQ (S-Tag ethtype 0x8a88) for the network"),
         )
 
+        pvlan_grp = parser.add_mutually_exclusive_group()
+        pvlan_grp.add_argument(
+            '--pvlan',
+            action='store_true',
+            help=_(
+                "Enable Private VLAN for the network "
+                "(PVLAN extension required)"
+            ),
+        )
+        pvlan_grp.add_argument(
+            '--no-pvlan',
+            action='store_true',
+            help=_(
+                "Disable Private VLAN for the network "
+                "(PVLAN extension required)"
+            ),
+        )
+
         _add_additional_network_options(parser)
         _tag.add_tag_option_to_parser_for_create(parser, _('network'))
         return parser
@@ -367,13 +392,26 @@ def take_action(
         if parsed_args.no_qinq_vlan:
             attrs['vlan_qinq'] = False
 
+        if parsed_args.pvlan:
+            attrs['pvlan'] = True
+        if parsed_args.no_pvlan:
+            attrs['pvlan'] = False
+
         if attrs.get('vlan_transparent') and attrs.get('vlan_qinq'):
             msg = _(
                 "--transparent-vlan and --qinq-vlan can not be both enabled "
                 "for the network."
             )
             raise exceptions.CommandError(msg)
 
+        if (
+            attrs.get('port_security_enabled') is False
+            and attrs.get('pvlan') is True
+        ):
+            msg = _(
+                "--disable-port-security and --pvlan can not be used together."
+            )
+            raise exceptions.CommandError(msg)
         if (
             parsed_args.segmentation_id
             and not parsed_args.provider_network_type
@@ -770,6 +808,23 @@ def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
             action='store_true',
             help=_("Remove the QoS policy attached to this network"),
         )
+        pvlan_grp = parser.add_mutually_exclusive_group()
+        pvlan_grp.add_argument(
+            '--pvlan',
+            action='store_true',
+            help=_(
+                "Enable Private VLAN for the network. PVLAN extension "
+                "required."
+            ),
+        )
+        pvlan_grp.add_argument(
+            '--no-pvlan',
+            action='store_true',
+            help=_(
+                "Disable Private VLAN for the network (Default). "
+                "PVLAN extension required."
+            ),
+        )
         _tag.add_tag_option_to_parser_for_set(parser, _('network'))
         _add_additional_network_options(parser)
         return parser
@@ -782,6 +837,14 @@ def take_action(self, parsed_args: argparse.Namespace) -> None:
         attrs.update(
             self._parse_extra_properties(parsed_args.extra_properties)
         )
+        if (
+            attrs.get('port_security_enabled') is False
+            and attrs.get('pvlan') is True
+        ):
+            msg = _(
+                "--disable-port-security and --pvlan can not be used together."
+            )
+            raise exceptions.CommandError(msg)
         if attrs:
             with common.check_missing_extension_if_error(
                 self.app.client_manager.network, attrs

openstackclient/network/v2/port.py

@@ -105,6 +105,8 @@ def _get_columns(item: Any) -> tuple[tuple[str, ...], tuple[str, ...]]:
         'port_security_enabled': 'is_port_security_enabled',
         'project_id': 'project_id',
         'propagate_uplink_status': 'propagate_uplink_status',
+        'pvlan_type': 'pvlan_type',
+        'pvlan_community': 'pvlan_community',
         'resource_request': 'resource_request',
         'revision_number': 'revision_number',
         'qos_network_policy_id': 'qos_network_policy_id',
@@ -256,9 +258,48 @@ def _get_attrs(
     if parsed_args.trusted:
         attrs['trusted'] = True
 
+    if 'pvlan_type' in parsed_args and parsed_args.pvlan_type is not None:
+        attrs['pvlan_type'] = parsed_args.pvlan_type
+    if (
+        'pvlan_community' in parsed_args
+        and parsed_args.pvlan_community is not None
+    ):
+        attrs['pvlan_community'] = parsed_args.pvlan_community
+
+    _validate_pvlan_port(attrs)
+
     return attrs
 
 
+def _validate_pvlan_port(attrs: dict[str, Any]) -> None:
+    if (attrs.get('pvlan_type') or attrs.get('pvlan_community')) and attrs.get(
+        'port_security_enabled'
+    ) is False:
+        msg = _(
+            "PVLAN attributes cannot be set when port security is disabled."
+        )
+        raise exceptions.CommandError(msg)
+
+    if attrs.get('pvlan_type') == 'community' and not attrs.get(
+        'pvlan_community'
+    ):
+        msg = _(
+            "--pvlan-community is required when --pvlan-type is 'community'."
+        )
+        raise exceptions.CommandError(msg)
+
+
+def _validate_pvlan_network_port(attrs: dict[str, Any], network: Any) -> None:
+    if not (attrs.get('pvlan_type') or attrs.get('pvlan_community')):
+        return
+    if not network.pvlan:
+        msg = _(
+            "PVLAN attributes cannot be set on a port whose "
+            "network does not have PVLAN enabled."
+        )
+        raise exceptions.CommandError(msg)
+
+
 def _prepare_fixed_ips(
     client_manager: Any, parsed_args: argparse.Namespace
 ) -> None:
@@ -445,6 +486,26 @@ def _add_updatable_args(
             "which expect it in this dictionary (for example, Nova)."
         ),
     )
+    parser.add_argument(
+        '--pvlan-type',
+        metavar='<type>',
+        choices=['promiscuous', 'isolated', 'community'],
+        dest='pvlan_type',
+        help=_(
+            "Set Private VLAN type for this port. Requires PVLAN service "
+            "plugin. Default: promiscuous."
+        ),
+    )
+    parser.add_argument(
+        '--pvlan-community',
+        metavar='<community>',
+        dest='pvlan_community',
+        help=_(
+            "Set PVLAN community name for this port. "
+            "Only applies when pvlan-type is 'community'. "
+            "Requires PVLAN service plugin. Default: None."
+        ),
+    )
 
 
 # TODO(abhiraut): Use the SDK resource mapped attribute names once the
@@ -732,6 +793,8 @@ def take_action(
             self._parse_extra_properties(parsed_args.extra_properties)
         )
 
+        _validate_pvlan_network_port(attrs, network)
+
         with common.check_missing_extension_if_error(network_client, attrs):
             obj = network_client.create_port(**attrs)
 
@@ -1232,6 +1295,10 @@ def take_action(self, parsed_args: argparse.Namespace) -> None:
             self._parse_extra_properties(parsed_args.extra_properties)
         )
 
+        if attrs.get('pvlan_type') or attrs.get('pvlan_community'):
+            network = client.find_network(obj.network_id, ignore_missing=False)
+            _validate_pvlan_network_port(attrs, network)
+
         if attrs:
             with common.check_missing_extension_if_error(
                 self.app.client_manager.network, attrs
@@ -1355,6 +1422,13 @@ def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
             default=False,
             help=_("Clear device owner for the port."),
         )
+        parser.add_argument(
+            '--pvlan-community',
+            action='store_true',
+            default=False,
+            dest='pvlan_community',
+            help=_("Clear PVLAN community name for the port."),
+        )
         _tag.add_tag_option_to_parser_for_unset(parser, _('port'))
         parser.add_argument(
             'port',
@@ -1425,6 +1499,8 @@ def take_action(self, parsed_args: argparse.Namespace) -> None:
             attrs['device_id'] = ''
         if parsed_args.device_owner:
             attrs['device_owner'] = ''
+        if parsed_args.pvlan_community:
+            attrs['pvlan_community'] = None
 
         attrs.update(
             self._parse_extra_properties(parsed_args.extra_properties)

openstackclient/tests/unit/network/v2/fakes.py

@@ -58,6 +58,8 @@
 from openstackclient.tests.unit import utils
 
 
+PVLAN_TYPE_COMMUNITY = 'community'
+PVLAN_COMMUNITY_NAME = 'community_1'
 RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit'
 RULE_TYPE_DSCP_MARKING = 'dscp-marking'
 RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth'
@@ -951,6 +953,7 @@ def create_one_network(attrs=None):
         'provider:network_type': 'vlan',
         'provider:physical_network': 'physnet1',
         'provider:segmentation_id': "400",
+        'pvlan': "False",
         'router:external': True,
         'availability_zones': [],
         'availability_zone_hints': [],
@@ -1211,6 +1214,8 @@ def create_one_port(attrs=None):
         'security_group_ids': [],
         'status': 'ACTIVE',
         'project_id': 'project-id-' + uuid.uuid4().hex,
+        'pvlan_type': PVLAN_TYPE_COMMUNITY,
+        'pvlan_community': PVLAN_COMMUNITY_NAME,
         'qos_network_policy_id': 'qos-policy-id-' + uuid.uuid4().hex,
         'qos_policy_id': 'qos-policy-id-' + uuid.uuid4().hex,
         'tags': [],

openstackclient/tests/unit/network/v2/test_network.py

@@ -70,6 +70,7 @@ class TestCreateNetworkIdentityV3(TestNetwork):
         'provider:network_type',
         'provider:physical_network',
         'provider:segmentation_id',
+        'pvlan',
         'qos_policy_id',
         'router:external',
         'shared',
@@ -99,6 +100,7 @@ class TestCreateNetworkIdentityV3(TestNetwork):
         _network.provider_network_type,
         _network.provider_physical_network,
         _network.provider_segmentation_id,
+        _network.pvlan,
         _network.qos_policy_id,
         network.RouterExternalColumn(_network.is_router_external),
         _network.is_shared,
@@ -189,6 +191,7 @@ def test_create_all_options(self):
             self.qos_policy.id,
             "--transparent-vlan",
             "--no-qinq-vlan",
+            "--no-pvlan",
             "--enable-port-security",
             "--dns-domain",
             "example.org.",
@@ -210,6 +213,7 @@ def test_create_all_options(self):
             ('qos_policy', self.qos_policy.id),
             ('transparent_vlan', True),
             ('qinq_vlan', False),
+            ('pvlan', False),
             ('enable_port_security', True),
             ('name', self._network.name),
             ('dns_domain', 'example.org.'),
@@ -235,6 +239,7 @@ def test_create_all_options(self):
                 'qos_policy_id': self.qos_policy.id,
                 'vlan_transparent': True,
                 'vlan_qinq': False,
+                'pvlan': False,
                 'port_security_enabled': True,
                 'dns_domain': 'example.org.',
             }
@@ -326,6 +331,19 @@ def test_create_with_vlan_qinq_and_transparency_enabled(self):
             exceptions.CommandError, self.cmd.take_action, parsed_args
         )
 
+    def test_create_with_pvlan_and_port_security_disabled(self):
+        arglist = [
+            "--disable-port-security",
+            "--pvlan",
+            self._network.name,
+        ]
+        verifylist = [('disable_port_security', True), ('pvlan', True)]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        self.assertRaises(
+            exceptions.CommandError, self.cmd.take_action, parsed_args
+        )
+
     def test_create_with_provider_segment_without_provider_type(self):
         arglist = [
             "--provider-segment",
@@ -371,6 +389,7 @@ class TestCreateNetworkIdentityV2(
         'name',
         'port_security_enabled',
         'project_id',
+        'pvlan',
         'provider:network_type',
         'provider:physical_network',
         'provider:segmentation_id',
@@ -400,6 +419,7 @@ class TestCreateNetworkIdentityV2(
         _network.name,
         _network.is_port_security_enabled,
         _network.project_id,
+        _network.pvlan,
         _network.provider_network_type,
         _network.provider_physical_network,
         _network.provider_segmentation_id,
@@ -996,6 +1016,23 @@ def setUp(self):
         # Get the command object to test
         self.cmd = network.SetNetwork(self.app, None)
 
+    def test_set_with_pvlan_and_port_security_disabled(self):
+        arglist = [
+            self._network.name,
+            '--disable-port-security',
+            '--pvlan',
+        ]
+        verifylist = [
+            ('network', self._network.name),
+            ('disable_port_security', True),
+            ('pvlan', True),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+        self.assertRaises(
+            exceptions.CommandError, self.cmd.take_action, parsed_args
+        )
+
     def test_set_this(self):
         arglist = [
             self._network.name,
@@ -1189,6 +1226,7 @@ class TestShowNetwork(TestNetwork):
         'provider:network_type',
         'provider:physical_network',
         'provider:segmentation_id',
+        'pvlan',
         'qos_policy_id',
         'router:external',
         'shared',
@@ -1218,6 +1256,7 @@ class TestShowNetwork(TestNetwork):
         _network.provider_network_type,
         _network.provider_physical_network,
         _network.provider_segmentation_id,
+        _network.pvlan,
         _network.qos_policy_id,
         network.RouterExternalColumn(_network.is_router_external),
         _network.is_shared,
@@ -1265,7 +1304,6 @@ def test_show_all_options(self):
         self.network_client.find_network.assert_called_once_with(
             self._network.name, ignore_missing=False
         )
-
         self.assertEqual(set(self.columns), set(columns))
         self.assertCountEqual(self.data, data)