mark/cdist
1
0
Fork 0
forked from ungleich-public/cdist
Code Issues Pull requests Projects Releases Wiki Activity

lib/cdist => cdist (pypi)

Browse source 
Signed-off-by: Nico Schottelius <nico@brief.schottelius.org>
This commit is contained in:
Nico Schottelius 2012-10-25 17:21:58 +02:00
commit c9f728e073
120 changed files with 0 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

79
lib/cdist/__init__.py
View file

@ -1,79 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2012 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import subprocess
try:
with open(os.devnull, 'w') as devnull:
here = os.path.dirname(os.path.realpath(__file__))
VERSION = subprocess.check_output(
'cd "%s" && git describe' % here,
stderr=devnull, shell=True).decode('utf-8')
except:
VERSION = "2.0.14"
BANNER = """
.. . .x+=:. s
dF @88> z` ^% :8
'88bu. %8P . <k .88
. '*88888bu . .@8Ned8" :888ooo
.udR88N ^"*8888N .@88u .@^%8888" -*8888888
<888'888k beWE "888L ''888E` x88: `)8b. 8888
9888 'Y" 888E 888E 888E 8888N=*8888 8888
9888 888E 888E 888E %8" R88 8888
9888 888E 888F 888E @8Wou 9% .8888Lu=
?8888u../ .888N..888 888& .888888P` ^%888*
"8888P' `"888*"" R888" ` ^"F 'Y"
"P' "" ""
"""
DOT_CDIST = ".cdist"
class Error(Exception):
"""Base exception class for this project"""
pass
class CdistObjectError(Error):
"""Something went wrong with an object"""
def __init__(self, cdist_object, message):
self.name = cdist_object.name
self.source = " ".join(cdist_object.source)
self.message = message
def __str__(self):
return '%s: %s (defined at %s)' % (self.name, self.message, self.source)
def file_to_list(filename):
"""Return list from \n seperated file"""
if os.path.isfile(filename):
file_fd = open(filename, "r")
lines = file_fd.readlines()
file_fd.close()
# Remove \n from all lines
lines = map(lambda s: s.strip(), lines)
else:
lines = []
return lines

32
lib/cdist/banner.py
View file

@ -1,32 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011-2012 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import sys
import cdist
log = logging.getLogger(__name__)
def banner(args):
"""Guess what :-)"""
print(cdist.BANNER)

26
lib/cdist/config.py
View file

@ -1,26 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2010-2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import cdist.config_install
class Config(cdist.config_install.ConfigInstall):
pass

148
lib/cdist/config_install.py
View file

@ -1,148 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2010-2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import stat
import shutil
import sys
import tempfile
import time
import itertools
import pprint
import cdist
from cdist import core
from cdist import resolver
class ConfigInstall(object):
"""Cdist main class to hold arbitrary data"""
def __init__(self, context):
self.context = context
self.log = logging.getLogger(self.context.target_host)
# For easy access
self.local = context.local
self.remote = context.remote
# Initialise local directory structure
self.local.create_directories()
# Initialise remote directory structure
self.remote.create_directories()
self.explorer = core.Explorer(self.context.target_host, self.local, self.remote)
self.manifest = core.Manifest(self.context.target_host, self.local)
self.code = core.Code(self.context.target_host, self.local, self.remote)
def cleanup(self):
# FIXME: move to local?
destination = os.path.join(self.local.cache_path, self.context.target_host)
self.log.debug("Saving " + self.local.out_path + " to " + destination)
if os.path.exists(destination):
shutil.rmtree(destination)
shutil.move(self.local.out_path, destination)
def deploy_to(self):
"""Mimic the old deploy to: Deploy to one host"""
self.stage_prepare()
self.stage_run()
def deploy_and_cleanup(self):
"""Do what is most often done: deploy & cleanup"""
start_time = time.time()
self.deploy_to()
self.cleanup()
self.log.info("Finished successful run in %s seconds",
time.time() - start_time)
def stage_prepare(self):
"""Do everything for a deploy, minus the actual code stage"""
self.local.link_emulator(self.context.exec_path)
self.explorer.run_global_explorers(self.local.global_explorer_out_path)
self.manifest.run_initial_manifest(self.context.initial_manifest)
self.log.info("Running object manifests and type explorers")
# Continue process until no new objects are created anymore
new_objects_created = True
while new_objects_created:
new_objects_created = False
for cdist_object in core.CdistObject.list_objects(self.local.object_path,
self.local.type_path):
if cdist_object.state == core.CdistObject.STATE_PREPARED:
self.log.debug("Skipping re-prepare of object %s", cdist_object)
continue
else:
self.object_prepare(cdist_object)
new_objects_created = True
def object_prepare(self, cdist_object):
"""Prepare object: Run type explorer + manifest"""
self.log.info("Running manifest and explorers for " + cdist_object.name)
self.explorer.run_type_explorers(cdist_object)
self.manifest.run_type_manifest(cdist_object)
cdist_object.state = core.CdistObject.STATE_PREPARED
def object_run(self, cdist_object):
"""Run gencode and code for an object"""
self.log.debug("Trying to run object " + cdist_object.name)
if cdist_object.state == core.CdistObject.STATE_DONE:
# TODO: remove once we are sure that this really never happens.
raise cdist.Error("Attempting to run an already finished object: %s", cdist_object)
cdist_type = cdist_object.cdist_type
# Generate
self.log.info("Generating and executing code for " + cdist_object.name)
cdist_object.code_local = self.code.run_gencode_local(cdist_object)
cdist_object.code_remote = self.code.run_gencode_remote(cdist_object)
if cdist_object.code_local or cdist_object.code_remote:
cdist_object.changed = True
# Execute
if cdist_object.code_local:
self.code.run_code_local(cdist_object)
if cdist_object.code_remote:
self.code.transfer_code_remote(cdist_object)
self.code.run_code_remote(cdist_object)
# Mark this object as done
self.log.debug("Finishing run of " + cdist_object.name)
cdist_object.state = core.CdistObject.STATE_DONE
def stage_run(self):
"""The final (and real) step of deployment"""
self.log.info("Generating and executing code")
objects = core.CdistObject.list_objects(
self.local.object_path,
self.local.type_path)
dependency_resolver = resolver.DependencyResolver(objects)
self.log.debug(pprint.pformat(dependency_resolver.dependencies))
for cdist_object in dependency_resolver:
self.log.debug("Run object: %s", cdist_object)
self.object_run(cdist_object)

100
lib/cdist/context.py
View file

@ -1,100 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2010-2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import sys
import tempfile
import shutil
from cdist.exec import local
from cdist.exec import remote
class Context(object):
"""Hold information about current context"""
def __init__(self,
target_host,
remote_copy,
remote_exec,
initial_manifest=False,
base_path=False,
exec_path=sys.argv[0],
debug=False):
self.debug = debug
self.target_host = target_host
# Only required for testing
self.exec_path = exec_path
# Context logging
self.log = logging.getLogger(self.target_host)
self.log.addFilter(self)
# Local base directory
self.base_path = (base_path or
os.path.abspath(os.path.join(os.path.dirname(__file__),
os.pardir, os.pardir)))
# Local temp directory
# FIXME: if __cdist_out_dir can be given from the outside, the same directory will be used for all hosts
if '__cdist_out_dir' in os.environ:
self.out_path = os.environ['__cdist_out_dir']
self.temp_dir = None
else:
self.temp_dir = tempfile.mkdtemp()
self.out_path = os.path.join(self.temp_dir, "out")
self.local = local.Local(self.target_host, self.base_path, self.out_path)
self.initial_manifest = (initial_manifest or
os.path.join(self.local.manifest_path, "init"))
self._init_remote(remote_copy, remote_exec)
# Remote stuff
def _init_remote(self, remote_copy, remote_exec):
self.remote_base_path = os.environ.get('__cdist_remote_out_dir', "/var/lib/cdist")
self.remote_copy = remote_copy
self.remote_exec = remote_exec
os.environ['__remote_copy'] = self.remote_copy
os.environ['__remote_exec'] = self.remote_exec
self.remote = remote.Remote(self.target_host, self.remote_base_path,
self.remote_exec, self.remote_copy)
def cleanup(self):
"""Remove temp stuff"""
if self.temp_dir:
shutil.rmtree(self.temp_dir)
def filter(self, record):
"""Add hostname to logs via logging Filter"""
record.msg = self.target_host + ": " + record.msg
return True

29
lib/cdist/core/__init__.py
View file

@ -1,29 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
from cdist.core.cdist_type import CdistType
from cdist.core.cdist_type import NoSuchTypeError
from cdist.core.cdist_object import CdistObject
from cdist.core.cdist_object import IllegalObjectIdError
from cdist.core.cdist_object import OBJECT_MARKER
from cdist.core.explorer import Explorer
from cdist.core.manifest import Manifest
from cdist.core.code import Code

211
lib/cdist/core/cdist_object.py
View file

@ -1,211 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011-2012 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import collections
import cdist
import cdist.core
from cdist.util import fsproperty
log = logging.getLogger(__name__)
OBJECT_MARKER = '.cdist'
class IllegalObjectIdError(cdist.Error):
def __init__(self, object_id, message=None):
self.object_id = object_id
self.message = message or 'Illegal object id'
def __str__(self):
return '%s: %s' % (self.message, self.object_id)
class CdistObject(object):
"""Represents a cdist object.
All interaction with objects in cdist should be done through this class.
Directly accessing an object through the file system from python code is
a bug.
"""
# Constants for use with Object.state
STATE_PREPARED = "prepared"
STATE_RUNNING = "running"
STATE_DONE = "done"
@classmethod
def list_objects(cls, object_base_path, type_base_path):
"""Return a list of object instances"""
for object_name in cls.list_object_names(object_base_path):
type_name, object_id = cls.split_name(object_name)
yield cls(cdist.core.CdistType(type_base_path, type_name), object_base_path, object_id=object_id)
@classmethod
def list_type_names(cls, object_base_path):
"""Return a list of type names"""
return os.listdir(object_base_path)
@classmethod
def list_object_names(cls, object_base_path):
"""Return a list of object names"""
for path, dirs, files in os.walk(object_base_path):
if OBJECT_MARKER in dirs:
yield os.path.relpath(path, object_base_path)
@staticmethod
def split_name(object_name):
"""split_name('__type_name/the/object_id') -> ('__type_name', 'the/object_id')
Split the given object name into it's type and object_id parts.
"""
type_name = object_name.split(os.sep)[0]
# FIXME: allow object without object_id? e.g. for singleton
object_id = os.sep.join(object_name.split(os.sep)[1:])
return type_name, object_id
@staticmethod
def join_name(type_name, object_id):
"""join_name('__type_name', 'the/object_id') -> __type_name/the/object_id'
Join the given type_name and object_id into an object name.
"""
return os.path.join(type_name, object_id)
def validate_object_id(self):
# FIXME: also check that there is no object ID when type is singleton?
"""Validate the given object_id and raise IllegalObjectIdError if it's not valid.
"""
if self.object_id:
if OBJECT_MARKER in self.object_id.split(os.sep):
raise IllegalObjectIdError(self.object_id, 'object_id may not contain \'%s\'' % OBJECT_MARKER)
if '//' in self.object_id:
raise IllegalObjectIdError(self.object_id, 'object_id may not contain //')
# If no object_id and type is not singleton => error out
if not self.object_id and not self.cdist_type.is_singleton:
raise IllegalObjectIdError(self.object_id,
"Missing object_id and type is not a singleton.")
def __init__(self, cdist_type, base_path, object_id=None):
self.cdist_type = cdist_type # instance of Type
self.base_path = base_path
self.object_id = object_id
self.validate_object_id()
self.sanitise_object_id()
self.name = self.join_name(self.cdist_type.name, self.object_id)
self.path = os.path.join(self.cdist_type.path, self.object_id, OBJECT_MARKER)
self.absolute_path = os.path.join(self.base_path, self.path)
self.code_local_path = os.path.join(self.path, "code-local")
self.code_remote_path = os.path.join(self.path, "code-remote")
self.parameter_path = os.path.join(self.path, "parameter")
def object_from_name(self, object_name):
"""Convenience method for creating an object instance from an object name.
Mainly intended to create objects when resolving requirements.
e.g:
<CdistObject __foo/bar>.object_from_name('__other/object') -> <CdistObject __other/object>
"""
base_path = self.base_path
type_path = self.cdist_type.base_path
type_name, object_id = self.split_name(object_name)
cdist_type = self.cdist_type.__class__(type_path, type_name)
return self.__class__(cdist_type, base_path, object_id=object_id)
def __repr__(self):
return '<CdistObject %s>' % self.name
def __eq__(self, other):
"""define equality as 'name is the same'"""
return self.name == other.name
def __hash__(self):
return hash(self.name)
def __lt__(self, other):
return isinstance(other, self.__class__) and self.name < other.name
def sanitise_object_id(self):
"""
Remove leading and trailing slash (one only)
"""
# Allow empty object id for singletons
if self.object_id:
# Remove leading slash
if self.object_id[0] == '/':
self.object_id = self.object_id[1:]
# Remove trailing slash
if self.object_id[-1] == '/':
self.object_id = self.object_id[:-1]
# FIXME: still needed?
@property
def explorer_path(self):
"""Create and return the relative path to this objects explorers"""
# create absolute path
path = os.path.join(self.absolute_path, "explorer")
if not os.path.isdir(path):
os.mkdir(path)
# return relative path
return os.path.join(self.path, "explorer")
requirements = fsproperty.FileListProperty(lambda obj: os.path.join(obj.absolute_path, 'require'))
autorequire = fsproperty.FileListProperty(lambda obj: os.path.join(obj.absolute_path, 'autorequire'))
parameters = fsproperty.DirectoryDictProperty(lambda obj: os.path.join(obj.base_path, obj.parameter_path))
explorers = fsproperty.DirectoryDictProperty(lambda obj: os.path.join(obj.base_path, obj.explorer_path))
changed = fsproperty.FileBooleanProperty(lambda obj: os.path.join(obj.absolute_path, "changed"))
state = fsproperty.FileStringProperty(lambda obj: os.path.join(obj.absolute_path, "state"))
source = fsproperty.FileListProperty(lambda obj: os.path.join(obj.absolute_path, "source"))
code_local = fsproperty.FileStringProperty(lambda obj: os.path.join(obj.base_path, obj.code_local_path))
code_remote = fsproperty.FileStringProperty(lambda obj: os.path.join(obj.base_path, obj.code_remote_path))
@property
def exists(self):
"""Checks wether this cdist object exists on the file systems."""
return os.path.exists(self.absolute_path)
def create(self):
"""Create this cdist object on the filesystem.
"""
try:
os.makedirs(self.absolute_path, exist_ok=False)
absolute_parameter_path = os.path.join(self.base_path, self.parameter_path)
os.makedirs(absolute_parameter_path, exist_ok=False)
except EnvironmentError as error:
raise cdist.Error('Error creating directories for cdist object: %s: %s' % (self, error))

