Commit 3b0e6412 by James Cammarata

Refactoring role spec stuff into a dedicated parsing class

Also reworking tests to cut down on the number of patches required
by sub-classing the DataLoader() class and reworking the base object's
structure a bit to allow its use
parent bd203a44
...@@ -91,6 +91,15 @@ class DataLoader(): ...@@ -91,6 +91,15 @@ class DataLoader():
return parsed_data return parsed_data
def path_exists(self, path):
return os.path.exists(path)
def is_directory(self, path):
return os.path.isdir(path)
def is_file(self, path):
return os.path.isfile(path)
def _safe_load(self, stream): def _safe_load(self, stream):
''' Implements yaml.safe_load(), except using our custom loader class. ''' ''' Implements yaml.safe_load(), except using our custom loader class. '''
return load(stream, AnsibleLoader) return load(stream, AnsibleLoader)
...@@ -100,7 +109,7 @@ class DataLoader(): ...@@ -100,7 +109,7 @@ class DataLoader():
Reads the file contents from the given file name, and will decrypt them Reads the file contents from the given file name, and will decrypt them
if they are found to be vault-encrypted. if they are found to be vault-encrypted.
''' '''
if not os.path.exists(file_name) or not os.path.isfile(file_name): if not self.path_exists(file_name) or not self.is_file(file_name):
raise AnsibleParserError("the file_name '%s' does not exist, or is not readable" % file_name) raise AnsibleParserError("the file_name '%s' does not exist, or is not readable" % file_name)
show_content = True show_content = True
......
...@@ -29,13 +29,11 @@ from ansible.parsing.yaml import DataLoader ...@@ -29,13 +29,11 @@ from ansible.parsing.yaml import DataLoader
class Base: class Base:
_tags = FieldAttribute(isa='list') def __init__(self):
_when = FieldAttribute(isa='list')
def __init__(self, loader=DataLoader): # initialize the data loader, this will be provided later
# when the object is actually loaded
# the data loader class is used to parse data from strings and files self._loader = None
self._loader = loader()
# each class knows attributes set upon it, see Task.py for example # each class knows attributes set upon it, see Task.py for example
self._attributes = dict() self._attributes = dict()
...@@ -61,11 +59,17 @@ class Base: ...@@ -61,11 +59,17 @@ class Base:
return ds return ds
def load_data(self, ds): def load_data(self, ds, loader=None):
''' walk the input datastructure and assign any values ''' ''' walk the input datastructure and assign any values '''
assert ds is not None assert ds is not None
# the data loader class is used to parse data from strings and files
if loader is not None:
self._loader = loader
else:
self._loader = DataLoader()
if isinstance(ds, string_types) or isinstance(ds, FileIO): if isinstance(ds, string_types) or isinstance(ds, FileIO):
ds = self._loader.load(ds) ds = self._loader.load(ds)
...@@ -89,6 +93,8 @@ class Base: ...@@ -89,6 +93,8 @@ class Base:
self.validate() self.validate()
return self return self
def get_loader(self):
return self._loader
def validate(self): def validate(self):
''' validation that is done at parse time, not load time ''' ''' validation that is done at parse time, not load time '''
......
...@@ -28,6 +28,8 @@ class Block(Base): ...@@ -28,6 +28,8 @@ class Block(Base):
_block = FieldAttribute(isa='list') _block = FieldAttribute(isa='list')
_rescue = FieldAttribute(isa='list') _rescue = FieldAttribute(isa='list')
_always = FieldAttribute(isa='list') _always = FieldAttribute(isa='list')
_tags = FieldAttribute(isa='list', default=[])
_when = FieldAttribute(isa='list', default=[])
# for future consideration? this would be functionally # for future consideration? this would be functionally
# similar to the 'else' clause for exceptions # similar to the 'else' clause for exceptions
...@@ -43,9 +45,9 @@ class Block(Base): ...@@ -43,9 +45,9 @@ class Block(Base):
return dict() return dict()
@staticmethod @staticmethod
def load(data, role=None): def load(data, role=None, loader=None):
b = Block(role=role) b = Block(role=role)
return b.load_data(data) return b.load_data(data, loader=loader)
def munge(self, ds): def munge(self, ds):
''' '''
......
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import iteritems, string_types
import os
from hashlib import md5
from types import NoneType
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.parsing.yaml import DataLoader
from ansible.playbook.attribute import FieldAttribute
from ansible.playbook.base import Base
from ansible.playbook.block import Block
from ansible.playbook.role.include import RoleInclude
from ansible.playbook.role.metadata import RoleMetadata
__all__ = ['Role', 'ROLE_CACHE']
# The role cache is used to prevent re-loading roles, which
# may already exist. Keys into this cache are the MD5 hash
# of the role definition (for dictionary definitions, this
# will be based on the repr() of the dictionary object)
ROLE_CACHE = dict()
class Role:
def __init__(self):
self._role_name = None
self._role_path = None
self._role_params = dict()
self._loader = None
self._metadata = None
self._parents = []
self._dependencies = []
self._task_blocks = []
self._handler_blocks = []
self._default_vars = dict()
self._role_vars = dict()
def __repr__(self):
return self.get_name()
def get_name(self):
return self._role_name
@staticmethod
def load(role_include, parent_role=None):
# FIXME: add back in the role caching support
try:
r = Role()
r._load_role_data(role_include, parent_role=parent_role)
except RuntimeError:
# FIXME: needs a better way to access the ds in the role include
raise AnsibleError("A recursion loop was detected with the roles specified. Make sure child roles do not have dependencies on parent roles", obj=role_include._ds)
return r
def _load_role_data(self, role_include, parent_role=None):
self._role_name = role_include.role
self._role_path = role_include.get_role_path()
self._role_params = role_include.get_role_params()
self._loader = role_include.get_loader()
if parent_role:
self.add_parent(parent_role)
# load the role's files, if they exist
metadata = self._load_role_yaml('meta')
if metadata:
self._metadata = RoleMetadata.load(metadata, owner=self, loader=self._loader)
self._dependencies = self._load_dependencies()
task_data = self._load_role_yaml('tasks')
if task_data:
self._task_blocks = self._load_list_of_blocks(task_data)
handler_data = self._load_role_yaml('handlers')
if handler_data:
self._handler_blocks = self._load_list_of_blocks(handler_data)
# vars and default vars are regular dictionaries
self._role_vars = self._load_role_yaml('vars')
if not isinstance(self._role_vars, (dict, NoneType)):
raise AnsibleParserError("The vars/main.yml file for role '%s' must contain a dictionary of variables" % self._role_name, obj=ds)
self._default_vars = self._load_role_yaml('defaults')
if not isinstance(self._default_vars, (dict, NoneType)):
raise AnsibleParserError("The default/main.yml file for role '%s' must contain a dictionary of variables" % self._role_name, obj=ds)
def _load_role_yaml(self, subdir):
file_path = os.path.join(self._role_path, subdir)
if self._loader.path_exists(file_path) and self._loader.is_directory(file_path):
main_file = self._resolve_main(file_path)
if self._loader.path_exists(main_file):
return self._loader.load_from_file(main_file)
return None
def _resolve_main(self, basepath):
''' flexibly handle variations in main filenames '''
possible_mains = (
os.path.join(basepath, 'main.yml'),
os.path.join(basepath, 'main.yaml'),
os.path.join(basepath, 'main.json'),
os.path.join(basepath, 'main'),
)
if sum([self._loader.is_file(x) for x in possible_mains]) > 1:
raise AnsibleError("found multiple main files at %s, only one allowed" % (basepath))
else:
for m in possible_mains:
if self._loader.is_file(m):
return m # exactly one main file
return possible_mains[0] # zero mains (we still need to return something)
def _load_list_of_blocks(self, ds):
'''
Given a list of mixed task/block data (parsed from YAML),
return a list of Block() objects, where implicit blocks
are created for each bare Task.
'''
assert type(ds) in (list, NoneType)
block_list = []
if ds:
for block in ds:
b = Block(block)
block_list.append(b)
return block_list
def _load_dependencies(self):
'''
Recursively loads role dependencies from the metadata list of
dependencies, if it exists
'''
deps = []
if self._metadata:
for role_include in self._metadata.dependencies:
r = Role.load(role_include, parent_role=self)
deps.append(r)
return deps
#------------------------------------------------------------------------------
# other functions
def add_parent(self, parent_role):
''' adds a role to the list of this roles parents '''
assert isinstance(parent_role, Role)
if parent_role not in self._parents:
self._parents.append(parent_role)
def get_parents(self):
return self._parents
# FIXME: not yet used
#def get_variables(self):
# # returns the merged variables for this role, including
# # recursively merging those of all child roles
# return dict()
def get_direct_dependencies(self):
return self._dependencies[:]
def get_all_dependencies(self):
# returns a list built recursively, of all deps from
# all child dependencies
child_deps = []
direct_deps = self.get_direct_dependencies()
for dep in direct_deps:
dep_deps = dep.get_all_dependencies()
for dep_dep in dep_deps:
if dep_dep not in child_deps:
child_deps.append(dep_dep)
return direct_deps + child_deps
# (c) 2014 Michael DeHaan, <michael@ansible.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import iteritems, string_types
import os
from ansible.errors import AnsibleError
from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject, AnsibleMapping
from ansible.playbook.attribute import Attribute, FieldAttribute
from ansible.playbook.base import Base
__all__ = ['RoleDefinition']
class RoleDefinition(Base):
_role = FieldAttribute(isa='string')
def __init__(self):
self._role_path = None
self._role_params = dict()
super(RoleDefinition, self).__init__()
def __repr__(self):
return 'ROLEDEF: ' + self._attributes.get('role', '<no name set>')
@staticmethod
def load(data, loader=None):
raise AnsibleError("not implemented")
def munge(self, ds):
assert isinstance(ds, dict) or isinstance(ds, string_types)
# we create a new data structure here, using the same
# object used internally by the YAML parsing code so we
# can preserve file:line:column information if it exists
new_ds = AnsibleMapping()
if isinstance(ds, AnsibleBaseYAMLObject):
new_ds.copy_position_info(ds)
# first we pull the role name out of the data structure,
# and then use that to determine the role path (which may
# result in a new role name, if it was a file path)
role_name = self._load_role_name(ds)
(role_name, role_path) = self._load_role_path(role_name)
# next, we split the role params out from the valid role
# attributes and update the new datastructure with that
# result and the role name
if isinstance(ds, dict):
(new_role_def, role_params) = self._split_role_params(ds)
new_ds.update(new_role_def)
self._role_params = role_params
# set the role name in the new ds
new_ds['role'] = role_name
# we store the role path internally
self._role_path = role_path
# save the original ds for use later
self._ds = ds
# and return the cleaned-up data structure
return new_ds
def _load_role_name(self, ds):
'''
Returns the role name (either the role: or name: field) from
the role definition, or (when the role definition is a simple
string), just that string
'''
if isinstance(ds, string_types):
return ds
role_name = ds.get('role', ds.get('name'))
if not role_name:
raise AnsibleError('role definitions must contain a role name', obj=ds)
return role_name
def _load_role_path(self, role_name):
'''
the 'role', as specified in the ds (or as a bare string), can either
be a simple name or a full path. If it is a full path, we use the
basename as the role name, otherwise we take the name as-given and
append it to the default role path
'''
# FIXME: this should use unfrackpath once the utils code has been sorted out
role_path = os.path.normpath(role_name)
if self._loader.path_exists(role_path):
role_name = os.path.basename(role_name)
return (role_name, role_path)
else:
# FIXME: this should search in the configured roles path
for path in ('./roles', '/etc/ansible/roles'):
role_path = os.path.join(path, role_name)
if self._loader.path_exists(role_path):
return (role_name, role_path)
# FIXME: make the parser smart about list/string entries
# in the yaml so the error line/file can be reported
# here
raise AnsibleError("the role '%s' was not found" % role_name)
def _split_role_params(self, ds):
'''
Splits any random role params off from the role spec and store
them in a dictionary of params for parsing later
'''
role_def = dict()
role_params = dict()
for (key, value) in iteritems(ds):
# use the list of FieldAttribute values to determine what is and is not
# an extra parameter for this role (or sub-class of this role)
if key not in [attr_name for (attr_name, attr_value) in self._get_base_attributes().iteritems()]:
# this key does not match a field attribute, so it must be a role param
role_params[key] = value
else:
# this is a field attribute, so copy it over directly
role_def[key] = value
return (role_def, role_params)
def get_role_params(self):
return self._role_params.copy()
def get_role_path(self):
return self._role_path
# (c) 2014 Michael DeHaan, <michael@ansible.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import iteritems, string_types
import os
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.playbook.attribute import Attribute, FieldAttribute
from ansible.playbook.role.definition import RoleDefinition
__all__ = ['RoleInclude']
class RoleInclude(RoleDefinition):
"""
FIXME: docstring
"""
_tags = FieldAttribute(isa='list', default=[])
_when = FieldAttribute(isa='list', default=[])
def __init__(self):
super(RoleInclude, self).__init__()
@staticmethod
def load(data, parent_role=None, loader=None):
assert isinstance(data, string_types) or isinstance(data, dict)
ri = RoleInclude()
return ri.load_data(data, loader=loader)
# (c) 2014 Michael DeHaan, <michael@ansible.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import iteritems, string_types
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.playbook.attribute import Attribute, FieldAttribute
from ansible.playbook.base import Base
from ansible.playbook.role.include import RoleInclude
__all__ = ['RoleMetadata']
class RoleMetadata(Base):
'''
This class wraps the parsing and validation of the optional metadata
within each Role (meta/main.yml).
'''
_allow_duplicates = FieldAttribute(isa='bool', default=False)
_dependencies = FieldAttribute(isa='list', default=[])
_galaxy_info = FieldAttribute(isa='GalaxyInfo')
def __init__(self):
self._owner = None
super(RoleMetadata, self).__init__()
@staticmethod
def load(data, owner, loader=None):
'''
Returns a new RoleMetadata object based on the datastructure passed in.
'''
if not isinstance(data, dict):
raise AnsibleParserError("the 'meta/main.yml' for role %s is not a dictionary" % owner.get_name())
m = RoleMetadata().load_data(data, loader=loader)
return m
def munge(self, ds):
# make sure there are no keys in the datastructure which
# do not map to attributes for this object
valid_attrs = [name for (name, attribute) in iteritems(self._get_base_attributes())]
for name in ds:
if name not in valid_attrs:
print("'%s' is not a valid attribute" % name)
raise AnsibleParserError("'%s' is not a valid attribute" % name, obj=ds)
return ds
def _load_dependencies(self, attr, ds):
'''
This is a helper loading function for the dependencis list,
which returns a list of RoleInclude objects
'''
assert isinstance(ds, list)
deps = []
for role_def in ds:
i = RoleInclude.load(role_def, loader=self._loader)
deps.append(i)
return deps
def _load_galaxy_info(self, attr, ds):
'''
This is a helper loading function for the galaxy info entry
in the metadata, which returns a GalaxyInfo object rather than
a simple dictionary.
'''
return ds
# (c) 2014 Michael DeHaan, <michael@ansible.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import iteritems, string_types
import os
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.playbook.role.definition import RoleDefinition
__all__ = ['RoleRequirement']
class RoleRequirement(RoleDefinition):
"""
FIXME: document various ways role specs can be specified
"""
def __init__(self):
pass
def _get_valid_spec_keys(self):
return (
'name',
'role',
'scm',
'src',
'version',
)
def parse(self, ds):
'''
FIXME: docstring
'''
assert type(ds) == dict or isinstance(ds, string_types)
role_name = ''
role_params = dict()
new_ds = dict()
if isinstance(ds, string_types):
role_name = ds
else:
ds = self._munge_role_spec(ds)
(new_ds, role_params) = self._split_role_params(ds)
# pull the role name out of the ds
role_name = new_ds.get('role_name')
del ds['role_name']
return (new_ds, role_name, role_params)
def _munge_role_spec(self, ds):
if 'role' in ds:
# Old style: {role: "galaxy.role,version,name", other_vars: "here" }
role_info = self._role_spec_parse(ds['role'])
if isinstance(role_info, dict):
# Warning: Slight change in behaviour here. name may be being
# overloaded. Previously, name was only a parameter to the role.
# Now it is both a parameter to the role and the name that
# ansible-galaxy will install under on the local system.
if 'name' in ds and 'name' in role_info:
del role_info['name']
ds.update(role_info)
else:
# New style: { src: 'galaxy.role,version,name', other_vars: "here" }
if 'github.com' in ds["src"] and 'http' in ds["src"] and '+' not in ds["src"] and not ds["src"].endswith('.tar.gz'):
ds["src"] = "git+" + ds["src"]
if '+' in ds["src"]:
(scm, src) = ds["src"].split('+')
ds["scm"] = scm
ds["src"] = src
if 'name' in role:
ds["role"] = ds["name"]
del ds["name"]
else:
ds["role"] = self._repo_url_to_role_name(ds["src"])
# set some values to a default value, if none were specified
ds.setdefault('version', '')
ds.setdefault('scm', None)
return ds
def _repo_url_to_role_name(self, repo_url):
# gets the role name out of a repo like
# http://git.example.com/repos/repo.git" => "repo"
if '://' not in repo_url and '@' not in repo_url:
return repo_url
trailing_path = repo_url.split('/')[-1]
if trailing_path.endswith('.git'):
trailing_path = trailing_path[:-4]
if trailing_path.endswith('.tar.gz'):
trailing_path = trailing_path[:-7]
if ',' in trailing_path:
trailing_path = trailing_path.split(',')[0]
return trailing_path
def _role_spec_parse(self, role_spec):
# takes a repo and a version like
# git+http://git.example.com/repos/repo.git,v1.0
# and returns a list of properties such as:
# {
# 'scm': 'git',
# 'src': 'http://git.example.com/repos/repo.git',
# 'version': 'v1.0',
# 'name': 'repo'
# }
default_role_versions = dict(git='master', hg='tip')
role_spec = role_spec.strip()
role_version = ''
if role_spec == "" or role_spec.startswith("#"):
return (None, None, None, None)
tokens = [s.strip() for s in role_spec.split(',')]
# assume https://github.com URLs are git+https:// URLs and not
# tarballs unless they end in '.zip'
if 'github.com/' in tokens[0] and not tokens[0].startswith("git+") and not tokens[0].endswith('.tar.gz'):
tokens[0] = 'git+' + tokens[0]
if '+' in tokens[0]:
(scm, role_url) = tokens[0].split('+')
else:
scm = None
role_url = tokens[0]
if len(tokens) >= 2:
role_version = tokens[1]
if len(tokens) == 3:
role_name = tokens[2]
else:
role_name = self._repo_url_to_role_name(tokens[0])
if scm and not role_version:
role_version = default_role_versions.get(scm, '')
return dict(scm=scm, src=role_url, version=role_version, role_name=role_name)
...@@ -83,14 +83,16 @@ class Task(Base): ...@@ -83,14 +83,16 @@ class Task(Base):
_sudo = FieldAttribute(isa='bool') _sudo = FieldAttribute(isa='bool')
_sudo_user = FieldAttribute(isa='string') _sudo_user = FieldAttribute(isa='string')
_sudo_pass = FieldAttribute(isa='string') _sudo_pass = FieldAttribute(isa='string')
_tags = FieldAttribute(isa='list', default=[])
_transport = FieldAttribute(isa='string') _transport = FieldAttribute(isa='string')
_until = FieldAttribute(isa='list') # ? _until = FieldAttribute(isa='list') # ?
_when = FieldAttribute(isa='list', default=[])
def __init__(self, block=None, role=None, loader=DataLoader): def __init__(self, block=None, role=None):
''' constructors a task, without the Task.load classmethod, it will be pretty blank ''' ''' constructors a task, without the Task.load classmethod, it will be pretty blank '''
self._block = block self._block = block
self._role = role self._role = role
super(Task, self).__init__(loader) super(Task, self).__init__()
def get_name(self): def get_name(self):
''' return the name of the task ''' ''' return the name of the task '''
...@@ -118,9 +120,9 @@ class Task(Base): ...@@ -118,9 +120,9 @@ class Task(Base):
return buf return buf
@staticmethod @staticmethod
def load(data, block=None, role=None): def load(data, block=None, role=None, loader=None):
t = Task(block=block, role=role) t = Task(block=block, role=role)
return t.load_data(data) return t.load_data(data, loader=loader)
def __repr__(self): def __repr__(self):
''' returns a human readable representation of the task ''' ''' returns a human readable representation of the task '''
......
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
from ansible.parsing.yaml import DataLoader
class DictDataLoader(DataLoader):
def __init__(self, file_mapping=dict()):
assert type(file_mapping) == dict
self._file_mapping = file_mapping
self._build_known_directories()
super(DictDataLoader, self).__init__()
def load_from_file(self, path):
if path in self._file_mapping:
return self.load(self._file_mapping[path], path)
return None
def path_exists(self, path):
return path in self._file_mapping or path in self._known_directories
def is_file(self, path):
return path in self._file_mapping
def is_directory(self, path):
return path in self._known_directories
def _add_known_directory(self, directory):
if directory not in self._known_directories:
self._known_directories.append(directory)
def _build_known_directories(self):
self._known_directories = []
for path in self._file_mapping:
dirname = os.path.dirname(path)
while dirname not in ('/', ''):
self._add_known_directory(dirname)
dirname = os.path.dirname(dirname)
def push(self, path, content):
rebuild_dirs = False
if path not in self._file_mapping:
rebuild_dirs = True
self._file_mapping[path] = content
if rebuild_dirs:
self._build_known_directories()
def pop(self, path):
if path in self._file_mapping:
del self._file_mapping[path]
self._build_known_directories()
def clear(self):
self._file_mapping = dict()
self._known_directories = []
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment