Fixes for type hinting issues. (#77170)

lib/ansible/executor/module_common.py

@@ -66,7 +66,7 @@ except NameError:
 
 display = Display()
 
-ModuleUtilsProcessEntry = namedtuple('ModuleUtilsInfo', ['name_parts', 'is_ambiguous', 'has_redirected_child', 'is_optional'])
+ModuleUtilsProcessEntry = namedtuple('ModuleUtilsProcessEntry', ['name_parts', 'is_ambiguous', 'has_redirected_child', 'is_optional'])
 
 REPLACER = b"#<<INCLUDE_ANSIBLE_MODULE_COMMON>>"
 REPLACER_VERSION = b"\"<<ANSIBLE_VERSION>>\""

lib/ansible/executor/process/worker.py

@@ -36,7 +36,7 @@ __all__ = ['WorkerProcess']
 display = Display()
 
 
-class WorkerProcess(multiprocessing_context.Process):
+class WorkerProcess(multiprocessing_context.Process):  # type: ignore[name-defined]
     '''
     The worker thread class, which uses TaskExecutor to run tasks
     read from a job queue and pushes results into a results queue

lib/ansible/galaxy/api.py

@@ -15,6 +15,7 @@ import stat
 import tarfile
 import time
 import threading
+import typing as t
 
 from urllib.error import HTTPError
 from urllib.parse import quote as urlquote, urlencode, urlparse, parse_qs, urljoin
@@ -298,7 +299,7 @@ class GalaxyAPI:
         return to_native(self.name)
 
     def __unicode__(self):
-        # type: (GalaxyAPI) -> unicode
+        # type: (GalaxyAPI) -> t.Text
         """Render GalaxyAPI as a unicode/text string representation."""
         return to_text(self.name)
 
@@ -314,7 +315,7 @@ class GalaxyAPI:
         )
 
     def __lt__(self, other_galaxy_api):
-        # type: (GalaxyAPI, GalaxyAPI) -> Union[bool, 'NotImplemented']
+        # type: (GalaxyAPI, GalaxyAPI) -> t.Union[bool, 'NotImplemented']
         """Return whether the instance priority is higher than other."""
         if not isinstance(other_galaxy_api, self.__class__):
             return NotImplemented

lib/ansible/galaxy/collection/__init__.py

@@ -1285,7 +1285,7 @@ def write_source_metadata(collection, b_collection_path, artifacts_manager):
 
 
 def verify_artifact_manifest(manifest_file, signatures, keyring):
-    # type: (str, str, List[str]) -> None
+    # type: (str, str, str) -> None
     failed_verify = False
     coll_path_parts = to_text(manifest_file, errors='surrogate_or_strict').split(os.path.sep)
     collection_name = '%s.%s' % (coll_path_parts[-3], coll_path_parts[-2])  # get 'ns' and 'coll' from /path/to/ns/coll/MANIFEST.json

lib/ansible/module_utils/basic.py

@@ -203,14 +203,14 @@ imap = map
 
 try:
     # Python 2
-    unicode
+    unicode  # type: ignore[has-type]
 except NameError:
     # Python 3
     unicode = text_type
 
 try:
     # Python 2
-    basestring
+    basestring  # type: ignore[has-type]
 except NameError:
     # Python 3
     basestring = string_types

lib/ansible/module_utils/facts/collector.py

@@ -59,7 +59,7 @@ class BaseFactCollector:
     _fact_ids = set()
 
     _platform = 'Generic'
-    name = None
+    name = None  # type: str | None
     required_facts = set()
 
     def __init__(self, collectors=None, namespace=None):

lib/ansible/module_utils/facts/packages.py

@@ -52,7 +52,7 @@ class PkgMgr(with_metaclass(ABCMeta, object)):
 
 class LibMgr(PkgMgr):
 
-    LIB = None
+    LIB = None  # type: str | None
 
     def __init__(self):
 
@@ -71,7 +71,7 @@ class LibMgr(PkgMgr):
 
 class CLIMgr(PkgMgr):
 
-    CLI = None
+    CLI = None  # type: str | None
 
     def __init__(self):
 

lib/ansible/modules/apt.py

@@ -376,7 +376,6 @@ CLEAN_OP_CHANGED_STR = dict(
     autoclean='Del ',
 )
 
-apt = apt_pkg = None  # keep pylint happy by declaring unconditionally
 
 HAS_PYTHON_APT = False
 try:
@@ -385,7 +384,7 @@ try:
     import apt_pkg
     HAS_PYTHON_APT = True
 except ImportError:
-    pass
+    apt = apt_pkg = None
 
 
 class PolicyRcD(object):

lib/ansible/modules/apt_repository.py

@@ -151,9 +151,6 @@ from ansible.module_utils._text import to_native
 from ansible.module_utils.six import PY3
 from ansible.module_utils.urls import fetch_url
 
-# init module names to keep pylint happy
-apt = apt_pkg = aptsources_distro = distro = None
-
 try:
     import apt
     import apt_pkg
@@ -163,6 +160,8 @@ try:
 
     HAVE_PYTHON_APT = True
 except ImportError:
+    apt = apt_pkg = aptsources_distro = distro = None
+
     HAVE_PYTHON_APT = False
 
 DEFAULT_SOURCES_PERM = 0o0644

lib/ansible/modules/group.py

@@ -129,7 +129,7 @@ class Group(object):
     """
 
     platform = 'Generic'
-    distribution = None
+    distribution = None  # type: str | None
     GROUPFILE = '/etc/group'
 
     def __new__(cls, *args, **kwargs):

lib/ansible/modules/hostname.py

@@ -71,6 +71,11 @@ import platform
 import socket
 import traceback
 
+try:
+    import typing as t
+except ImportError:
+    t = None
+
 from ansible.module_utils.basic import (
     AnsibleModule,
     get_distribution,
@@ -146,8 +151,8 @@ class Hostname(object):
     """
 
     platform = 'Generic'
-    distribution = None
-    strategy_class = UnimplementedStrategy
+    distribution = None  # type: str | None
+    strategy_class = UnimplementedStrategy  # type: t.Type[BaseStrategy]
 
     def __new__(cls, *args, **kwargs):
         new_cls = get_platform_subclass(Hostname)

lib/ansible/modules/unarchive.py

@@ -259,7 +259,7 @@ from ansible.module_utils.common.locale import get_best_parsable_locale
 from ansible.module_utils.urls import fetch_file
 
 try:  # python 3.3+
-    from shlex import quote
+    from shlex import quote  # type: ignore[attr-defined]
 except ImportError:  # older python
     from pipes import quote
 

lib/ansible/modules/user.py

@@ -496,9 +496,8 @@ class User(object):
 
     All subclasses MUST define platform and distribution (which may be None).
     """
-
     platform = 'Generic'
-    distribution = None
+    distribution = None  # type: str | None
     PASSWORDFILE = '/etc/passwd'
     SHADOWFILE = '/etc/shadow'
     SHADOWFILE_EXPIRE_INDEX = 7

lib/ansible/parsing/yaml/loader.py

@@ -41,7 +41,7 @@ else:
         def __init__(self, stream, file_name=None, vault_secrets=None):
             Reader.__init__(self, stream)
             Scanner.__init__(self)
-            Parser.__init__(self)  # pylint: disable=non-parent-init-called
+            Parser.__init__(self)  # type: ignore[call-arg]  # pylint: disable=non-parent-init-called
             Composer.__init__(self)
             AnsibleConstructor.__init__(self, file_name=file_name, vault_secrets=vault_secrets)
             Resolver.__init__(self)

lib/ansible/playbook/role_include.py

@@ -20,6 +20,8 @@ __metaclass__ = type
 
 from os.path import basename
 
+import typing as t
+
 import ansible.constants as C
 from ansible.errors import AnsibleParserError
 from ansible.playbook.attribute import FieldAttribute
@@ -43,9 +45,9 @@ class IncludeRole(TaskInclude):
     circumstances related to the `- include_role: ...`
     """
 
-    BASE = ('name', 'role')  # directly assigned
-    FROM_ARGS = ('tasks_from', 'vars_from', 'defaults_from', 'handlers_from')  # used to populate from dict in role
-    OTHER_ARGS = ('apply', 'public', 'allow_duplicates', 'rolespec_validate')  # assigned to matching property
+    BASE = ('name', 'role')  # type: t.Tuple[str, ...]  # directly assigned
+    FROM_ARGS = ('tasks_from', 'vars_from', 'defaults_from', 'handlers_from')  # type: t.Tuple[str, ...]  # used to populate from dict in role
+    OTHER_ARGS = ('apply', 'public', 'allow_duplicates', 'rolespec_validate')  # type: t.Tuple[str, ...]  # assigned to matching property
     VALID_ARGS = tuple(frozenset(BASE + FROM_ARGS + OTHER_ARGS))  # all valid args
 
     # =================================================================================

lib/ansible/plugins/become/__init__.py

@@ -23,7 +23,7 @@ def _gen_id(length=32):
 
 class BecomeBase(AnsiblePlugin):
 
-    name = None
+    name = None  # type: str | None
 
     # messages for detecting prompted password issues
     fail = tuple()

lib/ansible/plugins/connection/__init__.py

@@ -48,7 +48,7 @@ class ConnectionBase(AnsiblePlugin):
     # When running over this connection type, prefer modules written in a certain language
     # as discovered by the specified file extension.  An empty string as the
     # language means any language.
-    module_implementation_preferences = ('',)
+    module_implementation_preferences = ('',)  # type: tuple[str, ...]
     allow_executable = True
 
     # the following control whether or not the connection supports the

lib/ansible/utils/_junit_xml.py

@@ -167,7 +167,7 @@ class TestSuite:
     @property
     def time(self) -> decimal.Decimal:
         """The total time from all test cases."""
-        return sum(case.time for case in self.cases if case.time)
+        return t.cast(decimal.Decimal, sum(case.time for case in self.cases if case.time))
 
     def get_attributes(self) -> t.Dict[str, str]:
         """Return a dictionary of attributes for this instance."""
@@ -233,7 +233,7 @@ class TestSuites:
     @property
     def time(self) -> decimal.Decimal:
         """The total time from all test cases."""
-        return sum(suite.time for suite in self.suites)
+        return t.cast(decimal.Decimal, sum(suite.time for suite in self.suites))
 
     def get_attributes(self) -> t.Dict[str, str]:
         """Return a dictionary of attributes for this instance."""

lib/ansible/utils/collection_loader/_collection_finder.py

@@ -36,7 +36,7 @@ try:
     from importlib import reload as reload_module
 except ImportError:
     # 2.7 has a global reload function instead...
-    reload_module = reload  # pylint:disable=undefined-variable
+    reload_module = reload  # type: ignore[name-defined]  # pylint:disable=undefined-variable
 
 try:
     from importlib.util import spec_from_loader