197
lib/cdist/core/cdist_type.py
View file

@ -1,197 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import cdist
class NoSuchTypeError(cdist.Error):
def __init__(self, type_path, type_absolute_path):
self.type_path = type_path
self.type_absolute_path = type_absolute_path
def __str__(self):
return "Type '%s' does not exist at %s" % (self.type_path, self.type_absolute_path)
class CdistType(object):
"""Represents a cdist type.
All interaction with types in cdist should be done through this class.
Directly accessing an type through the file system from python code is
a bug.
"""
@classmethod
def list_types(cls, base_path):
"""Return a list of type instances"""
for name in cls.list_type_names(base_path):
yield cls(base_path, name)
@classmethod
def list_type_names(cls, base_path):
"""Return a list of type names"""
return os.listdir(base_path)
_instances = {}
def __new__(cls, *args, **kwargs):
"""only one instance of each named type may exist"""
# name is second argument
name = args[1]
if not name in cls._instances:
instance = super(CdistType, cls).__new__(cls)
cls._instances[name] = instance
# return instance so __init__ is called
return cls._instances[name]
def __init__(self, base_path, name):
self.base_path = base_path
self.name = name
self.path = self.name
self.absolute_path = os.path.join(self.base_path, self.path)
if not os.path.isdir(self.absolute_path):
raise NoSuchTypeError(self.path, self.absolute_path)
self.manifest_path = os.path.join(self.name, "manifest")
self.explorer_path = os.path.join(self.name, "explorer")
self.gencode_local_path = os.path.join(self.name, "gencode-local")
self.gencode_remote_path = os.path.join(self.name, "gencode-remote")
self.manifest_path = os.path.join(self.name, "manifest")
self.__explorers = None
self.__required_parameters = None
self.__required_multiple_parameters = None
self.__optional_parameters = None
self.__optional_multiple_parameters = None
self.__boolean_parameters = None
def __repr__(self):
return '<CdistType %s>' % self.name
def __eq__(self, other):
return isinstance(other, self.__class__) and self.name == other.name
def __lt__(self, other):
return isinstance(other, self.__class__) and self.name < other.name
@property
def is_singleton(self):
"""Check whether a type is a singleton."""
return os.path.isfile(os.path.join(self.absolute_path, "singleton"))
@property
def is_install(self):
"""Check whether a type is used for installation (if not: for configuration)"""
return os.path.isfile(os.path.join(self.absolute_path, "install"))
@property
def explorers(self):
"""Return a list of available explorers"""
if not self.__explorers:
try:
self.__explorers = os.listdir(os.path.join(self.absolute_path, "explorer"))
except EnvironmentError:
# error ignored
self.__explorers = []
return self.__explorers
@property
def required_parameters(self):
"""Return a list of required parameters"""
if not self.__required_parameters:
parameters = []
try:
with open(os.path.join(self.absolute_path, "parameter", "required")) as fd:
for line in fd:
parameters.append(line.strip())
except EnvironmentError:
# error ignored
pass
finally:
self.__required_parameters = parameters
return self.__required_parameters
@property
def required_multiple_parameters(self):
"""Return a list of required multiple parameters"""
if not self.__required_multiple_parameters:
parameters = []
try:
with open(os.path.join(self.absolute_path, "parameter", "required_multiple")) as fd:
for line in fd:
parameters.append(line.strip())
except EnvironmentError:
# error ignored
pass
finally:
self.__required_multiple_parameters = parameters
return self.__required_multiple_parameters
@property
def optional_parameters(self):
"""Return a list of optional parameters"""
if not self.__optional_parameters:
parameters = []
try:
with open(os.path.join(self.absolute_path, "parameter", "optional")) as fd:
for line in fd:
parameters.append(line.strip())
except EnvironmentError:
# error ignored
pass
finally:
self.__optional_parameters = parameters
return self.__optional_parameters
@property
def optional_multiple_parameters(self):
"""Return a list of optional multiple parameters"""
if not self.__optional_multiple_parameters:
parameters = []
try:
with open(os.path.join(self.absolute_path, "parameter", "optional_multiple")) as fd:
for line in fd:
parameters.append(line.strip())
except EnvironmentError:
# error ignored
pass
finally:
self.__optional_multiple_parameters = parameters
return self.__optional_multiple_parameters
@property
def boolean_parameters(self):
"""Return a list of boolean parameters"""
if not self.__boolean_parameters:
parameters = []
try:
with open(os.path.join(self.absolute_path, "parameter", "boolean")) as fd:
for line in fd:
parameters.append(line.strip())
except EnvironmentError:
# error ignored
pass
finally:
self.__boolean_parameters = parameters
return self.__boolean_parameters

139
lib/cdist/core/code.py
View file

@ -1,139 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import cdist
log = logging.getLogger(__name__)
'''
common:
runs only locally, does not need remote
env:
PATH: prepend directory with type emulator symlinks == local.bin_path
__target_host: the target host we are working on
__cdist_manifest: full qualified path of the manifest == script
__cdist_type_base_path: full qualified path to the directory where types are defined for use in type emulator
== local.type_path
gencode-local
script: full qualified path to a types gencode-local
env:
__target_host: the target host we are working on
__global: full qualified path to the global output dir == local.out_path
__object: full qualified path to the object's dir
__object_id: the objects id
__object_fq: full qualified object id, iow: $type.name + / + object_id
__type: full qualified path to the type's dir
returns: string containing the generated code or None
gencode-remote
script: full qualified path to a types gencode-remote
env:
__target_host: the target host we are working on
__global: full qualified path to the global output dir == local.out_path
__object: full qualified path to the object's dir
__object_id: the objects id
__object_fq: full qualified object id, iow: $type.name + / + object_id
__type: full qualified path to the type's dir
returns: string containing the generated code or None
code-local
script: full qualified path to object's code-local
- run script localy
returns: string containing the output
code-remote
script: full qualified path to object's code-remote
- copy script to remote
- run script remotely
returns: string containing the output
'''
class Code(object):
"""Generates and executes cdist code scripts.
"""
def __init__(self, target_host, local, remote):
self.target_host = target_host
self.local = local
self.remote = remote
self.env = {
'__target_host': self.target_host,
'__global': self.local.out_path,
}
def _run_gencode(self, cdist_object, which):
cdist_type = cdist_object.cdist_type
script = os.path.join(self.local.type_path, getattr(cdist_type, 'gencode_%s_path' % which))
if os.path.isfile(script):
env = os.environ.copy()
env.update(self.env)
env.update({
'__type': cdist_object.cdist_type.absolute_path,
'__object': cdist_object.absolute_path,
'__object_id': cdist_object.object_id,
'__object_name': cdist_object.name,
})
return self.local.run_script(script, env=env, return_output=True)
def run_gencode_local(self, cdist_object):
"""Run the gencode-local script for the given cdist object."""
return self._run_gencode(cdist_object, 'local')
def run_gencode_remote(self, cdist_object):
"""Run the gencode-remote script for the given cdist object."""
return self._run_gencode(cdist_object, 'remote')
def transfer_code_remote(self, cdist_object):
"""Transfer the code_remote script for the given object to the remote side."""
source = os.path.join(self.local.object_path, cdist_object.code_remote_path)
destination = os.path.join(self.remote.object_path, cdist_object.code_remote_path)
# FIXME: BUG: do not create destination, but top level of destination!
# FIXME: BUG2: we are called AFTER the code-remote has been transferred already:
# mkdir: cannot create directory `/var/lib/cdist/object/__directory/etc/acpi/actions/.cdist/code-remote': File exists
# OR: this is from previous run -> cleanup missing!
self.remote.mkdir(destination)
self.remote.transfer(source, destination)
def _run_code(self, cdist_object, which):
which_exec = getattr(self, which)
script = os.path.join(which_exec.object_path, getattr(cdist_object, 'code_%s_path' % which))
return which_exec.run_script(script)
def run_code_local(self, cdist_object):
"""Run the code-local script for the given cdist object."""
return self._run_code(cdist_object, 'local')
def run_code_remote(self, cdist_object):
"""Run the code-remote script for the given cdist object on the remote side."""
return self._run_code(cdist_object, 'remote')

163
lib/cdist/core/explorer.py
View file

