forked from ungleich-public/cdist
Merge branch 'master' of git://github.com/asteven/cdist
This commit is contained in:
Nico Schottelius
commit
9647933a40
|
|
@ -25,14 +25,14 @@ import os
|
|||
import collections
|
||||
|
||||
import cdist
|
||||
import cdist.core.property
|
||||
import cdist.core
|
||||
from cdist.util import fsproperty
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
DOT_CDIST = '.cdist'
|
||||
|
||||
# FIXME: i should not have to care about prefix directory, local, remote and such.
|
||||
# I know what my internals look like, the outside is none of my business.
|
||||
|
||||
class Object(object):
|
||||
"""Represents a cdist object.
|
||||
|
||||
|
|
@ -63,8 +63,8 @@ class Object(object):
|
|||
if DOT_CDIST in dirs:
|
||||
yield os.path.relpath(path, object_base_path)
|
||||
|
||||
def __init__(self, type, base_path, object_id=None):
|
||||
self.type = type # instance of Type
|
||||
def __init__(self, cdist_type, base_path, object_id=None):
|
||||
self.type = cdist_type # instance of Type
|
||||
self.base_path = base_path
|
||||
self.object_id = object_id
|
||||
self.name = os.path.join(self.type.name, self.object_id)
|
||||
|
|
@ -80,6 +80,10 @@ class Object(object):
|
|||
def __repr__(self):
|
||||
return '<Object %s>' % self.name
|
||||
|
||||
def __eq__(self, other):
|
||||
"""define equality as 'attributes are the same'"""
|
||||
return self.__dict__ == other.__dict__
|
||||
|
||||
@property
|
||||
def explorer_path(self):
|
||||
# create absolute path
|
||||
|
|
@ -89,97 +93,8 @@ class Object(object):
|
|||
# return relative path
|
||||
return os.path.join(self.path, "explorer")
|
||||
|
||||
|
||||
### requirements
|
||||
@property
|
||||
def requirements(self):
|
||||
if not self.__requirements:
|
||||
self.__requirements = cdist.core.property.FileList(os.path.join(self.path, "require"))
|
||||
return self.__requirements
|
||||
|
||||
@requirements.setter
|
||||
def requirements(self, value):
|
||||
if isinstance(value, cdist.core.property.FileList):
|
||||
self.__requirements = value
|
||||
else:
|
||||
self.__requirements = cdist.core.property.FileList(os.path.join(self.path, "require"), value)
|
||||
### /requirements
|
||||
|
||||
|
||||
### parameters
|
||||
@property
|
||||
def parameters(self):
|
||||
if not self.__parameters:
|
||||
self.__parameters = cdist.core.property.DirectoryDict(os.path.join(self.path, "parameter"))
|
||||
return self.__parameters
|
||||
|
||||
@parameters.setter
|
||||
def parameters(self, value):
|
||||
if isinstance(value, cdist.core.property.DirectoryDict):
|
||||
self.__parameters = value
|
||||
else:
|
||||
self.__parameters = cdist.core.property.DirectoryDict(os.path.join(self.path, "parameter"), value)
|
||||
### /parameters
|
||||
|
||||
|
||||
### changed
|
||||
@property
|
||||
def changed(self):
|
||||
"""Check whether the object has been changed."""
|
||||
return os.path.isfile(os.path.join(self.absolute_path, "changed"))
|
||||
|
||||
@changed.setter
|
||||
def changed(self, value):
|
||||
"""Change the objects changed status."""
|
||||
path = os.path.join(self.absolute_path, "changed")
|
||||
if value:
|
||||
open(path, "w").close()
|
||||
else:
|
||||
try:
|
||||
os.remove(path)
|
||||
except EnvironmentError:
|
||||
# ignore
|
||||
pass
|
||||
### /changed
|
||||
|
||||
|
||||
### prepared
|
||||
@property
|
||||
def prepared(self):
|
||||
"""Check whether the object has been prepared."""
|
||||
return os.path.isfile(os.path.join(self.absolute_path, "prepared"))
|
||||
|
||||
@prepared.setter
|
||||
def prepared(self, value):
|
||||
"""Change the objects prepared status."""
|
||||
path = os.path.join(self.absolute_path, "prepared")
|
||||
if value:
|
||||
open(path, "w").close()
|
||||
else:
|
||||
try:
|
||||
os.remove(path)
|
||||
except EnvironmentError:
|
||||
# ignore
|
||||
pass
|
||||
### /prepared
|
||||
|
||||
|
||||
### ran
|
||||
@property
|
||||
def ran(self):
|
||||
"""Check whether the object has been ran."""
|
||||
return os.path.isfile(os.path.join(self.absolute_path, "ran"))
|
||||
|
||||
@ran.setter
|
||||
def ran(self, value):
|
||||
"""Change the objects ran status."""
|
||||
path = os.path.join(self.absolute_path, "ran")
|
||||
if value:
|
||||
open(path, "w").close()
|
||||
else:
|
||||
try:
|
||||
os.remove(path)
|
||||
except EnvironmentError:
|
||||
# ignore
|
||||
pass
|
||||
### /ran
|
||||
requirements = fsproperty.FileListProperty(lambda obj: os.path.join(obj.absolute_path, 'require'))
|
||||
parameters = fsproperty.DirectoryDictProperty(lambda obj: os.path.join(obj.absolute_path, 'parameter'))
|
||||
changed = fsproperty.FileBooleanProperty(lambda obj: os.path.join(obj.absolute_path, "changed"))
|
||||
prepared = fsproperty.FileBooleanProperty(lambda obj: os.path.join(obj.absolute_path, "prepared"))
|
||||
ran = fsproperty.FileBooleanProperty(lambda obj: os.path.join(obj.absolute_path, "ran"))
|
||||
|
|
|
|||
|
|
@ -1,155 +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 collections
|
||||
|
||||
import cdist
|
||||
|
||||
|
||||
class FileList(collections.MutableSequence):
|
||||
"""A list that stores it's state in a file.
|
||||
|
||||
"""
|
||||
def __init__(self, path, initial=None):
|
||||
self._path = path
|
||||
if initial:
|
||||
# delete existing file
|
||||
os.unlink(self._path)
|
||||
for i in initial:
|
||||
self.append(i)
|
||||
|
||||
def __read(self):
|
||||
lines = []
|
||||
try:
|
||||
with open(self._path) as fd:
|
||||
for line in fd:
|
||||
lines.append(line.rstrip('\n'))
|
||||
except EnvironmentError as e:
|
||||
# error ignored
|
||||
pass
|
||||
return lines
|
||||
|
||||
def __write(self, lines):
|
||||
try:
|
||||
with open(self._path, 'w') as fd:
|
||||
for line in lines:
|
||||
fd.write(str(line) + '\n')
|
||||
except EnvironmentError as e:
|
||||
# error ignored
|
||||
raise
|
||||
|
||||
def __repr__(self):
|
||||
return repr(list(self))
|
||||
|
||||
def __getitem__(self, index):
|
||||
return self.__read()[index]
|
||||
|
||||
def __setitem__(self, index, value):
|
||||
lines = self.__read()
|
||||
lines[index] = value
|
||||
self.__write(lines)
|
||||
|
||||
def __delitem__(self, index):
|
||||
lines = self.__read()
|
||||
del lines[index]
|
||||
self.__write(lines)
|
||||
|
||||
def __len__(self):
|
||||
lines = self.__read()
|
||||
return len(lines)
|
||||
|
||||
def insert(self, index, value):
|
||||
lines = self.__read()
|
||||
lines.insert(index, value)
|
||||
self.__write(lines)
|
||||
|
||||
def sort(self):
|
||||
lines = sorted(self)
|
||||
self.__write(lines)
|
||||
|
||||
|
||||
class FileListProperty(FileList):
|
||||
# Descriptor Protocol
|
||||
def __get__(self, obj, objtype=None):
|
||||
if obj is None:
|
||||
return self.__class__
|
||||
return self
|
||||
|
||||
def __set__(self, obj, value):
|
||||
os.unlink(self._path)
|
||||
for item in value:
|
||||
self.append(item)
|
||||
|
||||
def __delete__(self, obj):
|
||||
raise AttributeError("can't delete attribute")
|
||||
|
||||
|
||||
class DirectoryDict(collections.MutableMapping):
|
||||
"""A dict that stores it's state in a directory.
|
||||
|
||||
"""
|
||||
def __init__(self, path, dict=None, **kwargs):
|
||||
self._path = path
|
||||
if dict is not None:
|
||||
self.update(dict)
|
||||
if len(kwargs):
|
||||
self.update(kwargs)
|
||||
|
||||
def __repr__(self):
|
||||
return repr(dict(self))
|
||||
|
||||
def __getitem__(self, key):
|
||||
try:
|
||||
with open(os.path.join(self._path, key), "r") as fd:
|
||||
return fd.read().rstrip('\n')
|
||||
except EnvironmentError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
with open(os.path.join(self._path, key), "w") as fd:
|
||||
fd.write(str(value))
|
||||
|
||||
def __delitem__(self, key):
|
||||
os.remove(os.path.join(self._path, key))
|
||||
|
||||
def __iter__(self):
|
||||
return iter(os.listdir(self._path))
|
||||
|
||||
def __len__(self):
|
||||
return len(os.listdir(self._path))
|
||||
|
||||
|
||||
class DirectoryDictProperty(DirectoryDict):
|
||||
# Descriptor Protocol
|
||||
def __get__(self, obj, objtype=None):
|
||||
if obj is None:
|
||||
return self.__class__
|
||||
return self
|
||||
|
||||
def __set__(self, obj, value):
|
||||
for name in self.keys():
|
||||
del self[name]
|
||||
if value is not None:
|
||||
self.update(value)
|
||||
|
||||
def __delete__(self, obj):
|
||||
raise AttributeError("can't delete attribute")
|
||||
|
|
@ -0,0 +1,117 @@
|
|||
# -*- 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 tempfile
|
||||
import unittest
|
||||
import shutil
|
||||
|
||||
import cdist.core
|
||||
|
||||
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(unittest.TestCase):
|
||||
|
||||
def test_list_object_names(self):
|
||||
object_names = list(cdist.core.Object.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.Object.list_type_names(object_base_path))
|
||||
self.assertEqual(type_names, ['__first', '__second', '__third'])
|
||||
|
||||
def test_list_objects(self):
|
||||
objects = list(cdist.core.Object.list_objects(object_base_path, type_base_path))
|
||||
objects_expected = [
|
||||
cdist.core.Object(cdist.core.Type(type_base_path, '__first'), object_base_path, 'man'),
|
||||
cdist.core.Object(cdist.core.Type(type_base_path, '__second'), object_base_path, 'on-the'),
|
||||
cdist.core.Object(cdist.core.Type(type_base_path, '__third'), object_base_path, 'moon'),
|
||||
]
|
||||
self.assertEqual(objects, objects_expected)
|
||||
|
||||
|
||||
class ObjectTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.cdist_type = cdist.core.Type(type_base_path, '__third')
|
||||
self.cdist_object = cdist.core.Object(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
|
||||
|
||||
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_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_prepared(self):
|
||||
self.assertFalse(self.cdist_object.prepared)
|
||||
|
||||
def test_prepared_after_changing(self):
|
||||
self.cdist_object.prepared = True
|
||||
self.assertTrue(self.cdist_object.prepared)
|
||||
|
||||
def test_ran(self):
|
||||
self.assertFalse(self.cdist_object.ran)
|
||||
|
||||
def test_ran_after_changing(self):
|
||||
self.cdist_object.ran = True
|
||||
self.assertTrue(self.cdist_object.ran)
|
||||
|
|
@ -0,0 +1 @@
|
|||
Prometheus
|
||||
|
|
@ -0,0 +1 @@
|
|||
Saturn
|
||||
|
|
@ -1,46 +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 tempfile
|
||||
import unittest
|
||||
import shutil
|
||||
|
||||
import os.path as op
|
||||
my_dir = op.abspath(op.dirname(__file__))
|
||||
base_path = op.join(my_dir, 'fixtures')
|
||||
|
||||
class ObjectTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# FIXME: use defined set of types for testing?
|
||||
# FIXME: generate object tree or use predefined?
|
||||
self.object_base_dir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.temp_dir)
|
||||
|
||||
'''
|
||||
suite = unittest.TestLoader().loadTestsFromTestCase(ObjectTestCase)
|
||||
|
||||
def suite():
|
||||
tests = []
|
||||
return unittest.TestSuite(map(ObjectTestCase, tests))
|
||||
'''
|
||||
|
|
@ -0,0 +1,146 @@
|
|||
# -*- 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 unittest
|
||||
|
||||
import cdist.core
|
||||
|
||||
import os.path as op
|
||||
my_dir = op.abspath(op.dirname(__file__))
|
||||
fixtures = op.join(my_dir, 'fixtures')
|
||||
|
||||
|
||||
class TypeTestCase(unittest.TestCase):
|
||||
|
||||
def test_list_type_names(self):
|
||||
base_path = op.join(fixtures, 'list_types')
|
||||
type_names = cdist.core.Type.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(cdist.core.Type.list_types(base_path))
|
||||
types_expected = [
|
||||
cdist.core.Type(base_path, '__first'),
|
||||
cdist.core.Type(base_path, '__second'),
|
||||
cdist.core.Type(base_path, '__third'),
|
||||
]
|
||||
self.assertEqual(types, types_expected)
|
||||
|
||||
def test_only_one_instance(self):
|
||||
base_path = fixtures
|
||||
cdist_type1 = cdist.core.Type(base_path, '__name_path')
|
||||
cdist_type2 = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(id(cdist_type1), id(cdist_type2))
|
||||
|
||||
def test_name(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(cdist_type.name, '__name_path')
|
||||
|
||||
def test_path(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(cdist_type.path, '__name_path')
|
||||
|
||||
def test_absolute_path(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(base_path, '__singleton')
|
||||
self.assertTrue(cdist_type.is_singleton)
|
||||
|
||||
def test_not_singleton_is_singleton(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__not_singleton')
|
||||
self.assertFalse(cdist_type.is_singleton)
|
||||
|
||||
def test_install_is_install(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__install')
|
||||
self.assertTrue(cdist_type.is_install)
|
||||
|
||||
def test_not_install_is_install(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__not_install')
|
||||
self.assertFalse(cdist_type.is_install)
|
||||
|
||||
def test_with_explorers(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_explorers')
|
||||
self.assertEqual(cdist_type.explorers, ['whatever'])
|
||||
|
||||
def test_without_explorers(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_explorers')
|
||||
self.assertEqual(cdist_type.explorers, [])
|
||||
|
||||
def test_with_required_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_required_parameters')
|
||||
self.assertEqual(cdist_type.required_parameters, ['required1', 'required2'])
|
||||
|
||||
def test_without_required_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_required_parameters')
|
||||
self.assertEqual(cdist_type.required_parameters, [])
|
||||
|
||||
def test_with_optional_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_optional_parameters')
|
||||
self.assertEqual(cdist_type.optional_parameters, ['optional1', 'optional2'])
|
||||
|
||||
def test_without_optional_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_optional_parameters')
|
||||
self.assertEqual(cdist_type.optional_parameters, [])
|
||||
|
||||
'''
|
||||
suite = unittest.TestLoader().loadTestsFromTestCase(ObjectTestCase)
|
||||
|
||||
def suite():
|
||||
tests = []
|
||||
return unittest.TestSuite(map(ObjectTestCase, tests))
|
||||
'''
|
||||
|
|
@ -1,146 +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 unittest
|
||||
|
||||
import cdist.core
|
||||
|
||||
import os.path as op
|
||||
my_dir = op.abspath(op.dirname(__file__))
|
||||
fixtures = op.join(my_dir, 'fixtures')
|
||||
|
||||
|
||||
class TypeTestCase(unittest.TestCase):
|
||||
|
||||
def test_list_type_names(self):
|
||||
base_path = op.join(fixtures, 'list_types')
|
||||
type_names = cdist.core.Type.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(cdist.core.Type.list_types(base_path))
|
||||
types_expected = [
|
||||
cdist.core.Type(base_path, '__first'),
|
||||
cdist.core.Type(base_path, '__second'),
|
||||
cdist.core.Type(base_path, '__third'),
|
||||
]
|
||||
self.assertEqual(types, types_expected)
|
||||
|
||||
def test_only_one_instance(self):
|
||||
base_path = fixtures
|
||||
cdist_type1 = cdist.core.Type(base_path, '__name_path')
|
||||
cdist_type2 = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(id(cdist_type1), id(cdist_type2))
|
||||
|
||||
def test_name(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(cdist_type.name, '__name_path')
|
||||
|
||||
def test_path(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__name_path')
|
||||
self.assertEqual(cdist_type.path, '__name_path')
|
||||
|
||||
def test_absolute_path(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(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 = cdist.core.Type(base_path, '__singleton')
|
||||
self.assertTrue(cdist_type.is_singleton)
|
||||
|
||||
def test_not_singleton_is_singleton(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__not_singleton')
|
||||
self.assertFalse(cdist_type.is_singleton)
|
||||
|
||||
def test_install_is_install(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__install')
|
||||
self.assertTrue(cdist_type.is_install)
|
||||
|
||||
def test_not_install_is_install(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__not_install')
|
||||
self.assertFalse(cdist_type.is_install)
|
||||
|
||||
def test_with_explorers(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_explorers')
|
||||
self.assertEqual(cdist_type.explorers, ['whatever'])
|
||||
|
||||
def test_without_explorers(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_explorers')
|
||||
self.assertEqual(cdist_type.explorers, [])
|
||||
|
||||
def test_with_required_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_required_parameters')
|
||||
self.assertEqual(cdist_type.required_parameters, ['required1', 'required2'])
|
||||
|
||||
def test_without_required_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_required_parameters')
|
||||
self.assertEqual(cdist_type.required_parameters, [])
|
||||
|
||||
def test_with_optional_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__with_optional_parameters')
|
||||
self.assertEqual(cdist_type.optional_parameters, ['optional1', 'optional2'])
|
||||
|
||||
def test_without_optional_parameters(self):
|
||||
base_path = fixtures
|
||||
cdist_type = cdist.core.Type(base_path, '__without_optional_parameters')
|
||||
self.assertEqual(cdist_type.optional_parameters, [])
|
||||
|
||||
'''
|
||||
suite = unittest.TestLoader().loadTestsFromTestCase(ObjectTestCase)
|
||||
|
||||
def suite():
|
||||
tests = []
|
||||
return unittest.TestSuite(map(ObjectTestCase, tests))
|
||||
'''
|
||||
|
|
@ -0,0 +1,272 @@
|
|||
# -*- 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 collections
|
||||
|
||||
import cdist
|
||||
|
||||
|
||||
class AbsolutePathRequiredError(cdist.Error):
|
||||
def __init__(self, path):
|
||||
self.path = path
|
||||
|
||||
def __str__(self):
|
||||
return 'Absolute path required, got: %s' % self.path
|
||||
|
||||
|
||||
class FileList(collections.MutableSequence):
|
||||
"""A list that stores it's state in a file.
|
||||
|
||||
"""
|
||||
def __init__(self, path, initial=None):
|
||||
if not os.path.isabs(path):
|
||||
raise AbsolutePathRequiredError(path)
|
||||
self.path = path
|
||||
if initial:
|
||||
# delete existing file
|
||||
os.unlink(self.path)
|
||||
for i in initial:
|
||||
self.append(i)
|
||||
|
||||
def __read(self):
|
||||
lines = []
|
||||
try:
|
||||
with open(self.path) as fd:
|
||||
for line in fd:
|
||||
lines.append(line.rstrip('\n'))
|
||||
except EnvironmentError as e:
|
||||
# error ignored
|
||||
pass
|
||||
return lines
|
||||
|
||||
def __write(self, lines):
|
||||
try:
|
||||
with open(self.path, 'w') as fd:
|
||||
for line in lines:
|
||||
fd.write(str(line) + '\n')
|
||||
except EnvironmentError as e:
|
||||
# error ignored
|
||||
raise
|
||||
|
||||
def __repr__(self):
|
||||
return repr(list(self))
|
||||
|
||||
def __getitem__(self, index):
|
||||
return self.__read()[index]
|
||||
|
||||
def __setitem__(self, index, value):
|
||||
lines = self.__read()
|
||||
lines[index] = value
|
||||
self.__write(lines)
|
||||
|
||||
def __delitem__(self, index):
|
||||
lines = self.__read()
|
||||
del lines[index]
|
||||
self.__write(lines)
|
||||
|
||||
def __len__(self):
|
||||
lines = self.__read()
|
||||
return len(lines)
|
||||
|
||||
def insert(self, index, value):
|
||||
lines = self.__read()
|
||||
lines.insert(index, value)
|
||||
self.__write(lines)
|
||||
|
||||
def sort(self):
|
||||
lines = sorted(self)
|
||||
self.__write(lines)
|
||||
|
||||
|
||||
class FileListProperty(FileList):
|
||||
|
||||
def __init__(self, path):
|
||||
"""
|
||||
:param path: string or callable
|
||||
|
||||
Usage:
|
||||
|
||||
class Foo(object):
|
||||
parameters = DirectoryDictProperty(lambda obj: os.path.join(obj.absolute_path, 'parameter'))
|
||||
other_dict = DirectoryDictProperty('/tmp/folder')
|
||||
|
||||
def __init__(self):
|
||||
self.absolute_path = '/tmp/foo'
|
||||
|
||||
"""
|
||||
self.path = None
|
||||
self.__path = path
|
||||
|
||||
def _set_path(self, *args, **kwargs):
|
||||
if self.path is None:
|
||||
path = self.__path
|
||||
if callable(path):
|
||||
path = path(*args, **kwargs)
|
||||
if not os.path.isabs(path):
|
||||
raise AbsolutePathRequiredError(path)
|
||||
self.path = path
|
||||
|
||||
# Descriptor Protocol
|
||||
def __get__(self, obj, objtype=None):
|
||||
if obj is None:
|
||||
return self.__class__
|
||||
self._set_path(obj)
|
||||
return self
|
||||
|
||||
def __set__(self, obj, value):
|
||||
self._set_path(obj)
|
||||
os.unlink(self.path)
|
||||
for item in value:
|
||||
self.append(item)
|
||||
|
||||
def __delete__(self, obj):
|
||||
raise AttributeError("can't delete attribute")
|
||||
|
||||
|
||||
class DirectoryDict(collections.MutableMapping):
|
||||
"""A dict that stores it's state in a directory.
|
||||
|
||||
"""
|
||||
def __init__(self, path, initial=None, **kwargs):
|
||||
if not os.path.isabs(path):
|
||||
raise AbsolutePathRequiredError(path)
|
||||
self.path = path
|
||||
if initial is not None:
|
||||
self.update(initial)
|
||||
if kwargs:
|
||||
self.update(kwargs)
|
||||
|
||||
def __repr__(self):
|
||||
return repr(dict(self))
|
||||
|
||||
def __getitem__(self, key):
|
||||
try:
|
||||
with open(os.path.join(self.path, key), "r") as fd:
|
||||
return fd.read().rstrip('\n')
|
||||
except EnvironmentError:
|
||||
raise KeyError(key)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
with open(os.path.join(self.path, key), "w") as fd:
|
||||
fd.write(str(value))
|
||||
|
||||
def __delitem__(self, key):
|
||||
os.remove(os.path.join(self.path, key))
|
||||
|
||||
def __iter__(self):
|
||||
return iter(os.listdir(self.path))
|
||||
|
||||
def __len__(self):
|
||||
return len(os.listdir(self.path))
|
||||
|
||||
|
||||
class DirectoryDictProperty(DirectoryDict):
|
||||
|
||||
def __init__(self, path):
|
||||
"""
|
||||
:param path: string or callable
|
||||
|
||||
Usage:
|
||||
|
||||
class Foo(object):
|
||||
parameters = DirectoryDictProperty(lambda obj: os.path.join(obj.absolute_path, 'parameter'))
|
||||
other_dict = DirectoryDictProperty('/tmp/folder')
|
||||
|
||||
def __init__(self):
|
||||
self.absolute_path = '/tmp/foo'
|
||||
|
||||
"""
|
||||
self.path = None
|
||||
self.__path = path
|
||||
|
||||
def _set_path(self, *args, **kwargs):
|
||||
if self.path is None:
|
||||
path = self.__path
|
||||
if callable(path):
|
||||
path = path(*args, **kwargs)
|
||||
if not os.path.isabs(path):
|
||||
raise AbsolutePathRequiredError(path)
|
||||
self.path = path
|
||||
|
||||
# Descriptor Protocol
|
||||
def __get__(self, obj, objtype=None):
|
||||
if obj is None:
|
||||
return self.__class__
|
||||
self._set_path(obj)
|
||||
return self
|
||||
|
||||
def __set__(self, obj, value):
|
||||
self._set_path(obj)
|
||||
if value is not None:
|
||||
for name in self.keys():
|
||||
del self[name]
|
||||
self.update(value)
|
||||
|
||||
def __delete__(self, obj):
|
||||
raise AttributeError("can't delete attribute")
|
||||
|
||||
|
||||
class FileBooleanProperty(object):
|
||||
def __init__(self, path):
|
||||
"""
|
||||
:param path: string or callable
|
||||
|
||||
Usage:
|
||||
|
||||
class Foo(object):
|
||||
changed = FileBoolean(lambda obj: os.path.join(obj.absolute_path, 'changed'))
|
||||
other_boolean = FileBoolean('/tmp/other_boolean')
|
||||
|
||||
def __init__(self):
|
||||
self.absolute_path = '/tmp/foo_boolean'
|
||||
|
||||
"""
|
||||
self._path = path
|
||||
|
||||
def _get_path(self, *args, **kwargs):
|
||||
path = self._path
|
||||
if callable(path):
|
||||
return path(*args, **kwargs)
|
||||
if not os.path.isabs(path):
|
||||
raise AbsolutePathRequiredError(path)
|
||||
return path
|
||||
|
||||
# Descriptor Protocol
|
||||
def __get__(self, obj, objtype=None):
|
||||
if obj is None:
|
||||
return self.__class__
|
||||
path = self._get_path(obj)
|
||||
return os.path.isfile(path)
|
||||
|
||||
def __set__(self, obj, value):
|
||||
path = self._get_path(obj)
|
||||
if value:
|
||||
open(path, "w").close()
|
||||
else:
|
||||
try:
|
||||
os.remove(path)
|
||||
except EnvironmentError:
|
||||
# ignore
|
||||
pass
|
||||
|
||||
def __delete__(self, obj):
|
||||
raise AttributeError("can't delete attribute")
|
||||
Loading…
Reference in New Issue