starred/ansible
1
0
Fork 0
mirror of https://github.com/ansible/ansible.git synced 2026-08-03 08:03:05 +02:00
Code Issues Packages Projects Releases Wiki Activity

ansible-test - Code cleanup. (#75774)

Browse Source 
* ansible-test - Remove unused code.
* ansible-test - More PEP 484 type hints.
This commit is contained in:
Matt Clay
2021-09-24 00:38:56 -07:00
committed by GitHub
parent 6c133da45e
commit 58b03be417
17 changed files with 130 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/lib/ansible_test/_internal/ansible_util.py
+2 -7
View File
@@ -69,13 +69,8 @@ def get_hosts(inventory, group_name): # type: (t.Dict[str, t.Any], str) -> t.Di
return hosts
def ansible_environment(args, color=True, ansible_config=None):
"""
:type args: CommonConfig
:type color: bool
:type ansible_config: str | None
:rtype: dict[str, str]
"""
def ansible_environment(args, color=True, ansible_config=None): # type: (CommonConfig, bool, t.Optional[str]) -> t.Dict[str, str]
"""Return a dictionary of environment variables to use when running Ansible commands."""
env = common_environment()
path = env['PATH']
test/lib/ansible_test/_internal/classification/common.py
+2 -5
View File
@@ -8,11 +8,8 @@ from ..data import (
)
def resolve_csharp_ps_util(import_name, path):
"""
:type import_name: str
:type path: str
"""
def resolve_csharp_ps_util(import_name, path): # type: (str, str) -> str
"""Return the fully qualified name of the given import if possible, otherwise return the original import name."""
if data_context().content.is_ansible or not import_name.startswith('.'):
# We don't support relative paths for builtin utils, there's no point.
return import_name
test/lib/ansible_test/_internal/classification/csharp.py
+9 -17
View File
@@ -21,14 +21,13 @@ from ..data import (
data_context,
)
from ..target import (
TestTarget,
)
def get_csharp_module_utils_imports(powershell_targets, csharp_targets):
"""Return a dictionary of module_utils names mapped to sets of powershell file paths.
:type powershell_targets: list[TestTarget] - C# files
:type csharp_targets: list[TestTarget] - PS files
:rtype: dict[str, set[str]]
"""
def get_csharp_module_utils_imports(powershell_targets, csharp_targets): # type: (t.List[TestTarget], t.List[TestTarget]) -> t.Dict[str, t.Set[str]]
"""Return a dictionary of module_utils names mapped to sets of powershell file paths."""
module_utils = enumerate_module_utils()
imports_by_target_path = {}
@@ -66,22 +65,15 @@ def get_csharp_module_utils_name(path): # type: (str) -> str
return name
def enumerate_module_utils():
"""Return a list of available module_utils imports.
:rtype: set[str]
"""
def enumerate_module_utils(): # type: () -> t.Set[str]
"""Return a set of available module_utils imports."""
return set(get_csharp_module_utils_name(p)
for p in data_context().content.walk_files(data_context().content.module_utils_csharp_path)
if os.path.splitext(p)[1] == '.cs')
def extract_csharp_module_utils_imports(path, module_utils, is_pure_csharp):
"""Return a list of module_utils imports found in the specified source file.
:type path: str
:type module_utils: set[str]
:type is_pure_csharp: bool
:rtype: set[str]
"""
def extract_csharp_module_utils_imports(path, module_utils, is_pure_csharp): # type: (str, t.Set[str], bool) -> t.Set[str]
"""Return a set of module_utils imports found in the specified source file."""
imports = set()
if is_pure_csharp:
pattern = re.compile(r'(?i)^using\s((?:Ansible|AnsibleCollections)\..+);$')
test/lib/ansible_test/_internal/classification/powershell.py
+9 -15
View File
@@ -21,13 +21,13 @@ from ..data import (
data_context,
)
from ..target import (
TestTarget,
)
def get_powershell_module_utils_imports(powershell_targets):
"""Return a dictionary of module_utils names mapped to sets of powershell file paths.
:type powershell_targets: list[TestTarget]
:rtype: dict[str, set[str]]
"""
def get_powershell_module_utils_imports(powershell_targets): # type: (t.List[TestTarget]) -> t.Dict[str, t.Set[str]]
"""Return a dictionary of module_utils names mapped to sets of powershell file paths."""
module_utils = enumerate_module_utils()
imports_by_target_path = {}
@@ -62,21 +62,15 @@ def get_powershell_module_utils_name(path): # type: (str) -> str
return name
def enumerate_module_utils():
"""Return a list of available module_utils imports.
:rtype: set[str]
"""
def enumerate_module_utils(): # type: () -> t.Set[str]
"""Return a set of available module_utils imports."""
return set(get_powershell_module_utils_name(p)
for p in data_context().content.walk_files(data_context().content.module_utils_powershell_path)
if os.path.splitext(p)[1] == '.psm1')
def extract_powershell_module_utils_imports(path, module_utils):
"""Return a list of module_utils imports found in the specified source file.
:type path: str
:type module_utils: set[str]
:rtype: set[str]
"""
def extract_powershell_module_utils_imports(path, module_utils): # type: (str, t.Set[str]) -> t.Set[str]
"""Return a set of module_utils imports found in the specified source file."""
imports = set()
code = read_text_file(path)
test/lib/ansible_test/_internal/classification/python.py
+15 -30
View File
@@ -20,17 +20,17 @@ from ..data import (
data_context,
)
from ..target import (
TestTarget,
)
VIRTUAL_PACKAGES = {
'ansible.module_utils.six',
}
def get_python_module_utils_imports(compile_targets):
"""Return a dictionary of module_utils names mapped to sets of python file paths.
:type compile_targets: list[TestTarget]
:rtype: dict[str, set[str]]
"""
def get_python_module_utils_imports(compile_targets): # type: (t.List[TestTarget]) -> t.Dict[str, t.Set[str]]
"""Return a dictionary of module_utils names mapped to sets of python file paths."""
module_utils = enumerate_module_utils()
virtual_utils = set(m for m in module_utils if any(m.startswith('%s.' % v) for v in VIRTUAL_PACKAGES))
@@ -163,12 +163,8 @@ def enumerate_module_utils():
return set(module_utils)
def extract_python_module_utils_imports(path, module_utils):
"""Return a list of module_utils imports found in the specified source file.
:type path: str
:type module_utils: set[str]
:rtype: set[str]
"""
def extract_python_module_utils_imports(path, module_utils): # type: (str, t.Set[str]) -> t.Set[str]
"""Return a list of module_utils imports found in the specified source file."""
# Python code must be read as bytes to avoid a SyntaxError when the source uses comments to declare the file encoding.
# See: https://www.python.org/dev/peps/pep-0263
# Specifically: If a Unicode string with a coding declaration is passed to compile(), a SyntaxError will be raised.
@@ -237,11 +233,7 @@ def relative_to_absolute(name, level, module, path, lineno): # type: (str, int,
class ModuleUtilFinder(ast.NodeVisitor):
"""AST visitor to find valid module_utils imports."""
def __init__(self, path, module_utils):
"""Return a list of module_utils imports found in the specified source file.
:type path: str
:type module_utils: set[str]
"""
def __init__(self, path, module_utils): # type: (str, t.Set[str]) -> None
self.path = path
self.module_utils = module_utils
self.imports = set()
@@ -287,10 +279,8 @@ class ModuleUtilFinder(ast.NodeVisitor):
# noinspection PyPep8Naming
# pylint: disable=locally-disabled, invalid-name
def visit_Import(self, node):
"""
:type node: ast.Import
"""
def visit_Import(self, node): # type: (ast.Import) -> None
"""Visit an import node."""
self.generic_visit(node)
# import ansible.module_utils.MODULE[.MODULE]
@@ -299,10 +289,8 @@ class ModuleUtilFinder(ast.NodeVisitor):
# noinspection PyPep8Naming
# pylint: disable=locally-disabled, invalid-name
def visit_ImportFrom(self, node):
"""
:type node: ast.ImportFrom
"""
def visit_ImportFrom(self, node): # type: (ast.ImportFrom) -> None
"""Visit an import from node."""
self.generic_visit(node)
if not node.module:
@@ -319,11 +307,8 @@ class ModuleUtilFinder(ast.NodeVisitor):
# from ansible_collections.{ns}.{col}.plugins.module_utils.MODULE[.MODULE] import MODULE[, MODULE]
self.add_imports(['%s.%s' % (module, alias.name) for alias in node.names], node.lineno)
def add_import(self, name, line_number):
"""
:type name: str
:type line_number: int
"""
def add_import(self, name, line_number): # type: (str, int) -> None
"""Record the specified import."""
import_name = name
while self.is_module_util_name(name):
test/lib/ansible_test/_internal/commands/coverage/combine.py
+8 -22
View File
@@ -61,11 +61,8 @@ from . import (
)
def command_coverage_combine(args):
"""Patch paths in coverage files and merge into a single file.
:type args: CoverageCombineConfig
:rtype: list[str]
"""
def command_coverage_combine(args): # type: (CoverageCombineConfig) -> None
"""Patch paths in coverage files and merge into a single file."""
host_state = prepare_profiles(args) # coverage combine
combine_coverage_files(args, host_state)
@@ -184,11 +181,8 @@ def _command_coverage_combine_python(args, host_state): # type: (CoverageCombin
return sorted(output_files)
def _command_coverage_combine_powershell(args):
"""
:type args: CoverageCombineConfig
:rtype: list[str]
"""
def _command_coverage_combine_powershell(args): # type: (CoverageCombineConfig) -> t.List[str]
"""Combine PowerShell coverage files and return a list of the output files."""
coverage_files = get_powershell_coverage_files()
def _default_stub_value(source_paths):
@@ -265,12 +259,8 @@ def _command_coverage_combine_powershell(args):
return sorted(output_files)
def _get_coverage_targets(args, walk_func):
"""
:type args: CoverageCombineConfig
:type walk_func: Func
:rtype: list[tuple[str, int]]
"""
def _get_coverage_targets(args, walk_func): # type: (CoverageCombineConfig, t.Callable) -> t.List[t.Tuple[str, int]]
"""Return a list of files to cover and the number of lines in each file, using the given function as the source of the files."""
sources = []
if args.all or args.stub:
@@ -321,12 +311,8 @@ def _build_stub_groups(args, sources, default_stub_value):
return groups
def get_coverage_group(args, coverage_file):
"""
:type args: CoverageCombineConfig
:type coverage_file: str
:rtype: str
"""
def get_coverage_group(args, coverage_file): # type: (CoverageCombineConfig, str) -> t.Optional[str]
"""Return the name of the coverage group for the specified coverage file, or None if no group was found."""
parts = os.path.basename(coverage_file).split('=', 4)
# noinspection PyTypeChecker
test/lib/ansible_test/_internal/commands/coverage/html.py
+2 -4
View File
@@ -29,10 +29,8 @@ from . import (
)
def command_coverage_html(args):
"""
:type args: CoverageHtmlConfig
"""
def command_coverage_html(args): # type: (CoverageHtmlConfig) -> None
"""Generate an HTML coverage report."""
host_state = prepare_profiles(args) # coverage html
output_files = combine_coverage_files(args, host_state)
test/lib/ansible_test/_internal/commands/coverage/report.py
+6 -14
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import os
import typing as t
from ...io import (
read_json_file,
@@ -29,10 +30,8 @@ from . import (
)
def command_coverage_report(args):
"""
:type args: CoverageReportConfig
"""
def command_coverage_report(args): # type: (CoverageReportConfig) -> None
"""Generate a console coverage report."""
host_state = prepare_profiles(args) # coverage report
output_files = combine_coverage_files(args, host_state)
@@ -57,12 +56,8 @@ def command_coverage_report(args):
run_coverage(args, host_state, output_file, 'report', options)
def _generate_powershell_output_report(args, coverage_file):
"""
:type args: CoverageReportConfig
:type coverage_file: str
:rtype: str
"""
def _generate_powershell_output_report(args, coverage_file): # type: (CoverageReportConfig, str) -> str
"""Generate and return a PowerShell coverage report for the given coverage file."""
coverage_info = read_json_file(coverage_file)
root_path = data_context().content.root + '/'
@@ -149,10 +144,7 @@ def _generate_powershell_output_report(args, coverage_file):
class CoverageReportConfig(CoverageCombineConfig):
"""Configuration for the coverage report command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args)
self.show_missing = args.show_missing # type: bool
test/lib/ansible_test/_internal/commands/coverage/xml.py
+7 -16
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import os
import time
import typing as t
from xml.etree.ElementTree import (
Comment,
@@ -47,10 +48,8 @@ from . import (
)
def command_coverage_xml(args):
"""
:type args: CoverageXmlConfig
"""
def command_coverage_xml(args): # type: (CoverageXmlConfig) -> None
"""Generate an XML coverage report."""
host_state = prepare_profiles(args) # coverage xml
output_files = combine_coverage_files(args, host_state)
@@ -70,11 +69,8 @@ def command_coverage_xml(args):
run_coverage(args, host_state, output_file, 'xml', ['-i', '-o', xml_path])
def _generate_powershell_xml(coverage_file):
"""
:type coverage_file: str
:rtype: Element
"""
def _generate_powershell_xml(coverage_file): # type: (str) -> Element
"""Generate a PowerShell coverage report XML element from the specified coverage file and return it."""
coverage_info = read_json_file(coverage_file)
content_root = data_context().content.root
@@ -135,13 +131,8 @@ def _generate_powershell_xml(coverage_file):
return elem_coverage
def _add_cobertura_package(packages, package_name, package_data):
"""
:type packages: SubElement
:type package_name: str
:type package_data: Dict[str, Dict[str, int]]
:rtype: Tuple[int, int]
"""
def _add_cobertura_package(packages, package_name, package_data): # type: (SubElement, str, t.Dict[str, t.Dict[str, int]]) -> t.Tuple[int, int]
"""Add a package element to the given packages element."""
elem_package = SubElement(packages, 'package')
elem_classes = SubElement(elem_package, 'classes')
test/lib/ansible_test/_internal/commands/sanity/pylint.py
+6 -17
View File
@@ -57,7 +57,6 @@ from ...host_configs import (
class PylintTest(SanitySingleVersion):
"""Sanity test using pylint."""
def __init__(self):
super().__init__()
self.optional_error_codes.update([
@@ -98,26 +97,16 @@ class PylintTest(SanitySingleVersion):
contexts = []
remaining_paths = set(paths)
def add_context(available_paths, context_name, context_filter):
"""
:type available_paths: set[str]
:type context_name: str
:type context_filter: (str) -> bool
"""
def add_context(available_paths, context_name, context_filter): # type: (t.Set[str], str, t.Callable[[str], bool]) -> None
"""Add the specified context to the context list, consuming available paths that match the given context filter."""
filtered_paths = set(p for p in available_paths if context_filter(p))
contexts.append((context_name, sorted(filtered_paths)))
available_paths -= filtered_paths
def filter_path(path_filter=None):
"""
:type path_filter: str
:rtype: (str) -> bool
"""
def context_filter(path_to_filter):
"""
:type path_to_filter: str
:rtype: bool
"""
def filter_path(path_filter=None): # type: (str) -> t.Callable[[str], bool]
"""Return a function that filters out paths which are not a subdirectory of the given path."""
def context_filter(path_to_filter): # type: (str) -> bool
"""Return true if the given path matches, otherwise return False."""
return is_subdir(path_to_filter, path_filter)
return context_filter
test/lib/ansible_test/_internal/commands/units/__init__.py
+2 -4
View File
@@ -92,10 +92,8 @@ class TestContext:
module_utils = 'module_utils'
def command_units(args):
"""
:type args: UnitsConfig
"""
def command_units(args): # type: (UnitsConfig) -> None
"""Run unit tests."""
handle_layout_messages(data_context().content.unit_messages)
changes = get_changes_filter(args)
test/lib/ansible_test/_internal/config.py
+8 -34
View File
@@ -194,11 +194,7 @@ class EnvironmentConfig(CommonConfig):
class TestConfig(EnvironmentConfig):
"""Configuration common to all test commands."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
def __init__(self, args, command): # type: (t.Any, str) -> None
super().__init__(args, command)
self.coverage = args.coverage # type: bool
@@ -239,10 +235,7 @@ class TestConfig(EnvironmentConfig):
class ShellConfig(EnvironmentConfig):
"""Configuration for the shell command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'shell')
self.raw = args.raw # type: bool
@@ -250,10 +243,7 @@ class ShellConfig(EnvironmentConfig):
class SanityConfig(TestConfig):
"""Configuration for the sanity command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'sanity')
self.test = args.test # type: t.List[str]
@@ -278,11 +268,7 @@ class SanityConfig(TestConfig):
class IntegrationConfig(TestConfig):
"""Configuration for the integration command."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
def __init__(self, args, command): # type: (t.Any, str) -> None
super().__init__(args, command)
self.start_at = args.start_at # type: str
@@ -326,28 +312,19 @@ TIntegrationConfig = t.TypeVar('TIntegrationConfig', bound=IntegrationConfig)
class PosixIntegrationConfig(IntegrationConfig):
"""Configuration for the posix integration command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'integration')
class WindowsIntegrationConfig(IntegrationConfig):
"""Configuration for the windows integration command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'windows-integration')
class NetworkIntegrationConfig(IntegrationConfig):
"""Configuration for the network integration command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'network-integration')
self.testcase = args.testcase # type: str
@@ -355,10 +332,7 @@ class NetworkIntegrationConfig(IntegrationConfig):
class UnitsConfig(TestConfig):
"""Configuration for the units command."""
def __init__(self, args):
"""
:type args: any
"""
def __init__(self, args): # type: (t.Any) -> None
super().__init__(args, 'units')
self.collect_only = args.collect_only # type: bool
test/lib/ansible_test/_internal/docker_util.py
+6 -8
View File
@@ -176,10 +176,11 @@ def is_docker_user_defined_network(network): # type: (str) -> bool
return network and network != 'bridge'
def docker_pull(args, image):
def docker_pull(args, image): # type: (EnvironmentConfig, str) -> None
"""
:type args: EnvironmentConfig
:type image: str
Pull the specified image if it is not available.
Images without a tag or digest will not be pulled.
Retries up to 10 times if the pull fails.
"""
if '@' not in image and ':' not in image:
display.info('Skipping pull of image without tag or digest: %s' % image, verbosity=2)
@@ -265,11 +266,8 @@ def docker_start(args, container_id, options=None): # type: (EnvironmentConfig,
raise ApplicationError('Failed to run docker container "%s".' % container_id)
def docker_rm(args, container_id):
"""
:type args: EnvironmentConfig
:type container_id: str
"""
def docker_rm(args, container_id): # type: (EnvironmentConfig, str) -> None
"""Remove the specified container."""
try:
docker_command(args, ['rm', '-f', container_id], capture=True)
except SubprocessError as ex:
test/lib/ansible_test/_internal/metadata.py
+18 -39
View File
@@ -26,10 +26,8 @@ class Metadata:
self.change_description = None # type: t.Optional[ChangeDescription]
self.ci_provider = None # type: t.Optional[str]
def populate_changes(self, diff):
"""
:type diff: list[str] | None
"""
def populate_changes(self, diff): # type: (t.Optional[t.List[str]]) -> None
"""Populate the changeset using the given diff."""
patches = parse_diff(diff)
patches = sorted(patches, key=lambda k: k.new.path) # type: t.List[FileDiff]
@@ -47,10 +45,8 @@ class Metadata:
# failed tests involving deleted files should be using line 0 since there is no content remaining
self.changes[path] = ((0, 0),)
def to_dict(self):
"""
:rtype: dict[str, any]
"""
def to_dict(self): # type: () -> t.Dict[str, t.Any]
"""Return a dictionary representation of the metadata."""
return dict(
changes=self.changes,
cloud_config=self.cloud_config,
@@ -58,10 +54,8 @@ class Metadata:
change_description=self.change_description.to_dict(),
)
def to_file(self, path):
"""
:type path: path
"""
def to_file(self, path): # type: (str) -> None
"""Write the metadata to the specified file."""
data = self.to_dict()
display.info('>>> Metadata: %s\n%s' % (path, data), verbosity=3)
@@ -69,20 +63,14 @@ class Metadata:
write_json_file(path, data)
@staticmethod
def from_file(path):
"""
:type path: str
:rtype: Metadata
"""
def from_file(path): # type: (str) -> Metadata
"""Return metadata loaded from the specified file."""
data = read_json_file(path)
return Metadata.from_dict(data)
@staticmethod
def from_dict(data):
"""
:type data: dict[str, any]
:rtype: Metadata
"""
def from_dict(data): # type: (t.Dict[str, t.Any]) -> Metadata
"""Return metadata loaded from the specified dictionary."""
metadata = Metadata()
metadata.changes = data['changes']
metadata.cloud_config = data['cloud_config']
@@ -103,23 +91,17 @@ class ChangeDescription:
self.no_integration_paths = [] # type: t.List[str]
@property
def targets(self):
"""
:rtype: list[str] | None
"""
def targets(self): # type: () -> t.Optional[t.List[str]]
"""Optional list of target names."""
return self.regular_command_targets.get(self.command)
@property
def focused_targets(self):
"""
:rtype: list[str] | None
"""
def focused_targets(self): # type: () -> t.Optional[t.List[str]]
"""Optional list of focused target names."""
return self.focused_command_targets.get(self.command)
def to_dict(self):
"""
:rtype: dict[str, any]
"""
def to_dict(self): # type: () -> t.Dict[str, t.Any]
"""Return a dictionary representation of the change description."""
return dict(
command=self.command,
changed_paths=self.changed_paths,
@@ -130,11 +112,8 @@ class ChangeDescription:
)
@staticmethod
def from_dict(data):
"""
:param data: dict[str, any]
:rtype: ChangeDescription
"""
def from_dict(data): # type: (t.Dict[str, t.Any]) -> ChangeDescription
"""Return a change description loaded from the given dictionary."""
changes = ChangeDescription()
changes.command = data['command']
changes.changed_paths = data['changed_paths']
test/lib/ansible_test/_internal/thread.py
+1 -4
View File
@@ -13,10 +13,7 @@ TCallable = t.TypeVar('TCallable', bound=t.Callable)
class WrappedThread(threading.Thread):
"""Wrapper around Thread which captures results and exceptions."""
def __init__(self, action):
"""
:type action: () -> any
"""
def __init__(self, action): # type: (t.Callable[[], t.Any]) -> None
super().__init__()
self._result = queue.Queue()
self.action = action
test/lib/ansible_test/_internal/util.py
+28 -76
View File
@@ -3,7 +3,6 @@ from __future__ import annotations
import errno
import fcntl
import hashlib
import inspect
import os
import pkgutil
@@ -31,7 +30,6 @@ from .encoding import (
from .io import (
open_binary_file,
read_binary_file,
read_text_file,
)
@@ -163,13 +161,11 @@ def exclude_none_values(data): # type: (t.Dict[TKey, t.Optional[TValue]]) -> t.
return dict((key, value) for key, value in data.items() if value is not None)
def find_executable(executable, cwd=None, path=None, required=True):
def find_executable(executable, cwd=None, path=None, required=True): # type: (str, t.Optional[str], t.Optional[str], t.Union[bool, str]) -> t.Optional[str]
"""
:type executable: str
:type cwd: str
:type path: str
:type required: bool | str
:rtype: str | None
Find the specified executable and return the full path, or None if it could not be found.
If required is True an exception will be raised if the executable is not found.
If required is set to 'warning' then a warning will be shown if the executable is not found.
"""
match = None
real_cwd = os.getcwd()
@@ -215,12 +211,11 @@ def find_executable(executable, cwd=None, path=None, required=True):
return match
def find_python(version, path=None, required=True):
def find_python(version, path=None, required=True): # type: (str, t.Optional[str], bool) -> t.Optional[str]
"""
:type version: str
:type path: str | None
:type required: bool
:rtype: str
Find and return the full path to the specified Python version.
If required, an exception will be raised not found.
If not required, None will be returned if not found.
"""
version_info = str_to_version(version)
@@ -415,10 +410,8 @@ def pass_vars(required, optional):
return env
def remove_tree(path):
"""
:type path: str
"""
def remove_tree(path): # type: (str) -> None
"""Remove the specified directory, siliently continuing if the directory does not exist."""
try:
shutil.rmtree(to_bytes(path))
except OSError as ex:
@@ -426,11 +419,8 @@ def remove_tree(path):
raise
def is_binary_file(path):
"""
:type path: str
:rtype: bool
"""
def is_binary_file(path): # type: (str) -> bool
"""Return True if the specified file is a binary file, otherwise return False."""
assume_text = {
'.cfg',
'.conf',
@@ -491,10 +481,8 @@ def generate_name(length=8): # type: (int) -> str
return ''.join(random.choice(string.ascii_letters + string.digits) for _idx in range(length))
def generate_password():
"""Generate a random password.
:rtype: str
"""
def generate_password(): # type: () -> str
"""Generate and return random password."""
chars = [
string.ascii_letters,
string.digits,
@@ -542,13 +530,11 @@ class Display:
if os.isatty(0):
self.rows, self.columns = unpack('HHHH', fcntl.ioctl(0, TIOCGWINSZ, pack('HHHH', 0, 0, 0, 0)))[:2]
def __warning(self, message):
"""
:type message: str
"""
def __warning(self, message): # type: (str) -> None
"""Internal implementation for displaying a warning message."""
self.print_message('WARNING: %s' % message, color=self.purple, fd=sys.stderr)
def review_warnings(self):
def review_warnings(self): # type: () -> None
"""Review all warnings which previously occurred."""
if not self.warnings:
return
@@ -558,12 +544,8 @@ class Display:
for warning in self.warnings:
self.__warning(warning)
def warning(self, message, unique=False, verbosity=0):
"""
:type message: str
:type unique: bool
:type verbosity: int
"""
def warning(self, message, unique=False, verbosity=0): # type: (str, bool, int) -> None
"""Display a warning level message."""
if verbosity > self.verbosity:
return
@@ -576,24 +558,16 @@ class Display:
self.__warning(message)
self.warnings.append(message)
def notice(self, message):
"""
:type message: str
"""
def notice(self, message): # type: (str) -> None
"""Display a notice level message."""
self.print_message('NOTICE: %s' % message, color=self.purple, fd=sys.stderr)
def error(self, message):
"""
:type message: str
"""
def error(self, message): # type: (str) -> None
"""Display an error level message."""
self.print_message('ERROR: %s' % message, color=self.red, fd=sys.stderr)
def info(self, message, verbosity=0, truncate=False):
"""
:type message: str
:type verbosity: int
:type truncate: bool
"""
def info(self, message, verbosity=0, truncate=False): # type: (str, int, bool) -> None
"""Display an info level message."""
if self.verbosity >= verbosity:
color = self.verbosity_colors.get(verbosity, self.yellow)
self.print_message(message, color=color, fd=sys.stderr if self.info_stderr else sys.stdout, truncate=truncate)
@@ -674,10 +648,7 @@ class SubprocessError(ApplicationError):
class MissingEnvironmentVariable(ApplicationError):
"""Error caused by missing environment variable."""
def __init__(self, name):
"""
:type name: str
"""
def __init__(self, name): # type: (str) -> None
super().__init__('Missing environment variable: %s' % name)
self.name = name
@@ -694,12 +665,8 @@ def retry(func, ex_type=SubprocessError, sleep=10, attempts=10):
return func()
def parse_to_list_of_dict(pattern, value):
"""
:type pattern: str
:type value: str
:return: list[dict[str, str]]
"""
def parse_to_list_of_dict(pattern, value): # type: (str, str) -> t.List[t.Dict[str, str]]
"""Parse lines from the given value using the specified pattern and return the extracted list of key/value pair dictionaries."""
matched = []
unmatched = []
@@ -833,21 +800,6 @@ def sanitize_host_name(name):
return re.sub('[^A-Za-z0-9]+', '-', name)[:63].strip('-')
def get_hash(path):
"""
:type path: str
:rtype: str | None
"""
if not os.path.exists(path):
return None
file_hash = hashlib.sha256()
file_hash.update(read_binary_file(path))
return file_hash.hexdigest()
@cache
def get_host_ip():
"""Return the host's IP address."""
test/lib/ansible_test/_internal/util_common.py
+1 -5
View File
@@ -123,11 +123,7 @@ ResultType._populate() # pylint: disable=protected-access
class CommonConfig:
"""Configuration common to all commands."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
def __init__(self, args, command): # type: (t.Any, str) -> None
self.command = command
self.success = None # type: t.Optional[bool]