@ -1,163 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import cdist
'''
common:
runs only remotely, needs local and remote to construct paths
env:
__explorer: full qualified path to other global explorers on remote side
-> remote.global_explorer_path
a global explorer is:
- a script
- executed on the remote side
- returns its output as a string
env:
creates: nothing, returns output
type explorer is:
- a script
- executed on the remote side for each object instance
- returns its output as a string
env:
__object: full qualified path to the object's remote dir
__object_id: the objects id
__object_fq: full qualified object id, iow: $type.name + / + object_id
__type_explorer: full qualified path to the other type explorers on remote side
creates: nothing, returns output
'''
class Explorer(object):
"""Executes cdist explorers.
"""
def __init__(self, target_host, local, remote):
self.target_host = target_host
self.log = logging.getLogger(target_host)
self.local = local
self.remote = remote
self.env = {
'__target_host': self.target_host,
'__explorer': self.remote.global_explorer_path,
}
self._type_explorers_transferred = []
### global
def list_global_explorer_names(self):
"""Return a list of global explorer names."""
return os.listdir(self.local.global_explorer_path)
def run_global_explorers(self, out_path):
"""Run global explorers and save output to files in the given
out_path directory.
"""
self.log.info("Running global explorers")
self.transfer_global_explorers()
for explorer in self.list_global_explorer_names():
output = self.run_global_explorer(explorer)
path = os.path.join(out_path, explorer)
with open(path, 'w') as fd:
fd.write(output)
def transfer_global_explorers(self):
"""Transfer the global explorers to the remote side."""
self.remote.mkdir(self.remote.global_explorer_path)
self.remote.transfer(self.local.global_explorer_path, self.remote.global_explorer_path)
def run_global_explorer(self, explorer):
"""Run the given global explorer and return it's output."""
script = os.path.join(self.remote.global_explorer_path, explorer)
return self.remote.run_script(script, env=self.env, return_output=True)
### type
def list_type_explorer_names(self, cdist_type):
"""Return a list of explorer names for the given type."""
source = os.path.join(self.local.type_path, cdist_type.explorer_path)
try:
return os.listdir(source)
except EnvironmentError:
return []
def run_type_explorers(self, cdist_object):
"""Run the type explorers for the given object and save their output
in the object.
"""
self.log.debug("Transfering type explorers for type: %s", cdist_object.cdist_type)
self.transfer_type_explorers(cdist_object.cdist_type)
self.log.debug("Transfering object parameters for object: %s", cdist_object.name)
self.transfer_object_parameters(cdist_object)
for explorer in self.list_type_explorer_names(cdist_object.cdist_type):
output = self.run_type_explorer(explorer, cdist_object)
self.log.debug("Running type explorer '%s' for object '%s'", explorer, cdist_object.name)
cdist_object.explorers[explorer] = output
def run_type_explorer(self, explorer, cdist_object):
"""Run the given type explorer for the given object and return it's output."""
cdist_type = cdist_object.cdist_type
env = self.env.copy()
env.update({
'__object': os.path.join(self.remote.object_path, cdist_object.path),
'__object_id': cdist_object.object_id,
'__object_name': cdist_object.name,
'__object_fq': cdist_object.path,
'__type_explorer': os.path.join(self.remote.type_path, cdist_type.explorer_path)
})
script = os.path.join(self.remote.type_path, cdist_type.explorer_path, explorer)
return self.remote.run_script(script, env=env, return_output=True)
def transfer_type_explorers(self, cdist_type):
"""Transfer the type explorers for the given type to the remote side."""
if cdist_type.explorers:
if cdist_type.name in self._type_explorers_transferred:
self.log.debug("Skipping retransfer of type explorers for: %s", cdist_type)
else:
source = os.path.join(self.local.type_path, cdist_type.explorer_path)
destination = os.path.join(self.remote.type_path, cdist_type.explorer_path)
self.remote.mkdir(destination)
self.remote.transfer(source, destination)
self._type_explorers_transferred.append(cdist_type.name)
def transfer_object_parameters(self, cdist_object):
"""Transfer the parameters for the given object to the remote side."""
if cdist_object.parameters:
source = os.path.join(self.local.object_path, cdist_object.parameter_path)
destination = os.path.join(self.remote.object_path, cdist_object.parameter_path)
self.remote.mkdir(destination)
self.remote.transfer(source, destination)

102
lib/cdist/core/manifest.py
View file

@ -1,102 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import cdist
'''
common:
runs only locally, does not need remote
env:
PATH: prepend directory with type emulator symlinks == local.bin_path
__target_host: the target host we are working on
__global: full qualified path to the global output dir == local.out_path
__cdist_manifest: full qualified path of the manifest == script
__cdist_type_base_path: full qualified path to the directory where types are defined for use in type emulator
== local.type_path
initial manifest is:
script: full qualified path to the initial manifest
env:
__manifest: path to .../conf/manifest/ == local.manifest_path
creates: new objects through type emulator
type manifeste is:
script: full qualified path to the type manifest
env:
__object: full qualified path to the object's dir
__object_id: the objects id
__object_fq: full qualified object id, iow: $type.name + / + object_id
__type: full qualified path to the type's dir
creates: new objects through type emulator
'''
class Manifest(object):
"""Executes cdist manifests.
"""
def __init__(self, target_host, local):
self.target_host = target_host
self.local = local
self.log = logging.getLogger(self.target_host)
self.env = {
'PATH': "%s:%s" % (self.local.bin_path, os.environ['PATH']),
'__target_host': self.target_host,
'__global': self.local.out_path,
'__cdist_type_base_path': self.local.type_path, # for use in type emulator
}
if self.log.getEffectiveLevel() == logging.DEBUG:
self.env.update({'__cdist_debug': "yes" })
def run_initial_manifest(self, script):
env = os.environ.copy()
env.update(self.env)
env['__manifest'] = self.local.manifest_path
env['__cdist_manifest'] = script
self.log.info("Running initial manifest " + self.local.manifest_path)
self.local.run_script(script, env=env)
def run_type_manifest(self, cdist_object):
script = os.path.join(self.local.type_path, cdist_object.cdist_type.manifest_path)
if os.path.isfile(script):
env = os.environ.copy()
env.update(self.env)
env.update({
'__manifest': self.local.manifest_path,
'__object': cdist_object.absolute_path,
'__object_id': cdist_object.object_id,
'__object_name': cdist_object.name,
'__type': cdist_object.cdist_type.absolute_path,
'__cdist_manifest': script,
})
self.local.run_script(script, env=env)

204
lib/cdist/emulator.py
View file

@ -1,204 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011-2012 Nico Schottelius (nico-cdist at schottelius.org)
# 2012 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import argparse
import logging
import os
import sys
import cdist
from cdist import core
class Emulator(object):
def __init__(self, argv):
self.argv = argv
self.object_id = False
self.global_path = os.environ['__global']
self.target_host = os.environ['__target_host']
# Internally only
self.object_source = os.environ['__cdist_manifest']
self.type_base_path = os.environ['__cdist_type_base_path']
self.object_base_path = os.path.join(self.global_path, "object")
self.type_name = os.path.basename(argv[0])
self.cdist_type = core.CdistType(self.type_base_path, self.type_name)
self.__init_log()
def filter(self, record):
"""Add hostname and object to logs via logging Filter"""
prefix = self.target_host + ": (emulator)"
if self.object_id:
prefix = prefix + " " + self.type_name + "/" + self.object_id
record.msg = prefix + ": " + record.msg
return True
def run(self):
"""Emulate type commands (i.e. __file and co)"""
if '__install' in os.environ:
if not self.cdist_type.is_install:
self.log.debug("Running in install mode, ignoring non install type")
return True
self.commandline()
self.setup_object()
self.save_stdin()
self.record_requirements()
self.record_auto_requirements()
self.log.debug("Finished %s %s" % (self.cdist_object.path, self.parameters))
def __init_log(self):
"""Setup logging facility"""
logformat = '%(levelname)s: %(message)s'
logging.basicConfig(format=logformat)
if '__cdist_debug' in os.environ:
logging.root.setLevel(logging.DEBUG)
else:
logging.root.setLevel(logging.INFO)
self.log = logging.getLogger(__name__)
self.log.addFilter(self)
def commandline(self):
"""Parse command line"""
parser = argparse.ArgumentParser(add_help=False, argument_default=argparse.SUPPRESS)
for parameter in self.cdist_type.required_parameters:
argument = "--" + parameter
parser.add_argument(argument, dest=parameter, action='store', required=True)
for parameter in self.cdist_type.required_multiple_parameters:
argument = "--" + parameter
parser.add_argument(argument, dest=parameter, action='append', required=True)
for parameter in self.cdist_type.optional_parameters:
argument = "--" + parameter
parser.add_argument(argument, dest=parameter, action='store', required=False)
for parameter in self.cdist_type.optional_multiple_parameters:
argument = "--" + parameter
parser.add_argument(argument, dest=parameter, action='append', required=False)
for parameter in self.cdist_type.boolean_parameters:
argument = "--" + parameter
parser.add_argument(argument, dest=parameter, action='store_const', const='')
# If not singleton support one positional parameter
if not self.cdist_type.is_singleton:
parser.add_argument("object_id", nargs=1)
# And finally parse/verify parameter
self.args = parser.parse_args(self.argv[1:])
self.log.debug('Args: %s' % self.args)
def setup_object(self):
# Setup object_id - FIXME: unset / do not setup anymore!
if self.cdist_type.is_singleton:
self.object_id = "singleton"
else:
self.object_id = self.args.object_id[0]
del self.args.object_id
# Instantiate the cdist object we are defining
self.cdist_object = core.CdistObject(self.cdist_type, self.object_base_path, self.object_id)
# Create object with given parameters
self.parameters = {}
for key,value in vars(self.args).items():
if value is not None:
self.parameters[key] = value
if self.cdist_object.exists:
if self.cdist_object.parameters != self.parameters:
raise cdist.Error("Object %s already exists with conflicting parameters:\n%s: %s\n%s: %s"
% (self.cdist_object.name, " ".join(self.cdist_object.source), self.cdist_object.parameters, self.object_source, self.parameters)
)
else:
self.cdist_object.create()
self.cdist_object.parameters = self.parameters
# Record / Append source
self.cdist_object.source.append(self.object_source)
chunk_size = 8192
def _read_stdin(self):
return sys.stdin.buffer.read(self.chunk_size)
def save_stdin(self):
"""If something is written to stdin, save it in the object as
$__object/stdin so it can be accessed in manifest and gencode-*
scripts.
"""
if not sys.stdin.isatty():
try:
# go directly to file instead of using CdistObject's api
# as that does not support streaming
path = os.path.join(self.cdist_object.absolute_path, 'stdin')
with open(path, 'wb') as fd:
chunk = self._read_stdin()
while chunk:
fd.write(chunk)
chunk = self._read_stdin()
except EnvironmentError as e:
raise cdist.Error('Failed to read from stdin: %s' % e)
def record_requirements(self):
"""record requirements"""
if "require" in os.environ:
requirements = os.environ['require']
self.log.debug("reqs = " + requirements)
for requirement in requirements.split(" "):
# Ignore empty fields - probably the only field anyway
if len(requirement) == 0: continue
# Raises an error, if object cannot be created
cdist_object = self.cdist_object.object_from_name(requirement)
self.log.debug("Recording requirement: " + requirement)
# Save the sanitised version, not the user supplied one
# (__file//bar => __file/bar)
# This ensures pattern matching is done against sanitised list
self.cdist_object.requirements.append(cdist_object.name)
def record_auto_requirements(self):
"""An object shall automatically depend on all objects that it defined in it's type manifest.
"""
# __object_name is the name of the object whose type manifest is currently executed
__object_name = os.environ.get('__object_name', None)
if __object_name:
# The object whose type manifest is currently run
parent = self.cdist_object.object_from_name(__object_name)
# The object currently being defined
current_object = self.cdist_object
# As parent defined current_object it shall automatically depend on it.
# But only if the user hasn't said otherwise.
# Must prevent circular dependencies.
if not parent.name in current_object.requirements:
parent.autorequire.append(current_object.name)

0
lib/cdist/exec/__init__.py
View file

124
lib/cdist/exec/local.py
View file

@ -1,124 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
# FIXME: common base class with Remote?
import io
import os
import sys
import subprocess
import shutil
import logging
import cdist
from cdist import core
class Local(object):
"""Execute commands locally.
All interaction with the local side should be done through this class.
Directly accessing the local side from python code is a bug.
"""
def __init__(self, target_host, local_base_path, out_path):
self.target_host = target_host
self.base_path = local_base_path
# Local input
self.cache_path = os.path.join(self.base_path, "cache")
self.conf_path = os.path.join(self.base_path, "conf")
self.global_explorer_path = os.path.join(self.conf_path, "explorer")
self.manifest_path = os.path.join(self.conf_path, "manifest")
self.type_path = os.path.join(self.conf_path, "type")
# FIXME: should not be needed anywhere
self.lib_path = os.path.join(self.base_path, "lib")
# Local output
self.out_path = out_path
self.bin_path = os.path.join(self.out_path, "bin")
self.global_explorer_out_path = os.path.join(self.out_path, "explorer")
self.object_path = os.path.join(self.out_path, "object")
self.log = logging.getLogger(self.target_host)
# Setup file permissions using umask
os.umask(0o077)
def create_directories(self):
self.mkdir(self.out_path)
self.mkdir(self.global_explorer_out_path)
self.mkdir(self.bin_path)
def rmdir(self, path):
"""Remove directory on the local side."""
self.log.debug("Local rmdir: %s", path)
shutil.rmtree(path)
def mkdir(self, path):
"""Create directory on the local side."""
self.log.debug("Local mkdir: %s", path)
os.makedirs(path, exist_ok=True)
def run(self, command, env=None, return_output=False):
"""Run the given command with the given environment.
Return the output as a string.
"""
assert isinstance(command, (list, tuple)), "list or tuple argument expected, got: %s" % command
self.log.debug("Local run: %s", command)
if env is None:
env = os.environ.copy()
# Export __target_host for use in __remote_{copy,exec} scripts
env['__target_host'] = self.target_host
try:
if return_output:
return subprocess.check_output(command, env=env).decode()
else:
subprocess.check_call(command, env=env)
except subprocess.CalledProcessError:
raise cdist.Error("Command failed: " + " ".join(command))
except OSError as error:
raise cdist.Error(" ".join(*args) + ": " + error.args[1])
def run_script(self, script, env=None, return_output=False):
"""Run the given script with the given environment.
Return the output as a string.
"""
command = ["/bin/sh", "-e"]
command.append(script)
return self.run(command, env, return_output)
def link_emulator(self, exec_path):
"""Link emulator to types"""
src = os.path.abspath(exec_path)
for cdist_type in core.CdistType.list_types(self.type_path):
dst = os.path.join(self.bin_path, cdist_type.name)
self.log.debug("Linking emulator: %s to %s", src, dst)
try:
os.symlink(src, dst)
except OSError as e:
raise cdist.Error("Linking emulator from %s to %s failed: %s" % (src, dst, e.__str__()))

