Add network trunk subport commands

The ``network subport list`` command requires the trunk to be passed
as a ``--trunk`` option, which reads backwards for a resource that
cannot exist without a parent trunk. Likewise, managing subports means
reaching for ``network trunk set --subport`` and ``network trunk unset
--subport``, which bury the operation in a generic set/unset command
and require a nested key=value blob to name the port.

Add three cmds that take the trunk as a positional argument instead:

  openstack network trunk subport list <trunk>
  openstack network trunk subport add <trunk> <port>
      [--segmentation-type <type>] [--segmentation-id <id>]
  openstack network trunk subport remove <trunk> <port> [<port> ...]

``add`` uses discrete ``--segmentation-type`` and ``--segmentation-id``
options rather than the nested key=value form, and
``--segmentation-id`` is parsed as an integer so argparse reports bad
input directly. ``remove`` accepts multiple ports, mirroring ``network
trunk delete``.

This change is purely additive. The existing ``network subport list``,
``network trunk set`` and ``network trunk unset`` commands are
unchanged, so nothing breaks for current users.

The subport listing body is extracted into a ``_list_trunk_subports()``
helper shared by both list commands. No documentation change is needed
as the existing ``network trunk *`` autoprogram-cliff glob already
picks up the new commands.

Assisted-By: Claude Opus 5
Change-Id: I85e07e963be41196452760b844a40e7f3223d379
Signed-off-by: Doug Goldstein <cardoe@cardoe.com>

Author: cardoe

openstackclient/network/v2/network_trunk.py

@@ -308,31 +308,116 @@ def take_action(
         self, parsed_args: argparse.Namespace
     ) -> tuple[tuple[str, ...], Iterable[tuple[Any, ...]]]:
         network_client = self.app.client_manager.network
-        trunk_id = network_client.find_trunk(
+        return _list_trunk_subports(network_client, parsed_args.trunk)
+
+
+class ListNetworkTrunkSubport(command.Lister):
+    """List all subports for a given network trunk"""
+
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            'trunk',
+            metavar="<trunk>",
+            help=_("List subports belonging to this trunk (name or ID)"),
+        )
+        return parser
+
+    def take_action(
+        self, parsed_args: argparse.Namespace
+    ) -> tuple[tuple[str, ...], Iterable[tuple[Any, ...]]]:
+        network_client = self.app.client_manager.network
+        return _list_trunk_subports(network_client, parsed_args.trunk)
+
+
+class AddNetworkTrunkSubport(command.Command):
+    """Add a subport to a given network trunk"""
+
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            'trunk',
+            metavar="<trunk>",
+            help=_("Trunk to add the subport to (name or ID)"),
+        )
+        parser.add_argument(
+            'port',
+            metavar="<port>",
+            help=_("Port to add as a subport (name or ID)"),
+        )
+        parser.add_argument(
+            '--segmentation-type',
+            metavar="<segmentation-type>",
+            help=_("Segmentation type of the subport, e.g. 'vlan'"),
+        )
+        parser.add_argument(
+            '--segmentation-id',
+            metavar="<segmentation-id>",
+            type=int,
+            help=_("Segmentation ID of the subport"),
+        )
+        return parser
+
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
+        network_client = self.app.client_manager.network
+        trunk = network_client.find_trunk(
             parsed_args.trunk,
             ignore_missing=False,
         )
