launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #06212
[Merge] lp:~jtv/maas/fake-cobbler into lp:maas
Jeroen T. Vermeulen has proposed merging lp:~jtv/maas/fake-cobbler into lp:maas.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~jtv/maas/fake-cobbler/+merge/90894
Test fake for a subset of the Cobbler API. This contains behaviours we can now test and compare to a real Cobbler.
We're still learning about the Cobbler XMLRPC API, and some things got simpler. In particular, object handles have disappeared from the API, and there are no longer separate "create object" and "modify object" calls — all you can do now is create and initialize an object, without caring whether it already existed.
Furthermore, CobblerSession's authenticate() method becomes private. The object knows when it needs to authenticate and does so on demand.
As far as I've been able to make out, cobbler uses a weird and dangerous mix of hyphens and underscores in identifiers, which interferes with the use of **kwargs in the API; I extended the lists of known attributes and used them to hide this wart. In MaaS you can simply use underscores all the time.
--
https://code.launchpad.net/~jtv/maas/fake-cobbler/+merge/90894
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jtv/maas/fake-cobbler into lp:maas.
=== modified file 'src/provisioningserver/cobblerclient.py'
--- src/provisioningserver/cobblerclient.py 2012-01-27 16:15:01 +0000
+++ src/provisioningserver/cobblerclient.py 2012-01-31 16:17:24 +0000
@@ -14,10 +14,11 @@
__metaclass__ = type
__all__ = [
- 'CobblerCommands',
'CobblerDistro',
'CobblerImage',
+ 'CobblerPreseeds',
'CobblerProfile',
+ 'CobblerRepo',
'CobblerSystem',
'DEFAULT_TIMEOUT',
]
@@ -84,10 +85,11 @@
return (self.connection_count, self.token)
@inlineCallbacks
- def authenticate(self, previous_state=None):
+ def _authenticate(self, previous_state=None):
"""Log in asynchronously.
- Call this when starting up, but also when an XMLRPC call result
+ This is called when an API function needs authentication when the
+ session isn't authenticated, but also when an XMLRPC call result
indicates that the authentication token used for a request has
expired.
@@ -121,7 +123,7 @@
finally:
self.authentication_lock.release()
- def substitute_token(self, arg):
+ def _substitute_token(self, arg):
"""Return `arg`, or the current auth token for `token_placeholder`."""
if arg is self.token_placeholder:
return self.token
@@ -155,7 +157,7 @@
substituted in its place.
:return: `Deferred`.
"""
- args = map(self.substitute_token, args)
+ args = map(self._substitute_token, args)
d = self._with_timeout(self.proxy.callRemote(method, *args))
return d
@@ -173,25 +175,24 @@
:return: A `Deferred` representing the call.
"""
original_state = self.record_state()
- authenticate = (self.token_placeholder in args)
-
- authentication_expired = (authenticate and self.token is None)
- if not authentication_expired:
+ uses_auth = (self.token_placeholder in args)
+ need_auth = (uses_auth and self.token is None)
+ if not need_auth:
# It looks like we're authenticated. Issue the call. If we
# then find out that our authentication token is invalid, we
# can retry it later.
try:
result = yield self._issue_call(method, *args)
except xmlrpclib.Fault as e:
- if authenticate and looks_like_auth_expiry(e):
- authentication_expired = True
+ if uses_auth and looks_like_auth_expiry(e):
+ need_auth = True
else:
raise
- if authentication_expired:
+ if need_auth:
# We weren't authenticated when we started, but we should be
# now. Make the final attempt.
- yield self.authenticate(original_state)
+ yield self._authenticate(original_state)
result = yield self._issue_call(method, *args)
returnValue(result)
@@ -207,7 +208,8 @@
Must be set in concrete derived classes.
:ivar object_type_plural: Optional plural for the type's identifier.
If not given, is derived by suffixing `object_type` with an "s".
- :ivar known_attributes: Attributes that this object is known to have.
+ :ivar known_attributes: Attributes that this type of object is known
+ to have.
"""
# What are objects of this type called in the Cobbler API?
@@ -219,29 +221,31 @@
# What attributes do we expect to support for this type of object?
# Only these attributes are allowed. This is here to force us to
# keep an accurate record of which attributes we use for which types
- # of objects. We may find that it helps us catch mistakes, or we
- # may want to let this go once we're comfortable and stable with the
- # API.
+ # of objects.
+ # Some attributes in Cobbler uses dashes as separators, others use
+ # underscores. In MaaS, use only underscores.
known_attributes = []
- def __init__(self, session, handle=None, name=None, values=None):
+ def __init__(self, session, name=None, values=None):
"""Reference an object in Cobbler.
:param session: A `CobblerSession`.
- :param handle: The object's handle, if known.
:param name: Name for this object, if known.
:param values: Attribute values for this object, if known.
"""
if values is None:
values = {}
self.session = session
- # Cache the handle; we need it when modifying or saving objects.
- self.handle = handle or values.get('handle')
# Cache the name; we need it when deleting objects.
self.name = name or values.get('name')
+ def _get_handle(self):
+ method = self._name_method('get_%s_handle')
+ return self.session.call(
+ method, self.name, self.session.token_placeholder)
+
@classmethod
- def name_method(cls, name_template, plural=False):
+ def _name_method(cls, name_template, plural=False):
"""Interpolate object_type into a method name template.
For example, on `CobblerSystem`, "get_%s_handle" would be
@@ -255,12 +259,33 @@
return name_template % type_name
@classmethod
- @inlineCallbacks
- def retrieve(cls, session, name):
- """Reference an object from Cobbler's database."""
- method = cls.name_method('get_%s_handle')
- handle = yield session.call(method, name, session.token_placeholder)
- returnValue(cls(session, handle, name=name))
+ def _normalize_attribute(cls, attribute_name):
+ """Normalize an attribute name.
+
+ Cobbler mixes dashes and underscores in attribute names. MaaS may
+ pass attributes as keyword arguments internally, where dashes are not
+ an option. Hide the distinction by looking up the proper name in
+ `known_attributes`.
+
+ :param attribute_name: An attribute name, possibly using underscores
+ where Cobbler expects dashes.
+ :return: A Cobbler-style attribute name, using either dashes or
+ underscores as used by Cobbler.
+ """
+ if attribute_name in cls.known_attributes:
+ # Already spelled the way Cobbler likes it.
+ return attribute_name
+
+ attribute_name = attribute_name.replace('_', '-')
+ if attribute_name in cls.known_attributes:
+ # Cobbler wants this one with dashes.
+ return attribute_name
+
+ attribute_name = attribute_name.replace('-', '_')
+ assert attribute_name in cls.known_attributes, (
+ "Unknown attribute for %s: %s."
+ % (cls.object_type, attribute_name))
+ return attribute_name
@classmethod
@inlineCallbacks
@@ -275,22 +300,38 @@
:return: A list of `cls` objects.
"""
if kwargs:
- method_template = "find_%s"
- args = (kwargs, )
+ method = cls._name_method("find_%s")
+ criteria = dict(
+ (cls._normalize_attribute(key), value)
+ for key, value in kwargs.items())
+ result = yield session.call(method, criteria)
else:
- method_template = "get_%s"
- args = ()
- method = cls.name_method(method_template, plural=True)
- result = yield session.call(method, *args)
+ method = cls._name_method("get_%s", plural=True)
+ result = yield session.call(method)
returnValue([cls(session, values=item) for item in result])
@classmethod
@inlineCallbacks
- def new(cls, session):
- """Create an object in Cobbler."""
- method = 'new_%s' % cls.object_type
- handle = yield session.call(method, session.token_placeholder)
- returnValue(cls(session, handle))
+ def new(cls, session, name, attributes):
+ """Create an object in Cobbler.
+
+ :param session: A `CobblerSession` to operate in.
+ :param name: Identifying name for the new object.
+ :param attributes: Dict mapping attribute names to values.
+ """
+ args = dict(
+ (cls._normalize_attribute(key), value)
+ for key, value in attributes.items())
+
+ # Overwrite any existing object of the same name. Unfortunately
+ # this parameter goes into the "attributes," and seems to be
+ # stored along with them. Its value doesn't matter.
+ args.setdefault('clobber', True)
+
+ yield session.call(
+ 'xapi_object_edit', cls.object_type, name, 'add', args,
+ session.token_placeholder)
+ returnValue(cls(session, name=name, values=args))
@inlineCallbacks
def delete(self, recurse=True):
@@ -300,62 +341,35 @@
"""
assert self.name is not None, (
"Can't delete %s; don't know its name." % self.object_type)
- method = self.name_method('remove_%s')
+ method = self._name_method('remove_%s')
yield self.session.call(
method, self.name, self.session.token_placeholder, recurse)
- @inlineCallbacks
- def _modify_attributes(self, attributes):
- """Attempt to modify the object's attributes."""
- method = 'modify_%s' % self.object_type
- for key, value in attributes.items():
- assert key in self.known_attributes, (
- "Unknown attribute for %s: %s." % (self.object_type, key))
- yield self.session.call(
- method, self.handle, key, value,
- self.session.token_placeholder)
- if key == 'name':
- self.name = value
-
- @inlineCallbacks
- def _save_attributes(self):
- """Save object's current state."""
- method = 'modify_%s' % self.object_type
- yield self.session.call(
- method, self.handle, self.session.token_placeholder)
-
- @inlineCallbacks
- def modify(self, **attributes):
- """Modify this object's attributes, and save.
-
- :param **attributes: Attribute values to set (as "attribute=value"
- keyword arguments).
- """
- original_state = self.session.record_state()
- yield self._modify_attributes(self, attributes)
- if self.session.record_state() != original_state:
- # Something went wrong and we had to re-authenticate our
- # session while we were modifying attributes. We can't be sure
- # that our changes all got through, so make them all again.
- original_state = self.session.record_state()
- yield self._modify_attributes(self, attributes)
- if self.session.record_state() != original_state:
- raise RuntimeError(
- "Cobbler session broke while modifying %s."
- % self.object_type)
-
- original_state = self.session.record_state()
- yield self._save_attributes()
- if self.session.record_state() != original_state:
- raise RuntimeError(
- "Cobbler session broke while saving %s." % self.object_type)
-
class CobblerProfile(CobblerObject):
"""A profile."""
object_type = 'profile'
known_attributes = [
+ 'distro',
+ 'comment',
+ 'enable-menu',
+ 'kickstart',
+ 'kopts',
+ 'kopts_post'
+ 'mgmt_classes',
'name',
+ 'name_servers',
+ 'name_servers_search',
+ 'owners',
+ 'repos',
+ 'template-files',
+ 'virt_auto_boot',
+ 'virt_bridge',
+ 'virt_cpus',
+ 'virt_file_size',
+ 'virt_disk_driver',
+ 'virt_path',
+ 'virt_ram',
]
@@ -363,7 +377,23 @@
"""An operating system image."""
object_type = "image"
known_attributes = [
+ 'arch',
+ # Set breed to 'debian' for Ubuntu.
+ 'breed',
+ 'comment',
+ 'file',
+ 'image_type',
'name',
+ 'os_version',
+ 'owners',
+ 'virt_auto_boot',
+ 'virt_bridge',
+ 'virt_cpus',
+ 'virt_disk_driver',
+ 'virt_file_size',
+ 'virt_path',
+ 'virt_ram',
+ 'virt_type',
]
@@ -371,12 +401,37 @@
"""A distribution."""
object_type = 'distro'
known_attributes = [
+ 'breed',
+ 'comment',
# Path to initrd image:
'initrd',
# Path to kernel:
'kernel',
+ 'kopts',
+ 'ksmeta',
+ 'mgmt-classes',
# Identifier:
'name',
+ 'os-version',
+ 'owners',
+ 'template-files',
+ ]
+
+
+class CobblerRepo(CobblerObject):
+ """A repository."""
+ object_type = 'repo'
+ known_attributes = [
+ 'arch',
+ 'comment',
+ 'createrepo_flags',
+ 'environment',
+ 'keep_updated',
+ 'mirror',
+ 'mirror_locally',
+ 'name',
+ 'owners',
+ 'priority',
]
@@ -384,18 +439,40 @@
"""A computer on the network."""
object_type = 'system'
known_attributes = [
+ 'boot_files',
+ 'comment',
+ 'fetchable_files',
+ 'gateway',
# FQDN:
'hostname',
# Space-separated key=value pairs:
'kernel_options'
+ 'kickstart',
+ 'kopts',
+ 'kopts_post',
# Space-separated key=value pairs for preseed:
'ks_meta',
+ 'mgmt_classes',
# A special dict; see below.
'modify_interface',
# Unqualified host name:
'name',
+ 'name_servers',
+ 'name_servers_search',
+ # Bool.
+ 'netboot_enabled',
+# TODO: Is this for ILO?
+ 'power_address',
+ 'power_id',
+ 'power_pass',
+ 'power_type',
+ 'power_user',
# Conventionally a distroseries-architecture combo.
'profile',
+ 'template_files',
+ 'uid',
+ 'virt_path',
+ 'virt_type',
]
# The modify_interface dict can contain:
@@ -403,72 +480,139 @@
# * "ipaddress-eth0" etc.
# * "dnsname-eth0" etc.
- @staticmethod
- def get_as_rendered(session, system_name):
- """Return system information in "blended" form.
-
- The blended form includes information "as koan (or PXE) (or
- templating) would evaluate it."
-
- I have no idea what this means, but it's in the cobbler API.
- """
- return session.call('get_system_as_rendered', system_name)
-
- @staticmethod
- def get_changed_systems(session, changed_since):
- """List systems changed since a given time."""
-# TODO: Who accounts for the race window?
- seconds_since_epoch = int(changed_since.strftime('%s'))
- return session.call('get_changed_systems', seconds_since_epoch)
-
+ @classmethod
+ def _callPowerMultiple(cls, session, operation, system_names):
+ """Call API's "background_power_system" method.
+
+ :return: Deferred.
+ """
+ d = session.call(
+ 'background_power_system',
+ {'power': operation, 'systems': system_names},
+ session.token_placeholder)
+ return d
+
+ @classmethod
+ def powerOnMultiple(cls, session, system_names):
+ """Initiate power-on for multiple systems.
+
+ There is no notification; we don't know if it succeeds or fails.
+
+ :return: Deferred.
+ """
+ return cls._callPowerMultiple(session, 'on', system_names)
+
+ @classmethod
+ def powerOffMultiple(cls, session, system_names):
+ """Initiate power-off for multiple systems.
+
+ There is no notification; we don't know if it succeeds or fails.
+
+ :return: Deferred.
+ """
+ return cls._callPowerMultiple(session, 'off', system_names)
+
+ @classmethod
+ def rebootMultiple(cls, session, system_names):
+ """Initiate reboot for multiple systems.
+
+ There is no notification; we don't know if it succeeds or fails.
+
+ :return: Deferred.
+ """
+ return cls._callPowerMultiple(session, 'reboot', system_names)
+
+ @inlineCallbacks
def _callPower(self, operation):
"""Call API's "power_system" method."""
- return self.session.call(
- 'power_system', operation, self.session.token_placeholder)
+ handle = yield self._get_handle()
+ yield self.session.call(
+ 'power_system', handle, operation,
+ self.session.token_placeholder)
def powerOn(self):
- """Turn system on."""
+ """Turn system on.
+
+ :return: Deferred.
+ """
return self._callPower('on')
def powerOff(self):
- """Turn system on."""
+ """Turn system on.
+
+ :return: Deferred.
+ """
return self._callPower('off')
def reboot(self):
- """Turn system on."""
+ """Turn system on.
+
+ :return: Deferred.
+ """
return self._callPower('reboot')
-class CobblerCommands:
- """Other miscellany: grab-bag of API leftovers."""
+class CobblerPreseeds:
+ """Deal with preseeds."""
def __init__(self, session):
self.session = session
- def read_preseed_template(self, path):
- """Read a preseed template."""
+ def read_template(self, path):
+ """Read a preseed template.
+
+ :return: Deferred.
+ """
return self.session.call(
'read_or_write_kickstart_template', path, True, '',
self.session.token_placeholder)
- def write_preseed_template(self, path, contents):
- """Write a preseed template."""
+ def write_template(self, path, contents):
+ """Write a preseed template.
+
+ :param path: Filesystem path for the template. Must be in
+ /var/lib/cobbler/kickstarts or /etc/cobbler
+ :param contents: Text of the template.
+ :return: Deferred.
+ """
return self.session.call(
'read_or_write_kickstart_template', path, False, contents,
self.session.token_placeholder)
- def read_preseed_snippet(self, path):
- """Read a preseed snippet."""
- return self.session.call(
- 'read_or_write_kickstart_snippet', path, True, '',
- self.session.token_placeholder)
-
- def write_preseed_snippet(self, path, contents):
- """Write a preseed snippet."""
- return self.session.call(
- 'read_or_write_kickstart_snippet', path, False, contents,
- self.session.token_placeholder)
+ def get_templates(self):
+ """Return the registered preseed templates."""
+ return self.session.call(
+ 'get_kickstart_templates', self.session.token_placeholder)
+
+ def read_snippet(self, path):
+ """Read a preseed snippet.
+
+ :return: Deferred.
+ """
+ return self.session.call(
+ 'read_or_write_snippet', path, True, '',
+ self.session.token_placeholder)
+
+ def write_snippet(self, path, contents):
+ """Write a preseed snippet.
+
+ :param path: Filesystem path for the snippet. Must be in
+ /var/lib/cobbler/snippets
+ :param contents: Text of the snippet.
+ :return: Deferred.
+ """
+ return self.session.call(
+ 'read_or_write_snippet', path, False, contents,
+ self.session.token_placeholder)
+
+ def get_snippets(self):
+ """Return the registered preseed snippets."""
+ return self.session.call(
+ 'get_snippets', self.session.token_placeholder)
def sync_netboot_configs(self):
- """Update netmasq and tftpd configurations."""
+ """Update netmasq and tftpd configurations.
+
+ :return: Deferred.
+ """
return self.session.call('sync', self.session.token_placeholder)
=== modified file 'src/provisioningserver/testing/fakecobbler.py'
--- src/provisioningserver/testing/fakecobbler.py 2012-01-26 10:36:14 +0000
+++ src/provisioningserver/testing/fakecobbler.py 2012-01-31 16:17:24 +0000
@@ -15,6 +15,7 @@
'fake_token',
]
+from fnmatch import fnmatch
from itertools import count
from random import randint
from xmlrpclib import Fault
@@ -58,12 +59,42 @@
class FakeCobbler:
"""Fake implementation of the Cobbler XMLRPC API.
- :param passwords: A dict mapping user names to their passwords.
-
- :ivar tokens: A dict mapping valid auth tokens to their users.
+ This does nothing useful, but tries to be internally consisten and
+ similar in use to a real Cobbler instance. Override as needed.
+
+ Public methods in this class represent Cobbler API calls, except ones
+ whose names start with `fake_`. Call those directly as needed, without
+ going through XMLRPC.
+
+ :ivar passwords: A dict mapping user names to their passwords.
+ :ivar tokens: Current authentication tokens, mapped to user names.
+ :ivar store: Store of Cobbler objects specific to each logged-in token,
+ or special value `None` for objects that have been saved to the
+ simulated Cobbler database. Address this dict as
+ `store[token][object_type][handle][attribute]`.
+ :ivar system_power: Last power action per system handle: 'on', 'off', or
+ 'reboot'.
+ :ivar preseed_templates: A dict mapping preseed templates' paths to
+ their contents.
+ :ivar preseed_snippets: A dict mapping preseed snippets' paths to
+ their contents.
"""
-
+ # Unlike the Cobbler-defined API methods, internal methods take
+ # subsets of these parameters, conventionally in this order:
+ # 0. self: duh.
+ # 1. token: represents a session, with its own local unsaved state.
+ # 2. object_type: type to operate on -- distro, system, profile etc.
+ # 3. operation: a sub-command to specialize the method further.
+ # 4. handle: an object's unique identifier.
+ # 5. <other>.
+ #
+ # Methods whose names start with "_api" are direct implementations
+ # of Cobbler API methods.
def __init__(self, passwords=None):
+ """Pretend to be a Cobbler instance.
+
+ :param passwords: A dict mapping user names to their passwords.
+ """
if passwords is None:
self.passwords = {}
else:
@@ -71,140 +102,348 @@
self.tokens = {}
- def fake_check_token(self, token):
- """Not part of the faked API: check token validity."""
+ # Store of Cobbler objects. This is a deeply nested dict:
+ # -> token for the session, or None for the saved objects
+ # -> object type (e.g. 'distro' or 'system')
+ # -> handle (assigned by FakeCobbler itself)
+ # -> attribute (an item in an object)
+ # -> value (which in some cases is another dict again).
+ self.store = {None: {}}
+
+ self.system_power = {}
+ self.preseed_templates = {}
+ self.preseed_snippets = {}
+
+ def _check_token(self, token):
if token not in self.tokens:
raise Fault(1, "invalid token: %s" % token)
+ def _raise_bad_handle(self, object_type, handle):
+ raise Fault(1, "Invalid %s handle: %s" % (object_type, handle))
+
+ def _register_type_for_token(self, token, object_type):
+ """Make room in dict for `object_type` when used with `token`."""
+ self.store.setdefault(token, {}).setdefault(object_type, {})
+
+ def _add_object_to_session(self, token, object_type, handle, obj_dict):
+ self._register_type_for_token(token, object_type)
+ self.store[token][object_type][handle] = obj_dict
+
+ def _remove_object_from_session(self, token, object_type, handle):
+ if handle in self.store.get(token, {}).get(object_type, {}):
+ del self.store[token][object_type][handle]
+
+ def _get_object_if_present(self, token, object_type, handle):
+ return self.store.get(token, {}).get(object_type, {}).get(handle)
+
+ def _get_handle_if_present(self, token, object_type, name):
+ candidates = self.store.get(token, {}).get(object_type, {})
+ for handle, candidate in candidates.items():
+ if candidate['name'] == name:
+ return handle
+ return None
+
+ def _matches(self, object_dict, criteria):
+ """Does `object_dict` satisfy the constraints in `criteria`?
+
+ :param object_dict: An object in dictionary form, as in the store.
+ :param criteria: A dict of constraints. Each is a glob.
+ :return: `True` if, for each key in `criteria`, there is a
+ corresponding key in `object_dict` whose value matches the
+ glob value as found in `criteria`.
+ """
+ # Assumption: None matches None.
+# TODO: If attribute is a list, match any item in the list.
+ return all(
+ fnmatch(object_dict.get(key), value)
+ for key, value in criteria.items())
+
+ def _api_new_object(self, token, object_type):
+ """Create object in the session's local store."""
+ self._check_token(token)
+ unique_id = next(unique_ints)
+ handle = "handle-%s-%d" % (object_type, unique_id)
+ name = "name-%s-%d" % (object_type, unique_id)
+ new_object = {
+ 'name': name,
+ }
+ self._add_object_to_session(token, object_type, handle, new_object)
+ return handle
+
+ def _api_remove_object(self, token, object_type, name):
+ """Remove object from the session and saved stores."""
+ # Assumption: removals don't need to be saved.
+ self._check_token(token)
+ handle = self._api_get_handle(token, object_type, name)
+ for session in [token, None]:
+ self._remove_object_from_session(session, object_type, handle)
+
+ def _api_get_handle(self, token, object_type, name):
+ """Get object handle by name.
+
+ Returns session-local version of the object if available, or
+ the saved version otherwise.
+ """
+ self._check_token(token)
+ handle = self._get_handle_if_present(token, object_type, name)
+ if handle is None:
+ handle = self._get_handle_if_present(None, object_type, name)
+ if handle is None:
+ raise Fault(1, "Unknown %s: %s." % (object_type, name))
+ return handle
+
+ def _api_find_objects(self, object_type, criteria):
+ """Find objects in the saved store that match `criteria`.
+
+ :return: A list of object dicts, as copied from the saved store.
+ """
+ # Assumption: these operations look only at saved objects.
+ location = self.store[None].get(object_type, {})
+ return [
+ dict(candidate)
+ for candidate in location.values()
+ if self._matches(candidate, criteria)]
+
+ def _api_get_objects(self, object_type):
+ """Return all saved objects of type `object_type`.
+
+ :return: A list of object dicts, as copied from the saved store.
+ """
+ # Assumption: these operations look only at saved objects.
+ return list(map(dict, self.store[object_type].values()))
+
+ def _api_modify_object(self, token, object_type, handle, key, value):
+ """Set an attribute on an object.
+
+ The object is copied into the session store if needed; the session
+ will see its locally modified version of the object until it saves
+ its changes. At that point, other sessions will get to see it too.
+ """
+ self._check_token(token)
+ session_obj = self._get_object_if_present(token, object_type, handle)
+ if session_obj is None:
+ # Draw a copy of the saved object into a session-local
+ # object.
+ saved_obj = self._get_object_if_present(None, object_type, handle)
+ if saved_obj is None:
+ self._raise_bad_handle(object_type, handle)
+ session_obj = dict(saved_obj)
+ self._add_object_to_session(
+ token, object_type, handle, session_obj)
+
+ session_obj[key] = value
+
+ def _api_save_object(self, token, object_type, handle):
+ """Save an object's modifications to the saved store."""
+ self._check_token(token)
+ saved_obj = self._get_object_if_present(None, object_type, handle)
+ session_obj = self._get_object_if_present(token, object_type, handle)
+
+ if session_obj is None and saved_obj is None:
+ self._raise_bad_handle(object_type, handle)
+ if session_obj is None:
+ # Object not modified. Nothing to do.
+ return True
+
+ name = session_obj['name']
+ other_handle = self._get_handle_if_present(token, object_type, name)
+ if other_handle is not None and other_handle != handle:
+ raise Fault(
+ 1, "The %s name '%s' is already in use."
+ % (object_type, name))
+
+ if saved_obj is None:
+ self._add_object_to_session(
+ None, object_type, handle, session_obj)
+ else:
+ saved_obj.update(session_obj)
+
+ self._remove_object_from_session(token, object_type, handle)
+ return True
+
+ def _api_access_preseed(self, token, read, preseeds_dict, path, contents):
+ """Read or write preseed template or snippet."""
+ assert read in [True, False], "Invalid 'read' value: %s." % read
+ self._check_token(token)
+ if read:
+ assert contents == '', "Pass empty contents when reading."
+ else:
+ preseeds_dict[path] = contents
+ return preseeds_dict[path]
+
+ def fake_retire_token(self, token):
+ """Pretend that `token` has expired."""
+ if token in self.store:
+ del self.store[token]
+ if token in self.tokens:
+ del self.tokens[token]
+
+ def fake_system_power(self, handle, power_status):
+ """Pretend that the given server has been turned on/off, or rebooted.
+
+ Use this, for example, to simulate completion of a reboot command.
+
+ :param handle: Handle for a system.
+ :param power_status: One of 'on', 'off', or 'reboot'.
+ """
+ self.system_power[handle] = power_status
+
def login(self, user, password):
if password != self.passwords.get(user, object()):
- raise Exception("login failed (%s)" % user)
+ raise Fault(1, "login failed (%s)" % user)
token = fake_token(user)
self.tokens[token] = user
return token
+ def xapi_object_edit(self, object_type, name, operation, attrs, token):
+ if operation == 'remove':
+ self._api_remove_object(token, object_type, name)
+ return True
+ elif operation == 'add':
+ handle = self._api_new_object(token, object_type)
+ obj_dict = self.store[token][object_type][handle]
+ obj_dict.update(attrs)
+ obj_dict['name'] = name
+ self._api_save_object(token, object_type, handle)
+ else:
+ raise NotImplemented(
+ "xapi_object_edit(%s, ..., %s, ...)"
+ % (object_type, operation))
+
def new_distro(self, token):
- self.fake_check_token(token)
- pass
+ return self._api_new_object(token, 'distro')
def remove_distro(self, name, token, recurse=True):
- self.fake_check_token(token)
- pass
+ self._api_remove_object(token, 'distro', name)
def get_distro_handle(self, name, token):
- self.fake_check_token(token)
- pass
+ return self._api_get_handle(token, 'distro', name)
- def find_distros(self, criteria):
- pass
+ def find_distro(self, criteria):
+ return self._api_find_objects('distro', criteria)
def get_distros(self):
- pass
+ return self._api_get_objects('distro')
def modify_distro(self, handle, key, value, token):
- self.fake_check_token(token)
- pass
+ self._api_modify_object(token, 'distro', handle, key, value)
def save_distro(self, handle, token):
- self.fake_check_token(token)
- pass
+ return self._api_save_object(token, 'distro', handle)
def new_image(self, token):
- self.fake_check_token(token)
- pass
+ return self._api_new_object(token, 'image')
def remove_image(self, name, token, recurse=True):
- self.fake_check_token(token)
- pass
+ self._api_remove_object(token, 'image', name)
def get_image_handle(self, name, token):
- self.fake_check_token(token)
- pass
+ return self._api_get_handle(token, 'image', name)
- def find_images(self, criteria):
- pass
+ def find_image(self, criteria):
+ return self._api_find_objects('image', criteria)
def get_images(self):
- pass
+ return self._api_get_objects('image')
def modify_image(self, handle, key, value, token):
- self.fake_check_token(token)
- pass
+ self._api_modify_object(token, 'image', handle, key, value)
def save_image(self, handle, token):
- self.fake_check_token(token)
- pass
+ return self._api_save_object(token, 'image', handle)
def new_profile(self, token):
- self.fake_check_token(token)
- pass
+ return self._api_new_object(token, 'profile')
def remove_profile(self, name, token, recurse=True):
- self.fake_check_token(token)
- pass
+ self._api_remove_object(token, 'profile', name)
def get_profile_handle(self, name, token):
- self.fake_check_token(token)
- pass
+ return self._api_get_handle(token, 'profile', name)
- def find_profiles(self, criteria):
- pass
+ def find_profile(self, criteria):
+ return self._api_find_objects('profile', criteria)
def get_profiles(self):
- pass
+ return self._api_get_objects('profile')
def modify_profile(self, handle, key, value, token):
- self.fake_check_token(token)
- pass
+ self._api_modify_object(token, 'profile', handle, key, value)
def save_profile(self, handle, token):
- self.fake_check_token(token)
- pass
+ return self._api_save_object(token, 'profile', handle)
+
+ def new_repo(self, token):
+ return self._api_new_object(token, 'repo')
+
+ def remove_repo(self, name, token, recurse=True):
+ self._api_remove_object(token, 'repo', name)
+
+ def get_repo_handle(self, name, token):
+ return self._api_get_handle(token, 'repo', name)
+
+ def find_repo(self, criteria):
+ return self._api_find_objects('repo', criteria)
+
+ def get_repos(self):
+ return self._api_get_objects('repo')
+
+ def modify_repo(self, handle, key, value, token):
+ self._api_modify_object(token, 'repo', handle, key, value)
+
+ def save_repo(self, handle, token):
+ return self._api_save_object(token, 'repo', handle)
def new_system(self, token):
- self.fake_check_token(token)
- pass
+ return self._api_new_object(token, 'system')
def remove_system(self, name, token, recurse=True):
- self.fake_check_token(token)
- pass
+ self._api_remove_object(token, 'system', name)
+
+ def background_power_system(self, args, token):
+ self._check_token(token)
+ operation = args['power']
+ system_names = args['systems']
+ handles = [
+ self.get_system_handle(name, token)
+ for name in system_names]
+ for handle in handles:
+ self.power_system(handle, operation, token)
def get_system_handle(self, name, token):
- self.fake_check_token(token)
- pass
+ return self._api_get_handle(token, 'system', name)
- def find_systems(self, criteria):
- pass
+ def find_system(self, criteria):
+ return self._api_find_objects('system', criteria)
def get_systems(self):
- pass
+ return self._api_get_objects('system')
def modify_system(self, handle, key, value, token):
- self.fake_check_token(token)
- pass
+ self._api_modify_object(token, 'system', handle, key, value)
def save_system(self, handle, token):
- self.fake_check_token(token)
- pass
-
- def get_system_as_rendered(self, name):
- pass
-
- def get_changed_systems(self, seconds_since_epoch):
- pass
-
- def power_system(self, operation, token):
- self.fake_check_token(token)
- pass
+ return self._api_save_object(token, 'system', handle)
+
+ def power_system(self, handle, operation, token):
+ self._check_token(token)
+ if operation not in ['on', 'off', 'reboot']:
+ raise Fault(1, "Invalid power operation: %s." % operation)
+ self.system_power[handle] = operation
def read_or_write_kickstart_template(self, path, read, contents, token):
- self.fake_check_token(token)
- pass
-
- def read_or_write_kickstart_snippet(self, path, read, contents, token):
- self.fake_check_token(token)
- pass
+ return self._api_access_preseed(
+ token, read, self.preseed_templates, path, contents)
+
+ def get_kickstart_templates(self, token=None):
+ return self.preseed_templates.keys()
+
+ def read_or_write_snippet(self, path, read, contents, token):
+ return self._api_access_preseed(
+ token, read, self.preseed_snippets, path, contents)
+
+ def get_snippets(self, token=None):
+ return self.preseed_snippets.keys()
def sync(self, token):
- self.fake_check_token(token)
- pass
+ self._check_token(token)
=== modified file 'src/provisioningserver/tests/test_cobblersession.py'
--- src/provisioningserver/tests/test_cobblersession.py 2012-01-27 16:23:18 +0000
+++ src/provisioningserver/tests/test_cobblersession.py 2012-01-31 16:17:24 +0000
@@ -11,11 +11,10 @@
__metaclass__ = type
__all__ = []
-import fixtures
-
from random import Random
from xmlrpclib import Fault
+import fixtures
from provisioningserver import cobblerclient
from provisioningserver.testing.fakecobbler import fake_token
from testtools.content import text_content
@@ -29,10 +28,7 @@
TestCase,
)
from twisted.internet import defer
-from twisted.internet.defer import (
- inlineCallbacks,
- returnValue,
- )
+from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import Clock
@@ -166,7 +162,7 @@
token = fake_token('authenticated')
session = self.make_recording_session(token=token)
self.assertEqual(None, session.token)
- yield session.authenticate()
+ yield session._authenticate()
self.assertEqual(token, session.token)
@inlineCallbacks
@@ -181,7 +177,7 @@
def test_authentication_changes_state_cookie(self):
session = self.make_recording_session()
old_cookie = session.record_state()
- yield session.authenticate()
+ yield session._authenticate()
self.assertNotEqual(old_cookie, session.record_state())
@inlineCallbacks
@@ -192,12 +188,12 @@
cookie_before_request_2 = session.record_state()
# Request 1 comes back with an authentication failure, and its
# callback refreshes the session's auth token.
- yield session.authenticate(cookie_before_request_1)
+ yield session._authenticate(cookie_before_request_1)
token_for_retrying_request_1 = session.token
# Request 2 also comes back an authentication failure, and its
# callback also asks the session to ensure that it is
# authenticated.
- yield session.authenticate(cookie_before_request_2)
+ yield session._authenticate(cookie_before_request_2)
token_for_retrying_request_2 = session.token
# The double authentication does not confuse the session; both
@@ -213,7 +209,7 @@
def test_substitute_token_substitutes_only_placeholder(self):
token = fake_token('for-substitution')
session = self.make_recording_session(token=token)
- yield session.authenticate()
+ yield session._authenticate()
arbitrary_number = pick_number()
arbitrary_string = 'string-%d' % pick_number()
inputs = [
@@ -229,7 +225,7 @@
None,
]
self.assertEqual(
- outputs, apply_to_all(session.substitute_token, inputs))
+ outputs, apply_to_all(session._substitute_token, inputs))
@inlineCallbacks
def test_call_calls_xmlrpc(self):
@@ -247,7 +243,7 @@
# If a call triggers an authentication error, call()
# re-authenticates and then re-issues the call.
session = self.make_recording_session()
- yield session.authenticate()
+ yield session._authenticate()
successful_return_value = pick_number()
session.proxy.set_return_values([
make_auth_failure(),
@@ -275,13 +271,13 @@
# call. This ensures that it will always notice a concurrent
# re-authentication that it needs to back off from.
session = self.make_recording_session()
- yield session.authenticate()
+ yield session._authenticate()
authenticate_cookies = []
def fake_authenticate(previous_state):
authenticate_cookies.append(previous_state)
- session.authenticate = fake_authenticate
+ session._authenticate = fake_authenticate
session.proxy.set_return_values([make_auth_failure()])
state_before_call = session.record_state()
yield session.call(
@@ -291,7 +287,7 @@
@inlineCallbacks
def test_call_raises_repeated_auth_failure(self):
session = self.make_recording_session()
- yield session.authenticate()
+ yield session._authenticate()
failures = [
# Initial operation fails: not authenticated.
make_auth_failure(),
@@ -307,7 +303,7 @@
@inlineCallbacks
def test_call_raises_general_failure(self):
session = self.make_recording_session()
- yield session.authenticate()
+ yield session._authenticate()
failure = Exception("Memory error. Where did I put it?")
session.proxy.set_return_values([failure])
with ExpectedException(Exception, failure.message):
@@ -334,7 +330,7 @@
fixtures.TestWithFixtures):
"""Tests for connection timeouts on `CobblerSession`."""
- run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted
+ run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted
def test__with_timeout_cancels(self):
# Winding a clock reactor past the timeout value should cancel
@@ -353,9 +349,11 @@
session = self.make_recording_session()
d = session._with_timeout(defer.succeed("frobnicle"), 1, clock)
clock.advance(2)
+
def result(value):
self.assertEqual(value, "frobnicle")
self.assertEqual([], clock.getDelayedCalls())
+
return d.addCallback(result)
def test__with_timeout_not_cancelled_unnecessarily(self):
@@ -375,7 +373,7 @@
session = self.make_recording_session(fake_proxy=DeadProxy)
d = session._issue_call("login", "foo")
- clock.advance(cobblerclient.DEFAULT_TIMEOUT+1)
+ clock.advance(cobblerclient.DEFAULT_TIMEOUT + 1)
return assert_fails_with(d, defer.CancelledError)
@@ -385,10 +383,9 @@
def test_name_method_inserts_type_name(self):
self.assertEqual(
'foo_system_bar',
- cobblerclient.CobblerSystem.name_method('foo_%s_bar'))
+ cobblerclient.CobblerSystem._name_method('foo_%s_bar'))
def test_name_method_appends_s_for_plural(self):
self.assertEqual(
'x_systems_y',
- cobblerclient.CobblerSystem.name_method('x_%s_y', plural=True))
-
+ cobblerclient.CobblerSystem._name_method('x_%s_y', plural=True))
=== modified file 'src/provisioningserver/tests/test_fakecobbler.py'
--- src/provisioningserver/tests/test_fakecobbler.py 2012-01-26 22:33:02 +0000
+++ src/provisioningserver/tests/test_fakecobbler.py 2012-01-31 16:17:24 +0000
@@ -11,7 +11,16 @@
__metaclass__ = type
__all__ = []
+from itertools import count
+from random import randint
+import xmlrpclib
+
from provisioningserver.cobblerclient import (
+ CobblerDistro,
+ CobblerImage,
+ CobblerPreseeds,
+ CobblerProfile,
+ CobblerRepo,
CobblerSession,
CobblerSystem,
)
@@ -31,6 +40,9 @@
)
+unique_ints = count(randint(0, 9999))
+
+
class FakeCobblerSession(CobblerSession):
"""A `CobblerSession` instrumented not to use real XMLRPC."""
@@ -48,10 +60,17 @@
@inlineCallbacks
def fake_cobbler_session(url=None, user=None, password=None,
fake_cobbler=None):
- """Fake a CobblerSession."""
+ """Fake a `CobblerSession`."""
+ unique_number = next(unique_ints)
+ if user is None:
+ user = "user%d" % unique_number
+ if password is None:
+ password = "password%d" % unique_number
+ if fake_cobbler is None:
+ fake_cobbler = FakeCobbler(passwords={user: password})
session = FakeCobblerSession(
url, user, password, fake_cobbler=fake_cobbler)
- yield session.authenticate()
+ yield session._authenticate()
returnValue(session)
@@ -66,33 +85,339 @@
@inlineCallbacks
def test_login_failure_raises_failure(self):
cobbler = FakeCobbler(passwords={'moi': 'potahto'})
- with ExpectedException(Exception):
+ with ExpectedException(xmlrpclib.Fault):
return_value = yield fake_cobbler_session(
user='moi', password='potayto', fake_cobbler=cobbler)
self.addDetail('return_value', text_content(repr(return_value)))
@inlineCallbacks
def test_expired_token_triggers_retry(self):
- cobbler = FakeCobbler(passwords={'user': 'pw'})
- session = yield fake_cobbler_session(
- user='user', password='pw', fake_cobbler=cobbler)
+ session = yield fake_cobbler_session()
# When an auth token expires, the server just forgets about it.
old_token = session.token
- del cobbler.tokens[session.token]
+ session.fake_proxy.fake_cobbler.fake_retire_token(old_token)
# Use of the token will now fail with an "invalid token"
# error. The Cobbler client notices this, re-authenticates, and
# re-runs the method.
- yield CobblerSystem.new(session)
+ yield CobblerSystem.new(session, 'my-session', {'comment': 'Mine'})
# The re-authentication results in a fresh token.
self.assertNotEqual(old_token, session.token)
@inlineCallbacks
def test_valid_token_does_not_raise_auth_error(self):
- cobbler = FakeCobbler(passwords={'user': 'password'})
- session = yield fake_cobbler_session(
- user='user', password='password', fake_cobbler=cobbler)
+ session = yield fake_cobbler_session()
old_token = session.token
- yield CobblerSystem.new(session)
+ yield CobblerSystem.new(session, 'some-session', {'comment': 'Boo'})
self.assertEqual(old_token, session.token)
+
+
+class CobblerObjectTestScenario:
+ """Generic tests for the various `CobblerObject` classes.
+
+ This will be run once for each of the classes in the hierarchy.
+ """
+
+ # The class to test
+ cobbler_class = None
+
+ def make_name(self):
+ return "name-%s-%d" % (
+ self.cobbler_class.object_type,
+ next(unique_ints),
+ )
+
+ def test_normalize_attribute_passes_on_simple_attribute_name(self):
+ self.assertEqual(
+ 'name', self.cobbler_class._normalize_attribute('name'))
+
+ def test_normalize_attribute_rejects_unknown_attribute(self):
+ self.assertRaises(
+ AssertionError,
+ self.cobbler_class._normalize_attribute, 'some-unknown-attribute')
+
+ @inlineCallbacks
+ def test_create_object(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ obj = yield self.cobbler_class.new(session, name, {})
+ self.assertEqual(name, obj.name)
+
+ @inlineCallbacks
+ def test_find_returns_empty_list_if_no_match(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ matches = yield self.cobbler_class.find(session, name=name)
+ self.assertEqual([], matches)
+
+ @inlineCallbacks
+ def test_find_matches_name(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ yield self.cobbler_class.new(session, name, {})
+ [by_comment] = yield self.cobbler_class.find(session, name=name)
+ self.assertEqual(name, by_comment.name)
+
+ @inlineCallbacks
+ def test_find_matches_attribute(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ yield self.cobbler_class.new(session, name, {'comment': 'Hi'})
+ [by_comment] = yield self.cobbler_class.find(session, comment='Hi')
+ self.assertEqual(name, by_comment.name)
+
+ @inlineCallbacks
+ def test_get_handle_finds_handle(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ obj = yield self.cobbler_class.new(session, name, {})
+ handle = yield obj._get_handle()
+ self.assertNotEqual(None, handle)
+
+ @inlineCallbacks
+ def test_get_handle_distinguishes_objects(self):
+ session = yield fake_cobbler_session()
+ obj1 = yield self.cobbler_class.new(session, self.make_name(), {})
+ handle1 = yield obj1._get_handle()
+ obj2 = yield self.cobbler_class.new(session, self.make_name(), {})
+ handle2 = yield obj2._get_handle()
+ self.assertNotEqual(handle1, handle2)
+
+ @inlineCallbacks
+ def test_get_handle_is_consistent(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ obj = yield self.cobbler_class.new(session, name, {})
+ handle1 = yield obj._get_handle()
+ handle2 = yield obj._get_handle()
+ self.assertEqual(handle1, handle2)
+
+ @inlineCallbacks
+ def test_delete_removes_object(self):
+ session = yield fake_cobbler_session()
+ name = self.make_name()
+ obj = yield self.cobbler_class.new(session, name, {'name': name})
+ obj.delete()
+ matches = yield self.cobbler_class.find(session, name=name)
+ self.assertEqual([], matches)
+
+
+class TestCobblerDistro(CobblerObjectTestScenario, TestCase):
+ """Tests for `CobblerDistro`. Uses generic `CobblerObject` scenario."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+ cobbler_class = CobblerDistro
+
+ def test_normalize_attribute_normalizes_separators(self):
+ # Based on the Cobbler source, Distro seems to use only dashes
+ # as separators in attribute names. The equivalents with
+ # underscores get translated to the ones Cobbler seems to
+ # expect.
+ inputs = [
+ 'mgmt-classes',
+ 'mgmt_classes',
+ ]
+ self.assertEqual(
+ ['mgmt-classes'] * 2,
+ list(map(self.cobbler_class._normalize_attribute, inputs)))
+
+
+class TestCobblerImage(CobblerObjectTestScenario, TestCase):
+ """Tests for `CobblerImage`. Uses generic `CobblerObject` scenario."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+ cobbler_class = CobblerImage
+
+ def test_normalize_attribute_normalizes_separators(self):
+ # Based on the Cobbler source, Image seems to use only
+ # underscores as separators in attribute names. The equivalents
+ # with dashes get translated to the ones Cobbler seems to
+ # expect.
+ inputs = [
+ 'image-type',
+ 'image_type',
+ ]
+ self.assertEqual(
+ ['image_type'] * 2,
+ list(map(self.cobbler_class._normalize_attribute, inputs)))
+
+
+class TestCobblerProfile(CobblerObjectTestScenario, TestCase):
+ """Tests for `CobblerProfile`. Uses generic `CobblerObject` scenario."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+ cobbler_class = CobblerProfile
+
+ def test_normalize_attribute_normalizes_separators(self):
+ # Based on the Cobbler source, Profile seems to use a mix of
+ # underscores and dashes in attribute names. The MaaS Cobbler
+ # wrapper ignores the difference, and uses whatever Cobbler
+ # seems to expect in either case.
+ inputs = [
+ 'enable-menu',
+ 'enable_menu',
+ 'name_servers',
+ 'name-servers',
+ ]
+ expected_outputs = [
+ 'enable-menu',
+ 'enable-menu',
+ 'name_servers',
+ 'name_servers',
+ ]
+ self.assertEqual(
+ expected_outputs,
+ list(map(self.cobbler_class._normalize_attribute, inputs)))
+
+
+class TestCobblerRepo(CobblerObjectTestScenario, TestCase):
+ """Tests for `CobblerRepo`. Uses generic `CobblerObject` scenario."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+ cobbler_class = CobblerRepo
+
+
+class TestCobblerSystem(CobblerObjectTestScenario, TestCase):
+ """Tests for `CobblerSystem`. Uses generic `CobblerObject` scenario."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+ cobbler_class = CobblerSystem
+
+ def make_name(self):
+ return 'system-%d' % next(unique_ints)
+
+ @inlineCallbacks
+ def make_systems(self, session, num_systems):
+ names = [self.make_name() for counter in range(num_systems)]
+ systems = []
+ for name in names:
+ new_system = yield self.cobbler_class.new(session, name, {})
+ systems.append(new_system)
+ returnValue((names, systems))
+
+ @inlineCallbacks
+ def get_handles(self, systems):
+ handles = []
+ for system in systems:
+ handle = yield system._get_handle()
+ handles.append(handle)
+ returnValue(handles)
+
+ @inlineCallbacks
+ def test_powerOnMultiple(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 3)
+ handles = yield self.get_handles(systems)
+ yield self.cobbler_class.powerOnMultiple(session, names[:2])
+ self.assertEqual(
+ dict((handle, 'on') for handle in handles[:2]),
+ session.fake_proxy.fake_cobbler.system_power)
+
+ @inlineCallbacks
+ def test_powerOffMultiple(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 3)
+ handles = yield self.get_handles(systems)
+ yield self.cobbler_class.powerOffMultiple(session, names[:2])
+ self.assertEqual(
+ dict((handle, 'off') for handle in handles[:2]),
+ session.fake_proxy.fake_cobbler.system_power)
+
+ @inlineCallbacks
+ def test_rebootMultiple(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 3)
+ handles = yield self.get_handles(systems)
+ yield self.cobbler_class.rebootMultiple(session, names[:2])
+ self.assertEqual(
+ dict((handle, 'reboot') for handle in handles[:2]),
+ session.fake_proxy.fake_cobbler.system_power)
+
+ @inlineCallbacks
+ def test_powerOn(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 2)
+ handle = yield systems[0]._get_handle()
+ yield systems[0].powerOn()
+ self.assertEqual(
+ {handle: 'on'}, session.fake_proxy.fake_cobbler.system_power)
+
+ @inlineCallbacks
+ def test_powerOff(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 2)
+ handle = yield systems[0]._get_handle()
+ yield systems[0].powerOff()
+ self.assertEqual(
+ {handle: 'off'}, session.fake_proxy.fake_cobbler.system_power)
+
+ @inlineCallbacks
+ def test_reboot(self):
+ session = yield fake_cobbler_session()
+ names, systems = yield self.make_systems(session, 2)
+ handle = yield systems[0]._get_handle()
+ yield systems[0].reboot()
+ self.assertEqual(
+ {handle: 'reboot'}, session.fake_proxy.fake_cobbler.system_power)
+
+
+class TestCobblerPreseeds(TestCase):
+ """Tests for `CobblerPreseeds`."""
+ run_tests_with = AsynchronousDeferredRunTest.make_factory()
+
+ @inlineCallbacks
+ def make_preseeds_api(self):
+ session = yield fake_cobbler_session()
+ returnValue(CobblerPreseeds(session))
+
+ @inlineCallbacks
+ def test_can_read_and_write_template(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/kickstarts/template-%d' % unique_int
+ text = "Template text #%d" % unique_int
+ yield preseeds.write_template(path, text)
+ contents = yield preseeds.read_template(path)
+ self.assertEqual(text, contents)
+
+ @inlineCallbacks
+ def test_can_read_and_write_snippets(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/snippets/snippet-%d' % unique_int
+ text = "Snippet text #%d" % unique_int
+ yield preseeds.write_snippet(path, text)
+ contents = yield preseeds.read_snippet(path)
+ self.assertEqual(text, contents)
+
+ @inlineCallbacks
+ def test_get_templates_lists_templates(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/template/template-%d' % unique_int
+ yield preseeds.write_template(path, "Text")
+ templates = yield preseeds.get_templates()
+ self.assertIn(path, templates)
+
+ @inlineCallbacks
+ def test_get_snippets_lists_snippets(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/snippets/snippet-%d' % unique_int
+ yield preseeds.write_snippet(path, "Text")
+ snippets = yield preseeds.get_snippets()
+ self.assertEqual([path], snippets)
+
+ @inlineCallbacks
+ def test_templates_do_not_show_up_as_snippets(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/template/template-%d' % unique_int
+ yield preseeds.write_template(path, "Text")
+ snippets = yield preseeds.get_snippets()
+ self.assertEqual([], snippets)
+
+ @inlineCallbacks
+ def test_snippets_do_not_show_up_as_templates(self):
+ preseeds = yield self.make_preseeds_api()
+ unique_int = next(unique_ints)
+ path = '/var/lib/cobbler/snippets/snippet-%d' % unique_int
+ yield preseeds.write_snippet(path, "Text")
+ templates = yield preseeds.get_templates()
+ self.assertEqual([], templates)
Follow ups