139
lib/cdist/exec/remote.py
View file

@ -1,139 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import io
import os
import sys
import subprocess
import logging
import cdist
class DecodeError(cdist.Error):
def __init__(self, command):
self.command = command
def __str__(self):
return "Cannot decode output of " + " ".join(self.command)
class Remote(object):
"""Execute commands remotely.
All interaction with the remote side should be done through this class.
Directly accessing the remote side from python code is a bug.
"""
def __init__(self, target_host, remote_base_path, remote_exec, remote_copy):
self.target_host = target_host
self.base_path = remote_base_path
self._exec = remote_exec
self._copy = remote_copy
self.conf_path = os.path.join(self.base_path, "conf")
self.object_path = os.path.join(self.base_path, "object")
self.type_path = os.path.join(self.conf_path, "type")
self.global_explorer_path = os.path.join(self.conf_path, "explorer")
self.log = logging.getLogger(self.target_host)
def create_directories(self):
self.rmdir(self.base_path)
self.mkdir(self.base_path)
self.mkdir(self.conf_path)
def rmdir(self, path):
"""Remove directory on the remote side."""
self.log.debug("Remote rmdir: %s", path)
self.run(["rm", "-rf", path])
def mkdir(self, path):
"""Create directory on the remote side."""
self.log.debug("Remote mkdir: %s", path)
self.run(["mkdir", "-p", path])
def transfer(self, source, destination):
"""Transfer a file or directory to the remote side."""
self.log.debug("Remote transfer: %s -> %s", source, destination)
self.rmdir(destination)
command = self._copy.split()
command.extend(["-r", source, self.target_host + ":" + destination])
self._run_command(command)
def run_script(self, script, env=None, return_output=False):
"""Run the given script with the given environment on the remote side.
Return the output as a string.
"""
command = ["/bin/sh", "-e"]
command.append(script)
return self.run(command, env, return_output)
def run(self, command, env=None, return_output=False):
"""Run the given command with the given environment on the remote side.
Return the output as a string.
"""
# prefix given command with remote_exec
cmd = self._exec.split()
cmd.append(self.target_host)
# Always call umask before actual call to ensure proper file permissions
cmd.append("umask 077;")
# FIXME: replace this by -o SendEnv name -o SendEnv name ... to ssh?
# can't pass environment to remote side, so prepend command with
# variable declarations
if env:
remote_env = ["%s=%s" % item for item in env.items()]
cmd.extend(remote_env)
cmd.extend(command)
return self._run_command(cmd, env=env, return_output=return_output)
def _run_command(self, command, env=None, return_output=False):
"""Run the given command with the given environment.
Return the output as a string.
"""
assert isinstance(command, (list, tuple)), "list or tuple argument expected, got: %s" % command
# export target_host for use in __remote_{exec,copy} scripts
os_environ = os.environ.copy()
os_environ['__target_host'] = self.target_host
self.log.debug("Remote run: %s", command)
try:
if return_output:
return subprocess.check_output(command, env=os_environ).decode()
else:
subprocess.check_call(command, env=os_environ)
except subprocess.CalledProcessError:
raise cdist.Error("Command failed: " + " ".join(command))
except OSError as error:
raise cdist.Error(" ".join(*args) + ": " + error.args[1])
except UnicodeDecodeError:
raise DecodeError(command)

33
lib/cdist/install.py
View file

@ -1,33 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2010-2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import cdist.config_install
class Install(cdist.config_install.ConfigInstall):
def __init__(self, *args, **kargs):
"""Enhance config install with install support"""
# Setup environ to be used in emulator
os.environ['__install'] = "yes"
super().__init__(*args, **kargs)

167
lib/cdist/resolver.py
View file

@ -1,167 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import logging
import os
import itertools
import fnmatch
import cdist
log = logging.getLogger(__name__)
class CircularReferenceError(cdist.Error):
def __init__(self, cdist_object, required_object):
self.cdist_object = cdist_object
self.required_object = required_object
def __str__(self):
return 'Circular reference detected: %s -> %s' % (self.cdist_object.name, self.required_object.name)
class RequirementNotFoundError(cdist.Error):
def __init__(self, requirement):
self.requirement = requirement
def __str__(self):
return 'Requirement could not be found: %s' % self.requirement
class DependencyResolver(object):
"""Cdist's dependency resolver.
Usage:
>> resolver = DependencyResolver(list_of_objects)
# Easy access to the objects we are working with
>> resolver.objects['__some_type/object_id']
<CdistObject __some_type/object_id>
# Easy access to a specific objects dependencies
>> resolver.dependencies['__some_type/object_id']
[<CdistObject __other_type/dependency>, <CdistObject __some_type/object_id>]
# Pretty print the dependency graph
>> from pprint import pprint
>> pprint(resolver.dependencies)
# Iterate over all existing objects in the correct order
>> for cdist_object in resolver:
>> do_something_with(cdist_object)
"""
def __init__(self, objects, logger=None):
self.objects = dict((o.name, o) for o in objects)
self._dependencies = None
self.log = logger or log
@property
def dependencies(self):
"""Build the dependency graph.
Returns a dict where the keys are the object names and the values are
lists of all dependencies including the key object itself.
"""
if self._dependencies is None:
self._dependencies = d = {}
self._preprocess_requirements()
for name,cdist_object in self.objects.items():
resolved = []
unresolved = []
self._resolve_object_dependencies(cdist_object, resolved, unresolved)
d[name] = resolved
return self._dependencies
def find_requirements_by_name(self, requirements):
"""Takes a list of requirement patterns and returns a list of matching object instances.
Patterns are expected to be Unix shell-style wildcards for use with fnmatch.filter.
find_requirements_by_name(['__type/object_id', '__other_type/*']) ->
[<Object __type/object_id>, <Object __other_type/any>, <Object __other_type/match>]
"""
object_names = self.objects.keys()
for pattern in requirements:
found = False
for requirement in fnmatch.filter(object_names, pattern):
found = True
yield self.objects[requirement]
if not found:
# FIXME: get rid of the singleton object_id, it should be invisible to the code -> hide it in Object
singleton = os.path.join(pattern, 'singleton')
if singleton in self.objects:
yield self.objects[singleton]
else:
raise RequirementNotFoundError(pattern)
def _preprocess_requirements(self):
"""Find all autorequire dependencies and merge them to be just requirements
for further processing.
"""
for cdist_object in self.objects.values():
if cdist_object.autorequire:
# The objects (children) that this cdist_object (parent) defined
# in it's type manifest shall inherit all explicit requirements
# that the parent has so that user defined requirements are
# fullfilled and processed in the expected order.
for auto_requirement in self.find_requirements_by_name(cdist_object.autorequire):
for requirement in cdist_object.requirements:
if requirement not in auto_requirement.requirements:
auto_requirement.requirements.append(requirement)
# On the other hand the parent shall depend on all the children
# it created so that the user can setup dependencies on it as a
# whole without having to know anything about the parents
# internals.
cdist_object.requirements.extend(cdist_object.autorequire)
# As we changed the object on disc, we have to ensure it is not
# preprocessed again if someone would call us multiple times.
cdist_object.autorequire = []
def _resolve_object_dependencies(self, cdist_object, resolved, unresolved):
"""Resolve all dependencies for the given cdist_object and store them
in the list which is passed as the 'resolved' arguments.
e.g.
resolved = []
unresolved = []
resolve_object_dependencies(some_object, resolved, unresolved)
print("Dependencies for %s: %s" % (some_object, resolved))
"""
self.log.debug('Resolving dependencies for: %s' % cdist_object.name)
try:
unresolved.append(cdist_object)
for required_object in self.find_requirements_by_name(cdist_object.requirements):
self.log.debug("Object %s requires %s", cdist_object, required_object)
if required_object not in resolved:
if required_object in unresolved:
raise CircularReferenceError(cdist_object, required_object)
self._resolve_object_dependencies(required_object, resolved, unresolved)
resolved.append(cdist_object)
unresolved.remove(cdist_object)
except RequirementNotFoundError as e:
raise cdist.CdistObjectError(cdist_object, "requires non-existing " + e.requirement)
def __iter__(self):
"""Iterate over all unique objects and yield them in the correct order.
"""
iterable = itertools.chain(*self.dependencies.values())
# Keep record of objects that have already been seen
seen = set()
seen_add = seen.add
for cdist_object in itertools.filterfalse(seen.__contains__, iterable):
seen_add(cdist_object)
yield cdist_object

38
lib/cdist/test/__init__.py
View file

@ -1,38 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import unittest
import tempfile
cdist_base_path = os.path.abspath(
os.path.join(os.path.dirname(os.path.realpath(__file__)), "../../../"))
cdist_exec_path = os.path.join(cdist_base_path, "bin/cdist")
class CdistTestCase(unittest.TestCase):
def mkdtemp(self, **kwargs):
return tempfile.mkdtemp(prefix='tmp.cdist.test.', **kwargs)
def mkstemp(self, **kwargs):
return tempfile.mkstemp(prefix='tmp.cdist.test.', **kwargs)

48
lib/cdist/test/__main__.py
View file

@ -1,48 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import imp
import os
import sys
import unittest
base_dir = os.path.dirname(os.path.realpath(__file__))
test_modules = []
for possible_test in os.listdir(base_dir):
filename = "__init__.py"
mod_path = os.path.join(base_dir, possible_test, filename)
if os.path.isfile(mod_path):
test_modules.append(possible_test)
suites = []
for test_module in test_modules:
module_parameters = imp.find_module(test_module, [base_dir])
module = imp.load_module("cdist.test." + test_module, *module_parameters)
suite = unittest.defaultTestLoader.loadTestsFromModule(module)
# print("Got suite: " + suite.__str__())
suites.append(suite)
all_suites = unittest.TestSuite(suites)
unittest.TextTestRunner(verbosity=2).run(all_suites)

78
lib/cdist/test/autorequire/__init__.py
View file

@ -1,78 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
import cdist
from cdist import test
from cdist.exec import local
from cdist import core
from cdist.core import manifest
from cdist import resolver
from cdist import config
import cdist.context
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
local_base_path = fixtures
class AutorequireTestCase(test.CdistTestCase):
def setUp(self):
self.orig_environ = os.environ
os.environ = os.environ.copy()
self.target_host = 'localhost'
self.temp_dir = self.mkdtemp()
os.environ['__cdist_out_dir'] = self.temp_dir
self.context = cdist.context.Context(
target_host=self.target_host,
base_path=local_base_path,
exec_path=test.cdist_exec_path,
debug=False)
self.config = config.Config(self.context)
def tearDown(self):
os.environ = self.orig_environ
shutil.rmtree(self.temp_dir)
def test_implicit_dependencies(self):
self.context.initial_manifest = os.path.join(self.config.local.manifest_path, 'implicit_dependencies')
self.config.stage_prepare()
objects = core.CdistObject.list_objects(self.config.local.object_path, self.config.local.type_path)
dependency_resolver = resolver.DependencyResolver(objects)
expected_dependencies = [
dependency_resolver.objects['__package_special/b'],
dependency_resolver.objects['__package/b'],
dependency_resolver.objects['__package_special/a']
]
resolved_dependencies = dependency_resolver.dependencies['__package_special/a']
self.assertEqual(resolved_dependencies, expected_dependencies)
def test_circular_dependency(self):
self.context.initial_manifest = os.path.join(self.config.local.manifest_path, 'circular_dependency')
self.config.stage_prepare()
# raises CircularDependecyError
self.config.stage_run()

