KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _pathctx method in localstack

Best Python code snippet using localstack_python
_collection_finder.py
Source:_collection_finder.py
1# (c) 2019 Ansible Project2# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)3from __future__ import (absolute_import, division, print_function)4__metaclass__ = type5import os6import os.path7import pkgutil8import re9import sys10# DO NOT add new non-stdlib import deps here, this loader is used by external tools (eg ansible-test import sanity)11# that only allow stdlib and module_utils12from ansible.module_utils.common.text.converters import to_native, to_text, to_bytes13from ansible.module_utils.six import string_types, PY314from ._collection_config import AnsibleCollectionConfig15from contextlib import contextmanager16from types import ModuleType17try:18    from importlib import import_module19except ImportError:20    def import_module(name):21        __import__(name)22        return sys.modules[name]23try:24    from importlib import reload as reload_module25except ImportError:26    # 2.7 has a global reload function instead...27    reload_module = reload  # pylint:disable=undefined-variable28# NB: this supports import sanity test providing a different impl29try:30    from ._collection_meta import _meta_yml_to_dict31except ImportError:32    _meta_yml_to_dict = None33class _AnsibleCollectionFinder:34    def __init__(self, paths=None, scan_sys_paths=True):35        # TODO: accept metadata loader override36        self._ansible_pkg_path = to_native(os.path.dirname(to_bytes(sys.modules['ansible'].__file__)))37        if isinstance(paths, string_types):38            paths = [paths]39        elif paths is None:40            paths = []41        # expand any placeholders in configured paths42        paths = [os.path.expanduser(to_native(p, errors='surrogate_or_strict')) for p in paths]43        if scan_sys_paths:44            # append all sys.path entries with an ansible_collections package45            for path in sys.path:46                if (47                        path not in paths and48                        os.path.isdir(to_bytes(49                            os.path.join(path, 'ansible_collections'),50                            errors='surrogate_or_strict',51                        ))52                ):53                    paths.append(path)54        self._n_configured_paths = paths55        self._n_cached_collection_paths = None56        self._n_cached_collection_qualified_paths = None57        self._n_playbook_paths = []58    @classmethod59    def _remove(cls):60        for mps in sys.meta_path:61            if isinstance(mps, _AnsibleCollectionFinder):62                sys.meta_path.remove(mps)63        # remove any path hooks that look like ours64        for ph in sys.path_hooks:65            if hasattr(ph, '__self__') and isinstance(ph.__self__, _AnsibleCollectionFinder):66                sys.path_hooks.remove(ph)67        # zap any cached path importer cache entries that might refer to us68        sys.path_importer_cache.clear()69        AnsibleCollectionConfig._collection_finder = None70        # validate via the public property that we really killed it71        if AnsibleCollectionConfig.collection_finder is not None:72            raise AssertionError('_AnsibleCollectionFinder remove did not reset AnsibleCollectionConfig.collection_finder')73    def _install(self):74        self._remove()75        sys.meta_path.insert(0, self)76        sys.path_hooks.insert(0, self._ansible_collection_path_hook)77        AnsibleCollectionConfig.collection_finder = self78    def _ansible_collection_path_hook(self, path):79        path = to_native(path)80        interesting_paths = self._n_cached_collection_qualified_paths81        if not interesting_paths:82            interesting_paths = [os.path.join(p, 'ansible_collections') for p in83                                 self._n_collection_paths]84            interesting_paths.insert(0, self._ansible_pkg_path)85            self._n_cached_collection_qualified_paths = interesting_paths86        if any(path.startswith(p) for p in interesting_paths):87            return _AnsiblePathHookFinder(self, path)88        raise ImportError('not interested')89    @property90    def _n_collection_paths(self):91        paths = self._n_cached_collection_paths92        if not paths:93            self._n_cached_collection_paths = paths = self._n_playbook_paths + self._n_configured_paths94        return paths95    def set_playbook_paths(self, playbook_paths):96        if isinstance(playbook_paths, string_types):97            playbook_paths = [playbook_paths]98        # track visited paths; we have to preserve the dir order as-passed in case there are duplicate collections (first one wins)99        added_paths = set()100        # de-dupe101        self._n_playbook_paths = [os.path.join(to_native(p), 'collections') for p in playbook_paths if not (p in added_paths or added_paths.add(p))]102        self._n_cached_collection_paths = None103        # HACK: playbook CLI sets this relatively late, so we've already loaded some packages whose paths might depend on this. Fix those up.104        # NB: this should NOT be used for late additions; ideally we'd fix the playbook dir setup earlier in Ansible init105        # to prevent this from occurring106        for pkg in ['ansible_collections', 'ansible_collections.ansible']:107            self._reload_hack(pkg)108    def _reload_hack(self, fullname):109        m = sys.modules.get(fullname)110        if not m:111            return112        reload_module(m)113    def find_module(self, fullname, path=None):114        # Figure out what's being asked for, and delegate to a special-purpose loader115        split_name = fullname.split('.')116        toplevel_pkg = split_name[0]117        module_to_find = split_name[-1]118        part_count = len(split_name)119        if toplevel_pkg not in ['ansible', 'ansible_collections']:120            # not interested in anything other than ansible_collections (and limited cases under ansible)121            return None122        # sanity check what we're getting from import, canonicalize path values123        if part_count == 1:124            if path:125                raise ValueError('path should not be specified for top-level packages (trying to find {0})'.format(fullname))126            else:127                # seed the path to the configured collection roots128                path = self._n_collection_paths129        if part_count > 1 and path is None:130            raise ValueError('path must be specified for subpackages (trying to find {0})'.format(fullname))131        # NB: actual "find"ing is delegated to the constructors on the various loaders; they'll ImportError if not found132        try:133            if toplevel_pkg == 'ansible':134                # something under the ansible package, delegate to our internal loader in case of redirections135                return _AnsibleInternalRedirectLoader(fullname=fullname, path_list=path)136            if part_count == 1:137                return _AnsibleCollectionRootPkgLoader(fullname=fullname, path_list=path)138            if part_count == 2:  # ns pkg eg, ansible_collections, ansible_collections.somens139                return _AnsibleCollectionNSPkgLoader(fullname=fullname, path_list=path)140            elif part_count == 3:  # collection pkg eg, ansible_collections.somens.somecoll141                return _AnsibleCollectionPkgLoader(fullname=fullname, path_list=path)142            # anything below the collection143            return _AnsibleCollectionLoader(fullname=fullname, path_list=path)144        except ImportError:145            # TODO: log attempt to load context146            return None147# Implements a path_hook finder for iter_modules (since it's only path based). This finder does not need to actually148# function as a finder in most cases, since our meta_path finder is consulted first for *almost* everything, except149# pkgutil.iter_modules, and under py2, pkgutil.get_data if the parent package passed has not been loaded yet.150class _AnsiblePathHookFinder:151    def __init__(self, collection_finder, pathctx):152        # when called from a path_hook, find_module doesn't usually get the path arg, so this provides our context153        self._pathctx = to_native(pathctx)154        self._collection_finder = collection_finder155        if PY3:156            # cache the native FileFinder (take advantage of its filesystem cache for future find/load requests)157            self._file_finder = None158    # class init is fun- this method has a self arg that won't get used159    def _get_filefinder_path_hook(self=None):160        _file_finder_hook = None161        if PY3:162            # try to find the FileFinder hook to call for fallback path-based imports in Py3163            _file_finder_hook = [ph for ph in sys.path_hooks if 'FileFinder' in repr(ph)]164            if len(_file_finder_hook) != 1:165                raise Exception('need exactly one FileFinder import hook (found {0})'.format(len(_file_finder_hook)))166            _file_finder_hook = _file_finder_hook[0]167        return _file_finder_hook168    _filefinder_path_hook = _get_filefinder_path_hook()169    def find_module(self, fullname, path=None):170        # we ignore the passed in path here- use what we got from the path hook init171        split_name = fullname.split('.')172        toplevel_pkg = split_name[0]173        if toplevel_pkg == 'ansible_collections':174            # collections content? delegate to the collection finder175            return self._collection_finder.find_module(fullname, path=[self._pathctx])176        else:177            # Something else; we'd normally restrict this to `ansible` descendent modules so that any weird loader178            # behavior that arbitrary Python modules have can be serviced by those loaders. In some dev/test179            # scenarios (eg a venv under a collection) our path_hook signs us up to load non-Ansible things, and180            # it's too late by the time we've reached this point, but also too expensive for the path_hook to figure181            # out what we *shouldn't* be loading with the limited info it has. So we'll just delegate to the182            # normal path-based loader as best we can to service it. This also allows us to take advantage of Python's183            # built-in FS caching and byte-compilation for most things.184            if PY3:185                # create or consult our cached file finder for this path186                if not self._file_finder:187                    try:188                        self._file_finder = _AnsiblePathHookFinder._filefinder_path_hook(self._pathctx)189                    except ImportError:190                        # FUTURE: log at a high logging level? This is normal for things like python36.zip on the path, but191                        # might not be in some other situation...192                        return None193                spec = self._file_finder.find_spec(fullname)194                if not spec:195                    return None196                return spec.loader197            else:198                # call py2's internal loader199                return pkgutil.ImpImporter(self._pathctx).find_module(fullname)200    def iter_modules(self, prefix):201        # NB: this currently represents only what's on disk, and does not handle package redirection202        return _iter_modules_impl([self._pathctx], prefix)203    def __repr__(self):204        return "{0}(path='{1}')".format(self.__class__.__name__, self._pathctx)205class _AnsibleCollectionPkgLoaderBase:206    _allows_package_code = False207    def __init__(self, fullname, path_list=None):208        self._fullname = fullname209        self._redirect_module = None210        self._split_name = fullname.split('.')211        self._rpart_name = fullname.rpartition('.')212        self._parent_package_name = self._rpart_name[0]  # eg ansible_collections for ansible_collections.somens, '' for toplevel213        self._package_to_load = self._rpart_name[2]  # eg somens for ansible_collections.somens214        self._source_code_path = None215        self._decoded_source = None216        self._compiled_code = None217        self._validate_args()218        self._candidate_paths = self._get_candidate_paths([to_native(p) for p in path_list])219        self._subpackage_search_paths = self._get_subpackage_search_paths(self._candidate_paths)220        self._validate_final()221    # allow subclasses to validate args and sniff split values before we start digging around222    def _validate_args(self):223        if self._split_name[0] != 'ansible_collections':224            raise ImportError('this loader can only load packages from the ansible_collections package, not {0}'.format(self._fullname))225    # allow subclasses to customize candidate path filtering226    def _get_candidate_paths(self, path_list):227        return [os.path.join(p, self._package_to_load) for p in path_list]228    # allow subclasses to customize finding paths229    def _get_subpackage_search_paths(self, candidate_paths):230        # filter candidate paths for existence (NB: silently ignoring package init code and same-named modules)231        return [p for p in candidate_paths if os.path.isdir(to_bytes(p))]232    # allow subclasses to customize state validation/manipulation before we return the loader instance233    def _validate_final(self):234        return235    @staticmethod236    @contextmanager237    def _new_or_existing_module(name, **kwargs):238        # handle all-or-nothing sys.modules creation/use-existing/delete-on-exception-if-created behavior239        created_module = False240        module = sys.modules.get(name)241        try:242            if not module:243                module = ModuleType(name)244                created_module = True245                sys.modules[name] = module246            # always override the values passed, except name (allow reference aliasing)247            for attr, value in kwargs.items():248                setattr(module, attr, value)249            yield module250        except Exception:251            if created_module:252                if sys.modules.get(name):253                    sys.modules.pop(name)254            raise255    # basic module/package location support256    # NB: this does not support distributed packages!257    @staticmethod258    def _module_file_from_path(leaf_name, path):259        has_code = True260        package_path = os.path.join(to_native(path), to_native(leaf_name))261        module_path = None262        # if the submodule is a package, assemble valid submodule paths, but stop looking for a module263        if os.path.isdir(to_bytes(package_path)):264            # is there a package init?265            module_path = os.path.join(package_path, '__init__.py')266            if not os.path.isfile(to_bytes(module_path)):267                module_path = os.path.join(package_path, '__synthetic__')268                has_code = False269        else:270            module_path = package_path + '.py'271            package_path = None272            if not os.path.isfile(to_bytes(module_path)):273                raise ImportError('{0} not found at {1}'.format(leaf_name, path))274        return module_path, has_code, package_path275    def load_module(self, fullname):276        # short-circuit redirect; we've already imported the redirected module, so just alias it and return it277        if self._redirect_module:278            sys.modules[self._fullname] = self._redirect_module279            return self._redirect_module280        # we're actually loading a module/package281        module_attrs = dict(282            __loader__=self,283            __file__=self.get_filename(fullname),284            __package__=self._parent_package_name  # sane default for non-packages285        )286        # eg, I am a package287        if self._subpackage_search_paths is not None:  # empty is legal288            module_attrs['__path__'] = self._subpackage_search_paths289            module_attrs['__package__'] = fullname  # per PEP366290        with self._new_or_existing_module(fullname, **module_attrs) as module:291            # execute the module's code in its namespace292            code_obj = self.get_code(fullname)293            if code_obj is not None:  # things like NS packages that can't have code on disk will return None294                exec(code_obj, module.__dict__)295            return module296    def is_package(self, fullname):297        if fullname != self._fullname:298            raise ValueError('this loader cannot answer is_package for {0}, only {1}'.format(fullname, self._fullname))299        return self._subpackage_search_paths is not None300    def get_source(self, fullname):301        if self._decoded_source:302            return self._decoded_source303        if fullname != self._fullname:304            raise ValueError('this loader cannot load source for {0}, only {1}'.format(fullname, self._fullname))305        if not self._source_code_path:306            return None307        # FIXME: what do we want encoding/newline requirements to be?308        self._decoded_source = self.get_data(self._source_code_path)309        return self._decoded_source310    def get_data(self, path):311        if not path:312            raise ValueError('a path must be specified')313        # TODO: ensure we're being asked for a path below something we own314        # TODO: try to handle redirects internally?315        if not path[0] == '/':316            # relative to current package, search package paths if possible (this may not be necessary)317            # candidate_paths = [os.path.join(ssp, path) for ssp in self._subpackage_search_paths]318            raise ValueError('relative resource paths not supported')319        else:320            candidate_paths = [path]321        for p in candidate_paths:322            b_path = to_bytes(p)323            if os.path.isfile(b_path):324                with open(b_path, 'rb') as fd:325                    return fd.read()326            # HACK: if caller asks for __init__.py and the parent dir exists, return empty string (this keep consistency327            # with "collection subpackages don't require __init__.py" working everywhere with get_data328            elif b_path.endswith(b'__init__.py') and os.path.isdir(os.path.dirname(b_path)):329                return ''330        return None331    def _synthetic_filename(self, fullname):332        return '<ansible_synthetic_collection_package>'333    def get_filename(self, fullname):334        if fullname != self._fullname:335            raise ValueError('this loader cannot find files for {0}, only {1}'.format(fullname, self._fullname))336        filename = self._source_code_path337        if not filename and self.is_package(fullname):338            if len(self._subpackage_search_paths) == 1:339                filename = os.path.join(self._subpackage_search_paths[0], '__synthetic__')340            else:341                filename = self._synthetic_filename(fullname)342        return filename343    def get_code(self, fullname):344        if self._compiled_code:345            return self._compiled_code346        # this may or may not be an actual filename, but it's the value we'll use for __file__347        filename = self.get_filename(fullname)348        if not filename:349            filename = '<string>'350        source_code = self.get_source(fullname)351        # for things like synthetic modules that really have no source on disk, don't return a code object at all352        # vs things like an empty package init (which has an empty string source on disk)353        if source_code is None:354            return None355        self._compiled_code = compile(source=source_code, filename=filename, mode='exec', flags=0, dont_inherit=True)356        return self._compiled_code357    def iter_modules(self, prefix):358        return _iter_modules_impl(self._subpackage_search_paths, prefix)359    def __repr__(self):360        return '{0}(path={1})'.format(self.__class__.__name__, self._subpackage_search_paths or self._source_code_path)361class _AnsibleCollectionRootPkgLoader(_AnsibleCollectionPkgLoaderBase):362    def _validate_args(self):363        super(_AnsibleCollectionRootPkgLoader, self)._validate_args()364        if len(self._split_name) != 1:365            raise ImportError('this loader can only load the ansible_collections toplevel package, not {0}'.format(self._fullname))366# Implements Ansible's custom namespace package support.367# The ansible_collections package and one level down (collections namespaces) are Python namespace packages368# that search across all configured collection roots. The collection package (two levels down) is the first one found369# on the configured collection root path, and Python namespace package aggregation is not allowed at or below370# the collection. Implements implicit package (package dir) support for both Py2/3. Package init code is ignored371# by this loader.372class _AnsibleCollectionNSPkgLoader(_AnsibleCollectionPkgLoaderBase):373    def _validate_args(self):374        super(_AnsibleCollectionNSPkgLoader, self)._validate_args()375        if len(self._split_name) != 2:376            raise ImportError('this loader can only load collections namespace packages, not {0}'.format(self._fullname))377    def _validate_final(self):378        # special-case the `ansible` namespace, since `ansible.builtin` is magical379        if not self._subpackage_search_paths and self._package_to_load != 'ansible':380            raise ImportError('no {0} found in {1}'.format(self._package_to_load, self._candidate_paths))381# handles locating the actual collection package and associated metadata382class _AnsibleCollectionPkgLoader(_AnsibleCollectionPkgLoaderBase):383    def _validate_args(self):384        super(_AnsibleCollectionPkgLoader, self)._validate_args()385        if len(self._split_name) != 3:386            raise ImportError('this loader can only load collection packages, not {0}'.format(self._fullname))387    def _validate_final(self):388        if self._split_name[1:3] == ['ansible', 'builtin']:389            # we don't want to allow this one to have on-disk search capability390            self._subpackage_search_paths = []391        elif not self._subpackage_search_paths:392            raise ImportError('no {0} found in {1}'.format(self._package_to_load, self._candidate_paths))393        else:394            # only search within the first collection we found395            self._subpackage_search_paths = [self._subpackage_search_paths[0]]396    def load_module(self, fullname):397        if not _meta_yml_to_dict:398            raise ValueError('ansible.utils.collection_loader._meta_yml_to_dict is not set')399        module = super(_AnsibleCollectionPkgLoader, self).load_module(fullname)400        module._collection_meta = {}401        # TODO: load collection metadata, cache in __loader__ state402        collection_name = '.'.join(self._split_name[1:3])403        if collection_name == 'ansible.builtin':404            # ansible.builtin is a synthetic collection, get its routing config from the Ansible distro405            ansible_pkg_path = os.path.dirname(import_module('ansible').__file__)406            metadata_path = os.path.join(ansible_pkg_path, 'config/ansible_builtin_runtime.yml')407            with open(to_bytes(metadata_path), 'rb') as fd:408                raw_routing = fd.read()409        else:410            b_routing_meta_path = to_bytes(os.path.join(module.__path__[0], 'meta/runtime.yml'))411            if os.path.isfile(b_routing_meta_path):412                with open(b_routing_meta_path, 'rb') as fd:413                    raw_routing = fd.read()414            else:415                raw_routing = ''416        try:417            if raw_routing:418                routing_dict = _meta_yml_to_dict(raw_routing, (collection_name, 'runtime.yml'))419                module._collection_meta = self._canonicalize_meta(routing_dict)420        except Exception as ex:421            raise ValueError('error parsing collection metadata: {0}'.format(to_native(ex)))422        AnsibleCollectionConfig.on_collection_load.fire(collection_name=collection_name, collection_path=os.path.dirname(module.__file__))423        return module424    def _canonicalize_meta(self, meta_dict):425        # TODO: rewrite import keys and all redirect targets that start with .. (current namespace) and . (current collection)426        # OR we could do it all on the fly?427        # if not meta_dict:428        #     return {}429        #430        # ns_name = '.'.join(self._split_name[0:2])431        # collection_name = '.'.join(self._split_name[0:3])432        #433        # #434        # for routing_type, routing_type_dict in iteritems(meta_dict.get('plugin_routing', {})):435        #     for plugin_key, plugin_dict in iteritems(routing_type_dict):436        #         redirect = plugin_dict.get('redirect', '')437        #         if redirect.startswith('..'):438        #             redirect =  redirect[2:]439        action_groups = meta_dict.pop('action_groups', {})440        meta_dict['action_groups'] = {}441        for group_name in action_groups:442            for action_name in action_groups[group_name]:443                if action_name in meta_dict['action_groups']:444                    meta_dict['action_groups'][action_name].append(group_name)445                else:446                    meta_dict['action_groups'][action_name] = [group_name]447        return meta_dict448# loads everything under a collection, including handling redirections defined by the collection449class _AnsibleCollectionLoader(_AnsibleCollectionPkgLoaderBase):450    # HACK: stash this in a better place451    _redirected_package_map = {}452    _allows_package_code = True453    def _validate_args(self):454        super(_AnsibleCollectionLoader, self)._validate_args()455        if len(self._split_name) < 4:456            raise ValueError('this loader is only for sub-collection modules/packages, not {0}'.format(self._fullname))457    def _get_candidate_paths(self, path_list):458        if len(path_list) != 1 and self._split_name[1:3] != ['ansible', 'builtin']:459            raise ValueError('this loader requires exactly one path to search')460        return path_list461    def _get_subpackage_search_paths(self, candidate_paths):462        collection_name = '.'.join(self._split_name[1:3])463        collection_meta = _get_collection_metadata(collection_name)464        # check for explicit redirection, as well as ancestor package-level redirection (only load the actual code once!)465        redirect = None466        explicit_redirect = False467        routing_entry = _nested_dict_get(collection_meta, ['import_redirection', self._fullname])468        if routing_entry:469            redirect = routing_entry.get('redirect')470        if redirect:471            explicit_redirect = True472        else:473            redirect = _get_ancestor_redirect(self._redirected_package_map, self._fullname)474        # NB: package level redirection requires hooking all future imports beneath the redirected source package475        # in order to ensure sanity on future relative imports. We always import everything under its "real" name,476        # then add a sys.modules entry with the redirected name using the same module instance. If we naively imported477        # the source for each redirection, most submodules would import OK, but we'd have N runtime copies of the module478        # (one for each name), and relative imports that ascend above the redirected package would break (since they'd479        # see the redirected ancestor package contents instead of the package where they actually live).480        if redirect:481            # FIXME: wrap this so we can be explicit about a failed redirection482            self._redirect_module = import_module(redirect)483            if explicit_redirect and hasattr(self._redirect_module, '__path__') and self._redirect_module.__path__:484                # if the import target looks like a package, store its name so we can rewrite future descendent loads485                self._redirected_package_map[self._fullname] = redirect486            # if we redirected, don't do any further custom package logic487            return None488        # we're not doing a redirect- try to find what we need to actually load a module/package489        # this will raise ImportError if we can't find the requested module/package at all490        if not candidate_paths:491            # noplace to look, just ImportError492            raise ImportError('package has no paths')493        found_path, has_code, package_path = self._module_file_from_path(self._package_to_load, candidate_paths[0])494        # still here? we found something to load...495        if has_code:496            self._source_code_path = found_path497        if package_path:498            return [package_path]  # always needs to be a list499        return None500# This loader only answers for intercepted Ansible Python modules. Normal imports will fail here and be picked up later501# by our path_hook importer (which proxies the built-in import mechanisms, allowing normal caching etc to occur)502class _AnsibleInternalRedirectLoader:503    def __init__(self, fullname, path_list):504        self._redirect = None505        split_name = fullname.split('.')506        toplevel_pkg = split_name[0]507        module_to_load = split_name[-1]508        if toplevel_pkg != 'ansible':509            raise ImportError('not interested')510        builtin_meta = _get_collection_metadata('ansible.builtin')511        routing_entry = _nested_dict_get(builtin_meta, ['import_redirection', fullname])512        if routing_entry:513            self._redirect = routing_entry.get('redirect')514        if not self._redirect:515            raise ImportError('not redirected, go ask path_hook')516    def load_module(self, fullname):517        # since we're delegating to other loaders, this should only be called for internal redirects where we answered518        # find_module with this loader, in which case we'll just directly import the redirection target, insert it into519        # sys.modules under the name it was requested by, and return the original module.520        # should never see this521        if not self._redirect:522            raise ValueError('no redirect found for {0}'.format(fullname))523        # FIXME: smuggle redirection context, provide warning/error that we tried and failed to redirect524        mod = import_module(self._redirect)525        sys.modules[fullname] = mod526        return mod527class AnsibleCollectionRef:528    # FUTURE: introspect plugin loaders to get these dynamically?529    VALID_REF_TYPES = frozenset(to_text(r) for r in ['action', 'become', 'cache', 'callback', 'cliconf', 'connection',530                                                     'doc_fragments', 'filter', 'httpapi', 'inventory', 'lookup',531                                                     'module_utils', 'modules', 'netconf', 'role', 'shell', 'strategy',532                                                     'terminal', 'test', 'vars'])533    # FIXME: tighten this up to match Python identifier reqs, etc534    VALID_COLLECTION_NAME_RE = re.compile(to_text(r'^(\w+)\.(\w+)$'))535    VALID_SUBDIRS_RE = re.compile(to_text(r'^\w+(\.\w+)*$'))536    VALID_FQCR_RE = re.compile(to_text(r'^\w+\.\w+\.\w+(\.\w+)*$'))  # can have 0-N included subdirs as well537    def __init__(self, collection_name, subdirs, resource, ref_type):538        """539        Create an AnsibleCollectionRef from components540        :param collection_name: a collection name of the form 'namespace.collectionname'541        :param subdirs: optional subdir segments to be appended below the plugin type (eg, 'subdir1.subdir2')542        :param resource: the name of the resource being references (eg, 'mymodule', 'someaction', 'a_role')543        :param ref_type: the type of the reference, eg 'module', 'role', 'doc_fragment'544        """545        collection_name = to_text(collection_name, errors='strict')546        if subdirs is not None:547            subdirs = to_text(subdirs, errors='strict')548        resource = to_text(resource, errors='strict')549        ref_type = to_text(ref_type, errors='strict')550        if not self.is_valid_collection_name(collection_name):551            raise ValueError('invalid collection name (must be of the form namespace.collection): {0}'.format(to_native(collection_name)))552        if ref_type not in self.VALID_REF_TYPES:553            raise ValueError('invalid collection ref_type: {0}'.format(ref_type))554        self.collection = collection_name555        if subdirs:556            if not re.match(self.VALID_SUBDIRS_RE, subdirs):557                raise ValueError('invalid subdirs entry: {0} (must be empty/None or of the form subdir1.subdir2)'.format(to_native(subdirs)))558            self.subdirs = subdirs559        else:560            self.subdirs = u''561        self.resource = resource562        self.ref_type = ref_type563        package_components = [u'ansible_collections', self.collection]564        fqcr_components = [self.collection]565        self.n_python_collection_package_name = to_native('.'.join(package_components))566        if self.ref_type == u'role':567            package_components.append(u'roles')568        else:569            # we assume it's a plugin570            package_components += [u'plugins', self.ref_type]571        if self.subdirs:572            package_components.append(self.subdirs)573            fqcr_components.append(self.subdirs)574        if self.ref_type == u'role':575            # roles are their own resource576            package_components.append(self.resource)577        fqcr_components.append(self.resource)578        self.n_python_package_name = to_native('.'.join(package_components))579        self._fqcr = u'.'.join(fqcr_components)580    def __repr__(self):581        return 'AnsibleCollectionRef(collection={0!r}, subdirs={1!r}, resource={2!r})'.format(self.collection, self.subdirs, self.resource)582    @property583    def fqcr(self):584        return self._fqcr585    @staticmethod586    def from_fqcr(ref, ref_type):587        """588        Parse a string as a fully-qualified collection reference, raises ValueError if invalid589        :param ref: collection reference to parse (a valid ref is of the form 'ns.coll.resource' or 'ns.coll.subdir1.subdir2.resource')590        :param ref_type: the type of the reference, eg 'module', 'role', 'doc_fragment'591        :return: a populated AnsibleCollectionRef object592        """593        # assuming the fq_name is of the form (ns).(coll).(optional_subdir_N).(resource_name),594        # we split the resource name off the right, split ns and coll off the left, and we're left with any optional595        # subdirs that need to be added back below the plugin-specific subdir we'll add. So:596        # ns.coll.resource -> ansible_collections.ns.coll.plugins.(plugintype).resource597        # ns.coll.subdir1.resource -> ansible_collections.ns.coll.plugins.subdir1.(plugintype).resource598        # ns.coll.rolename -> ansible_collections.ns.coll.roles.rolename599        if not AnsibleCollectionRef.is_valid_fqcr(ref):600            raise ValueError('{0} is not a valid collection reference'.format(to_native(ref)))601        ref = to_text(ref, errors='strict')602        ref_type = to_text(ref_type, errors='strict')603        resource_splitname = ref.rsplit(u'.', 1)604        package_remnant = resource_splitname[0]605        resource = resource_splitname[1]606        # split the left two components of the collection package name off, anything remaining is plugin-type607        # specific subdirs to be added back on below the plugin type608        package_splitname = package_remnant.split(u'.', 2)609        if len(package_splitname) == 3:610            subdirs = package_splitname[2]611        else:612            subdirs = u''613        collection_name = u'.'.join(package_splitname[0:2])614        return AnsibleCollectionRef(collection_name, subdirs, resource, ref_type)615    @staticmethod616    def try_parse_fqcr(ref, ref_type):617        """618        Attempt to parse a string as a fully-qualified collection reference, returning None on failure (instead of raising an error)619        :param ref: collection reference to parse (a valid ref is of the form 'ns.coll.resource' or 'ns.coll.subdir1.subdir2.resource')620        :param ref_type: the type of the reference, eg 'module', 'role', 'doc_fragment'621        :return: a populated AnsibleCollectionRef object on successful parsing, else None622        """623        try:624            return AnsibleCollectionRef.from_fqcr(ref, ref_type)625        except ValueError:626            pass627    @staticmethod628    def legacy_plugin_dir_to_plugin_type(legacy_plugin_dir_name):629        """630        Utility method to convert from a PluginLoader dir name to a plugin ref_type631        :param legacy_plugin_dir_name: PluginLoader dir name (eg, 'action_plugins', 'library')632        :return: the corresponding plugin ref_type (eg, 'action', 'role')633        """634        legacy_plugin_dir_name = to_text(legacy_plugin_dir_name)635        plugin_type = legacy_plugin_dir_name.replace(u'_plugins', u'')636        if plugin_type == u'library':637            plugin_type = u'modules'638        if plugin_type not in AnsibleCollectionRef.VALID_REF_TYPES:639            raise ValueError('{0} cannot be mapped to a valid collection ref type'.format(to_native(legacy_plugin_dir_name)))640        return plugin_type641    @staticmethod642    def is_valid_fqcr(ref, ref_type=None):643        """644        Validates if is string is a well-formed fully-qualified collection reference (does not look up the collection itself)645        :param ref: candidate collection reference to validate (a valid ref is of the form 'ns.coll.resource' or 'ns.coll.subdir1.subdir2.resource')646        :param ref_type: optional reference type to enable deeper validation, eg 'module', 'role', 'doc_fragment'647        :return: True if the collection ref passed is well-formed, False otherwise648        """649        ref = to_text(ref)650        if not ref_type:651            return bool(re.match(AnsibleCollectionRef.VALID_FQCR_RE, ref))652        return bool(AnsibleCollectionRef.try_parse_fqcr(ref, ref_type))653    @staticmethod654    def is_valid_collection_name(collection_name):655        """656        Validates if the given string is a well-formed collection name (does not look up the collection itself)657        :param collection_name: candidate collection name to validate (a valid name is of the form 'ns.collname')658        :return: True if the collection name passed is well-formed, False otherwise659        """660        collection_name = to_text(collection_name)661        return bool(re.match(AnsibleCollectionRef.VALID_COLLECTION_NAME_RE, collection_name))662def _get_collection_role_path(role_name, collection_list=None):663    acr = AnsibleCollectionRef.try_parse_fqcr(role_name, 'role')664    if acr:665        # looks like a valid qualified collection ref; skip the collection_list666        collection_list = [acr.collection]667        subdirs = acr.subdirs668        resource = acr.resource669    elif not collection_list:670        return None  # not a FQ role and no collection search list spec'd, nothing to do671    else:672        resource = role_name  # treat as unqualified, loop through the collection search list to try and resolve673        subdirs = ''674    for collection_name in collection_list:675        try:676            acr = AnsibleCollectionRef(collection_name=collection_name, subdirs=subdirs, resource=resource, ref_type='role')677            # FIXME: error handling/logging; need to catch any import failures and move along678            pkg = import_module(acr.n_python_package_name)679            if pkg is not None:680                # the package is now loaded, get the collection's package and ask where it lives681                path = os.path.dirname(to_bytes(sys.modules[acr.n_python_package_name].__file__, errors='surrogate_or_strict'))682                return resource, to_text(path, errors='surrogate_or_strict'), collection_name683        except IOError:684            continue685        except Exception as ex:686            # FIXME: pick out typical import errors first, then error logging687            continue688    return None689def _get_collection_name_from_path(path):690    """691    Return the containing collection name for a given path, or None if the path is not below a configured collection, or692    the collection cannot be loaded (eg, the collection is masked by another of the same name higher in the configured693    collection roots).694    :param path: path to evaluate for collection containment695    :return: collection name or None696    """697    # FIXME: mess with realpath canonicalization or not?698    path = to_native(path)699    path_parts = path.split('/')700    if path_parts.count('ansible_collections') != 1:701        return None702    ac_pos = path_parts.index('ansible_collections')703    # make sure it's followed by at least a namespace and collection name704    if len(path_parts) < ac_pos + 3:705        return None706    candidate_collection_name = '.'.join(path_parts[ac_pos + 1:ac_pos + 3])707    try:708        # we've got a name for it, now see if the path prefix matches what the loader sees709        imported_pkg_path = to_native(os.path.dirname(to_bytes(import_module('ansible_collections.' + candidate_collection_name).__file__)))710    except ImportError:711        return None712    # reassemble the original path prefix up the collection name, and it should match what we just imported. If not713    # this is probably a collection root that's not configured.714    original_path_prefix = os.path.join('/', *path_parts[0:ac_pos + 3])715    if original_path_prefix != imported_pkg_path:716        return None717    return candidate_collection_name718def _get_import_redirect(collection_meta_dict, fullname):719    if not collection_meta_dict:720        return None721    return _nested_dict_get(collection_meta_dict, ['import_redirection', fullname, 'redirect'])722def _get_ancestor_redirect(redirected_package_map, fullname):723    # walk the requested module's ancestor packages to see if any have been previously redirected724    cur_pkg = fullname725    while cur_pkg:726        cur_pkg = cur_pkg.rpartition('.')[0]727        ancestor_redirect = redirected_package_map.get(cur_pkg)728        if ancestor_redirect:729            # rewrite the prefix on fullname so we import the target first, then alias it730            redirect = ancestor_redirect + fullname[len(cur_pkg):]731            return redirect732    return None733def _nested_dict_get(root_dict, key_list):734    cur_value = root_dict735    for key in key_list:736        cur_value = cur_value.get(key)737        if not cur_value:738            return None739    return cur_value740def _iter_modules_impl(paths, prefix=''):741    # NB: this currently only iterates what's on disk- redirected modules are not considered742    if not prefix:743        prefix = ''744    else:745        prefix = to_native(prefix)746    # yield (module_loader, name, ispkg) for each module/pkg under path747    # TODO: implement ignore/silent catch for unreadable?748    for b_path in map(to_bytes, paths):749        if not os.path.isdir(b_path):750            continue751        for b_basename in sorted(os.listdir(b_path)):752            b_candidate_module_path = os.path.join(b_path, b_basename)753            if os.path.isdir(b_candidate_module_path):754                # exclude things that obviously aren't Python package dirs755                # FIXME: this dir is adjustable in py3.8+, check for it756                if b'.' in b_basename or b_basename == b'__pycache__':757                    continue758                # TODO: proper string handling?759                yield prefix + to_native(b_basename), True760            else:761                # FIXME: match builtin ordering for package/dir/file, support compiled?762                if b_basename.endswith(b'.py') and b_basename != b'__init__.py':763                    yield prefix + to_native(os.path.splitext(b_basename)[0]), False764def _get_collection_metadata(collection_name):765    collection_name = to_native(collection_name)766    if not collection_name or not isinstance(collection_name, string_types) or len(collection_name.split('.')) != 2:767        raise ValueError('collection_name must be a non-empty string of the form namespace.collection')768    try:769        collection_pkg = import_module('ansible_collections.' + collection_name)770    except ImportError:771        raise ValueError('unable to locate collection {0}'.format(collection_name))772    _collection_meta = getattr(collection_pkg, '_collection_meta', None)773    if _collection_meta is None:774        raise ValueError('collection metadata was not loaded for collection {0}'.format(collection_name))...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.