ansible-test - Defer loading of completion entries. (#76852)

* ansible-test - Defer loading of completion entries.

This avoids a traceback when running ansible-test outside of a supported directory.

changelogs/fragments/ansible-test-unsupported-directory-traceback.yaml

@@ -0,0 +1,2 @@
+bugfixes:
+  - ansible-test - Show an error message instead of a traceback when running outside of a supported directory.

test/integration/targets/ansible-test/collection-tests/unsupported-directory.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+
+set -eux -o pipefail
+
+cd "${WORK_DIR}"
+
+if ansible-test --help 1>stdout 2>stderr; then
+  echo "ansible-test did not fail"
+  exit 1
+fi
+
+grep '^Current working directory: ' stderr
+
+if grep raise stderr; then
+  echo "ansible-test failed with a traceback instead of an error message"
+  exit 2
+fi

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

@@ -26,8 +26,8 @@ from ..docker_util import (
 )
 
 from ..completion import (
-    DOCKER_COMPLETION,
-    REMOTE_COMPLETION,
+    docker_completion,
+    remote_completion,
     filter_completion,
 )
 
@@ -68,7 +68,7 @@ def controller_python(version):  # type: (t.Optional[str]) -> t.Optional[str]
 def get_fallback_remote_controller():  # type: () -> str
     """Return the remote fallback platform for the controller."""
     platform = 'freebsd'  # lower cost than RHEL and macOS
-    candidates = [item for item in filter_completion(REMOTE_COMPLETION).values() if item.controller_supported and item.platform == platform]
+    candidates = [item for item in filter_completion(remote_completion()).values() if item.controller_supported and item.platform == platform]
     fallback = sorted(candidates, key=lambda value: str_to_version(value.version), reverse=True)[0]
     return fallback.name
 
@@ -316,7 +316,7 @@ def get_legacy_host_config(
             targets = [ControllerConfig(python=VirtualPythonConfig(version=options.python or 'default',
                                                                    system_site_packages=options.venv_system_site_packages))]
     elif options.docker:
-        docker_config = filter_completion(DOCKER_COMPLETION).get(options.docker)
+        docker_config = filter_completion(docker_completion()).get(options.docker)
 
         if docker_config:
             if options.python and options.python not in docker_config.supported_pythons:
@@ -350,7 +350,7 @@ def get_legacy_host_config(
                 targets = [DockerConfig(name=options.docker, python=native_python(options),
                                         privileged=options.docker_privileged, seccomp=options.docker_seccomp, memory=options.docker_memory)]
     elif options.remote:
-        remote_config = filter_completion(REMOTE_COMPLETION).get(options.remote)
+        remote_config = filter_completion(remote_completion()).get(options.remote)
         context, reason = None, None
 
         if remote_config:

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

@@ -14,10 +14,10 @@ from ..constants import (
 )
 
 from ..completion import (
-    DOCKER_COMPLETION,
-    NETWORK_COMPLETION,
-    REMOTE_COMPLETION,
-    WINDOWS_COMPLETION,
+    docker_completion,
+    network_completion,
+    remote_completion,
+    windows_completion,
     filter_completion,
 )
 
@@ -425,9 +425,9 @@ def add_environment_docker(
 ):  # type: (...) -> None
     """Add environment arguments for running in docker containers."""
     if target_mode in (TargetMode.POSIX_INTEGRATION, TargetMode.SHELL):
-        docker_images = sorted(filter_completion(DOCKER_COMPLETION))
+        docker_images = sorted(filter_completion(docker_completion()))
     else:
-        docker_images = sorted(filter_completion(DOCKER_COMPLETION, controller_only=True))
+        docker_images = sorted(filter_completion(docker_completion(), controller_only=True))
 
     exclusive_parser.add_argument(
         '--docker',
@@ -538,7 +538,7 @@ def complete_windows(prefix: str, parsed_args: argparse.Namespace, **_) -> t.Lis
 
 def complete_network_platform(prefix: str, parsed_args: argparse.Namespace, **_) -> t.List[str]:
     """Return a list of supported network platforms matching the given prefix, excluding platforms already parsed from the command line."""
-    images = sorted(filter_completion(NETWORK_COMPLETION))
+    images = sorted(filter_completion(network_completion()))
 
     return [i for i in images if i.startswith(prefix) and (not parsed_args.platform or i not in parsed_args.platform)]
 
@@ -546,7 +546,7 @@ def complete_network_platform(prefix: str, parsed_args: argparse.Namespace, **_)
 def complete_network_platform_collection(prefix: str, parsed_args: argparse.Namespace, **_) -> t.List[str]:
     """Return a list of supported network platforms matching the given prefix, excluding collection platforms already parsed from the command line."""
     left = prefix.split('=')[0]
-    images = sorted(set(image.platform for image in filter_completion(NETWORK_COMPLETION).values()))
+    images = sorted(set(image.platform for image in filter_completion(network_completion()).values()))
 
     return [i + '=' for i in images if i.startswith(left) and (not parsed_args.platform_collection or i not in [x[0] for x in parsed_args.platform_collection])]
 
@@ -554,21 +554,21 @@ def complete_network_platform_collection(prefix: str, parsed_args: argparse.Name
 def complete_network_platform_connection(prefix: str, parsed_args: argparse.Namespace, **_) -> t.List[str]:
     """Return a list of supported network platforms matching the given prefix, excluding connection platforms already parsed from the command line."""
     left = prefix.split('=')[0]
-    images = sorted(set(image.platform for image in filter_completion(NETWORK_COMPLETION).values()))
+    images = sorted(set(image.platform for image in filter_completion(network_completion()).values()))
 
     return [i + '=' for i in images if i.startswith(left) and (not parsed_args.platform_connection or i not in [x[0] for x in parsed_args.platform_connection])]
 
 
 def get_remote_platform_choices(controller=False):  # type: (bool) -> t.List[str]
     """Return a list of supported remote platforms matching the given prefix."""
-    return sorted(filter_completion(REMOTE_COMPLETION, controller_only=controller))
+    return sorted(filter_completion(remote_completion(), controller_only=controller))
 
 
 def get_windows_platform_choices():  # type: () -> t.List[str]
     """Return a list of supported Windows versions matching the given prefix."""
-    return sorted(f'windows/{windows.version}' for windows in filter_completion(WINDOWS_COMPLETION).values())
+    return sorted(f'windows/{windows.version}' for windows in filter_completion(windows_completion()).values())
 
 
 def get_windows_version_choices():  # type: () -> t.List[str]
     """Return a list of supported Windows versions."""
-    return sorted(windows.version for windows in filter_completion(WINDOWS_COMPLETION).values())
+    return sorted(windows.version for windows in filter_completion(windows_completion()).values())

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

@@ -9,8 +9,8 @@ from ...constants import (
 )
 
 from ...completion import (
-    DOCKER_COMPLETION,
-    REMOTE_COMPLETION,
+    docker_completion,
+    remote_completion,
     filter_completion,
 )
 
@@ -23,7 +23,7 @@ from ...host_configs import (
 
 def get_docker_pythons(name, controller, strict):  # type: (str, bool, bool) -> t.List[str]
     """Return a list of docker instance Python versions supported by the specified host config."""
-    image_config = filter_completion(DOCKER_COMPLETION).get(name)
+    image_config = filter_completion(docker_completion()).get(name)
     available_pythons = CONTROLLER_PYTHON_VERSIONS if controller else SUPPORTED_PYTHON_VERSIONS
 
     if not image_config:
@@ -36,7 +36,7 @@ def get_docker_pythons(name, controller, strict):  # type: (str, bool, bool) ->
 
 def get_remote_pythons(name, controller, strict):  # type: (str, bool, bool) -> t.List[str]
     """Return a list of remote instance Python versions supported by the specified host config."""
-    platform_config = filter_completion(REMOTE_COMPLETION).get(name)
+    platform_config = filter_completion(remote_completion()).get(name)
     available_pythons = CONTROLLER_PYTHON_VERSIONS if controller else SUPPORTED_PYTHON_VERSIONS
 
     if not platform_config:

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

@@ -4,10 +4,10 @@ from __future__ import annotations
 import typing as t
 
 from ...completion import (
-    DOCKER_COMPLETION,
-    NETWORK_COMPLETION,
-    REMOTE_COMPLETION,
-    WINDOWS_COMPLETION,
+    docker_completion,
+    network_completion,
+    remote_completion,
+    windows_completion,
     filter_completion,
 )
 
@@ -108,7 +108,7 @@ class DockerParser(PairParser):
 
     def get_left_parser(self, state):  # type: (ParserState) -> Parser
         """Return the parser for the left side."""
-        return NamespaceWrappedParser('name', ChoicesParser(list(filter_completion(DOCKER_COMPLETION, controller_only=self.controller)),
+        return NamespaceWrappedParser('name', ChoicesParser(list(filter_completion(docker_completion(), controller_only=self.controller)),
                                                             conditions=MatchConditions.CHOICE | MatchConditions.ANY))
 
     def get_right_parser(self, choice):  # type: (t.Any) -> Parser
@@ -128,7 +128,7 @@ class DockerParser(PairParser):
         """Generate and return documentation for this parser."""
         default = 'default'
         content = '\n'.join([f'  {image} ({", ".join(get_docker_pythons(image, self.controller, False))})'
-                             for image, item in filter_completion(DOCKER_COMPLETION, controller_only=self.controller).items()])
+                             for image, item in filter_completion(docker_completion(), controller_only=self.controller).items()])
 
         content += '\n'.join([
             '',
@@ -151,7 +151,7 @@ class PosixRemoteParser(PairParser):
 
     def get_left_parser(self, state):  # type: (ParserState) -> Parser
         """Return the parser for the left side."""
-        return NamespaceWrappedParser('name', PlatformParser(list(filter_completion(REMOTE_COMPLETION, controller_only=self.controller))))
+        return NamespaceWrappedParser('name', PlatformParser(list(filter_completion(remote_completion(), controller_only=self.controller))))
 
     def get_right_parser(self, choice):  # type: (t.Any) -> Parser
         """Return the parser for the right side."""
@@ -170,7 +170,7 @@ class PosixRemoteParser(PairParser):
         """Generate and return documentation for this parser."""
         default = get_fallback_remote_controller()
         content = '\n'.join([f'  {name} ({", ".join(get_remote_pythons(name, self.controller, False))})'
-                             for name, item in filter_completion(REMOTE_COMPLETION, controller_only=self.controller).items()])
+                             for name, item in filter_completion(remote_completion(), controller_only=self.controller).items()])
 
         content += '\n'.join([
             '',
@@ -190,7 +190,7 @@ class WindowsRemoteParser(PairParser):
 
     def get_left_parser(self, state):  # type: (ParserState) -> Parser
         """Return the parser for the left side."""
-        names = list(filter_completion(WINDOWS_COMPLETION))
+        names = list(filter_completion(windows_completion()))
 
         for target in state.root_namespace.targets or []:  # type: WindowsRemoteConfig
             names.remove(target.name)
@@ -203,7 +203,7 @@ class WindowsRemoteParser(PairParser):
 
     def document(self, state):  # type: (DocumentationState) -> t.Optional[str]
         """Generate and return documentation for this parser."""
-        content = '\n'.join([f'  {name}' for name, item in filter_completion(WINDOWS_COMPLETION).items()])
+        content = '\n'.join([f'  {name}' for name, item in filter_completion(windows_completion()).items()])
 
         content += '\n'.join([
             '',
@@ -223,7 +223,7 @@ class NetworkRemoteParser(PairParser):
 
     def get_left_parser(self, state):  # type: (ParserState) -> Parser
         """Return the parser for the left side."""
-        names = list(filter_completion(NETWORK_COMPLETION))
+        names = list(filter_completion(network_completion()))
 
         for target in state.root_namespace.targets or []:  # type: NetworkRemoteConfig
             names.remove(target.name)
@@ -236,7 +236,7 @@ class NetworkRemoteParser(PairParser):
 
     def document(self, state):  # type: (DocumentationState) -> t.Optional[str]
         """Generate and return documentation for this parser."""
-        content = '\n'.join([f'  {name}' for name, item in filter_completion(NETWORK_COMPLETION).items()])
+        content = '\n'.join([f'  {name}' for name, item in filter_completion(network_completion()).items()])
 
         content += '\n'.join([
             '',

test/lib/ansible_test/_internal/completion.py

@@ -13,6 +13,7 @@ from .constants import (
 
 from .util import (
     ANSIBLE_TEST_DATA_ROOT,
+    cache,
     read_lines_without_comments,
 )
 
@@ -220,7 +221,25 @@ def filter_completion(
     return completion
 
 
-DOCKER_COMPLETION = load_completion('docker', DockerCompletionConfig)
-REMOTE_COMPLETION = load_completion('remote', PosixRemoteCompletionConfig)
-WINDOWS_COMPLETION = load_completion('windows', WindowsRemoteCompletionConfig)
-NETWORK_COMPLETION = load_completion('network', NetworkRemoteCompletionConfig)
+@cache
+def docker_completion():  # type: () -> t.Dict[str, DockerCompletionConfig]
+    """Return docker completion entries."""
+    return load_completion('docker', DockerCompletionConfig)
+
+
+@cache
+def remote_completion():  # type: () -> t.Dict[str, PosixRemoteCompletionConfig]
+    """Return remote completion entries."""
+    return load_completion('remote', PosixRemoteCompletionConfig)
+
+
+@cache
+def windows_completion():  # type: () -> t.Dict[str, WindowsRemoteCompletionConfig]
+    """Return windows completion entries."""
+    return load_completion('windows', WindowsRemoteCompletionConfig)
+
+
+@cache
+def network_completion():  # type: () -> t.Dict[str, NetworkRemoteCompletionConfig]
+    """Return network completion entries."""
+    return load_completion('network', NetworkRemoteCompletionConfig)

test/lib/ansible_test/_internal/host_configs.py

@@ -19,17 +19,17 @@ from .io import (
 
 from .completion import (
     CompletionConfig,
-    DOCKER_COMPLETION,
+    docker_completion,
     DockerCompletionConfig,
     InventoryCompletionConfig,
-    NETWORK_COMPLETION,
+    network_completion,
     NetworkRemoteCompletionConfig,
     PosixCompletionConfig,
     PosixRemoteCompletionConfig,
     PosixSshCompletionConfig,
-    REMOTE_COMPLETION,
+    remote_completion,
     RemoteCompletionConfig,
-    WINDOWS_COMPLETION,
+    windows_completion,
     WindowsRemoteCompletionConfig,
     filter_completion,
 )
@@ -277,7 +277,7 @@ class DockerConfig(ControllerHostConfig, PosixConfig):
 
     def get_defaults(self, context):  # type: (HostContext) -> DockerCompletionConfig
         """Return the default settings."""
-        return filter_completion(DOCKER_COMPLETION).get(self.name) or DockerCompletionConfig(
+        return filter_completion(docker_completion()).get(self.name) or DockerCompletionConfig(
             name=self.name,
             image=self.name,
             placeholder=True,
@@ -285,7 +285,7 @@ class DockerConfig(ControllerHostConfig, PosixConfig):
 
     def get_default_targets(self, context):  # type: (HostContext) -> t.List[ControllerConfig]
         """Return the default targets for this host config."""
-        if self.name in filter_completion(DOCKER_COMPLETION):
+        if self.name in filter_completion(docker_completion()):
             defaults = self.get_defaults(context)
             pythons = {version: defaults.get_python_path(version) for version in defaults.supported_pythons}
         else:
@@ -327,14 +327,14 @@ class PosixRemoteConfig(RemoteConfig, ControllerHostConfig, PosixConfig):
 
     def get_defaults(self, context):  # type: (HostContext) -> PosixRemoteCompletionConfig
         """Return the default settings."""
-        return filter_completion(REMOTE_COMPLETION).get(self.name) or REMOTE_COMPLETION.get(self.platform) or PosixRemoteCompletionConfig(
+        return filter_completion(remote_completion()).get(self.name) or remote_completion().get(self.platform) or PosixRemoteCompletionConfig(
             name=self.name,
             placeholder=True,
         )
 
     def get_default_targets(self, context):  # type: (HostContext) -> t.List[ControllerConfig]
         """Return the default targets for this host config."""
-        if self.name in filter_completion(REMOTE_COMPLETION):
+        if self.name in filter_completion(remote_completion()):
             defaults = self.get_defaults(context)
             pythons = {version: defaults.get_python_path(version) for version in defaults.supported_pythons}
         else:
@@ -358,7 +358,7 @@ class WindowsRemoteConfig(RemoteConfig, WindowsConfig):
     """Configuration for a remoe Windows host."""
     def get_defaults(self, context):  # type: (HostContext) -> WindowsRemoteCompletionConfig
         """Return the default settings."""
-        return filter_completion(WINDOWS_COMPLETION).get(self.name) or WindowsRemoteCompletionConfig(
+        return filter_completion(windows_completion()).get(self.name) or WindowsRemoteCompletionConfig(
             name=self.name,
         )
 
@@ -381,7 +381,7 @@ class NetworkRemoteConfig(RemoteConfig, NetworkConfig):
 
     def get_defaults(self, context):  # type: (HostContext) -> NetworkRemoteCompletionConfig
         """Return the default settings."""
-        return filter_completion(NETWORK_COMPLETION).get(self.name) or NetworkRemoteCompletionConfig(
+        return filter_completion(network_completion()).get(self.name) or NetworkRemoteCompletionConfig(
             name=self.name,
         )