0
lib/cdist/test/autorequire/fixtures/conf/explorer/.keep
View file

2
lib/cdist/test/autorequire/fixtures/conf/manifest/circular_dependency
View file

@ -1,2 +0,0 @@
# this has triggered CircularReferenceError
__nfsroot_client test

3
lib/cdist/test/autorequire/fixtures/conf/manifest/implicit_dependencies
View file

@ -1,3 +0,0 @@
# this creates implicit dependencies through autorequire.
# this failed because autorequired dependencies where not aware of their anchestors dependencies
__top test

0
lib/cdist/test/autorequire/fixtures/conf/type/__addifnosuchline/.keep
View file

0
lib/cdist/test/autorequire/fixtures/conf/type/__directory/.keep
View file

3
lib/cdist/test/autorequire/fixtures/conf/type/__nfsroot_client/manifest
View file

@ -1,3 +0,0 @@
__user root
__root_ssh_authorized_key john
__root_ssh_authorized_key frank

1
lib/cdist/test/autorequire/fixtures/conf/type/__package/manifest
View file

@ -1 +0,0 @@
__package_special "$__object_id"

0
lib/cdist/test/autorequire/fixtures/conf/type/__package_special/.keep
View file

4
lib/cdist/test/autorequire/fixtures/conf/type/__root_ssh_authorized_key/manifest
View file

@ -1,4 +0,0 @@
user="$__object_id"
__directory /root/.ssh
require="__directory/root/.ssh" \
__addifnosuchline "ssh-root-$user"

2
lib/cdist/test/autorequire/fixtures/conf/type/__top/manifest
View file

@ -1,2 +0,0 @@
__package b
require="__package/b" __package a

0
lib/cdist/test/autorequire/fixtures/conf/type/__user/.keep
View file

42
lib/cdist/test/banner/__init__.py
View file

@ -1,42 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import io
import sys
import unittest
import cdist
import cdist.banner
class Banner(unittest.TestCase):
def setUp(self):
self.banner = cdist.BANNER + "\n"
def test_banner_output(self):
"""Check that printed banner equals saved banner"""
output = io.StringIO()
sys.stdout = output
cdist.banner.banner(None)
self.assertEqual(output.getvalue(), self.banner)

110
lib/cdist/test/code/__init__.py
View file

@ -1,110 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
import getpass
import logging
import cdist
from cdist import core
from cdist import test
from cdist.exec import local
from cdist.exec import remote
from cdist.core import code
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
local_base_path = fixtures
class CodeTestCase(test.CdistTestCase):
def setUp(self):
self.target_host = 'localhost'
self.local_base_path = local_base_path
self.out_path = self.mkdtemp()
self.local = local.Local(self.target_host, self.local_base_path, self.out_path)
self.local.create_directories()
self.remote_base_path = self.mkdtemp()
self.user = getpass.getuser()
remote_exec = "ssh -o User=%s -q" % self.user
remote_copy = "scp -o User=%s -q" % self.user
self.remote = remote.Remote(self.target_host, self.remote_base_path, remote_exec, remote_copy)
self.code = code.Code(self.target_host, self.local, self.remote)
self.cdist_type = core.CdistType(self.local.type_path, '__dump_environment')
self.cdist_object = core.CdistObject(self.cdist_type, self.local.object_path, 'whatever')
self.cdist_object.create()
self.log = logging.getLogger("cdist")
def tearDown(self):
shutil.rmtree(self.out_path)
shutil.rmtree(self.remote_base_path)
def test_run_gencode_local_environment(self):
output_string = self.code.run_gencode_local(self.cdist_object)
output_dict = {}
for line in output_string.split('\n'):
if line:
junk,value = line.split(': ')
key = junk.split(' ')[1]
output_dict[key] = value
self.assertEqual(output_dict['__target_host'], self.local.target_host)
self.assertEqual(output_dict['__global'], self.local.out_path)
self.assertEqual(output_dict['__type'], self.cdist_type.absolute_path)
self.assertEqual(output_dict['__object'], self.cdist_object.absolute_path)
self.assertEqual(output_dict['__object_id'], self.cdist_object.object_id)
self.assertEqual(output_dict['__object_name'], self.cdist_object.name)
def test_run_gencode_remote_environment(self):
output_string = self.code.run_gencode_remote(self.cdist_object)
output_dict = {}
for line in output_string.split('\n'):
if line:
junk,value = line.split(': ')
key = junk.split(' ')[1]
output_dict[key] = value
self.assertEqual(output_dict['__target_host'], self.local.target_host)
self.assertEqual(output_dict['__global'], self.local.out_path)
self.assertEqual(output_dict['__type'], self.cdist_type.absolute_path)
self.assertEqual(output_dict['__object'], self.cdist_object.absolute_path)
self.assertEqual(output_dict['__object_id'], self.cdist_object.object_id)
self.assertEqual(output_dict['__object_name'], self.cdist_object.name)
def test_transfer_code_remote(self):
self.cdist_object.code_remote = self.code.run_gencode_remote(self.cdist_object)
self.code.transfer_code_remote(self.cdist_object)
destination = os.path.join(self.remote.object_path, self.cdist_object.code_remote_path)
self.assertTrue(os.path.isfile(destination))
def test_run_code_local(self):
self.cdist_object.code_local = self.code.run_gencode_local(self.cdist_object)
self.code.run_code_local(self.cdist_object)
def test_run_code_remote_environment(self):
self.cdist_object.code_remote = self.code.run_gencode_remote(self.cdist_object)
self.code.transfer_code_remote(self.cdist_object)
self.code.run_code_remote(self.cdist_object)

8
lib/cdist/test/code/fixtures/conf/type/__dump_environment/gencode-local
View file

@ -1,8 +0,0 @@
#!/bin/sh
echo "echo __target_host: $__target_host"
echo "echo __global: $__global"
echo "echo __type: $__type"
echo "echo __object: $__object"
echo "echo __object_id: $__object_id"
echo "echo __object_name: $__object_name"

1
lib/cdist/test/code/fixtures/conf/type/__dump_environment/gencode-remote
View file

@ -1 +0,0 @@
gencode-local

266
lib/cdist/test/emulator/__init__.py
View file

@ -1,266 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
import string
import filecmp
import random
import cdist
from cdist import test
from cdist.exec import local
from cdist import emulator
from cdist import core
from cdist import config
import cdist.context
local_base_path = test.cdist_base_path
class EmulatorTestCase(test.CdistTestCase):
def setUp(self):
self.orig_environ = os.environ
os.environ = os.environ.copy()
self.temp_dir = self.mkdtemp()
handle, self.script = self.mkstemp(dir=self.temp_dir)
os.close(handle)
self.target_host = 'localhost'
out_path = self.temp_dir
self.local = local.Local(self.target_host, local_base_path, out_path)
self.local.create_directories()
self.env = {
'PATH': "%s:%s" % (self.local.bin_path, os.environ['PATH']),
'__target_host': self.target_host,
'__global': self.local.out_path,
'__cdist_type_base_path': self.local.type_path, # for use in type emulator
'__manifest': self.local.manifest_path,
'__cdist_manifest': self.script,
}
def tearDown(self):
os.environ = self.orig_environ
shutil.rmtree(self.temp_dir)
def test_nonexistent_type_exec(self):
argv = ['__does-not-exist']
os.environ.update(self.env)
self.assertRaises(core.NoSuchTypeError, emulator.Emulator, argv)
def test_nonexistent_type_requirement(self):
argv = ['__file', '/tmp/foobar']
os.environ.update(self.env)
os.environ['require'] = '__does-not-exist/some-id'
emu = emulator.Emulator(argv)
self.assertRaises(core.NoSuchTypeError, emu.run)
def test_illegal_object_id_requirement(self):
argv = ['__file', '/tmp/foobar']
os.environ.update(self.env)
os.environ['require'] = '__file/bad/id/with/.cdist/inside'
emu = emulator.Emulator(argv)
self.assertRaises(core.IllegalObjectIdError, emu.run)
def test_missing_object_id_requirement(self):
argv = ['__file', '/tmp/foobar']
os.environ.update(self.env)
os.environ['require'] = '__file'
emu = emulator.Emulator(argv)
self.assertRaises(core.IllegalObjectIdError, emu.run)
def test_singleton_object_requirement(self):
argv = ['__file', '/tmp/foobar']
os.environ.update(self.env)
os.environ['require'] = '__issue'
emu = emulator.Emulator(argv)
emu.run()
# if we get here all is fine
def test_requirement_pattern(self):
argv = ['__file', '/tmp/foobar']
os.environ.update(self.env)
os.environ['require'] = '__file/etc/*'
emu = emulator.Emulator(argv)
# if we get here all is fine
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
class AutoRequireEmulatorTestCase(test.CdistTestCase):
def setUp(self):
self.temp_dir = self.mkdtemp()
self.target_host = 'localhost'
out_path = self.temp_dir
_local_base_path = fixtures
self.local = local.Local(self.target_host, _local_base_path, out_path)
self.local.create_directories()
self.local.link_emulator(cdist.test.cdist_exec_path)
self.manifest = core.Manifest(self.target_host, self.local)
def tearDown(self):
shutil.rmtree(self.temp_dir)
def test_autorequire(self):
initial_manifest = os.path.join(self.local.manifest_path, "init")
self.manifest.run_initial_manifest(initial_manifest)
cdist_type = core.CdistType(self.local.type_path, '__saturn')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'singleton')
self.manifest.run_type_manifest(cdist_object)
expected = ['__planet/Saturn', '__moon/Prometheus']
self.assertEqual(sorted(cdist_object.autorequire), sorted(expected))
class ArgumentsTestCase(test.CdistTestCase):
def setUp(self):
self.temp_dir = self.mkdtemp()
self.target_host = 'localhost'
out_path = self.temp_dir
handle, self.script = self.mkstemp(dir=self.temp_dir)
os.close(handle)
_local_base_path = fixtures
self.local = local.Local(self.target_host, _local_base_path, out_path)
self.local.create_directories()
self.local.link_emulator(test.cdist_exec_path)
self.env = {
'PATH': "%s:%s" % (self.local.bin_path, os.environ['PATH']),
'__target_host': self.target_host,
'__global': self.local.out_path,
'__cdist_type_base_path': self.local.type_path, # for use in type emulator
'__manifest': self.local.manifest_path,
'__cdist_manifest': self.script,
}
def tearDown(self):
shutil.rmtree(self.temp_dir)
def test_arguments_with_dashes(self):
argv = ['__arguments_with_dashes', 'some-id', '--with-dash', 'some value']
os.environ.update(self.env)
emu = emulator.Emulator(argv)
emu.run()
cdist_type = core.CdistType(self.local.type_path, '__arguments_with_dashes')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'some-id')
self.assertTrue('with-dash' in cdist_object.parameters)
def test_boolean(self):
type_name = '__arguments_boolean'
object_id = 'some-id'
argv = [type_name, object_id, '--boolean1']
os.environ.update(self.env)
emu = emulator.Emulator(argv)
emu.run()
cdist_type = core.CdistType(self.local.type_path, type_name)
cdist_object = core.CdistObject(cdist_type, self.local.object_path, object_id)
self.assertTrue('boolean1' in cdist_object.parameters)
self.assertFalse('boolean2' in cdist_object.parameters)
# empty file -> True
self.assertTrue(cdist_object.parameters['boolean1'] == '')
def test_required(self):
type_name = '__arguments_required'
object_id = 'some-id'
value = 'some value'
argv = [type_name, object_id, '--required1', value, '--required2', value]
os.environ.update(self.env)
emu = emulator.Emulator(argv)
emu.run()
cdist_type = core.CdistType(self.local.type_path, type_name)
cdist_object = core.CdistObject(cdist_type, self.local.object_path, object_id)
self.assertTrue('required1' in cdist_object.parameters)
self.assertTrue('required2' in cdist_object.parameters)
self.assertEqual(cdist_object.parameters['required1'], value)
self.assertEqual(cdist_object.parameters['required2'], value)
# def test_required_missing(self):
# type_name = '__arguments_required'
# object_id = 'some-id'
# value = 'some value'
# argv = [type_name, object_id, '--required1', value]
# os.environ.update(self.env)
# emu = emulator.Emulator(argv)
#
# self.assertRaises(SystemExit, emu.run)
def test_optional(self):
type_name = '__arguments_optional'
object_id = 'some-id'
value = 'some value'
argv = [type_name, object_id, '--optional1', value]
os.environ.update(self.env)
emu = emulator.Emulator(argv)
emu.run()
cdist_type = core.CdistType(self.local.type_path, type_name)
cdist_object = core.CdistObject(cdist_type, self.local.object_path, object_id)
self.assertTrue('optional1' in cdist_object.parameters)
self.assertFalse('optional2' in cdist_object.parameters)
self.assertEqual(cdist_object.parameters['optional1'], value)
class StdinTestCase(test.CdistTestCase):
def setUp(self):
self.orig_environ = os.environ
os.environ = os.environ.copy()
self.target_host = 'localhost'
self.temp_dir = self.mkdtemp()
os.environ['__cdist_out_dir'] = self.temp_dir
local_base_path = fixtures
self.context = cdist.context.Context(
target_host=self.target_host,
remote_copy='scp -o User=root -q',
remote_exec='ssh -o User=root -q',
base_path=local_base_path,
exec_path=test.cdist_exec_path,
debug=False)
self.config = config.Config(self.context)
def tearDown(self):
os.environ = self.orig_environ
shutil.rmtree(self.temp_dir)
def test_file_from_stdin(self):
handle, destination = self.mkstemp(dir=self.temp_dir)
os.close(handle)
source_handle, source = self.mkstemp(dir=self.temp_dir)
candidates = string.ascii_letters+string.digits
with os.fdopen(source_handle, 'w') as fd:
for x in range(100):
fd.write(''.join(random.sample(candidates, len(candidates))))
handle, initial_manifest = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, 'w') as fd:
fd.write('__file_from_stdin %s --source %s\n' % (destination, source))
self.context.initial_manifest = initial_manifest
self.config.stage_prepare()
cdist_type = core.CdistType(self.config.local.type_path, '__file')
cdist_object = core.CdistObject(cdist_type, self.config.local.object_path, destination)
# Test weither stdin has been stored correctly
self.assertTrue(filecmp.cmp(source, os.path.join(cdist_object.absolute_path, 'stdin')))

