archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

ansible-test - Clean up indentation and spaces (#79980)

Browse Source
pull/79985/head
Matt Clay 2 years ago committed by GitHub
parent 71f2e777ed
commit 58d84933fc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
57 changed files with 534 additions and 534 deletions
Show Stats Download Patch File Download Diff File

10
test/lib/ansible_test/_internal/ansible_util.py
Unescape Escape View File

@ -284,11 +284,11 @@ def get_collection_detail(python: PythonConfig) -> CollectionDetail:
def run_playbook(
args: EnvironmentConfig,
inventory_path: str,
playbook: str,
capture: bool,
variables: t.Optional[dict[str, t.Any]] = None,
args: EnvironmentConfig,
inventory_path: str,
playbook: str,
capture: bool,
variables: t.Optional[dict[str, t.Any]] = None,
) -> None:
"""Run the specified playbook using the given inventory file and playbook variables."""
playbook_path = os.path.join(ANSIBLE_TEST_DATA_ROOT, 'playbooks', playbook)

54
test/lib/ansible_test/_internal/classification/__init__.py
Unescape Escape View File

@ -379,9 +379,9 @@ class PathMapper:
if is_subdir(path, data_context().content.integration_path):
if dirname == data_context().content.integration_path:
for command in (
'integration',
'windows-integration',
'network-integration',
'integration',
'windows-integration',
'network-integration',
):
if name == command and ext == '.cfg':
return {
@ -641,19 +641,19 @@ class PathMapper:
if '/' not in path:
if path in (
'.gitignore',
'COPYING',
'LICENSE',
'Makefile',
'.gitignore',
'COPYING',
'LICENSE',
'Makefile',
):
return minimal
if ext in (
'.in',
'.md',
'.rst',
'.toml',
'.txt',
'.in',
'.md',
'.rst',
'.toml',
'.txt',
):
return minimal
@ -757,17 +757,17 @@ class PathMapper:
if path.startswith('test/lib/ansible_test/_data/requirements/'):
if name in (
'integration',
'network-integration',
'windows-integration',
'integration',
'network-integration',
'windows-integration',
):
return {
name: self.integration_all_target,
}
if name in (
'sanity',
'units',
'sanity',
'units',
):
return {
name: 'all',
@ -826,11 +826,11 @@ class PathMapper:
if '/' not in path:
if path in (
'.gitattributes',
'.gitignore',
'.mailmap',
'COPYING',
'Makefile',
'.gitattributes',
'.gitignore',
'.mailmap',
'COPYING',
'Makefile',
):
return minimal
@ -840,11 +840,11 @@ class PathMapper:
return all_tests(self.args) # broad impact, run all tests
if ext in (
'.in',
'.md',
'.rst',
'.toml',
'.txt',
'.in',
'.md',
'.rst',
'.toml',
'.txt',
):
return minimal

46
test/lib/ansible_test/_internal/cli/argparsing/__init__.py
Unescape Escape View File

@ -40,11 +40,11 @@ class RegisteredCompletionFinder(OptionCompletionFinder):
self.registered_completions: t.Optional[list[str]] = None
def completer(
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
**kwargs,
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
**kwargs,
) -> list[str]:
"""
Return a list of completions for the specified prefix and action.
@ -63,10 +63,10 @@ class RegisteredCompletionFinder(OptionCompletionFinder):
@abc.abstractmethod
def get_completions(
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
) -> list[str]:
"""
Return a list of completions for the specified prefix and action.
@ -89,9 +89,9 @@ class CompositeAction(argparse.Action, metaclass=abc.ABCMeta):
documentation_state: dict[t.Type[CompositeAction], DocumentationState] = {}
def __init__(
self,
*args,
**kwargs,
self,
*args,
**kwargs,
):
self.definition = self.create_parser()
self.documentation_state[type(self)] = documentation_state = DocumentationState()
@ -108,11 +108,11 @@ class CompositeAction(argparse.Action, metaclass=abc.ABCMeta):
"""Return a namespace parser to parse the argument associated with this action."""
def __call__(
self,
parser,
namespace,
values,
option_string=None,
self,
parser,
namespace,
values,
option_string=None,
):
state = ParserState(mode=ParserMode.PARSE, namespaces=[namespace], remainder=values)
@ -135,10 +135,10 @@ class CompositeAction(argparse.Action, metaclass=abc.ABCMeta):
class CompositeActionCompletionFinder(RegisteredCompletionFinder):
"""Completion finder with support for composite argument parsing."""
def get_completions(
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
self,
prefix: str,
action: argparse.Action,
parsed_args: argparse.Namespace,
) -> list[str]:
"""Return a list of completions appropriate for the given prefix and action, taking into account the arguments that have already been parsed."""
assert isinstance(action, CompositeAction)
@ -232,8 +232,8 @@ def detect_false_file_completion(value: str, mode: ParserMode) -> bool:
def complete(
completer: Parser,
state: ParserState,
completer: Parser,
state: ParserState,
) -> Completion:
"""Perform argument completion using the given completer and return the completion result."""
value = state.remainder

4
test/lib/ansible_test/_internal/cli/commands/__init__.py
Unescape Escape View File

@ -44,8 +44,8 @@ from .units import (
def do_commands(
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for all commands."""
common = argparse.ArgumentParser(add_help=False)

8
test/lib/ansible_test/_internal/cli/commands/coverage/__init__.py
Unescape Escape View File

@ -37,9 +37,9 @@ from .xml import (
def do_coverage(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for all `coverage` commands."""
coverage_common = argparse.ArgumentParser(add_help=False, parents=[parent])
@ -61,7 +61,7 @@ def do_coverage(
def add_coverage_common(
parser: argparse.ArgumentParser,
parser: argparse.ArgumentParser,
):
"""Add common coverage arguments."""
parser.add_argument(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/__init__.py
Unescape Escape View File

@ -13,9 +13,9 @@ from ....environments import (
def do_analyze(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for all `coverage analyze` commands."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/__init__.py
Unescape Escape View File

@ -29,9 +29,9 @@ from .missing import (
def do_targets(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for all `coverage analyze targets` commands."""
targets = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/combine.py
Unescape Escape View File

@ -17,9 +17,9 @@ from .....environments import (
def do_combine(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `coverage analyze targets combine` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/expand.py
Unescape Escape View File

@ -17,9 +17,9 @@ from .....environments import (
def do_expand(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `coverage analyze targets expand` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/filter.py
Unescape Escape View File

@ -17,9 +17,9 @@ from .....environments import (
def do_filter(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `coverage analyze targets filter` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/generate.py
Unescape Escape View File

@ -17,9 +17,9 @@ from .....environments import (
def do_generate(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `coverage analyze targets generate` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/analyze/targets/missing.py
Unescape Escape View File

@ -17,9 +17,9 @@ from .....environments import (
def do_missing(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `coverage analyze targets missing` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/coverage/combine.py
Unescape Escape View File

@ -19,10 +19,10 @@ from ...environments import (
def do_combine(
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for the `coverage combine` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/coverage/erase.py
Unescape Escape View File

@ -17,9 +17,9 @@ from ...environments import (
def do_erase(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for the `coverage erase` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/coverage/html.py
Unescape Escape View File

@ -19,10 +19,10 @@ from ...environments import (
def do_html(
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for the `coverage html` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/coverage/report.py
Unescape Escape View File

@ -19,10 +19,10 @@ from ...environments import (
def do_report(
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for the `coverage report` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/coverage/xml.py
Unescape Escape View File

@ -19,10 +19,10 @@ from ...environments import (
def do_xml(
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_coverage_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
) -> None:
"""Command line parsing for the `coverage xml` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/env.py
Unescape Escape View File

@ -17,9 +17,9 @@ from ..environments import (
def do_env(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `env` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/integration/__init__.py
Unescape Escape View File

@ -26,9 +26,9 @@ from .windows import (
def do_integration(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for all integration commands."""
parser = argparse.ArgumentParser(
@ -42,7 +42,7 @@ def do_integration(
def add_integration_common(
parser: argparse.ArgumentParser,
parser: argparse.ArgumentParser,
):
"""Add common integration arguments."""
register_completer(parser.add_argument(

8
test/lib/ansible_test/_internal/cli/commands/integration/network.py
Unescape Escape View File

@ -35,10 +35,10 @@ from ...completers import (
def do_network_integration(
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `network-integration` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/integration/posix.py
Unescape Escape View File

@ -26,10 +26,10 @@ from ...environments import (
def do_posix_integration(
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `integration` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

8
test/lib/ansible_test/_internal/cli/commands/integration/windows.py
Unescape Escape View File

@ -26,10 +26,10 @@ from ...environments import (
def do_windows_integration(
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
add_integration_common: c.Callable[[argparse.ArgumentParser], None],
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `windows-integration` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/sanity.py
Unescape Escape View File

@ -29,9 +29,9 @@ from ..environments import (
def do_sanity(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `sanity` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/shell.py
Unescape Escape View File

@ -20,9 +20,9 @@ from ..environments import (
def do_shell(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `shell` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

6
test/lib/ansible_test/_internal/cli/commands/units.py
Unescape Escape View File

@ -24,9 +24,9 @@ from ..environments import (
def do_units(
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
subparsers,
parent: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
):
"""Command line parsing for the `units` command."""
parser: argparse.ArgumentParser = subparsers.add_parser(

16
test/lib/ansible_test/_internal/cli/compat.py
Unescape Escape View File

@ -191,9 +191,9 @@ class TargetMode(enum.Enum):
def convert_legacy_args(
argv: list[str],
args: t.Union[argparse.Namespace, types.SimpleNamespace],
mode: TargetMode,
argv: list[str],
args: t.Union[argparse.Namespace, types.SimpleNamespace],
mode: TargetMode,
) -> HostSettings:
"""Convert pre-split host arguments in the given namespace to their split counterparts."""
old_options = LegacyHostOptions.create(args)
@ -262,9 +262,9 @@ def convert_legacy_args(
def controller_targets(
mode: TargetMode,
options: LegacyHostOptions,
controller: ControllerHostConfig,
mode: TargetMode,
options: LegacyHostOptions,
controller: ControllerHostConfig,
) -> list[HostConfig]:
"""Return the configuration for controller targets."""
python = native_python(options)
@ -288,8 +288,8 @@ def native_python(options: LegacyHostOptions) -> t.Optional[NativePythonConfig]:
def get_legacy_host_config(
mode: TargetMode,
options: LegacyHostOptions,
mode: TargetMode,
options: LegacyHostOptions,
) -> tuple[ControllerHostConfig, list[HostConfig], t.Optional[FallbackDetail]]:
"""
Returns controller and target host configs derived from the provided legacy host options.

64
test/lib/ansible_test/_internal/cli/environments.py
Unescape Escape View File

@ -81,10 +81,10 @@ class ControllerMode(enum.Enum):
def add_environments(
parser: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
controller_mode: ControllerMode,
target_mode: TargetMode,
parser: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
controller_mode: ControllerMode,
target_mode: TargetMode,
) -> None:
"""Add arguments for the environments used to run ansible-test and commands it invokes."""
no_environment = controller_mode == ControllerMode.NO_DELEGATION and target_mode == TargetMode.NO_TARGETS
@ -114,8 +114,8 @@ def add_environments(
def add_global_options(
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
):
"""Add global options for controlling the test environment that work with both the legacy and composite options."""
global_parser = t.cast(argparse.ArgumentParser, parser.add_argument_group(title='global environment arguments'))
@ -156,10 +156,10 @@ def add_global_options(
def add_composite_environment_options(
parser: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
controller_mode: ControllerMode,
target_mode: TargetMode,
parser: argparse.ArgumentParser,
completer: CompositeActionCompletionFinder,
controller_mode: ControllerMode,
target_mode: TargetMode,
) -> list[t.Type[CompositeAction]]:
"""Add composite options for controlling the test environment."""
composite_parser = t.cast(argparse.ArgumentParser, parser.add_argument_group(
@ -246,9 +246,9 @@ def add_composite_environment_options(
def add_legacy_environment_options(
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
target_mode: TargetMode,
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
target_mode: TargetMode,
):
"""Add legacy options for controlling the test environment."""
environment: argparse.ArgumentParser = parser.add_argument_group( # type: ignore[assignment] # real type private
@ -259,8 +259,8 @@ def add_legacy_environment_options(
def add_environments_python(
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
) -> None:
"""Add environment arguments to control the Python version(s) used."""
python_versions: tuple[str, ...]
@ -285,9 +285,9 @@ def add_environments_python(
def add_environments_host(
environments_parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
target_mode: TargetMode,
environments_parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
target_mode: TargetMode,
) -> None:
"""Add environment arguments for the given host and argument modes."""
environments_exclusive_group: argparse.ArgumentParser = environments_parser.add_mutually_exclusive_group() # type: ignore[assignment] # real type private
@ -341,7 +341,7 @@ def add_environment_network(
def add_environment_windows(
environments_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
) -> None:
"""Add environment arguments for running on a windows host."""
register_completer(environments_parser.add_argument(
@ -359,7 +359,7 @@ def add_environment_windows(
def add_environment_local(
exclusive_parser: argparse.ArgumentParser,
exclusive_parser: argparse.ArgumentParser,
) -> None:
"""Add environment arguments for running on the local (origin) host."""
exclusive_parser.add_argument(
@ -370,8 +370,8 @@ def add_environment_local(
def add_environment_venv(
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
) -> None:
"""Add environment arguments for running in ansible-test managed virtual environments."""
exclusive_parser.add_argument(
@ -387,8 +387,8 @@ def add_environment_venv(
def add_global_docker(
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
) -> None:
"""Add global options for Docker."""
if controller_mode != ControllerMode.DELEGATED:
@ -450,9 +450,9 @@ def add_global_docker(
def add_environment_docker(
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
) -> None:
"""Add environment arguments for running in docker containers."""
if target_mode in (TargetMode.POSIX_INTEGRATION, TargetMode.SHELL):
@ -490,8 +490,8 @@ def add_environment_docker(
def add_global_remote(
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
parser: argparse.ArgumentParser,
controller_mode: ControllerMode,
) -> None:
"""Add global options for remote instances."""
if controller_mode != ControllerMode.DELEGATED:
@ -529,9 +529,9 @@ def add_global_remote(
def add_environment_remote(
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
exclusive_parser: argparse.ArgumentParser,
environments_parser: argparse.ArgumentParser,
target_mode: TargetMode,
) -> None:
"""Add environment arguments for running in ansible-core-ci provisioned remote virtual machines."""
if target_mode == TargetMode.POSIX_INTEGRATION:

26
test/lib/ansible_test/_internal/commands/coverage/__init__.py
Unescape Escape View File

@ -36,7 +36,7 @@ from ...python_requirements import (
install_requirements,
)
from ... target import (
from ...target import (
walk_module_targets,
)
@ -158,11 +158,11 @@ def get_python_modules() -> dict[str, str]:
def enumerate_python_arcs(
path: str,
coverage: coverage_module,
modules: dict[str, str],
collection_search_re: t.Optional[t.Pattern],
collection_sub_re: t.Optional[t.Pattern],
path: str,
coverage: coverage_module,
modules: dict[str, str],
collection_search_re: t.Optional[t.Pattern],
collection_sub_re: t.Optional[t.Pattern],
) -> c.Generator[tuple[str, set[tuple[int, int]]], None, None]:
"""Enumerate Python code coverage arcs in the given file."""
if os.path.getsize(path) == 0:
@ -235,9 +235,9 @@ def read_python_coverage_legacy(path: str) -> PythonArcs:
def enumerate_powershell_lines(
path: str,
collection_search_re: t.Optional[t.Pattern],
collection_sub_re: t.Optional[t.Pattern],
path: str,
collection_search_re: t.Optional[t.Pattern],
collection_sub_re: t.Optional[t.Pattern],
) -> c.Generator[tuple[str, dict[int, int]], None, None]:
"""Enumerate PowerShell code coverage lines in the given file."""
if os.path.getsize(path) == 0:
@ -274,10 +274,10 @@ def enumerate_powershell_lines(
def sanitize_filename(
filename: str,
modules: t.Optional[dict[str, str]] = None,
collection_search_re: t.Optional[t.Pattern] = None,
collection_sub_re: t.Optional[t.Pattern] = None,
filename: str,
modules: t.Optional[dict[str, str]] = None,
collection_search_re: t.Optional[t.Pattern] = None,
collection_sub_re: t.Optional[t.Pattern] = None,
) -> t.Optional[str]:
"""Convert the given code coverage path to a local absolute path and return its, or None if the path is not valid."""
ansible_path = os.path.abspath('lib/ansible/') + '/'

6
test/lib/ansible_test/_internal/commands/coverage/analyze/targets/__init__.py
Unescape Escape View File

@ -119,9 +119,9 @@ def get_target_index(name: str, target_indexes: TargetIndexes) -> int:
def expand_indexes(
source_data: IndexedPoints,
source_index: list[str],
format_func: c.Callable[[TargetKey], TFlexKey],
source_data: IndexedPoints,
source_index: list[str],
format_func: c.Callable[[TargetKey], TFlexKey],
) -> dict[str, dict[TFlexKey, set[str]]]:
"""Expand indexes from the source into target names for easier processing of the data (arcs or lines)."""
combined_data: dict[str, dict[TFlexKey, set[str]]] = {}

8
test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py
Unescape Escape View File

@ -58,10 +58,10 @@ def command_coverage_analyze_targets_combine(args: CoverageAnalyzeTargetsCombine
def merge_indexes(
source_data: IndexedPoints,
source_index: list[str],
combined_data: IndexedPoints,
combined_index: TargetIndexes,
source_data: IndexedPoints,
source_index: list[str],
combined_data: IndexedPoints,
combined_index: TargetIndexes,
) -> None:
"""Merge indexes from the source into the combined data set (arcs or lines)."""
for covered_path, covered_points in source_data.items():

6
test/lib/ansible_test/_internal/commands/coverage/analyze/targets/filter.py
Unescape Escape View File

@ -97,9 +97,9 @@ def command_coverage_analyze_targets_filter(args: CoverageAnalyzeTargetsFilterCo
def filter_data(
data: NamedPoints,
path_filter_func: c.Callable[[str], bool],
target_filter_func: c.Callable[[set[str]], set[str]],
data: NamedPoints,
path_filter_func: c.Callable[[str], bool],
target_filter_func: c.Callable[[set[str]], set[str]],
) -> NamedPoints:
"""Filter the data set using the specified filter function."""
result: NamedPoints = {}

20
test/lib/ansible_test/_internal/commands/coverage/analyze/targets/generate.py
Unescape Escape View File

@ -75,10 +75,10 @@ def command_coverage_analyze_targets_generate(args: CoverageAnalyzeTargetsGenera
def analyze_python_coverage(
args: CoverageAnalyzeTargetsGenerateConfig,
host_state: HostState,
path: str,
target_indexes: TargetIndexes,
args: CoverageAnalyzeTargetsGenerateConfig,
host_state: HostState,
path: str,
target_indexes: TargetIndexes,
) -> Arcs:
"""Analyze Python code coverage."""
results: Arcs = {}
@ -107,9 +107,9 @@ def analyze_python_coverage(
def analyze_powershell_coverage(
args: CoverageAnalyzeTargetsGenerateConfig,
path: str,
target_indexes: TargetIndexes,
args: CoverageAnalyzeTargetsGenerateConfig,
path: str,
target_indexes: TargetIndexes,
) -> Lines:
"""Analyze PowerShell code coverage"""
results: Lines = {}
@ -136,9 +136,9 @@ def analyze_powershell_coverage(
def prune_invalid_filenames(
args: CoverageAnalyzeTargetsGenerateConfig,
results: dict[str, t.Any],
collection_search_re: t.Optional[t.Pattern] = None,
args: CoverageAnalyzeTargetsGenerateConfig,
results: dict[str, t.Any],
collection_search_re: t.Optional[t.Pattern] = None,
) -> None:
"""Remove invalid filenames from the given result set."""
path_checker = PathChecker(args, collection_search_re)

22
test/lib/ansible_test/_internal/commands/coverage/analyze/targets/missing.py
Unescape Escape View File

@ -66,11 +66,11 @@ def command_coverage_analyze_targets_missing(args: CoverageAnalyzeTargetsMissing
def find_gaps(
from_data: IndexedPoints,
from_index: list[str],
to_data: IndexedPoints,
target_indexes: TargetIndexes,
only_exists: bool,
from_data: IndexedPoints,
from_index: list[str],
to_data: IndexedPoints,
target_indexes: TargetIndexes,
only_exists: bool,
) -> IndexedPoints:
"""Find gaps in coverage between the from and to data sets."""
target_data: IndexedPoints = {}
@ -91,12 +91,12 @@ def find_gaps(
def find_missing(
from_data: IndexedPoints,
from_index: list[str],
to_data: IndexedPoints,
to_index: list[str],
target_indexes: TargetIndexes,
only_exists: bool,
from_data: IndexedPoints,
from_index: list[str],
to_data: IndexedPoints,
to_index: list[str],
target_indexes: TargetIndexes,
only_exists: bool,
) -> IndexedPoints:
"""Find coverage in from_data not present in to_data (arcs or lines)."""
target_data: IndexedPoints = {}

6
test/lib/ansible_test/_internal/commands/coverage/combine.py
Unescape Escape View File

@ -285,9 +285,9 @@ def _get_coverage_targets(args: CoverageCombineConfig, walk_func: c.Callable) ->
def _build_stub_groups(
args: CoverageCombineConfig,
sources: list[tuple[str, int]],
default_stub_value: c.Callable[[list[str]], dict[str, TValue]],
args: CoverageCombineConfig,
sources: list[tuple[str, int]],
default_stub_value: c.Callable[[list[str]], dict[str, TValue]],
) -> dict[str, dict[str, TValue]]:
"""
Split the given list of sources with line counts into groups, maintaining a maximum line count for each group.

92
test/lib/ansible_test/_internal/commands/integration/__init__.py
Unescape Escape View File

@ -243,9 +243,9 @@ def delegate_inventory(args: IntegrationConfig, inventory_path_src: str) -> None
@contextlib.contextmanager
def integration_test_environment(
args: IntegrationConfig,
target: IntegrationTarget,
inventory_path_src: str,
args: IntegrationConfig,
target: IntegrationTarget,
inventory_path_src: str,
) -> c.Iterator[IntegrationEnvironment]:
"""Context manager that prepares the integration test environment and cleans it up."""
ansible_config_src = args.get_ansible_config()
@ -346,9 +346,9 @@ def integration_test_environment(
@contextlib.contextmanager
def integration_test_config_file(
args: IntegrationConfig,
env_config: CloudEnvironmentConfig,
integration_dir: str,
args: IntegrationConfig,
env_config: CloudEnvironmentConfig,
integration_dir: str,
) -> c.Iterator[t.Optional[str]]:
"""Context manager that provides a config file for integration tests, if needed."""
if not env_config:
@ -375,10 +375,10 @@ def integration_test_config_file(
def create_inventory(
args: IntegrationConfig,
host_state: HostState,
inventory_path: str,
target: IntegrationTarget,
args: IntegrationConfig,
host_state: HostState,
inventory_path: str,
target: IntegrationTarget,
) -> None:
"""Create inventory."""
if isinstance(args, PosixIntegrationConfig):
@ -401,13 +401,13 @@ def create_inventory(
def command_integration_filtered(
args: IntegrationConfig,
host_state: HostState,
targets: tuple[IntegrationTarget, ...],
all_targets: tuple[IntegrationTarget, ...],
inventory_path: str,
pre_target: t.Optional[c.Callable[[IntegrationTarget], None]] = None,
post_target: t.Optional[c.Callable[[IntegrationTarget], None]] = None,
args: IntegrationConfig,
host_state: HostState,
targets: tuple[IntegrationTarget, ...],
all_targets: tuple[IntegrationTarget, ...],
inventory_path: str,
pre_target: t.Optional[c.Callable[[IntegrationTarget], None]] = None,
post_target: t.Optional[c.Callable[[IntegrationTarget], None]] = None,
):
"""Run integration tests for the specified targets."""
found = False
@ -580,12 +580,12 @@ def command_integration_filtered(
def command_integration_script(
args: IntegrationConfig,
host_state: HostState,
target: IntegrationTarget,
test_dir: str,
inventory_path: str,
coverage_manager: CoverageManager,
args: IntegrationConfig,
host_state: HostState,
target: IntegrationTarget,
test_dir: str,
inventory_path: str,
coverage_manager: CoverageManager,
):
"""Run an integration test script."""
display.info('Running %s integration test script' % target.name)
@ -632,13 +632,13 @@ def command_integration_script(
def command_integration_role(
args: IntegrationConfig,
host_state: HostState,
target: IntegrationTarget,
start_at_task: t.Optional[str],
test_dir: str,
inventory_path: str,
coverage_manager: CoverageManager,
args: IntegrationConfig,
host_state: HostState,
target: IntegrationTarget,
start_at_task: t.Optional[str],
test_dir: str,
inventory_path: str,
coverage_manager: CoverageManager,
):
"""Run an integration test role."""
display.info('Running %s integration test role' % target.name)
@ -751,15 +751,15 @@ def command_integration_role(
def run_setup_targets(
args: IntegrationConfig,
host_state: HostState,
test_dir: str,
target_names: c.Sequence[str],
targets_dict: dict[str, IntegrationTarget],
targets_executed: set[str],
inventory_path: str,
coverage_manager: CoverageManager,
always: bool,
args: IntegrationConfig,
host_state: HostState,
test_dir: str,
target_names: c.Sequence[str],
targets_dict: dict[str, IntegrationTarget],
targets_executed: set[str],
inventory_path: str,
coverage_manager: CoverageManager,
always: bool,
):
"""Run setup targets."""
for target_name in target_names:
@ -782,13 +782,13 @@ def run_setup_targets(
def integration_environment(
args: IntegrationConfig,
target: IntegrationTarget,
test_dir: str,
inventory_path: str,
ansible_config: t.Optional[str],
env_config: t.Optional[CloudEnvironmentConfig],
test_env: IntegrationEnvironment,
args: IntegrationConfig,
target: IntegrationTarget,
test_dir: str,
inventory_path: str,
ansible_config: t.Optional[str],
env_config: t.Optional[CloudEnvironmentConfig],
test_env: IntegrationEnvironment,
) -> dict[str, str]:
"""Return a dictionary of environment variables to use when running the given integration test target."""
env = ansible_environment(args, ansible_config=ansible_config)

2
test/lib/ansible_test/_internal/commands/integration/cloud/cs.py
Unescape Escape View File

@ -136,7 +136,7 @@ class CsCloudProvider(CloudProvider):
# noinspection PyBroadException
try:
json.loads(value)
except Exception: # pylint: disable=broad-except
except Exception: # pylint: disable=broad-except
return False # sometimes the file exists but is not yet valid JSON
return True

12
test/lib/ansible_test/_internal/commands/integration/filters.py
Unescape Escape View File

@ -67,12 +67,12 @@ class TargetFilter(t.Generic[THostConfig], metaclass=abc.ABCMeta):
return self.configs[0]
def skip(
self,
skip: str,
reason: str,
targets: list[IntegrationTarget],
exclude: set[str],
override: t.Optional[list[str]] = None,
self,
skip: str,
reason: str,
targets: list[IntegrationTarget],
exclude: set[str],
override: t.Optional[list[str]] = None,
) -> None:
"""Apply the specified skip rule to the given targets by updating the provided exclude list."""
if skip.startswith('skip/'):

20
test/lib/ansible_test/_internal/commands/sanity/__init__.py
Unescape Escape View File

@ -635,11 +635,11 @@ class SanitySkipped(TestSkipped):
class SanityFailure(TestFailure):
"""Sanity test failure."""
def __init__(
self,
test: str,
python_version: t.Optional[str] = None,
messages: t.Optional[c.Sequence[SanityMessage]] = None,
summary: t.Optional[str] = None,
self,
test: str,
python_version: t.Optional[str] = None,
messages: t.Optional[c.Sequence[SanityMessage]] = None,
summary: t.Optional[str] = None,
) -> None:
super().__init__(COMMAND, test, python_version, messages, summary)
@ -1101,11 +1101,11 @@ def sanity_get_tests() -> tuple[SanityTest, ...]:
def create_sanity_virtualenv(
args: SanityConfig,
python: PythonConfig,
name: str,
coverage: bool = False,
minimize: bool = False,
args: SanityConfig,
python: PythonConfig,
name: str,
coverage: bool = False,
minimize: bool = False,
) -> t.Optional[VirtualPythonConfig]:
"""Return an existing sanity virtual environment matching the requested parameters or create a new one."""
commands = collect_requirements( # create_sanity_virtualenv()

4
test/lib/ansible_test/_internal/commands/sanity/import.py
Unescape Escape View File

@ -122,8 +122,8 @@ class ImportTest(SanityMultipleVersion):
messages = []
for import_type, test in (
('module', _get_module_test(True)),
('plugin', _get_module_test(False)),
('module', _get_module_test(True)),
('plugin', _get_module_test(False)),
):
if import_type == 'plugin' and python.version in REMOTE_ONLY_PYTHON_VERSIONS:
continue

8
test/lib/ansible_test/_internal/commands/sanity/integration_aliases.py
Unescape Escape View File

@ -322,10 +322,10 @@ class IntegrationAliasesTest(SanitySingleVersion):
return messages
def check_ci_group(
self,
targets: tuple[CompletionTarget, ...],
find: str,
find_incidental: t.Optional[list[str]] = None,
self,
targets: tuple[CompletionTarget, ...],
find: str,
find_incidental: t.Optional[list[str]] = None,
) -> list[SanityMessage]:
"""Check the CI groups set in the provided targets and return a list of messages with any issues found."""
all_paths = set(target.path for target in targets)

10
test/lib/ansible_test/_internal/commands/sanity/mypy.py
Unescape Escape View File

@ -176,11 +176,11 @@ class MypyTest(SanityMultipleVersion):
@staticmethod
def test_context(
args: SanityConfig,
virtualenv_python: VirtualPythonConfig,
python: PythonConfig,
context: MyPyContext,
paths: list[str],
args: SanityConfig,
virtualenv_python: VirtualPythonConfig,
python: PythonConfig,
context: MyPyContext,
paths: list[str],
) -> list[SanityMessage]:
"""Run mypy tests for the specified context."""
context_paths = [path for path in paths if any(is_subdir(path, match_path) for match_path in context.paths)]

14
test/lib/ansible_test/_internal/commands/sanity/pylint.py
Unescape Escape View File

@ -189,13 +189,13 @@ class PylintTest(SanitySingleVersion):
@staticmethod
def pylint(
args: SanityConfig,
context: str,
paths: list[str],
plugin_dir: str,
plugin_names: list[str],
python: PythonConfig,
collection_detail: CollectionDetail,
args: SanityConfig,
context: str,
paths: list[str],
plugin_dir: str,
plugin_names: list[str],
python: PythonConfig,
collection_detail: CollectionDetail,
) -> list[dict[str, str]]:
"""Run pylint using the config specified by the context on the specified paths."""
rcfile = os.path.join(SANITY_ROOT, 'pylint', 'config', context.split('/')[0] + '.cfg')

6
test/lib/ansible_test/_internal/completion.py
Unescape Escape View File

@ -270,9 +270,9 @@ def parse_completion_entry(value: str) -> tuple[str, dict[str, str]]:
def filter_completion(
completion: dict[str, TCompletionConfig],
controller_only: bool = False,
include_defaults: bool = False,
completion: dict[str, TCompletionConfig],
controller_only: bool = False,
include_defaults: bool = False,
) -> dict[str, TCompletionConfig]:
"""Return the given completion dictionary, filtering out configs which do not support the controller if controller_only is specified."""
if controller_only:

80
test/lib/ansible_test/_internal/containers.py
Unescape Escape View File

@ -108,19 +108,19 @@ class CleanupMode(enum.Enum):
def run_support_container(
args: EnvironmentConfig,
context: str,
image: str,
name: str,
ports: list[int],
aliases: t.Optional[list[str]] = None,
start: bool = True,
allow_existing: bool = False,
cleanup: t.Optional[CleanupMode] = None,
cmd: t.Optional[list[str]] = None,
env: t.Optional[dict[str, str]] = None,
options: t.Optional[list[str]] = None,
publish_ports: bool = True,
args: EnvironmentConfig,
context: str,
image: str,
name: str,
ports: list[int],
aliases: t.Optional[list[str]] = None,
start: bool = True,
allow_existing: bool = False,
cleanup: t.Optional[CleanupMode] = None,
cmd: t.Optional[list[str]] = None,
env: t.Optional[dict[str, str]] = None,
options: t.Optional[list[str]] = None,
publish_ports: bool = True,
) -> t.Optional[ContainerDescriptor]:
"""
Start a container used to support tests, but not run them.
@ -236,12 +236,12 @@ def run_support_container(
def run_container(
args: EnvironmentConfig,
image: str,
name: str,
options: t.Optional[list[str]],
cmd: t.Optional[list[str]] = None,
create_only: bool = False,
args: EnvironmentConfig,
image: str,
name: str,
options: t.Optional[list[str]],
cmd: t.Optional[list[str]] = None,
create_only: bool = False,
) -> str:
"""Run a container using the given docker image."""
options = list(options or [])
@ -594,8 +594,8 @@ class SupportContainerContext:
@contextlib.contextmanager
def support_container_context(
args: EnvironmentConfig,
ssh: t.Optional[SshConnectionDetail],
args: EnvironmentConfig,
ssh: t.Optional[SshConnectionDetail],
) -> c.Iterator[t.Optional[ContainerDatabase]]:
"""Create a context manager for integration tests that use support containers."""
if not isinstance(args, (IntegrationConfig, UnitsConfig, SanityConfig, ShellConfig)):
@ -617,9 +617,9 @@ def support_container_context(
def create_support_container_context(
args: EnvironmentConfig,
ssh: t.Optional[SshConnectionDetail],
containers: ContainerDatabase,
args: EnvironmentConfig,
ssh: t.Optional[SshConnectionDetail],
containers: ContainerDatabase,
) -> SupportContainerContext:
"""Context manager that provides SSH port forwards. Returns updated container metadata."""
host_type = HostType.control
@ -819,9 +819,9 @@ def create_hosts_entries(context: dict[str, ContainerAccess]) -> list[str]:
def create_container_hooks(
args: IntegrationConfig,
control_connections: list[SshConnectionDetail],
managed_connections: t.Optional[list[SshConnectionDetail]],
args: IntegrationConfig,
control_connections: list[SshConnectionDetail],
managed_connections: t.Optional[list[SshConnectionDetail]],
) -> tuple[t.Optional[c.Callable[[IntegrationTarget], None]], t.Optional[c.Callable[[IntegrationTarget], None]]]:
"""Return pre and post target callbacks for enabling and disabling container access for each test target."""
containers = get_container_database(args)
@ -873,13 +873,13 @@ def create_managed_contexts(control_contexts: dict[str, dict[str, ContainerAcces
def forward_ssh_ports(
args: IntegrationConfig,
ssh_connections: t.Optional[list[SshConnectionDetail]],
playbook: str,
target_state: dict[str, tuple[list[str], list[SshProcess]]],
target: IntegrationTarget,
host_type: str,
contexts: dict[str, dict[str, ContainerAccess]],
args: IntegrationConfig,
ssh_connections: t.Optional[list[SshConnectionDetail]],
playbook: str,
target_state: dict[str, tuple[list[str], list[SshProcess]]],
target: IntegrationTarget,
host_type: str,
contexts: dict[str, dict[str, ContainerAccess]],
) -> None:
"""Configure port forwarding using SSH and write hosts file entries."""
if ssh_connections is None:
@ -944,12 +944,12 @@ def forward_ssh_ports(
def cleanup_ssh_ports(
args: IntegrationConfig,
ssh_connections: list[SshConnectionDetail],
playbook: str,
target_state: dict[str, tuple[list[str], list[SshProcess]]],
target: IntegrationTarget,
host_type: str,
args: IntegrationConfig,
ssh_connections: list[SshConnectionDetail],
playbook: str,
target_state: dict[str, tuple[list[str], list[SshProcess]]],
target: IntegrationTarget,
host_type: str,
) -> None:
"""Stop previously configured SSH port forwarding and remove previously written hosts file entries."""
state = target_state.pop(target.name, None)

8
test/lib/ansible_test/_internal/core_ci.py
Unescape Escape View File

@ -115,10 +115,10 @@ class AnsibleCoreCI:
DEFAULT_ENDPOINT = 'https://ansible-core-ci.testing.ansible.com'
def __init__(
self,
args: EnvironmentConfig,
resource: Resource,
load: bool = True,
self,
args: EnvironmentConfig,
resource: Resource,
load: bool = True,
) -> None:
self.args = args
self.resource = resource

22
test/lib/ansible_test/_internal/coverage_util.py
Unescape Escape View File

@ -143,14 +143,14 @@ def get_sqlite_schema_version(path: str) -> int:
def cover_python(
args: TestConfig,
python: PythonConfig,
cmd: list[str],
target_name: str,
env: dict[str, str],
capture: bool,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
args: TestConfig,
python: PythonConfig,
cmd: list[str],
target_name: str,
env: dict[str, str],
capture: bool,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run a command while collecting Python code coverage."""
if args.coverage:
@ -176,9 +176,9 @@ def get_coverage_platform(config: HostConfig) -> str:
def get_coverage_environment(
args: TestConfig,
target_name: str,
version: str,
args: TestConfig,
target_name: str,
version: str,
) -> dict[str, str]:
"""Return environment variables needed to collect code coverage."""
# unit tests, sanity tests and other special cases (localhost only)

20
test/lib/ansible_test/_internal/delegation.py
Unescape Escape View File

@ -266,12 +266,12 @@ def download_results(args: EnvironmentConfig, con: Connection, content_root: str
def generate_command(
args: EnvironmentConfig,
python: PythonConfig,
ansible_bin_path: str,
content_root: str,
exclude: list[str],
require: list[str],
args: EnvironmentConfig,
python: PythonConfig,
ansible_bin_path: str,
content_root: str,
exclude: list[str],
require: list[str],
) -> list[str]:
"""Generate the command necessary to delegate ansible-test."""
cmd = [os.path.join(ansible_bin_path, 'ansible-test')]
@ -318,10 +318,10 @@ def generate_command(
def filter_options(
args: EnvironmentConfig,
argv: list[str],
exclude: list[str],
require: list[str],
args: EnvironmentConfig,
argv: list[str],
exclude: list[str],
require: list[str],
) -> c.Iterable[str]:
"""Return an iterable that filters out unwanted CLI options and injects new ones as requested."""
replace: list[tuple[str, int, t.Optional[t.Union[bool, str, list[str]]]]] = [

72
test/lib/ansible_test/_internal/docker_util.py
Unescape Escape View File

@ -401,11 +401,11 @@ def detect_host_properties(args: CommonConfig) -> ContainerHostProperties:
def run_utility_container(
args: CommonConfig,
name: str,
cmd: list[str],
options: list[str],
data: t.Optional[str] = None,
args: CommonConfig,
name: str,
cmd: list[str],
options: list[str],
data: t.Optional[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command using the ansible-test utility container, returning stdout and stderr."""
options = options + [
@ -670,30 +670,30 @@ def docker_cp_to(args: CommonConfig, container_id: str, src: str, dst: str) -> N
def docker_create(
args: CommonConfig,
image: str,
options: list[str],
cmd: list[str] = None,
args: CommonConfig,
image: str,
options: list[str],
cmd: list[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Create a container using the given docker image."""
return docker_command(args, ['create'] + options + [image] + cmd, capture=True)
def docker_run(
args: CommonConfig,
image: str,
options: list[str],
cmd: list[str] = None,
data: t.Optional[str] = None,
args: CommonConfig,
image: str,
options: list[str],
cmd: list[str] = None,
data: t.Optional[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run a container using the given docker image."""
return docker_command(args, ['run'] + options + [image] + cmd, data=data, capture=True)
def docker_start(
args: CommonConfig,
container_id: str,
options: list[str],
args: CommonConfig,
container_id: str,
options: list[str],
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Start a container by name or ID."""
return docker_command(args, ['start'] + options + [container_id], capture=True)
@ -943,16 +943,16 @@ def docker_logs(args: CommonConfig, container_id: str) -> None:
def docker_exec(
args: CommonConfig,
container_id: str,
cmd: list[str],
capture: bool,
options: t.Optional[list[str]] = None,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
data: t.Optional[str] = None,
args: CommonConfig,
container_id: str,
cmd: list[str],
capture: bool,
options: t.Optional[list[str]] = None,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
data: t.Optional[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Execute the given command in the specified container."""
if not options:
@ -966,15 +966,15 @@ def docker_exec(
def docker_command(
args: CommonConfig,
cmd: list[str],
capture: bool,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
always: bool = False,
data: t.Optional[str] = None,
args: CommonConfig,
cmd: list[str],
capture: bool,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
always: bool = False,
data: t.Optional[str] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified docker command."""
env = docker_environment()

6
test/lib/ansible_test/_internal/host_profiles.py
Unescape Escape View File

@ -1411,9 +1411,9 @@ def get_config_profile_type_map() -> dict[t.Type[HostConfig], t.Type[HostProfile
def create_host_profile(
args: EnvironmentConfig,
config: HostConfig,
controller: bool,
args: EnvironmentConfig,
config: HostConfig,
controller: bool,
) -> HostProfile:
"""Create and return a host profile from the given host configuration."""
profile_type = get_config_profile_type_map()[type(config)]

8
test/lib/ansible_test/_internal/provisioning.py
Unescape Escape View File

@ -97,10 +97,10 @@ class HostState:
def prepare_profiles(
args: TEnvironmentConfig,
targets_use_pypi: bool = False,
skip_setup: bool = False,
requirements: t.Optional[c.Callable[[HostProfile], None]] = None,
args: TEnvironmentConfig,
targets_use_pypi: bool = False,
skip_setup: bool = False,
requirements: t.Optional[c.Callable[[HostProfile], None]] = None,
) -> HostState:
"""
Create new profiles, or load existing ones, and return them.

50
test/lib/ansible_test/_internal/python_requirements.py
Unescape Escape View File

@ -122,14 +122,14 @@ class PipBootstrap(PipCommand):
def install_requirements(
args: EnvironmentConfig,
python: PythonConfig,
ansible: bool = False,
command: bool = False,
coverage: bool = False,
virtualenv: bool = False,
controller: bool = True,
connection: t.Optional[Connection] = None,
args: EnvironmentConfig,
python: PythonConfig,
ansible: bool = False,
command: bool = False,
coverage: bool = False,
virtualenv: bool = False,
controller: bool = True,
connection: t.Optional[Connection] = None,
) -> None:
"""Install requirements for the given Python using the specified arguments."""
create_result_directories(args)
@ -197,15 +197,15 @@ def collect_bootstrap(python: PythonConfig) -> list[PipCommand]:
def collect_requirements(
python: PythonConfig,
controller: bool,
ansible: bool,
cryptography: bool,
coverage: bool,
virtualenv: bool,
minimize: bool,
command: t.Optional[str],
sanity: t.Optional[str],
python: PythonConfig,
controller: bool,
ansible: bool,
cryptography: bool,
coverage: bool,
virtualenv: bool,
minimize: bool,
command: t.Optional[str],
sanity: t.Optional[str],
) -> list[PipCommand]:
"""Collect requirements for the given Python using the specified arguments."""
commands: list[PipCommand] = []
@ -252,10 +252,10 @@ def collect_requirements(
def run_pip(
args: EnvironmentConfig,
python: PythonConfig,
commands: list[PipCommand],
connection: t.Optional[Connection],
args: EnvironmentConfig,
python: PythonConfig,
commands: list[PipCommand],
connection: t.Optional[Connection],
) -> None:
"""Run the specified pip commands for the given Python, and optionally the specified host."""
connection = connection or LocalConnection(args)
@ -367,10 +367,10 @@ def collect_integration_install(command: str, controller: bool) -> list[PipInsta
def collect_install(
requirements_paths: list[tuple[str, str]],
constraints_paths: list[tuple[str, str]],
packages: t.Optional[list[str]] = None,
constraints: bool = True,
requirements_paths: list[tuple[str, str]],
constraints_paths: list[tuple[str, str]],
packages: t.Optional[list[str]] = None,
constraints: bool = True,
) -> list[PipInstall]:
"""Build a pip install list from the given requirements, constraints and packages."""
# listing content constraints first gives them priority over constraints provided by ansible-test

30
test/lib/ansible_test/_internal/ssh.py
Unescape Escape View File

@ -151,10 +151,10 @@ class SshProcess:
def create_ssh_command(
ssh: SshConnectionDetail,
options: t.Optional[dict[str, t.Union[str, int]]] = None,
cli_args: list[str] = None,
command: t.Optional[str] = None,
ssh: SshConnectionDetail,
options: t.Optional[dict[str, t.Union[str, int]]] = None,
cli_args: list[str] = None,
command: t.Optional[str] = None,
) -> list[str]:
"""Create an SSH command using the specified options."""
cmd = [
@ -207,11 +207,11 @@ def ssh_options_to_str(options: t.Union[dict[str, t.Union[int, str]], dict[str,
def run_ssh_command(
args: EnvironmentConfig,
ssh: SshConnectionDetail,
options: t.Optional[dict[str, t.Union[str, int]]] = None,
cli_args: list[str] = None,
command: t.Optional[str] = None,
args: EnvironmentConfig,
ssh: SshConnectionDetail,
options: t.Optional[dict[str, t.Union[str, int]]] = None,
cli_args: list[str] = None,
command: t.Optional[str] = None,
) -> SshProcess:
"""Run the specified SSH command, returning the created SshProcess instance created."""
cmd = create_ssh_command(ssh, options, cli_args, command)
@ -233,9 +233,9 @@ def run_ssh_command(
def create_ssh_port_forwards(
args: EnvironmentConfig,
ssh: SshConnectionDetail,
forwards: list[tuple[str, int]],
args: EnvironmentConfig,
ssh: SshConnectionDetail,
forwards: list[tuple[str, int]],
) -> SshProcess:
"""
Create SSH port forwards using the provided list of tuples (target_host, target_port).
@ -257,9 +257,9 @@ def create_ssh_port_forwards(
def create_ssh_port_redirects(
args: EnvironmentConfig,
ssh: SshConnectionDetail,
redirects: list[tuple[int, str, int]],
args: EnvironmentConfig,
ssh: SshConnectionDetail,
redirects: list[tuple[int, str, int]],
) -> SshProcess:
"""Create SSH port redirections using the provided list of tuples (bind_port, target_host, target_port)."""
options: dict[str, t.Union[str, int]] = {}

34
test/lib/ansible_test/_internal/target.py
Unescape Escape View File

@ -65,10 +65,10 @@ def walk_completion_targets(targets: c.Iterable[CompletionTarget], prefix: str,
def walk_internal_targets(
targets: c.Iterable[TCompletionTarget],
includes: t.Optional[list[str]] = None,
excludes: t.Optional[list[str]] = None,
requires: t.Optional[list[str]] = None,
targets: c.Iterable[TCompletionTarget],
includes: t.Optional[list[str]] = None,
excludes: t.Optional[list[str]] = None,
requires: t.Optional[list[str]] = None,
) -> tuple[TCompletionTarget, ...]:
"""Return a tuple of matching completion targets."""
targets = tuple(targets)
@ -256,13 +256,13 @@ def load_integration_prefixes() -> dict[str, str]:
def walk_test_targets(
path: t.Optional[str] = None,
module_path: t.Optional[str] = None,
extensions: t.Optional[tuple[str, ...]] = None,
prefix: t.Optional[str] = None,
extra_dirs: t.Optional[tuple[str, ...]] = None,
include_symlinks: bool = False,
include_symlinked_directories: bool = False,
path: t.Optional[str] = None,
module_path: t.Optional[str] = None,
extensions: t.Optional[tuple[str, ...]] = None,
prefix: t.Optional[str] = None,
extra_dirs: t.Optional[tuple[str, ...]] = None,
include_symlinks: bool = False,
include_symlinked_directories: bool = False,
) -> c.Iterable[TestTarget]:
"""Iterate over available test targets."""
if path:
@ -436,12 +436,12 @@ class CompletionTarget(metaclass=abc.ABCMeta):
class TestTarget(CompletionTarget):
"""Generic test target."""
def __init__(
self,
path: str,
module_path: t.Optional[str],
module_prefix: t.Optional[str],
base_path: str,
symlink: t.Optional[bool] = None,
self,
path: str,
module_path: t.Optional[str],
module_prefix: t.Optional[str],
base_path: str,
symlink: t.Optional[bool] = None,
) -> None:
super().__init__()

28
test/lib/ansible_test/_internal/test.py
Unescape Escape View File

@ -215,12 +215,12 @@ class TestSkipped(TestResult):
class TestFailure(TestResult):
"""Test failure."""
def __init__(
self,
command: str,
test: str,
python_version: t.Optional[str] = None,
messages: t.Optional[c.Sequence[TestMessage]] = None,
summary: t.Optional[str] = None,
self,
command: str,
test: str,
python_version: t.Optional[str] = None,
messages: t.Optional[c.Sequence[TestMessage]] = None,
summary: t.Optional[str] = None,
):
super().__init__(command, test, python_version)
@ -379,14 +379,14 @@ class TestFailure(TestResult):
class TestMessage:
"""Single test message for one file."""
def __init__(
self,
message: str,
path: str,
line: int = 0,
column: int = 0,
level: str = 'error',
code: t.Optional[str] = None,
confidence: t.Optional[int] = None,
self,
message: str,
path: str,
line: int = 0,
column: int = 0,
level: str = 'error',
code: t.Optional[str] = None,
confidence: t.Optional[int] = None,
):
self.__path = path
self.__line = line

62
test/lib/ansible_test/_internal/util.py
Unescape Escape View File

@ -345,19 +345,19 @@ def get_available_python_versions() -> dict[str, str]:
def raw_command(
cmd: c.Iterable[str],
capture: bool,
env: t.Optional[dict[str, str]] = None,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
explain: bool = False,
stdin: t.Optional[t.Union[t.IO[bytes], int]] = None,
stdout: t.Optional[t.Union[t.IO[bytes], int]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
cmd_verbosity: int = 1,
str_errors: str = 'strict',
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
cmd: c.Iterable[str],
capture: bool,
env: t.Optional[dict[str, str]] = None,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
explain: bool = False,
stdin: t.Optional[t.Union[t.IO[bytes], int]] = None,
stdout: t.Optional[t.Union[t.IO[bytes], int]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
cmd_verbosity: int = 1,
str_errors: str = 'strict',
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return stdout and stderr as a tuple."""
output_stream = output_stream or OutputStream.AUTO
@ -496,12 +496,12 @@ def raw_command(
def communicate_with_process(
process: subprocess.Popen,
stdin: t.Optional[bytes],
stdout: bool,
stderr: bool,
capture: bool,
output_stream: OutputStream,
process: subprocess.Popen,
stdin: t.Optional[bytes],
stdout: bool,
stderr: bool,
capture: bool,
output_stream: OutputStream,
) -> tuple[bytes, bytes]:
"""Communicate with the specified process, handling stdin/stdout/stderr as requested."""
threads: list[WrappedThread] = []
@ -855,11 +855,11 @@ class Display:
self.print_message(message, color=color, truncate=truncate)
def print_message( # pylint: disable=locally-disabled, invalid-name
self,
message: str,
color: t.Optional[str] = None,
stderr: bool = False,
truncate: bool = False,
self,
message: str,
color: t.Optional[str] = None,
stderr: bool = False,
truncate: bool = False,
) -> None:
"""Display a message."""
if self.redact and self.sensitive:
@ -901,13 +901,13 @@ class ApplicationWarning(Exception):
class SubprocessError(ApplicationError):
"""Error resulting from failed subprocess execution."""
def __init__(
self,
cmd: list[str],
status: int = 0,
stdout: t.Optional[str] = None,
stderr: t.Optional[str] = None,
runtime: t.Optional[float] = None,
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
self,
cmd: list[str],
status: int = 0,
stdout: t.Optional[str] = None,
stderr: t.Optional[str] = None,
runtime: t.Optional[float] = None,
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
) -> None:
message = 'Command "%s" returned exit status %s.\n' % (shlex.join(cmd), status)

44
test/lib/ansible_test/_internal/util_common.py
Unescape Escape View File

@ -396,14 +396,14 @@ def cleanup_python_paths() -> None:
def intercept_python(
args: CommonConfig,
python: PythonConfig,
cmd: list[str],
env: dict[str, str],
capture: bool,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
always: bool = False,
args: CommonConfig,
python: PythonConfig,
cmd: list[str],
env: dict[str, str],
capture: bool,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
always: bool = False,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""
Run a command while intercepting invocations of Python to control the version used.
@ -428,20 +428,20 @@ def intercept_python(
def run_command(
args: CommonConfig,
cmd: c.Iterable[str],
capture: bool,
env: t.Optional[dict[str, str]] = None,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
always: bool = False,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
cmd_verbosity: int = 1,
str_errors: str = 'strict',
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
args: CommonConfig,
cmd: c.Iterable[str],
capture: bool,
env: t.Optional[dict[str, str]] = None,
data: t.Optional[str] = None,
cwd: t.Optional[str] = None,
always: bool = False,
stdin: t.Optional[t.IO[bytes]] = None,
stdout: t.Optional[t.IO[bytes]] = None,
interactive: bool = False,
output_stream: t.Optional[OutputStream] = None,
cmd_verbosity: int = 1,
str_errors: str = 'strict',
error_callback: t.Optional[c.Callable[[SubprocessError], None]] = None,
) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return stdout and stderr as a tuple."""
explain = args.explain and not always

4
test/lib/ansible_test/_internal/venv.py
Unescape Escape View File

@ -41,8 +41,8 @@ from .python_requirements import (
def get_virtual_python(
args: EnvironmentConfig,
python: VirtualPythonConfig,
args: EnvironmentConfig,
python: VirtualPythonConfig,
) -> VirtualPythonConfig:
"""Create a virtual environment for the given Python and return the path to its root."""
if python.system_site_packages:

Write Preview
Loading…
Cancel
Save