mirror of https://github.com/ansible/ansible.git
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
![]()
5 years ago
|
"""Provider (plugin) infrastructure for ansible-test."""
|
||
|
from __future__ import (absolute_import, division, print_function)
|
||
|
__metaclass__ = type
|
||
|
|
||
|
import abc
|
||
|
import os
|
||
|
|
||
![]()
5 years ago
|
from .. import types as t
|
||
![]()
5 years ago
|
|
||
![]()
5 years ago
|
from ..util import (
|
||
![]()
5 years ago
|
ABC,
|
||
|
ApplicationError,
|
||
|
get_subclasses,
|
||
|
)
|
||
|
|
||
|
|
||
|
try:
|
||
|
C = t.TypeVar('C', 'PathProvider', 'PathProvider')
|
||
|
except AttributeError:
|
||
|
pass
|
||
|
|
||
|
|
||
|
def get_path_provider_classes(provider_type): # type: (t.Type[C]) -> t.List[t.Type[C]]
|
||
|
"""Return a list of path provider classes of the given type."""
|
||
|
return sorted(get_subclasses(provider_type), key=lambda c: (c.priority, c.__name__))
|
||
|
|
||
|
|
||
|
def find_path_provider(provider_type, provider_classes, path, walk): # type: (t.Type[C], t.List[t.Type[C]], str, bool) -> C
|
||
|
"""Return the first found path provider of the given type for the given path."""
|
||
|
sequences = sorted(set(pc.sequence for pc in provider_classes if pc.sequence > 0))
|
||
|
|
||
|
for sequence in sequences:
|
||
|
candidate_path = path
|
||
|
tier_classes = [pc for pc in provider_classes if pc.sequence == sequence]
|
||
|
|
||
|
while True:
|
||
|
for provider_class in tier_classes:
|
||
|
if provider_class.is_content_root(candidate_path):
|
||
|
return provider_class(candidate_path)
|
||
|
|
||
|
if not walk:
|
||
|
break
|
||
|
|
||
|
parent_path = os.path.dirname(candidate_path)
|
||
|
|
||
|
if parent_path == candidate_path:
|
||
|
break
|
||
|
|
||
|
candidate_path = parent_path
|
||
|
|
||
|
raise ProviderNotFoundForPath(provider_type, path)
|
||
|
|
||
|
|
||
|
class ProviderNotFoundForPath(ApplicationError):
|
||
|
"""Exception generated when a path based provider cannot be found for a given path."""
|
||
|
def __init__(self, provider_type, path): # type: (t.Type, str) -> None
|
||
|
super(ProviderNotFoundForPath, self).__init__('No %s found for path: %s' % (provider_type.__name__, path))
|
||
|
|
||
|
self.provider_type = provider_type
|
||
|
self.path = path
|
||
|
|
||
|
|
||
|
class PathProvider(ABC):
|
||
|
"""Base class for provider plugins that are path based."""
|
||
|
sequence = 500
|
||
|
priority = 500
|
||
|
|
||
|
def __init__(self, root): # type: (str) -> None
|
||
|
self.root = root
|
||
|
|
||
|
@staticmethod
|
||
|
@abc.abstractmethod
|
||
|
def is_content_root(path): # type: (str) -> bool
|
||
|
"""Return True if the given path is a content root for this provider."""
|