0
lib/cdist/test/emulator/fixtures/conf/explorer/.keep
View file

3
lib/cdist/test/emulator/fixtures/conf/manifest/init
View file

@ -1,3 +0,0 @@
#!/bin/sh
__saturn

2
lib/cdist/test/emulator/fixtures/conf/type/__arguments_boolean/parameter/boolean
View file

@ -1,2 +0,0 @@
boolean1
boolean2

1
lib/cdist/test/emulator/fixtures/conf/type/__arguments_optional/parameter/optional
View file

@ -1 +0,0 @@
optional1

2
lib/cdist/test/emulator/fixtures/conf/type/__arguments_required/parameter/required
View file

@ -1,2 +0,0 @@
required1
required2

1
lib/cdist/test/emulator/fixtures/conf/type/__arguments_with_dashes/parameter/required
View file

@ -1 +0,0 @@
with-dash

1
lib/cdist/test/emulator/fixtures/conf/type/__file
View file

@ -1 +0,0 @@
../../../../../../../conf/type/__file

4
lib/cdist/test/emulator/fixtures/conf/type/__file_from_stdin/manifest
View file

@ -1,4 +0,0 @@
#!/bin/sh
source="$(cat "$__object/parameter/source")"
cat "$source" | __file "/$__object_id" --source /dev/null

1
lib/cdist/test/emulator/fixtures/conf/type/__file_from_stdin/parameter/required
View file

@ -1 +0,0 @@
source

8
lib/cdist/test/emulator/fixtures/conf/type/__moon/manifest
View file

@ -1,8 +0,0 @@
#!/bin/sh
if [ -f "$__object/parameter/name" ]; then
name="(cat "$__object/parameter/name")"
else
name="$__object_id"
echo "$name" > "$__object/parameter/name"
fi

1
lib/cdist/test/emulator/fixtures/conf/type/__moon/parameter/optional
View file

@ -1 +0,0 @@
name

1
lib/cdist/test/emulator/fixtures/conf/type/__moon/parameter/required
View file

@ -1 +0,0 @@
planet

8
lib/cdist/test/emulator/fixtures/conf/type/__planet/manifest
View file

@ -1,8 +0,0 @@
#!/bin/sh
if [ -f "$__object/parameter/name" ]; then
name="(cat "$__object/parameter/name")"
else
name="$__object_id"
echo "$name" > "$__object/parameter/name"
fi

1
lib/cdist/test/emulator/fixtures/conf/type/__planet/parameter/optional
View file

@ -1 +0,0 @@
name

4
lib/cdist/test/emulator/fixtures/conf/type/__saturn/manifest
View file

@ -1,4 +0,0 @@
#!/bin/sh
__planet Saturn
require="__planet/Saturn" __moon Prometheus --planet Saturn

0
lib/cdist/test/emulator/fixtures/conf/type/__saturn/singleton
View file

0
lib/cdist/test/exec/__init__.py
View file

123
lib/cdist/test/exec/local.py
View file

@ -1,123 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import getpass
import shutil
import string
import random
#import logging
#logging.basicConfig(level=logging.DEBUG, format='%(levelname)s: %(message)s')
import cdist
from cdist import test
from cdist.exec import local
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
local_base_path = fixtures
class LocalTestCase(test.CdistTestCase):
def setUp(self):
self.temp_dir = self.mkdtemp()
target_host = 'localhost'
self.out_path = self.temp_dir
self.base_path = local_base_path
self.local = local.Local(target_host, self.base_path, self.out_path)
def tearDown(self):
shutil.rmtree(self.temp_dir)
### test api
def test_cache_path(self):
self.assertEqual(self.local.cache_path, os.path.join(self.base_path, "cache"))
def test_conf_path(self):
self.assertEqual(self.local.conf_path, os.path.join(self.base_path, "conf"))
def test_global_explorer_path(self):
self.assertEqual(self.local.global_explorer_path, os.path.join(self.base_path, "conf", "explorer"))
def test_manifest_path(self):
self.assertEqual(self.local.manifest_path, os.path.join(self.base_path, "conf", "manifest"))
def test_type_path(self):
self.assertEqual(self.local.type_path, os.path.join(self.base_path, "conf", "type"))
def test_out_path(self):
self.assertEqual(self.local.out_path, self.out_path)
def test_bin_path(self):
self.assertEqual(self.local.bin_path, os.path.join(self.out_path, "bin"))
def test_global_explorer_out_path(self):
self.assertEqual(self.local.global_explorer_out_path, os.path.join(self.out_path, "explorer"))
def test_object_path(self):
self.assertEqual(self.local.object_path, os.path.join(self.out_path, "object"))
### /test api
def test_run_success(self):
self.local.run(['/bin/true'])
def test_run_fail(self):
self.assertRaises(cdist.Error, self.local.run, ['/bin/false'])
def test_run_script_success(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "/bin/true"])
self.local.run_script(script)
def test_run_script_fail(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "/bin/false"])
self.assertRaises(local.LocalScriptError, self.local.run_script, script)
def test_run_script_get_output(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "echo foobar"])
self.assertEqual(self.local.run_script(script, return_output=True), "foobar\n")
def test_mkdir(self):
temp_dir = self.mkdtemp(dir=self.temp_dir)
os.rmdir(temp_dir)
self.local.mkdir(temp_dir)
self.assertTrue(os.path.isdir(temp_dir))
def test_rmdir(self):
temp_dir = self.mkdtemp(dir=self.temp_dir)
self.local.rmdir(temp_dir)
self.assertFalse(os.path.isdir(temp_dir))
def test_create_directories(self):
self.local.create_directories()
self.assertTrue(os.path.isdir(self.local.out_path))
self.assertTrue(os.path.isdir(self.local.bin_path))

142
lib/cdist/test/exec/remote.py
View file

@ -1,142 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import getpass
import shutil
import string
import random
import cdist
from cdist import test
from cdist.exec import remote
class RemoteTestCase(test.CdistTestCase):
def setUp(self):
self.temp_dir = self.mkdtemp()
self.target_host = 'localhost'
self.base_path = self.temp_dir
user = getpass.getuser()
remote_exec = "ssh -o User=%s -q" % user
remote_copy = "scp -o User=%s -q" % user
self.remote = remote.Remote(self.target_host, self.base_path, remote_exec, remote_copy)
def tearDown(self):
shutil.rmtree(self.temp_dir)
### test api
def test_conf_path(self):
self.assertEqual(self.remote.conf_path, os.path.join(self.base_path, "conf"))
def test_object_path(self):
self.assertEqual(self.remote.object_path, os.path.join(self.base_path, "object"))
def test_type_path(self):
self.assertEqual(self.remote.type_path, os.path.join(self.base_path, "conf", "type"))
def test_global_explorer_path(self):
self.assertEqual(self.remote.global_explorer_path, os.path.join(self.base_path, "conf", "explorer"))
### /test api
def test_run_success(self):
self.remote.run(['/bin/true'])
def test_run_fail(self):
self.assertRaises(cdist.Error, self.remote.run, ['/bin/false'])
def test_run_script_success(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "/bin/true"])
self.remote.run_script(script)
def test_run_script_fail(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "/bin/false"])
self.assertRaises(remote.RemoteScriptError, self.remote.run_script, script)
def test_run_script_get_output(self):
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "echo foobar"])
self.assertEqual(self.remote.run_script(script, return_output=True), "foobar\n")
def test_mkdir(self):
temp_dir = self.mkdtemp(dir=self.temp_dir)
os.rmdir(temp_dir)
self.remote.mkdir(temp_dir)
self.assertTrue(os.path.isdir(temp_dir))
def test_rmdir(self):
temp_dir = self.mkdtemp(dir=self.temp_dir)
self.remote.rmdir(temp_dir)
self.assertFalse(os.path.isdir(temp_dir))
def test_transfer_file(self):
handle, source = self.mkstemp(dir=self.temp_dir)
os.close(handle)
target = self.mkdtemp(dir=self.temp_dir)
self.remote.transfer(source, target)
self.assertTrue(os.path.isfile(target))
def test_transfer_dir(self):
source = self.mkdtemp(dir=self.temp_dir)
# put a file in the directory as payload
handle, source_file = self.mkstemp(dir=source)
os.close(handle)
source_file_name = os.path.split(source_file)[-1]
target = self.mkdtemp(dir=self.temp_dir)
self.remote.transfer(source, target)
# test if the payload file is in the target directory
self.assertTrue(os.path.isfile(os.path.join(target, source_file_name)))
def test_create_directories(self):
self.remote.create_directories()
self.assertTrue(os.path.isdir(self.remote.base_path))
self.assertTrue(os.path.isdir(self.remote.conf_path))
def test_run_target_host_in_env(self):
handle, remote_exec_path = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, 'w') as fd:
fd.writelines(["#!/bin/sh\n", "echo $__target_host"])
os.chmod(remote_exec_path, 0o755)
remote_exec = remote_exec_path
remote_copy = "echo"
r = remote.Remote(self.target_host, self.base_path, remote_exec, remote_copy)
self.assertEqual(r.run('/bin/true', return_output=True), "%s\n" % self.target_host)
def test_run_script_target_host_in_env(self):
handle, remote_exec_path = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, 'w') as fd:
fd.writelines(["#!/bin/sh\n", "echo $__target_host"])
os.chmod(remote_exec_path, 0o755)
remote_exec = remote_exec_path
remote_copy = "echo"
r = remote.Remote(self.target_host, self.base_path, remote_exec, remote_copy)
handle, script = self.mkstemp(dir=self.temp_dir)
with os.fdopen(handle, "w") as fd:
fd.writelines(["#!/bin/sh\n", "/bin/true"])
self.assertEqual(r.run_script(script, return_output=True), "%s\n" % self.target_host)

133
lib/cdist/test/explorer/__init__.py
View file

