typing: Add hints to network commands

Change-Id: Ie3d3745461c361befd0b61bc0893bd39814035a0
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>

Author: stephenfin

openstackclient/network/client.py

@@ -10,7 +10,9 @@
 #   License for the specific language governing permissions and limitations
 #   under the License.
 
+import argparse
 import logging
+from typing import Any
 
 from osc_lib import utils
 
@@ -26,7 +28,7 @@
 API_EXTENSIONS = ('fwaas', 'taas')
 
 
-def make_client(instance):
+def make_client(instance: Any) -> Any:
     """Returns a network proxy"""
     LOG.debug(
         'Network client initialized using OpenStack SDK: %s',
@@ -35,7 +37,9 @@ def make_client(instance):
     return instance.sdk_connection.network
 
 
-def build_option_parser(parser):
+def build_option_parser(
+    parser: argparse.ArgumentParser,
+) -> argparse.ArgumentParser:
     """Hook to add global options"""
     parser.add_argument(
         '--os-network-api-version',

openstackclient/network/common.py

@@ -12,11 +12,12 @@
 #
 
 import abc
+import argparse
+from collections.abc import Generator
 import contextlib
 import logging
 from typing import Any
 
-from cliff import _argparse
 import openstack.exceptions
 from osc_lib.cli import parseractions
 from osc_lib import exceptions
@@ -43,7 +44,9 @@
 
 
 @contextlib.contextmanager
-def check_missing_extension_if_error(client_manager, attrs):
+def check_missing_extension_if_error(
+    client_manager: Any, attrs: dict[str, Any]
+) -> Generator[None, None, None]:
     # If specified option requires extension, then try to
     # find out if it exists. If it does not exist,
     # then an exception with the appropriate message
@@ -68,7 +71,7 @@ class NetDetectionMixin(command.Command, metaclass=abc.ABCMeta):
     """
 
     @property
-    def _network_type(self):
+    def _network_type(self) -> str | None:
         """Discover whether the running cloud is using neutron or nova-network.
 
         :return:
@@ -96,31 +99,31 @@ def _network_type(self):
         return self._net_type
 
     @property
-    def is_neutron(self):
+    def is_neutron(self) -> bool:
         return self._network_type is _NET_TYPE_NEUTRON
 
     @property
-    def is_nova_network(self):
+    def is_nova_network(self) -> bool:
         return self._network_type is _NET_TYPE_COMPUTE
 
     @property
-    def is_docs_build(self):
+    def is_docs_build(self) -> bool:
         return self._network_type is None
 
-    def enhance_help_neutron(self, _help):
+    def enhance_help_neutron(self, _help: str) -> str:
         if self.is_docs_build:
             # Why can't we say 'neutron'?
             return _QUALIFIER_FMT % (_help, _("Network version 2 only"))
         return _help
 
-    def enhance_help_nova_network(self, _help):
+    def enhance_help_nova_network(self, _help: str) -> str:
         if self.is_docs_build:
             # Why can't we say 'nova-network'?
             return _QUALIFIER_FMT % (_help, _("Compute version 2 only"))
         return _help
 
     @staticmethod
-    def split_help(network_help, compute_help):
+    def split_help(network_help: str, compute_help: str) -> str:
         return (
             "*{network_qualifier}:*\n  {network_help}\n\n"
             "*{compute_qualifier}:*\n  {compute_help}".format(
@@ -133,7 +136,7 @@ def split_help(network_help, compute_help):
             )
         )
 
-    def get_parser(self, prog_name: str) -> _argparse.ArgumentParser:
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         LOG.debug('get_parser(%s)', prog_name)
         parser = super().get_parser(prog_name)
         parser = self.update_parser_common(parser)
@@ -145,19 +148,25 @@ def get_parser(self, prog_name: str) -> _argparse.ArgumentParser:
             parser = self.update_parser_compute(parser)
         return parser
 
-    def update_parser_common(self, parser):
+    def update_parser_common(
+        self, parser: argparse.ArgumentParser
+    ) -> argparse.ArgumentParser:
         """Default is no updates to parser."""
         return parser
 
-    def update_parser_network(self, parser):
+    def update_parser_network(
+        self, parser: argparse.ArgumentParser
+    ) -> argparse.ArgumentParser:
         """Default is no updates to parser."""
         return parser
 
-    def update_parser_compute(self, parser):
+    def update_parser_compute(
+        self, parser: argparse.ArgumentParser
+    ) -> argparse.ArgumentParser:
         """Default is no updates to parser."""
         return parser
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> Any:
         if self.is_neutron:
             return self.take_action_network(
                 self.app.client_manager.network,
@@ -169,11 +178,15 @@ def take_action(self, parsed_args):
                 parsed_args,
             )
 
-    def take_action_network(self, client, parsed_args):
+    def take_action_network(
+        self, client: Any, parsed_args: argparse.Namespace
+    ) -> Any:
         """Override to do something useful."""
         pass
 
-    def take_action_compute(self, client, parsed_args):
+    def take_action_compute(
+        self, client: Any, parsed_args: argparse.Namespace
+    ) -> Any:
         """Override to do something useful."""
         pass
 
@@ -204,7 +217,7 @@ class NetworkAndComputeDelete(NetworkAndComputeCommand, metaclass=abc.ABCMeta):
 
     resource: str
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> None:
         ret = 0
         resources = getattr(parsed_args, self.resource, [])
 
@@ -267,7 +280,7 @@ class NetworkAndComputeShowOne(
     arguments.
     """
 
-    def take_action(self, parsed_args):
+    def take_action(self, parsed_args: argparse.Namespace) -> Any:
         try:
             if self.app.client_manager.is_network_endpoint_enabled():
                 return self.take_action_network(
@@ -300,7 +313,7 @@ class NeutronCommandWithExtraArgs(command.Command):
         'str': str,
     }
 
-    def _get_property_converter(self, _property):
+    def _get_property_converter(self, _property: dict[str, Any]) -> Any:
         if 'type' in _property:
             converter = self._allowed_types_dict.get(_property['type'])
         else:
@@ -316,15 +329,17 @@ def _get_property_converter(self, _property):
             )
         return converter
 
-    def _parse_extra_properties(self, extra_properties):
+    def _parse_extra_properties(
+        self, extra_properties: list[dict[str, Any]] | None
+    ) -> dict[str, Any]:
         result: dict[str, Any] = {}
         if extra_properties:
             for _property in extra_properties:
                 converter = self._get_property_converter(_property)
                 result[_property['name']] = converter(_property['value'])
         return result
 
-    def get_parser(self, prog_name):
+    def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
         parser = super().get_parser(prog_name)
         parser.add_argument(
             '--extra-property',
@@ -349,7 +364,9 @@ def get_parser(self, prog_name):
 
 
 class NeutronUnsetCommandWithExtraArgs(NeutronCommandWithExtraArgs):
-    def _parse_extra_properties(self, extra_properties):
+    def _parse_extra_properties(
+        self, extra_properties: list[dict[str, Any]] | None
+    ) -> dict[str, Any]:
         result: dict[str, Any] = {}
         if extra_properties:
             for _property in extra_properties:

openstackclient/network/utils.py

@@ -11,13 +11,18 @@
 #   under the License.
 #
 
+from collections.abc import Iterable, Sequence
+from typing import Any, cast
+
 from osc_lib import exceptions
 
 from openstackclient.i18n import _
 
 
 # Transform compute security group rule for display.
-def transform_compute_security_group_rule(sg_rule):
+def transform_compute_security_group_rule(
+    sg_rule: dict[str, Any],
+) -> dict[str, Any]:
     info = {}
     info.update(sg_rule)
     from_port = info.pop('from_port')
@@ -45,14 +50,14 @@ def transform_compute_security_group_rule(sg_rule):
     return info
 
 
-def str2bool(strbool):
+def str2bool(strbool: str | None) -> bool | None:
     if strbool is None:
         return None
     return strbool.lower() == 'true'
 
 
-def str2list(strlist):
-    result = []
+def str2list(strlist: str | None) -> list[str]:
+    result: list[str] = []
     if strlist:
         result = strlist.split(';')
     return result
@@ -83,12 +88,15 @@ def str2dict(strdict: str) -> dict[str, str]:
     return result
 
 
-def format_security_group_rule_show(obj):
+def format_security_group_rule_show(
+    obj: dict[str, Any],
+) -> tuple[Sequence[str], Iterable[Any]]:
     data = transform_compute_security_group_rule(obj)
-    return zip(*sorted(data.items()))
+    col_headers, col_data = zip(*sorted(data.items()))
+    return col_headers, col_data
 
 
-def format_network_port_range(rule):
+def format_network_port_range(rule: dict[str, Any]) -> str:
     # Display port range or ICMP type and code. For example:
     # - ICMP type: 'type=3'
     # - ICMP type and code: 'type=3:code=0'
@@ -114,8 +122,8 @@ def format_network_port_range(rule):
     return port_range
 
 
-def format_remote_ip_prefix(rule):
-    remote_ip_prefix = rule['remote_ip_prefix']
+def format_remote_ip_prefix(rule: dict[str, Any]) -> str | None:
+    remote_ip_prefix = cast(str | None, rule['remote_ip_prefix'])
     if remote_ip_prefix is None:
         ethertype = rule['ether_type']
         if ethertype == 'IPv4':
@@ -125,15 +133,15 @@ def format_remote_ip_prefix(rule):
     return remote_ip_prefix
 
 
-def convert_ipvx_case(string):
+def convert_ipvx_case(string: str) -> str:
     if string.lower() == 'ipv4':
         return 'IPv4'
     if string.lower() == 'ipv6':
         return 'IPv6'
     return string
 
 
-def is_icmp_protocol(protocol):
+def is_icmp_protocol(protocol: str | None) -> bool:
     # NOTE(rtheis): Neutron has deprecated protocol icmpv6.
     # However, while the OSC CLI doesn't document the protocol,
     # the code must still handle it. In addition, handle both
@@ -144,12 +152,14 @@ def is_icmp_protocol(protocol):
         return False
 
 
-def convert_to_lowercase(string):
+def convert_to_lowercase(string: str) -> str:
     return string.lower()
 
 
-def get_protocol(parsed_args, default_protocol='any'):
-    protocol = default_protocol
+def get_protocol(
+    parsed_args: Any, default_protocol: str = 'any'
+) -> str | None:
+    protocol: str | None = default_protocol
     if parsed_args.protocol is not None:
         protocol = parsed_args.protocol
     if hasattr(parsed_args, "proto") and parsed_args.proto is not None:
@@ -159,7 +169,7 @@ def get_protocol(parsed_args, default_protocol='any'):
     return protocol
 
 
-def get_ethertype(parsed_args, protocol):
+def get_ethertype(parsed_args: Any, protocol: str | None) -> str:
     ethertype = 'IPv4'
     if parsed_args.ethertype is not None:
         ethertype = parsed_args.ethertype
@@ -168,7 +178,7 @@ def get_ethertype(parsed_args, protocol):
     return ethertype
 
 
-def is_ipv6_protocol(protocol):
+def is_ipv6_protocol(protocol: str | None) -> bool:
     # NOTE(rtheis): Neutron has deprecated protocol icmpv6.
     # However, while the OSC CLI doesn't document the protocol,
     # the code must still handle it. In addition, handle both