-        data = network_client.get_trunk_subports(trunk_id)
-        headers: tuple[str, ...] = (
-            'Port',
-            'Segmentation Type',
-            'Segmentation ID',
+        subport: dict[str, Any] = {
+            'port_id': network_client.find_port(
+                parsed_args.port,
+                ignore_missing=False,
+            ).id
+        }
+        if parsed_args.segmentation_type is not None:
+            subport['segmentation_type'] = parsed_args.segmentation_type
+        if parsed_args.segmentation_id is not None:
+            subport['segmentation_id'] = parsed_args.segmentation_id
+        try:
+            network_client.add_trunk_subports(trunk, [subport])
+        except Exception as e:
+            msg = _("Failed to add subport to trunk '%(t)s': %(e)s") % {
+                't': parsed_args.trunk,
+                'e': e,
+            }
+            raise exceptions.CommandError(msg)
+
+
+class RemoveNetworkTrunkSubport(command.Command):
+    """Remove subport(s) from a given network trunk"""
+
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
+        parser = super().get_parser(prog_name)
+        parser.add_argument(
+            'trunk',
+            metavar="<trunk>",
+            help=_("Trunk to remove the subport(s) from (name or ID)"),
         )
-        columns: tuple[str, ...] = (
-            'port_id',
-            'segmentation_type',
-            'segmentation_id',
+        parser.add_argument(
+            'port',
+            metavar="<port>",
+            nargs="+",
+            help=_("Port(s) to remove as a subport (name or ID)"),
         )
-        return (
-            headers,
-            (
-                osc_utils.get_dict_properties(
-                    s,
-                    columns,
-                )
-                for s in data[SUB_PORTS]
-            ),
+        return parser
+
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
+        network_client = self.app.client_manager.network
+        trunk = network_client.find_trunk(
+            parsed_args.trunk,
+            ignore_missing=False,
         )
+        attrs = [
+            {
+                'port_id': network_client.find_port(
+                    port,
+                    ignore_missing=False,
+                ).id
+            }
+            for port in parsed_args.port
+        ]
+        network_client.delete_trunk_subports(trunk, attrs)
 
 
 class UnsetNetworkTrunk(command.Command):
@@ -383,6 +468,36 @@ def _get_columns(
     )
 
 
+def _list_trunk_subports(
+    network_client: network_v2.Proxy, trunk: str
+) -> tuple[tuple[str, ...], Iterable[tuple[Any, ...]]]:
+    trunk_obj = network_client.find_trunk(
+        trunk,
+        ignore_missing=False,
+    )
+    data = network_client.get_trunk_subports(trunk_obj)
+    headers: tuple[str, ...] = (
+        'Port',
+        'Segmentation Type',
+        'Segmentation ID',
+    )
+    columns: tuple[str, ...] = (
+        'port_id',
+        'segmentation_type',
+        'segmentation_id',
+    )
+    return (
+        headers,
+        (
+            osc_utils.get_dict_properties(
+                s,
+                columns,
+            )
+            for s in data[SUB_PORTS]
+        ),
+    )
+
+
 def _get_attrs_for_trunk(
     network_client: network_v2.Proxy,
     identity_client: identity_v2.Proxy | identity_v3.Proxy,

openstackclient/tests/functional/network/v2/test_network_trunk.py

@@ -146,3 +146,41 @@ def test_network_trunk_list_subports(self):
             ],
             json_output,
         )
+
+    def test_network_trunk_subport_add_remove_list(self):
+        trunk_name = uuid.uuid4().hex
+        json_output = json.loads(
+            self.openstack(
+                f'network trunk create {trunk_name} --parent-port {self.parent_port_name} -f json '
+            )
+        )
+        self.addCleanup(self.openstack, 'network trunk delete ' + trunk_name)
+        self.assertEqual(trunk_name, json_output['name'])
+
+        # Add subport to trunk
+        self.openstack(
+            f'network trunk subport add {trunk_name} {self.sub_port_name} '
+            '--segmentation-type vlan --segmentation-id 42'
+        )
+        json_output = json.loads(
+            self.openstack(f'network trunk subport list {trunk_name} -f json')
+        )
+        self.assertEqual(
+            [
+                {
+                    'Port': self.sub_port_id,
+                    'Segmentation ID': 42,
+                    'Segmentation Type': 'vlan',
+                }
+            ],
+            json_output,
+        )
+
+        # Remove subport from trunk
+        self.openstack(
+            f'network trunk subport remove {trunk_name} {self.sub_port_name}'
+        )
+        json_output = json.loads(
+            self.openstack(f'network trunk subport list {trunk_name} -f json')
+        )
+        self.assertEqual([], json_output)