"""Test target identification, iteration and inclusion/exclusion.""" from __future__ import annotations import collections import collections.abc as c import enum import os import re import itertools import abc import typing as t from .encoding import ( to_bytes, to_text, ) from .io import ( read_text_file, ) from .util import ( ApplicationError, display, read_lines_without_comments, is_subdir, ) from .data import ( data_context, content_plugins, ) MODULE_EXTENSIONS = '.py', '.ps1' def find_target_completion(target_func: c.Callable[[], c.Iterable[CompletionTarget]], prefix: str, short: bool) -> list[str]: """Return a list of targets from the given target function which match the given prefix.""" try: targets = target_func() matches = list(walk_completion_targets(targets, prefix, short)) return matches except Exception as ex: # pylint: disable=locally-disabled, broad-except return ['%s' % ex] def walk_completion_targets(targets: c.Iterable[CompletionTarget], prefix: str, short: bool = False) -> tuple[str, ...]: """Return a tuple of targets from the given target iterable which match the given prefix.""" aliases = set(alias for target in targets for alias in target.aliases) if prefix.endswith('/') and prefix in aliases: aliases.remove(prefix) matches = [alias for alias in aliases if alias.startswith(prefix) and '/' not in alias[len(prefix):-1]] if short: offset = len(os.path.dirname(prefix)) if offset: offset += 1 relative_matches = [match[offset:] for match in matches if len(match) > offset] if len(relative_matches) > 1: matches = relative_matches return tuple(sorted(matches)) 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, ) -> tuple[TCompletionTarget, ...]: """Return a tuple of matching completion targets.""" targets = tuple(targets) include_targets = sorted(filter_targets(targets, includes), key=lambda include_target: include_target.name) if requires: require_targets = set(filter_targets(targets, requires)) include_targets = [require_target for require_target in include_targets if require_target in require_targets] if excludes: list(filter_targets(targets, excludes, include=False)) internal_targets = set(filter_targets(include_targets, excludes, errors=False, include=False)) return tuple(sorted(internal_targets, key=lambda sort_target: sort_target.name)) def filter_targets(targets: c.Iterable[TCompletionTarget], patterns: list[str], include: bool = True, errors: bool = True, ) -> c.Iterable[TCompletionTarget]: """Iterate over the given targets and filter them based on the supplied arguments.""" unmatched = set(patterns or ()) compiled_patterns = dict((p, re.compile('^%s$' % p)) for p in patterns) if patterns else None for target in targets: matched_directories = set() match = False if patterns: for alias in target.aliases: for pattern in patterns: if compiled_patterns[pattern].match(alias): match = True try: unmatched.remove(pattern) except KeyError: pass if alias.endswith('/'): if target.base_path and len(target.base_path) > len(alias): matched_directories.add(target.base_path) else: matched_directories.add(alias) elif include: match = True if not target.base_path: matched_directories.add('.') for alias in target.aliases: if alias.endswith('/'): if target.base_path and len(target.base_path) > len(alias): matched_directories.add(target.base_path) else: matched_directories.add(alias) if match != include: continue yield target if errors: if unmatched: raise TargetPatternsNotMatched(unmatched) def walk_module_targets() -> c.Iterable[TestTarget]: """Iterate through the module test targets.""" for target in walk_test_targets(path=data_context().content.module_path, module_path=data_context().content.module_path, extensions=MODULE_EXTENSIONS): if not target.module: continue yield target def walk_units_targets() -> c.Iterable[TestTarget]: """Return an iterable of units targets.""" return walk_test_targets(path=data_context().content.unit_path, module_path=data_context().content.unit_module_path, extensions=('.py',), prefix='test_') def walk_compile_targets(include_symlinks: bool = True) -> c.Iterable[TestTarget]: """Return an iterable of compile targets.""" return walk_test_targets(module_path=data_context().content.module_path, extensions=('.py',), extra_dirs=('bin',), include_symlinks=include_symlinks) def walk_powershell_targets(include_symlinks: bool = True) -> c.Iterable[TestTarget]: """Return an iterable of PowerShell targets.""" return walk_test_targets(module_path=data_context().content.module_path, extensions=('.ps1', '.psm1'), include_symlinks=include_symlinks) def walk_sanity_targets() -> c.Iterable[TestTarget]: """Return an iterable of sanity targets.""" return walk_test_targets(module_path=data_context().content.module_path, include_symlinks=True, include_symlinked_directories=True) def walk_posix_integration_targets(include_hidden: bool = False) -> c.Iterable[IntegrationTarget]: """Return an iterable of POSIX integration targets.""" for target in walk_integration_targets(): if 'posix/' in target.aliases or (include_hidden and 'hidden/posix/' in target.aliases): yield target def walk_network_integration_targets(include_hidden: bool = False) -> c.Iterable[IntegrationTarget]: """Return an iterable of network integration targets.""" for target in walk_integration_targets(): if 'network/' in target.aliases or (include_hidden and 'hidden/network/' in target.aliases): yield target def walk_windows_integration_targets(include_hidden: bool = False) -> c.Iterable[IntegrationTarget]: """Return an iterable of windows integration targets.""" for target in walk_integration_targets(): if 'windows/' in target.aliases or (include_hidden and 'hidden/windows/' in target.aliases): yield target def walk_integration_targets() -> c.Iterable[IntegrationTarget]: """Return an iterable of integration targets.""" path = data_context().content.integration_targets_path modules = frozenset(target.module for target in walk_module_targets()) paths = data_context().content.walk_files(path) prefixes = load_integration_prefixes() targets_path_tuple = tuple(path.split(os.path.sep)) entry_dirs = ( 'defaults', 'files', 'handlers', 'meta', 'tasks', 'templates', 'vars', ) entry_files = ( 'main.yml', 'main.yaml', ) entry_points = [] for entry_dir in entry_dirs: for entry_file in entry_files: entry_points.append(os.path.join(os.path.sep, entry_dir, entry_file)) # any directory with at least one file is a target path_tuples = set(tuple(os.path.dirname(p).split(os.path.sep)) for p in paths) # also detect targets which are ansible roles, looking for standard entry points path_tuples.update(tuple(os.path.dirname(os.path.dirname(p)).split(os.path.sep)) for p in paths if any(p.endswith(entry_point) for entry_point in entry_points)) # remove the top-level directory if it was included if targets_path_tuple in path_tuples: path_tuples.remove(targets_path_tuple) previous_path_tuple = None paths = [] for path_tuple in sorted(path_tuples): if previous_path_tuple and previous_path_tuple == path_tuple[:len(previous_path_tuple)]: # ignore nested directories continue previous_path_tuple = path_tuple paths.append(os.path.sep.join(path_tuple)) for path in paths: yield IntegrationTarget(to_text(path), modules, prefixes) def load_integration_prefixes() -> dict[str, str]: """Load and return the integration test prefixes.""" path = data_context().content.integration_path file_paths = sorted(f for f in data_context().content.get_files(path) if os.path.splitext(os.path.basename(f))[0] == 'target-prefixes') prefixes = {} for file_path in file_paths: prefix = os.path.splitext(file_path)[1][1:] prefixes.update(dict((k, prefix) for k in read_text_file(file_path).splitlines())) return prefixes 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, ) -> c.Iterable[TestTarget]: """Iterate over available test targets.""" if path: file_paths = data_context().content.walk_files(path, include_symlinked_directories=include_symlinked_directories) else: file_paths = data_context().content.all_files(include_symlinked_directories=include_symlinked_directories) for file_path in file_paths: name, ext = os.path.splitext(os.path.basename(file_path)) if extensions and ext not in extensions: continue if prefix and not name.startswith(prefix): continue symlink = os.path.islink(to_bytes(file_path.rstrip(os.path.sep))) if symlink and not include_symlinks: continue yield TestTarget(to_text(file_path), module_path, prefix, path, symlink) file_paths = [] if extra_dirs: for extra_dir in extra_dirs: for file_path in data_context().content.get_files(extra_dir): file_paths.append(file_path) for file_path in file_paths: symlink = os.path.islink(to_bytes(file_path.rstrip(os.path.sep))) if symlink and not include_symlinks: continue yield TestTarget(file_path, module_path, prefix, path, symlink) def analyze_integration_target_dependencies(integration_targets: list[IntegrationTarget]) -> dict[str, set[str]]: """Analyze the given list of integration test targets and return a dictionary expressing target names and the target names which depend on them.""" real_target_root = os.path.realpath(data_context().content.integration_targets_path) + '/' role_targets = [target for target in integration_targets if target.type == 'role'] hidden_role_target_names = set(target.name for target in role_targets if 'hidden/' in target.aliases) dependencies: collections.defaultdict[str, set[str]] = collections.defaultdict(set) # handle setup dependencies for target in integration_targets: for setup_target_name in target.setup_always + target.setup_once: dependencies[setup_target_name].add(target.name) # handle target dependencies for target in integration_targets: for need_target in target.needs_target: dependencies[need_target].add(target.name) # handle symlink dependencies between targets # this use case is supported, but discouraged for target in integration_targets: for path in data_context().content.walk_files(target.path): if not os.path.islink(to_bytes(path.rstrip(os.path.sep))): continue real_link_path = os.path.realpath(path) if not real_link_path.startswith(real_target_root): continue link_target = real_link_path[len(real_target_root):].split('/')[0] if link_target == target.name: continue dependencies[link_target].add(target.name) # intentionally primitive analysis of role meta to avoid a dependency on pyyaml # script based targets are scanned as they may execute a playbook with role dependencies for target in integration_targets: meta_dir = os.path.join(target.path, 'meta') if not os.path.isdir(meta_dir): continue meta_paths = data_context().content.get_files(meta_dir) for meta_path in meta_paths: if os.path.exists(meta_path): # try and decode the file as a utf-8 string, skip if it contains invalid chars (binary file) try: meta_lines = read_text_file(meta_path).splitlines() except UnicodeDecodeError: continue for meta_line in meta_lines: if re.search(r'^ *#.*$', meta_line): continue if not meta_line.strip(): continue for hidden_target_name in hidden_role_target_names: if hidden_target_name in meta_line: dependencies[hidden_target_name].add(target.name) while True: changes = 0 for dummy, dependent_target_names in dependencies.items(): for dependent_target_name in list(dependent_target_names): new_target_names = dependencies.get(dependent_target_name) if new_target_names: for new_target_name in new_target_names: if new_target_name not in dependent_target_names: dependent_target_names.add(new_target_name) changes += 1 if not changes: break for target_name in sorted(dependencies): consumers = dependencies[target_name] if not consumers: continue display.info('%s:' % target_name, verbosity=4) for consumer in sorted(consumers): display.info(' %s' % consumer, verbosity=4) return dependencies class CompletionTarget(metaclass=abc.ABCMeta): """Command-line argument completion target base class.""" def __init__(self) -> None: self.name = '' self.path = '' self.base_path: t.Optional[str] = None self.modules: tuple[str, ...] = tuple() self.aliases: tuple[str, ...] = tuple() def __eq__(self, other): if isinstance(other, CompletionTarget): return self.__repr__() == other.__repr__() return False def __ne__(self, other): return not self.__eq__(other) def __lt__(self, other): return self.name.__lt__(other.name) def __gt__(self, other): return self.name.__gt__(other.name) def __hash__(self): return hash(self.__repr__()) def __repr__(self): if self.modules: return '%s (%s)' % (self.name, ', '.join(self.modules)) return self.name 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, ) -> None: super().__init__() if symlink is None: symlink = os.path.islink(to_bytes(path.rstrip(os.path.sep))) self.name = path self.path = path self.base_path = base_path + '/' if base_path else None self.symlink = symlink name, ext = os.path.splitext(os.path.basename(self.path)) if module_path and is_subdir(path, module_path) and name != '__init__' and ext in MODULE_EXTENSIONS: self.module = name[len(module_prefix or ''):].lstrip('_') self.modules = (self.module,) else: self.module = None self.modules = tuple() aliases = [self.path, self.module] parts = self.path.split('/') for i in range(1, len(parts)): alias = '%s/' % '/'.join(parts[:i]) aliases.append(alias) aliases = [a for a in aliases if a] self.aliases = tuple(sorted(aliases)) class IntegrationTargetType(enum.Enum): """Type of integration test target.""" CONTROLLER = enum.auto() TARGET = enum.auto() UNKNOWN = enum.auto() CONFLICT = enum.auto() def extract_plugin_references(name: str, aliases: list[str]) -> list[tuple[str, str]]: """Return a list of plugin references found in the given integration test target name and aliases.""" plugins = content_plugins() found: list[tuple[str, str]] = [] for alias in [name] + aliases: plugin_type = 'modules' plugin_name = alias if plugin_name in plugins.get(plugin_type, {}): found.append((plugin_type, plugin_name)) parts = alias.split('_') for type_length in (1, 2): if len(parts) > type_length: plugin_type = '_'.join(parts[:type_length]) plugin_name = '_'.join(parts[type_length:]) if plugin_name in plugins.get(plugin_type, {}): found.append((plugin_type, plugin_name)) return found def categorize_integration_test(name: str, aliases: list[str], force_target: bool) -> tuple[IntegrationTargetType, IntegrationTargetType]: """Return the integration test target types (used and actual) based on the given target name and aliases.""" context_controller = f'context/{IntegrationTargetType.CONTROLLER.name.lower()}' in aliases context_target = f'context/{IntegrationTargetType.TARGET.name.lower()}' in aliases or force_target actual_type = None strict_mode = data_context().content.is_ansible if context_controller and context_target: target_type = IntegrationTargetType.CONFLICT elif context_controller and not context_target: target_type = IntegrationTargetType.CONTROLLER elif context_target and not context_controller: target_type = IntegrationTargetType.TARGET else: target_types = {IntegrationTargetType.TARGET if plugin_type in ('modules', 'module_utils') else IntegrationTargetType.CONTROLLER for plugin_type, plugin_name in extract_plugin_references(name, aliases)} if len(target_types) == 1: target_type = target_types.pop() elif not target_types: actual_type = IntegrationTargetType.UNKNOWN target_type = actual_type if strict_mode else IntegrationTargetType.TARGET else: target_type = IntegrationTargetType.CONFLICT return target_type, actual_type or target_type class IntegrationTarget(CompletionTarget): """Integration test target.""" non_posix = frozenset(( 'network', 'windows', )) categories = frozenset(non_posix | frozenset(( 'posix', 'module', 'needs', 'skip', ))) def __init__(self, path: str, modules: frozenset[str], prefixes: dict[str, str]) -> None: super().__init__() self.relative_path = os.path.relpath(path, data_context().content.integration_targets_path) self.name = self.relative_path.replace(os.path.sep, '.') self.path = path # script_path and type file_paths = data_context().content.get_files(path) runme_path = os.path.join(path, 'runme.sh') if runme_path in file_paths: self.type = 'script' self.script_path = runme_path else: self.type = 'role' # ansible will consider these empty roles, so ansible-test should as well self.script_path = None # static_aliases aliases_path = os.path.join(path, 'aliases') if aliases_path in file_paths: static_aliases = tuple(read_lines_without_comments(aliases_path, remove_blank_lines=True)) else: static_aliases = tuple() # modules if self.name in modules: module_name = self.name elif self.name.startswith('win_') and self.name[4:] in modules: module_name = self.name[4:] else: module_name = None self.modules = tuple(sorted(a for a in static_aliases + tuple([module_name]) if a in modules)) # groups groups = [self.type] groups += [a for a in static_aliases if a not in modules] groups += ['module/%s' % m for m in self.modules] if data_context().content.is_ansible and (self.name == 'ansible-test' or self.name.startswith('ansible-test-')): groups.append('ansible-test') if not self.modules: groups.append('non_module') if 'destructive' not in groups: groups.append('non_destructive') if 'needs/httptester' in groups: groups.append('cloud/httptester') # backwards compatibility for when it was not a cloud plugin if '_' in self.name: prefix = self.name[:self.name.find('_')] else: prefix = None if prefix in prefixes: group = prefixes[prefix] if group != prefix: group = '%s/%s' % (group, prefix) groups.append(group) if self.name.startswith('win_'): groups.append('windows') if self.name.startswith('connection_'): groups.append('connection') if self.name.startswith('setup_') or self.name.startswith('prepare_'): groups.append('hidden') if self.type not in ('script', 'role'): groups.append('hidden') targets_relative_path = data_context().content.integration_targets_path # Collect skip entries before group expansion to avoid registering more specific skip entries as less specific versions. self.skips = tuple(g for g in groups if g.startswith('skip/')) # Collect file paths before group expansion to avoid including the directories. # Ignore references to test targets, as those must be defined using `needs/target/*` or other target references. self.needs_file = tuple(sorted(set('/'.join(g.split('/')[2:]) for g in groups if g.startswith('needs/file/') and not g.startswith('needs/file/%s/' % targets_relative_path)))) # network platform networks = [g.split('/')[1] for g in groups if g.startswith('network/')] self.network_platform = networks[0] if networks else None for group in itertools.islice(groups, 0, len(groups)): if '/' in group: parts = group.split('/') for i in range(1, len(parts)): groups.append('/'.join(parts[:i])) if not any(g in self.non_posix for g in groups): groups.append('posix') # target type # targets which are non-posix test against the target, even if they also support posix force_target = any(group in self.non_posix for group in groups) target_type, actual_type = categorize_integration_test(self.name, list(static_aliases), force_target) groups.extend(['context/', f'context/{target_type.name.lower()}']) if target_type != actual_type: # allow users to query for the actual type groups.extend(['context/', f'context/{actual_type.name.lower()}']) self.target_type = target_type self.actual_type = actual_type # aliases aliases = [self.name] + \ ['%s/' % g for g in groups] + \ ['%s/%s' % (g, self.name) for g in groups if g not in self.categories] if 'hidden/' in aliases: aliases = ['hidden/'] + ['hidden/%s' % a for a in aliases if not a.startswith('hidden/')] self.aliases = tuple(sorted(set(aliases))) # configuration self.retry_never = 'retry/never/' in self.aliases self.setup_once = tuple(sorted(set(g.split('/')[2] for g in groups if g.startswith('setup/once/')))) self.setup_always = tuple(sorted(set(g.split('/')[2] for g in groups if g.startswith('setup/always/')))) self.needs_target = tuple(sorted(set(g.split('/')[2] for g in groups if g.startswith('needs/target/')))) class TargetPatternsNotMatched(ApplicationError): """One or more targets were not matched when a match was required.""" def __init__(self, patterns: set[str]) -> None: self.patterns = sorted(patterns) if len(patterns) > 1: message = 'Target patterns not matched:\n%s' % '\n'.join(self.patterns) else: message = 'Target pattern not matched: %s' % self.patterns[0] super().__init__(message) TCompletionTarget = t.TypeVar('TCompletionTarget', bound=CompletionTarget) TIntegrationTarget = t.TypeVar('TIntegrationTarget', bound=IntegrationTarget)