ansible-test - Code cleanup. (#76540)

* Add missing typing imports.
* Remove unnecessary non-capturing group.
* Whitespace.
* Add type hints.
* Ignore PyCharm false positives.
* Remove Python 2.x plugin loading logic.
* Remove Python 2.x exception handling.
* Remove Python 2.x display logic.
* Add changelog.

changelogs/fragments/ansible-test-code-cleanup.yml

@@ -0,0 +1,3 @@
+minor_changes:
+  - ansible-test - Miscellaneous code cleanup and type hint fixes.
+  - ansible-test - Remove unused Python 2.x compatibility code.

test/lib/ansible_test/_internal/__init__.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import os
 import sys
+import typing as t
 
 # This import should occur as early as possible.
 # It must occur before subprocess has been imported anywhere in the current process.

test/lib/ansible_test/_internal/classification/powershell.py

@@ -83,7 +83,7 @@ def extract_powershell_module_utils_imports(path, module_utils):  # type: (str,
 
     for line in lines:
         line_number += 1
-        match = re.search(r'(?i)^#\s*(?:requires\s+-module(?:s?)|ansiblerequires\s+-powershell)\s*((?:Ansible|ansible_collections|\.)\..+)', line)
+        match = re.search(r'(?i)^#\s*(?:requires\s+-modules?|ansiblerequires\s+-powershell)\s*((?:Ansible|ansible_collections|\.)\..+)', line)
 
         if not match:
             continue

test/lib/ansible_test/_internal/cli/__init__.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import argparse
 import os
 import sys
+import typing as t
 
 from .argparsing import (
     CompositeActionCompletionFinder,
@@ -42,6 +43,7 @@ def parse_args(argv=None):  # type: (t.Optional[t.List[str]]) -> argparse.Namesp
         argv = sys.argv[1:]
     else:
         argv = argv[1:]
+
     args = parser.parse_args(argv)
 
     if args.explain and not args.verbosity:

test/lib/ansible_test/_internal/cli/parsers/__init__.py

@@ -73,7 +73,7 @@ class DelegatedControllerParser(ControllerNamespaceParser, TypeParser):
     """Composite argument parser for the controller when delegation is supported."""
     def get_stateless_parsers(self):  # type: () -> t.Dict[str, Parser]
         """Return a dictionary of type names and type parsers."""
-        parsers = dict(
+        parsers: t.Dict[str, Parser] = dict(
             origin=OriginParser(),
             docker=DockerParser(controller=True),
         )
@@ -99,7 +99,7 @@ class PosixTargetParser(TargetNamespaceParser, TypeParser):
     """Composite argument parser for a POSIX target."""
     def get_stateless_parsers(self):  # type: () -> t.Dict[str, Parser]
         """Return a dictionary of type names and type parsers."""
-        parsers = dict(
+        parsers: t.Dict[str, Parser] = dict(
             controller=ControllerParser(),
             docker=DockerParser(controller=False),
         )

test/lib/ansible_test/_internal/commands/integration/__init__.py

@@ -311,6 +311,7 @@ def integration_test_environment(
             display.info('Copying %s/ to %s/' % (dir_src, dir_dst), verbosity=2)
 
             if not args.explain:
+                # noinspection PyTypeChecker
                 shutil.copytree(to_bytes(dir_src), to_bytes(dir_dst), symlinks=True)
 
         for file_src, file_dst in file_copies:

test/lib/ansible_test/_internal/containers.py

@@ -286,7 +286,7 @@ class ContainerAccess:
 
     def to_dict(self):  # type: () -> t.Dict[str, t.Any]
         """Return a dict of the current instance."""
-        value = dict(
+        value: t.Dict[str, t.Any] = dict(
             host_ip=self.host_ip,
             names=self.names,
         )

test/lib/ansible_test/_internal/init.py

@@ -1,6 +1,7 @@
 """Early initialization for ansible-test before most other imports have been performed."""
 from __future__ import annotations
 
+# noinspection PyCompatibility
 import resource
 
 from .constants import (

test/lib/ansible_test/_internal/thread.py

@@ -41,10 +41,7 @@ class WrappedThread(threading.Thread):
         result, exception = self._result.get()
 
         if exception:
-            if sys.version_info[0] > 2:
+            raise exception[1].with_traceback(exception[2])
-                raise exception[1].with_traceback(exception[2])
-            # noinspection PyRedundantParentheses
-            exec('raise exception[0], exception[1], exception[2]')  # pylint: disable=locally-disabled, exec-used
 
         self.result = result
 

test/lib/ansible_test/_internal/util.py

@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 import errno
+# noinspection PyCompatibility
 import fcntl
+import importlib.util
 import inspect
 import os
 import pkgutil
@@ -20,6 +22,7 @@ import shlex
 import typing as t
 
 from struct import unpack, pack
+# noinspection PyCompatibility
 from termios import TIOCGWINSZ
 
 from .encoding import (
@@ -587,9 +590,6 @@ class Display:
             message = message.replace(self.clear, color)
             message = '%s%s%s' % (color, message, self.clear)
 
-        if sys.version_info[0] == 2:
-            message = to_bytes(message)
-
         print(message, file=fd)
         fd.flush()
 
@@ -768,23 +768,12 @@ def load_module(path, name):  # type: (str, str) -> None
     if name in sys.modules:
         return
 
-    if sys.version_info >= (3, 4):
+    spec = importlib.util.spec_from_file_location(name, path)
-        import importlib.util
+    module = importlib.util.module_from_spec(spec)
-
+    # noinspection PyUnresolvedReferences
-        spec = importlib.util.spec_from_file_location(name, path)
+    spec.loader.exec_module(module)
-        module = importlib.util.module_from_spec(spec)
-        # noinspection PyUnresolvedReferences
-        spec.loader.exec_module(module)
-
-        sys.modules[name] = module
-    else:
-        # noinspection PyDeprecation
-        import imp  # pylint: disable=deprecated-module
 
-        # load_source (and thus load_module) require a file opened with `open` in text mode
+    sys.modules[name] = module
-        with open(to_bytes(path)) as module_file:
-            # noinspection PyDeprecation
-            imp.load_module(name, module_file, path, ('.py', 'r', imp.PY_SOURCE))
 
 
 def sanitize_host_name(name):