1"""Provider (plugin) infrastructure for ansible-test.""" 2from __future__ import (absolute_import, division, print_function) 3__metaclass__ = type 4 5import abc 6import os 7 8from .. import types as t 9 10from ..util import ( 11 ABC, 12 ApplicationError, 13 get_subclasses, 14) 15 16 17try: 18 TPathProvider = t.TypeVar('TPathProvider', bound='PathProvider') 19except AttributeError: 20 TPathProvider = None # pylint: disable=invalid-name 21 22 23def get_path_provider_classes(provider_type): # type: (t.Type[TPathProvider]) -> t.List[t.Type[TPathProvider]] 24 """Return a list of path provider classes of the given type.""" 25 return sorted(get_subclasses(provider_type), key=lambda c: (c.priority, c.__name__)) 26 27 28def find_path_provider(provider_type, # type: t.Type[TPathProvider], 29 provider_classes, # type: t.List[t.Type[TPathProvider]] 30 path, # type: str 31 walk, # type: bool 32 ): # type: (...) -> TPathProvider 33 """Return the first found path provider of the given type for the given path.""" 34 sequences = sorted(set(pc.sequence for pc in provider_classes if pc.sequence > 0)) 35 36 for sequence in sequences: 37 candidate_path = path 38 tier_classes = [pc for pc in provider_classes if pc.sequence == sequence] 39 40 while True: 41 for provider_class in tier_classes: 42 if provider_class.is_content_root(candidate_path): 43 return provider_class(candidate_path) 44 45 if not walk: 46 break 47 48 parent_path = os.path.dirname(candidate_path) 49 50 if parent_path == candidate_path: 51 break 52 53 candidate_path = parent_path 54 55 raise ProviderNotFoundForPath(provider_type, path) 56 57 58class ProviderNotFoundForPath(ApplicationError): 59 """Exception generated when a path based provider cannot be found for a given path.""" 60 def __init__(self, provider_type, path): # type: (t.Type, str) -> None 61 super(ProviderNotFoundForPath, self).__init__('No %s found for path: %s' % (provider_type.__name__, path)) 62 63 self.provider_type = provider_type 64 self.path = path 65 66 67class PathProvider(ABC): 68 """Base class for provider plugins that are path based.""" 69 sequence = 500 70 priority = 500 71 72 def __init__(self, root): # type: (str) -> None 73 self.root = root 74 75 @staticmethod 76 @abc.abstractmethod 77 def is_content_root(path): # type: (str) -> bool 78 """Return True if the given path is a content root for this provider.""" 79