@ -1,133 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
# 2011 Nico Schottelius (nico-cdist at schottelius.org)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
import getpass
import logging
import cdist
from cdist import core
from cdist import test
from cdist.exec import local
from cdist.exec import remote
from cdist.core import explorer
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
local_base_path = fixtures
class ExplorerClassTestCase(test.CdistTestCase):
def setUp(self):
self.target_host = 'localhost'
self.local_base_path = local_base_path
self.out_path = self.mkdtemp()
self.local = local.Local(self.target_host, self.local_base_path, self.out_path)
self.local.create_directories()
self.remote_base_path = self.mkdtemp()
self.user = getpass.getuser()
remote_exec = "ssh -o User=%s -q" % self.user
remote_copy = "scp -o User=%s -q" % self.user
self.remote = remote.Remote(self.target_host, self.remote_base_path, remote_exec, remote_copy)
self.explorer = explorer.Explorer(self.target_host, self.local, self.remote)
self.log = logging.getLogger(self.target_host)
def tearDown(self):
shutil.rmtree(self.out_path)
shutil.rmtree(self.remote_base_path)
def test_list_global_explorer_names(self):
expected = ['foobar', 'global']
self.assertEqual(self.explorer.list_global_explorer_names(), expected)
def test_transfer_global_explorers(self):
self.explorer.transfer_global_explorers()
source = self.local.global_explorer_path
destination = self.remote.global_explorer_path
self.assertEqual(sorted(os.listdir(source)), sorted(os.listdir(destination)))
def test_run_global_explorer(self):
self.explorer.transfer_global_explorers()
output = self.explorer.run_global_explorer('global')
self.assertEqual(output, 'global\n')
def test_run_global_explorers(self):
out_path = self.mkdtemp()
self.explorer.run_global_explorers(out_path)
self.assertEqual(sorted(os.listdir(out_path)), sorted(['foobar', 'global']))
shutil.rmtree(out_path)
def test_list_type_explorer_names(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
expected = cdist_type.explorers
self.assertEqual(self.explorer.list_type_explorer_names(cdist_type), expected)
def test_transfer_type_explorers(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
self.explorer.transfer_type_explorers(cdist_type)
source = os.path.join(self.local.type_path, cdist_type.explorer_path)
destination = os.path.join(self.remote.type_path, cdist_type.explorer_path)
self.assertEqual(os.listdir(source), os.listdir(destination))
def test_transfer_type_explorers_only_once(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
# first transfer
self.explorer.transfer_type_explorers(cdist_type)
source = os.path.join(self.local.type_path, cdist_type.explorer_path)
destination = os.path.join(self.remote.type_path, cdist_type.explorer_path)
self.assertEqual(os.listdir(source), os.listdir(destination))
# nuke destination folder content, but recreate directory
shutil.rmtree(destination)
os.makedirs(destination)
# second transfer, should not happen
self.explorer.transfer_type_explorers(cdist_type)
self.assertFalse(os.listdir(destination))
def test_transfer_object_parameters(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'whatever')
cdist_object.create()
cdist_object.parameters = {'first': 'first value', 'second': 'second value'}
self.explorer.transfer_object_parameters(cdist_object)
source = os.path.join(self.local.object_path, cdist_object.parameter_path)
destination = os.path.join(self.remote.object_path, cdist_object.parameter_path)
self.assertEqual(sorted(os.listdir(source)), sorted(os.listdir(destination)))
def test_run_type_explorer(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'whatever')
self.explorer.transfer_type_explorers(cdist_type)
output = self.explorer.run_type_explorer('world', cdist_object)
self.assertEqual(output, 'hello\n')
def test_run_type_explorers(self):
cdist_type = core.CdistType(self.local.type_path, '__test_type')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'whatever')
cdist_object.create()
self.explorer.run_type_explorers(cdist_object)
self.assertEqual(cdist_object.explorers, {'world': 'hello'})

2
lib/cdist/test/explorer/fixtures/conf/explorer/foobar
View file

@ -1,2 +0,0 @@
#!/bin/sh
echo foobar

2
lib/cdist/test/explorer/fixtures/conf/explorer/global
View file

@ -1,2 +0,0 @@
#!/bin/sh
echo global

2
lib/cdist/test/explorer/fixtures/conf/type/__test_type/explorer/world
View file

@ -1,2 +0,0 @@
#!/bin/sh
echo hello

3
lib/cdist/test/explorer/fixtures/conf/type/__test_type_object_parameter/explorer/test_parameter
View file

@ -1,3 +0,0 @@
#!/bin/sh
cat "$__object/parameter/test"

108
lib/cdist/test/manifest/__init__.py
View file

@ -1,108 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import getpass
import shutil
import string
import random
import logging
import io
import sys
import cdist
from cdist import test
from cdist.exec import local
from cdist import core
from cdist.core import manifest
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
local_base_path = fixtures
class ManifestTestCase(test.CdistTestCase):
def setUp(self):
self.orig_environ = os.environ
os.environ = os.environ.copy()
self.temp_dir = self.mkdtemp()
self.target_host = 'localhost'
out_path = self.temp_dir
self.local = local.Local(self.target_host, local_base_path, out_path)
self.local.create_directories()
self.local.link_emulator(cdist.test.cdist_exec_path)
self.manifest = manifest.Manifest(self.target_host, self.local)
self.log = logging.getLogger(self.target_host)
def tearDown(self):
os.environ = self.orig_environ
shutil.rmtree(self.temp_dir)
def test_initial_manifest_environment(self):
initial_manifest = os.path.join(self.local.manifest_path, "dump_environment")
handle, output_file = self.mkstemp(dir=self.temp_dir)
os.close(handle)
os.environ['__cdist_test_out'] = output_file
self.manifest.run_initial_manifest(initial_manifest)
with open(output_file, 'r') as fd:
output_string = fd.read()
output_dict = {}
for line in output_string.split('\n'):
if line:
key,value = line.split(': ')
output_dict[key] = value
self.assertTrue(output_dict['PATH'].startswith(self.local.bin_path))
self.assertEqual(output_dict['__target_host'], self.local.target_host)
self.assertEqual(output_dict['__global'], self.local.out_path)
self.assertEqual(output_dict['__cdist_type_base_path'], self.local.type_path)
self.assertEqual(output_dict['__manifest'], self.local.manifest_path)
def test_type_manifest_environment(self):
cdist_type = core.CdistType(self.local.type_path, '__dump_environment')
cdist_object = core.CdistObject(cdist_type, self.local.object_path, 'whatever')
handle, output_file = self.mkstemp(dir=self.temp_dir)
os.close(handle)
os.environ['__cdist_test_out'] = output_file
self.manifest.run_type_manifest(cdist_object)
with open(output_file, 'r') as fd:
output_string = fd.read()
output_dict = {}
for line in output_string.split('\n'):
if line:
key,value = line.split(': ')
output_dict[key] = value
self.assertTrue(output_dict['PATH'].startswith(self.local.bin_path))
self.assertEqual(output_dict['__target_host'], self.local.target_host)
self.assertEqual(output_dict['__global'], self.local.out_path)
self.assertEqual(output_dict['__cdist_type_base_path'], self.local.type_path)
self.assertEqual(output_dict['__type'], cdist_type.absolute_path)
self.assertEqual(output_dict['__object'], cdist_object.absolute_path)
self.assertEqual(output_dict['__object_id'], cdist_object.object_id)
self.assertEqual(output_dict['__object_name'], cdist_object.name)
def test_debug_env_setup(self):
self.log.setLevel(logging.DEBUG)
manifest = cdist.core.manifest.Manifest(self.target_host, self.local)
self.assertTrue("__cdist_debug" in manifest.env)

9
lib/cdist/test/manifest/fixtures/conf/manifest/dump_environment
View file

@ -1,9 +0,0 @@
#!/bin/sh
cat > $__cdist_test_out << DONE
PATH: $PATH
__target_host: $__target_host
__global: $__global
__cdist_type_base_path: $__cdist_type_base_path
__manifest: $__manifest
DONE

4
lib/cdist/test/manifest/fixtures/conf/manifest/init
View file

@ -1,4 +0,0 @@
#!/bin/sh
__planet Saturn
__moon Prometheus --planet Saturn

13
lib/cdist/test/manifest/fixtures/conf/type/__dump_environment/manifest
View file

@ -1,13 +0,0 @@
#!/bin/sh
cat > $__cdist_test_out << DONE
PATH: $PATH
__target_host: $__target_host
__global: $__global
__cdist_type_base_path: $__cdist_type_base_path
__type: $__type
__self: $__self
__object: $__object
__object_id: $__object_id
__object_name: $__object_name
DONE

0
lib/cdist/test/manifest/fixtures/conf/type/__moon/.keep
View file

8
lib/cdist/test/manifest/fixtures/conf/type/__moon/manifest
View file

@ -1,8 +0,0 @@
#!/bin/sh
if [ -f "$__object/parameter/name" ]; then
name="(cat "$__object/parameter/name")"
else
name="$__object_id"
echo "$name" > "$__object/parameter/name"
fi

1
lib/cdist/test/manifest/fixtures/conf/type/__moon/parameter/optional
View file

@ -1 +0,0 @@
name

1
lib/cdist/test/manifest/fixtures/conf/type/__moon/parameter/required
View file

@ -1 +0,0 @@
planet

0
lib/cdist/test/manifest/fixtures/conf/type/__planet/.keep
View file

8
lib/cdist/test/manifest/fixtures/conf/type/__planet/manifest
View file

@ -1,8 +0,0 @@
#!/bin/sh
if [ -f "$__object/parameter/name" ]; then
name="(cat "$__object/parameter/name")"
else
name="$__object_id"
echo "$name" > "$__object/parameter/name"
fi

1
lib/cdist/test/manifest/fixtures/conf/type/__planet/parameter/optional
View file

@ -1 +0,0 @@
name

202
lib/cdist/test/object/__init__.py
View file

@ -1,202 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
from cdist import test
from cdist import core
import cdist
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
object_base_path = op.join(fixtures, 'object')
type_base_path = op.join(fixtures, 'type')
class ObjectClassTestCase(test.CdistTestCase):
def test_list_object_names(self):
object_names = list(core.CdistObject.list_object_names(object_base_path))
self.assertEqual(object_names, ['__first/man', '__second/on-the', '__third/moon'])
def test_list_type_names(self):
type_names = list(cdist.core.CdistObject.list_type_names(object_base_path))
self.assertEqual(type_names, ['__first', '__second', '__third'])
def test_list_objects(self):
objects = list(core.CdistObject.list_objects(object_base_path, type_base_path))
objects_expected = [
core.CdistObject(core.CdistType(type_base_path, '__first'), object_base_path, 'man'),
core.CdistObject(core.CdistType(type_base_path, '__second'), object_base_path, 'on-the'),
core.CdistObject(core.CdistType(type_base_path, '__third'), object_base_path, 'moon'),
]
self.assertEqual(objects, objects_expected)
class ObjectIdTestCase(test.CdistTestCase):
def test_object_id_contains_double_slash(self):
cdist_type = core.CdistType(type_base_path, '__third')
illegal_object_id = '/object_id//may/not/contain/double/slash'
with self.assertRaises(core.IllegalObjectIdError):
core.CdistObject(cdist_type, object_base_path, illegal_object_id)
def test_object_id_contains_object_marker(self):
cdist_type = core.CdistType(type_base_path, '__third')
illegal_object_id = 'object_id/may/not/contain/%s/anywhere' % core.OBJECT_MARKER
with self.assertRaises(core.IllegalObjectIdError):
core.CdistObject(cdist_type, object_base_path, illegal_object_id)
def test_object_id_contains_object_marker_string(self):
cdist_type = core.CdistType(type_base_path, '__third')
illegal_object_id = 'object_id/may/contain_%s_in_filename' % core.OBJECT_MARKER
core.CdistObject(cdist_type, object_base_path, illegal_object_id)
# if we get here, the test passed
class ObjectTestCase(test.CdistTestCase):
def setUp(self):
self.cdist_type = core.CdistType(type_base_path, '__third')
self.cdist_object = core.CdistObject(self.cdist_type, object_base_path, 'moon')
def tearDown(self):
self.cdist_object.changed = False
self.cdist_object.prepared = False
self.cdist_object.ran = False
self.cdist_object.source = []
self.cdist_object.code_local = ''
self.cdist_object.code_remote = ''
self.cdist_object.state = ''
def test_name(self):
self.assertEqual(self.cdist_object.name, '__third/moon')
def test_object_id(self):
self.assertEqual(self.cdist_object.object_id, 'moon')
def test_path(self):
self.assertEqual(self.cdist_object.path, '__third/moon/.cdist')
def test_absolute_path(self):
self.assertEqual(self.cdist_object.absolute_path, os.path.join(object_base_path, '__third/moon/.cdist'))
def test_code_local_path(self):
self.assertEqual(self.cdist_object.code_local_path, '__third/moon/.cdist/code-local')
def test_code_remote_path(self):
self.assertEqual(self.cdist_object.code_remote_path, '__third/moon/.cdist/code-remote')
def test_parameter_path(self):
self.assertEqual(self.cdist_object.parameter_path, '__third/moon/.cdist/parameter')
def test_explorer_path(self):
self.assertEqual(self.cdist_object.explorer_path, '__third/moon/.cdist/explorer')
def test_parameters(self):
expected_parameters = {'planet': 'Saturn', 'name': 'Prometheus'}
self.assertEqual(self.cdist_object.parameters, expected_parameters)
def test_explorers(self):
self.assertEqual(self.cdist_object.explorers, {})
# FIXME: actually testing fsproperty.DirectoryDictProperty here, move to their own test case
def test_explorers_assign_dict(self):
expected = {'first': 'foo', 'second': 'bar'}
# when set, written to file
self.cdist_object.explorers = expected
object_explorer_path = os.path.join(self.cdist_object.base_path, self.cdist_object.explorer_path)
self.assertTrue(os.path.isdir(object_explorer_path))
# when accessed, read from file
self.assertEqual(self.cdist_object.explorers, expected)
# remove dynamically created folder
self.cdist_object.explorers = {}
os.rmdir(os.path.join(self.cdist_object.base_path, self.cdist_object.explorer_path))
# FIXME: actually testing fsproperty.DirectoryDictProperty here, move to their own test case
def test_explorers_assign_key_value(self):
expected = {'first': 'foo', 'second': 'bar'}
object_explorer_path = os.path.join(self.cdist_object.base_path, self.cdist_object.explorer_path)
for key,value in expected.items():
# when set, written to file
self.cdist_object.explorers[key] = value
self.assertTrue(os.path.isfile(os.path.join(object_explorer_path, key)))
# when accessed, read from file
self.assertEqual(self.cdist_object.explorers, expected)
# remove dynamically created folder
self.cdist_object.explorers = {}
os.rmdir(os.path.join(self.cdist_object.base_path, self.cdist_object.explorer_path))
def test_requirements(self):
expected = []
self.assertEqual(list(self.cdist_object.requirements), expected)
def test_changed(self):
self.assertFalse(self.cdist_object.changed)
def test_changed_after_changing(self):
self.cdist_object.changed = True
self.assertTrue(self.cdist_object.changed)
def test_state(self):
self.assertEqual(self.cdist_object.state, '')
def test_state_prepared(self):
self.cdist_object.state = core.CdistObject.STATE_PREPARED
self.assertEqual(self.cdist_object.state, core.CdistObject.STATE_PREPARED)
def test_state_running(self):
self.cdist_object.state = core.CdistObject.STATE_RUNNING
self.assertEqual(self.cdist_object.state, core.CdistObject.STATE_RUNNING)
def test_state_done(self):
self.cdist_object.state = core.CdistObject.STATE_DONE
self.assertEqual(self.cdist_object.state, core.CdistObject.STATE_DONE)
def test_source(self):
self.assertEqual(list(self.cdist_object.source), [])
def test_source_after_changing(self):
self.cdist_object.source = ['/path/to/manifest']
self.assertEqual(list(self.cdist_object.source), ['/path/to/manifest'])
def test_code_local(self):
self.assertEqual(self.cdist_object.code_local, '')
def test_code_local_after_changing(self):
self.cdist_object.code_local = 'Hello World'
self.assertEqual(self.cdist_object.code_local, 'Hello World')
def test_code_remote(self):
self.assertEqual(self.cdist_object.code_remote, '')
def test_code_remote_after_changing(self):
self.cdist_object.code_remote = 'Hello World'
self.assertEqual(self.cdist_object.code_remote, 'Hello World')
def test_object_from_name(self):
self.cdist_object.code_remote = 'Hello World'
other_name = '__first/man'
other_object = self.cdist_object.object_from_name(other_name)
self.assertTrue(isinstance(other_object, core.CdistObject))
self.assertEqual(other_object.cdist_type.name, '__first')
self.assertEqual(other_object.object_id, 'man')

0
lib/cdist/test/object/fixtures/object/__first/.keep
View file

0
lib/cdist/test/object/fixtures/object/__first/man/.cdist/.keep
View file

0
lib/cdist/test/object/fixtures/object/__second/.keep
View file

0
lib/cdist/test/object/fixtures/object/__second/on-the/.cdist/.keep
View file

0
lib/cdist/test/object/fixtures/object/__third/.keep
View file

0
lib/cdist/test/object/fixtures/object/__third/moon/.cdist/.keep
View file

1
lib/cdist/test/object/fixtures/object/__third/moon/.cdist/parameter/name
View file

@ -1 +0,0 @@
Prometheus

1
lib/cdist/test/object/fixtures/object/__third/moon/.cdist/parameter/planet
View file

@ -1 +0,0 @@
Saturn

0
lib/cdist/test/object/fixtures/type/__first/.keep
View file

0
lib/cdist/test/object/fixtures/type/__second/.keep
View file

0
lib/cdist/test/object/fixtures/type/__third/.keep
View file

88
lib/cdist/test/resolver/__init__.py
View file

@ -1,88 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
import shutil
import cdist
from cdist import test
from cdist import core
from cdist import resolver
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
object_base_path = op.join(fixtures, 'object')
type_base_path = op.join(fixtures, 'type')
class ResolverTestCase(test.CdistTestCase):
def setUp(self):
self.objects = list(core.CdistObject.list_objects(object_base_path, type_base_path))
self.object_index = dict((o.name, o) for o in self.objects)
self.dependency_resolver = resolver.DependencyResolver(self.objects)
def tearDown(self):
for o in self.objects:
o.requirements = []
def test_find_requirements_by_name_string(self):
requirements = ['__first/man', '__second/on-the', '__third/moon']
required_objects = [self.object_index[name] for name in requirements]
self.assertEqual(sorted(list(self.dependency_resolver.find_requirements_by_name(requirements))),
sorted(required_objects))
def test_find_requirements_by_name_pattern(self):
requirements = ['__first/*', '__second/*-the', '__third/moon']
requirements_expanded = [
'__first/child', '__first/dog', '__first/man', '__first/woman',
'__second/on-the', '__second/under-the',
'__third/moon'
]
required_objects = [self.object_index[name] for name in requirements_expanded]
self.assertEqual(sorted(list(self.dependency_resolver.find_requirements_by_name(requirements))),
sorted(required_objects))
def test_dependency_resolution(self):
first_man = self.object_index['__first/man']
second_on_the = self.object_index['__second/on-the']
third_moon = self.object_index['__third/moon']
first_man.requirements = [second_on_the.name]
second_on_the.requirements = [third_moon.name]
self.assertEqual(
self.dependency_resolver.dependencies['__first/man'],
[third_moon, second_on_the, first_man]
)
def test_circular_reference(self):
first_man = self.object_index['__first/man']
first_woman = self.object_index['__first/woman']
first_man.requirements = [first_woman.name]
first_woman.requirements = [first_man.name]
with self.assertRaises(resolver.CircularReferenceError):
self.dependency_resolver.dependencies
def test_requirement_not_found(self):
first_man = self.object_index['__first/man']
first_man.requirements = ['__does/not/exist']
with self.assertRaises(cdist.Error):
self.dependency_resolver.dependencies

0
lib/cdist/test/resolver/fixtures/object/__first/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__first/child/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__first/dog/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__first/man/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__first/woman/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__second/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__second/on-the/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__second/under-the/.cdist/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__third/.keep
View file

0
lib/cdist/test/resolver/fixtures/object/__third/moon/.cdist/.keep
View file

1
lib/cdist/test/resolver/fixtures/object/__third/moon/.cdist/parameter/name
View file

@ -1 +0,0 @@
Prometheus

1
lib/cdist/test/resolver/fixtures/object/__third/moon/.cdist/parameter/planet
View file

@ -1 +0,0 @@
Saturn

0
lib/cdist/test/resolver/fixtures/type/__first/.keep
View file

0
lib/cdist/test/resolver/fixtures/type/__second/.keep
View file

0
lib/cdist/test/resolver/fixtures/type/__third/.keep
View file

158
lib/cdist/test/type/__init__.py
View file

@ -1,158 +0,0 @@
# -*- coding: utf-8 -*-
#
# 2010-2011 Steven Armstrong (steven-cdist at armstrong.cc)
#
# This file is part of cdist.
#
# cdist 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.
#
# cdist 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 cdist. If not, see <http://www.gnu.org/licenses/>.
#
#
import os
from cdist import test
from cdist import core
import os.path as op
my_dir = op.abspath(op.dirname(__file__))
fixtures = op.join(my_dir, 'fixtures')
class TypeTestCase(test.CdistTestCase):
def test_list_type_names(self):
base_path = op.join(fixtures, 'list_types')
type_names = core.CdistType.list_type_names(base_path)
self.assertEqual(type_names, ['__first', '__second', '__third'])
def test_list_types(self):
base_path = op.join(fixtures, 'list_types')
types = list(core.CdistType.list_types(base_path))
types_expected = [
core.CdistType(base_path, '__first'),
core.CdistType(base_path, '__second'),
core.CdistType(base_path, '__third'),
]
self.assertEqual(types, types_expected)
def test_only_one_instance(self):
base_path = fixtures
cdist_type1 = core.CdistType(base_path, '__name_path')
cdist_type2 = core.CdistType(base_path, '__name_path')
self.assertEqual(id(cdist_type1), id(cdist_type2))
def test_nonexistent_type(self):
base_path = fixtures
self.assertRaises(core.NoSuchTypeError, core.CdistType, base_path, '__i-dont-exist')
def test_name(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.name, '__name_path')
def test_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.path, '__name_path')
def test_base_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.base_path, base_path)
def test_absolute_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.absolute_path, os.path.join(base_path, '__name_path'))
def test_manifest_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.manifest_path, os.path.join('__name_path', 'manifest'))
def test_explorer_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.explorer_path, os.path.join('__name_path', 'explorer'))
def test_gencode_local_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.gencode_local_path, os.path.join('__name_path', 'gencode-local'))
def test_gencode_remote_path(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__name_path')
self.assertEqual(cdist_type.gencode_remote_path, os.path.join('__name_path', 'gencode-remote'))
def test_singleton_is_singleton(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__singleton')
self.assertTrue(cdist_type.is_singleton)
def test_not_singleton_is_singleton(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__not_singleton')
self.assertFalse(cdist_type.is_singleton)
def test_install_is_install(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__install')
self.assertTrue(cdist_type.is_install)
def test_not_install_is_install(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__not_install')
self.assertFalse(cdist_type.is_install)
def test_with_explorers(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__with_explorers')
self.assertEqual(cdist_type.explorers, ['whatever'])
def test_without_explorers(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__without_explorers')
self.assertEqual(cdist_type.explorers, [])
def test_with_required_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__with_required_parameters')
self.assertEqual(cdist_type.required_parameters, ['required1', 'required2'])
def test_without_required_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__without_required_parameters')
self.assertEqual(cdist_type.required_parameters, [])
def test_with_optional_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__with_optional_parameters')
self.assertEqual(cdist_type.optional_parameters, ['optional1', 'optional2'])
def test_without_optional_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__without_optional_parameters')
self.assertEqual(cdist_type.optional_parameters, [])
def test_with_boolean_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__with_boolean_parameters')
self.assertEqual(cdist_type.boolean_parameters, ['boolean1', 'boolean2'])
def test_without_boolean_parameters(self):
base_path = fixtures
cdist_type = core.CdistType(base_path, '__without_boolean_parameters')
self.assertEqual(cdist_type.boolean_parameters, [])

Some files were not shown because too many files have changed in this diff Show more