1 files changed, 197 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/commands/env/__init__.py b/test/lib/ansible_test/_internal/commands/env/__init__.py
new file mode 100644
index 0000000..44f229f
--- /dev/null
+++ b/test/lib/ansible_test/_internal/commands/env/__init__.py
@@ -0,0 +1,197 @@
+"""Show information about the test environment."""
+from __future__ import annotations
+
+import datetime
+import os
+import platform
+import sys
+import typing as t
+
+from ...config import (
+    CommonConfig,
+)
+
+from ...io import (
+    write_json_file,
+)
+
+from ...util import (
+    display,
+    get_ansible_version,
+    get_available_python_versions,
+    ApplicationError,
+)
+
+from ...util_common import (
+    data_context,
+    write_json_test_results,
+    ResultType,
+)
+
+from ...docker_util import (
+    get_docker_command,
+    get_docker_info,
+    get_docker_container_id,
+)
+
+from ...constants import (
+    TIMEOUT_PATH,
+)
+
+from ...ci import (
+    get_ci_provider,
+)
+
+
+class EnvConfig(CommonConfig):
+    """Configuration for the `env` command."""
+    def __init__(self, args: t.Any) -> None:
+        super().__init__(args, 'env')
+
+        self.show = args.show
+        self.dump = args.dump
+        self.timeout = args.timeout
+        self.list_files = args.list_files
+
+        if not self.show and not self.dump and self.timeout is None and not self.list_files:
+            # default to --show if no options were given
+            self.show = True
+
+
+def command_env(args: EnvConfig) -> None:
+    """Entry point for the `env` command."""
+    show_dump_env(args)
+    list_files_env(args)
+    set_timeout(args)
+
+
+def show_dump_env(args: EnvConfig) -> None:
+    """Show information about the current environment and/or write the information to disk."""
+    if not args.show and not args.dump:
+        return
+
+    container_id = get_docker_container_id()
+
+    data = dict(
+        ansible=dict(
+            version=get_ansible_version(),
+        ),
+        docker=get_docker_details(args),
+        container_id=container_id,
+        environ=os.environ.copy(),
+        location=dict(
+            pwd=os.environ.get('PWD', None),
+            cwd=os.getcwd(),
+        ),
+        git=get_ci_provider().get_git_details(args),
+        platform=dict(
+            datetime=datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'),
+            platform=platform.platform(),
+            uname=platform.uname(),
+        ),
+        python=dict(
+            executable=sys.executable,
+            version=platform.python_version(),
+        ),
+        interpreters=get_available_python_versions(),
+    )
+
+    if args.show:
+        verbose = {
+            'docker': 3,
+            'docker.executable': 0,
+            'environ': 2,
+            'platform.uname': 1,
+        }
+
+        show_dict(data, verbose)
+
+    if args.dump and not args.explain:
+        write_json_test_results(ResultType.BOT, 'data-environment.json', data)
+
+
+def list_files_env(args: EnvConfig) -> None:
+    """List files on stdout."""
+    if not args.list_files:
+        return
+
+    for path in data_context().content.all_files():
+        display.info(path)
+
+
+def set_timeout(args: EnvConfig) -> None:
+    """Set an execution timeout for subsequent ansible-test invocations."""
+    if args.timeout is None:
+        return
+
+    if args.timeout:
+        deadline = (datetime.datetime.utcnow() + datetime.timedelta(minutes=args.timeout)).strftime('%Y-%m-%dT%H:%M:%SZ')
+
+        display.info('Setting a %d minute test timeout which will end at: %s' % (args.timeout, deadline), verbosity=1)
+    else:
+        deadline = None
+
+        display.info('Clearing existing test timeout.', verbosity=1)
+
+    if args.explain:
+        return
+
+    if deadline:
+        data = dict(
+            duration=args.timeout,
+            deadline=deadline,
+        )
+
+        write_json_file(TIMEOUT_PATH, data)
+    elif os.path.exists(TIMEOUT_PATH):
+        os.remove(TIMEOUT_PATH)
+
+
+def show_dict(data: dict[str, t.Any], verbose: dict[str, int], root_verbosity: int = 0, path: t.Optional[list[str]] = None) -> None:
+    """Show a dict with varying levels of verbosity."""
+    path = path if path else []
+
+    for key, value in sorted(data.items()):
+        indent = '  ' * len(path)
+        key_path = path + [key]
+        key_name = '.'.join(key_path)
+        verbosity = verbose.get(key_name, root_verbosity)
+
+        if isinstance(value, (tuple, list)):
+            display.info(indent + '%s:' % key, verbosity=verbosity)
+            for item in value:
+                display.info(indent + '  - %s' % item, verbosity=verbosity)
+        elif isinstance(value, dict):
+            min_verbosity = min([verbosity] + [v for k, v in verbose.items() if k.startswith('%s.' % key)])
+            display.info(indent + '%s:' % key, verbosity=min_verbosity)
+            show_dict(value, verbose, verbosity, key_path)
+        else:
+            display.info(indent + '%s: %s' % (key, value), verbosity=verbosity)
+
+
+def get_docker_details(args: EnvConfig) -> dict[str, t.Any]:
+    """Return details about docker."""
+    docker = get_docker_command()
+
+    executable = None
+    info = None
+    version = None
+
+    if docker:
+        executable = docker.executable
+
+        try:
+            docker_info = get_docker_info(args)
+        except ApplicationError as ex:
+            display.warning(str(ex))
+        else:
+            info = docker_info.info
+            version = docker_info.version
+
+    docker_details = dict(
+        executable=executable,
+        info=info,
+        version=version,
+    )
+
+    return docker_details