launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #11011
[Merge] lp:~jtv/maas/kill-cobblerclient into lp:maas
Jeroen T. Vermeulen has proposed merging lp:~jtv/maas/kill-cobblerclient into lp:maas.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~jtv/maas/kill-cobblerclient/+merge/119839
I'm stripping Cobbler contact out of the provisioning server A first attempt broke with strange import problems, not to mention a dramatically oversized branch, so I'm trying it in stages this time. Producing a branch that is still dramatically oversized, but not as dramatically as the last time. And the tests pass.
The obvious change is that CobblerSession, the handling of Cobbler XMLRPC errors, and the test fixtures for fake or real cobblers go away. The one interesting detail is that in plugin.py, the “site” service is no longer set up. Basically pserv is no longer a daemon a daemon of its own at this point, as far as I can see.
More to follow.
Jeroen
--
https://code.launchpad.net/~jtv/maas/kill-cobblerclient/+merge/119839
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jtv/maas/kill-cobblerclient into lp:maas.
=== modified file 'src/provisioningserver/api.py'
--- src/provisioningserver/api.py 2012-05-21 04:55:16 +0000
+++ src/provisioningserver/api.py 2012-08-16 06:46:19 +0000
@@ -14,7 +14,6 @@
"ProvisioningAPI",
]
-from base64 import b64encode
from functools import partial
from itertools import (
chain,
@@ -24,11 +23,6 @@
)
from maasserver.utils import map_enum
-from provisioningserver.cobblerclient import (
- CobblerDistro,
- CobblerProfile,
- CobblerSystem,
- )
from provisioningserver.enum import POWER_TYPE
from provisioningserver.interfaces import IProvisioningAPI
from provisioningserver.utils import deferred
@@ -185,22 +179,13 @@
assert isinstance(name, basestring)
assert isinstance(initrd, basestring)
assert isinstance(kernel, basestring)
- distro = yield CobblerDistro.new(
- self.session, name, {
- "initrd": initrd,
- "kernel": kernel,
- })
yield self.sync()
- returnValue(distro.name)
@inlineCallbacks
def add_profile(self, name, distro):
assert isinstance(name, basestring)
assert isinstance(distro, basestring)
- profile = yield CobblerProfile.new(
- self.session, name, {"distro": distro})
yield self.sync()
- returnValue(profile.name)
@inlineCallbacks
def add_node(self, name, hostname, profile, power_type, preseed_data):
@@ -209,43 +194,18 @@
assert isinstance(profile, basestring)
assert power_type in POWER_TYPE_VALUES
assert isinstance(preseed_data, basestring)
- attributes = {
- "hostname": hostname,
- "profile": profile,
- "ks_meta": {"MAAS_PRESEED": b64encode(preseed_data)},
- "power_type": power_type,
- }
- system = yield CobblerSystem.new(self.session, name, attributes)
yield self.sync()
- returnValue(system.name)
@inlineCallbacks
def modify_distros(self, deltas):
- for name, delta in deltas.items():
- yield CobblerDistro(self.session, name).modify(delta)
yield self.sync()
@inlineCallbacks
def modify_profiles(self, deltas):
- for name, delta in deltas.items():
- yield CobblerProfile(self.session, name).modify(delta)
yield self.sync()
@inlineCallbacks
def modify_nodes(self, deltas):
- for name, delta in deltas.items():
- system = CobblerSystem(self.session, name)
- if "mac_addresses" in delta:
- # This needs to be handled carefully.
- mac_addresses = delta.pop("mac_addresses")
- system_state = yield system.get_values()
- hostname = system_state.get("hostname", "")
- interfaces = system_state.get("interfaces", {})
- interface_modifications = gen_cobbler_interface_deltas(
- interfaces, hostname, mac_addresses)
- for interface_modification in interface_modifications:
- yield system.modify(interface_modification)
- yield system.modify(delta)
yield self.sync()
@inlineCallbacks
@@ -268,18 +228,15 @@
@deferred
def get_distros_by_name(self, names):
- d = self.get_objects_by_name(CobblerDistro, names)
- return d.addCallback(cobbler_mapping_to_papi_distros)
+ pass
@deferred
def get_profiles_by_name(self, names):
- d = self.get_objects_by_name(CobblerProfile, names)
- return d.addCallback(cobbler_mapping_to_papi_profiles)
+ pass
@deferred
def get_nodes_by_name(self, names):
- d = self.get_objects_by_name(CobblerSystem, names)
- return d.addCallback(cobbler_mapping_to_papi_nodes)
+ pass
@inlineCallbacks
def delete_objects_by_name(self, object_type, names):
@@ -298,43 +255,32 @@
@deferred
def delete_distros_by_name(self, names):
- return self.delete_objects_by_name(CobblerDistro, names)
+ pass
@deferred
def delete_profiles_by_name(self, names):
- return self.delete_objects_by_name(CobblerProfile, names)
+ pass
@deferred
def delete_nodes_by_name(self, names):
- return self.delete_objects_by_name(CobblerSystem, names)
+ pass
@deferred
def get_distros(self):
- # WARNING: This could return a large number of results. Consider
- # adding filtering options to this function before using it in anger.
- d = CobblerDistro.get_all_values(self.session)
- return d.addCallback(cobbler_mapping_to_papi_distros)
+ pass
@deferred
def get_profiles(self):
- # WARNING: This could return a large number of results. Consider
- # adding filtering options to this function before using it in anger.
- d = CobblerProfile.get_all_values(self.session)
- return d.addCallback(cobbler_mapping_to_papi_profiles)
+ pass
@deferred
def get_nodes(self):
- # WARNING: This could return a *huge* number of results. Consider
- # adding filtering options to this function before using it in anger.
- d = CobblerSystem.get_all_values(self.session)
- return d.addCallback(cobbler_mapping_to_papi_nodes)
+ pass
@deferred
def start_nodes(self, names):
- d = CobblerSystem.powerOnMultiple(self.session, names)
- return d
+ pass
@deferred
def stop_nodes(self, names):
- d = CobblerSystem.powerOffMultiple(self.session, names)
- return d
+ pass
=== removed file 'src/provisioningserver/cobblercatcher.py'
--- src/provisioningserver/cobblercatcher.py 2012-04-16 10:00:51 +0000
+++ src/provisioningserver/cobblercatcher.py 1970-01-01 00:00:00 +0000
@@ -1,98 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Helping hands for dealing with Cobbler exceptions."""
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-__metaclass__ = type
-__all__ = [
- 'convert_cobbler_exception',
- 'ProvisioningError',
- ]
-
-import re
-from xmlrpclib import Fault
-
-from provisioningserver.enum import PSERV_FAULT
-
-
-class ProvisioningError(Fault):
- """Fault, as passed from provisioning server to maasserver.
-
- This acts as a marker class within the provisioning server. As far as
- maasserver is concerned, when it sees an exception of this kind, it will
- just be a Fault with a more telling faultCode.
-
- The faultCode is one of :class:`PSERV_FAULT`.
- """
-
-
-def extract_text(fault_string):
- """Get the actual error message out of a Cobbler fault string.
-
- This removes exception type information that may have been added on
- the Cobbler side.
-
- :param fault_string: A `Fault.faultString` as found on an exception
- raised while working with Cobbler.
- :type fault_string: basestring
- :return: Extracted error message.
- :rtype: basestring
- """
- match = re.match(
- "<class 'cobbler\.cexceptions\.CX'>:'(.*)'", fault_string)
- if match is None:
- return fault_string
- else:
- return match.groups(0)[0]
-
-
-def divine_fault_code(err_str):
- """Parse error string to figure out what kind of error it is.
-
- :param err_str: An error string, as extracted from a `Fault` by
- `extract_text`.
- :type err_str: basestring
- :return: A fault code from :class:`PSERV_FAULT`, for use as a
- `Fault.faultCode`.
- :rtype: int
- """
- prefixes = [
- ("login failed", PSERV_FAULT.COBBLER_AUTH_FAILED),
- ("invalid token:", PSERV_FAULT.COBBLER_AUTH_ERROR),
- ("invalid profile name", PSERV_FAULT.NO_SUCH_PROFILE),
- ("", PSERV_FAULT.GENERIC_COBBLER_ERROR),
- ]
- for prefix, code in prefixes:
- if err_str.startswith(prefix):
- return code
- assert False, "No prefix matched fault string '%s'." % err_str
-
-
-def convert_cobbler_exception(fault):
- """Convert a :class:`Fault` from Cobbler to a :class:`ProvisioningError`.
-
- :param fault: The original exception, as raised by code that tried to
- talk to Cobbler.
- :type fault: Fault
- :return: A more descriptive exception, for consumption by maasserver.
- :rtype: :class:`ProvisioningError`
- """
- assert isinstance(fault, Fault)
-
- if isinstance(fault, ProvisioningError):
- raise AssertionError(
- "Fault %r went through double conversion." % fault)
-
- err_str = extract_text(fault.faultString)
- if fault.faultCode != 1:
- fault_code = PSERV_FAULT.NO_COBBLER
- else:
- fault_code = divine_fault_code(err_str)
-
- return ProvisioningError(faultCode=fault_code, faultString=err_str)
=== removed file 'src/provisioningserver/cobblerclient.py'
--- src/provisioningserver/cobblerclient.py 2012-04-16 10:00:51 +0000
+++ src/provisioningserver/cobblerclient.py 1970-01-01 00:00:00 +0000
@@ -1,878 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-"""Wrapper for the Cobbler XMLRPC API, using Twisted.
-
-The API looks synchronous, but under the covers, calls yield to the Twisted
-reactor so that it can service other callbacks.
-
-To use Cobbler, create a `CobblerSession` (which connects and authenticates
-as appropriate). Create, query, or manipulate cobbler objects through the
-various classes derived from `CobblerObject`: `CobblerDistro` for a distro,
-`CobblerImage` for an image, `CobblerProfile` for a profile, `CobblerSystem`
-for a node system, and so on. In addition, `CobblerPreseeds` manages
-preseeds.
-"""
-
-__metaclass__ = type
-__all__ = [
- 'CobblerDistro',
- 'CobblerImage',
- 'CobblerPreseeds',
- 'CobblerProfile',
- 'CobblerRepo',
- 'CobblerSystem',
- 'DEFAULT_TIMEOUT',
- ]
-
-from functools import partial
-from urlparse import urlparse
-import xmlrpclib
-
-from provisioningserver.cobblercatcher import (
- convert_cobbler_exception,
- ProvisioningError,
- )
-from provisioningserver.enum import PSERV_FAULT
-from twisted.internet import reactor as default_reactor
-from twisted.internet.defer import (
- DeferredLock,
- inlineCallbacks,
- returnValue,
- )
-from twisted.internet.error import DNSLookupError
-from twisted.python.log import msg
-from twisted.web.xmlrpc import Proxy
-
-# Default timeout in seconds for xmlrpc requests to cobbler.
-DEFAULT_TIMEOUT = 30
-
-
-def tilde_to_None(data):
- """Repair Cobbler's XML-RPC response.
-
- Cobbler has an annoying function, `cobbler.utils.strip_none`, that is
- applied to every data structure that it sends back through its XML-RPC API
- service. It "encodes" `None` as `"~"`, and does so recursively in `list`s
- and `dict`s. It also forces all dictionary keys to be `str`, so `None`
- keys become `"None"`.
-
- This function attempts to repair this damage. Sadly, it may get things
- wrong - it will "repair" genuine tildes to `None` - but it's likely to be
- more correct than doing nothing - and having tildes everwhere.
-
- This also does not attempt to repair `"None"` dictionary keys.
- """
- if data == "~":
- return None
- elif isinstance(data, list):
- return [tilde_to_None(value) for value in data]
- elif isinstance(data, dict):
- return {key: tilde_to_None(value) for key, value in data.items()}
- else:
- return data
-
-
-class CobblerXMLRPCProxy(Proxy):
- """An XML-RPC proxy that attempts to repair Cobbler's broken responses.
-
- See `tilde_to_None` for an explanation.
- """
-
- def callRemote(self, method, *args):
- """See `Proxy.callRemote`.
-
- Uses `tilde_to_None` to repair the response.
- """
- d = Proxy.callRemote(self, method, *args)
- d.addCallback(tilde_to_None)
- return d
-
-
-def looks_like_object_not_found(exception):
- """Does `exception` look like an unknown object failure?"""
- if not hasattr(exception, 'faultString'):
- # An unknown object failure would come as an xmlrpclib.Fault.
- return False
- else:
- return "internal error, unknown " in exception.faultString
-
-
-class CobblerSession:
- """A session on the Cobbler XMLRPC API.
-
- The session can be used for many asynchronous requests, all of them
- sharing a single authentication token.
-
- If you're just using Cobbler's services, treat this class as a
- "connection" class: create it with the right service url, user name,
- and password, and pass it around for the various other Cobbler* classes
- to use. After creation it's a black box except to the Cobbler client
- code.
- """
-
- # In an arguments list, this means "insert security token here."
- token_placeholder = object()
-
- def __init__(self, url, user, password):
- self.url = url
- self.user = user
- self.password = password
- self.proxy = self._make_twisted_proxy()
- self.token = None
- self.connection_count = 0
- self.authentication_lock = DeferredLock()
-
- def _make_twisted_proxy(self):
- """Create a Twisted XMRLPC proxy.
-
- For internal use only, but overridable for test purposes.
- """
- # Twisted does not encode the URL, and breaks with "Data must
- # not be unicode" if it's in Unicode. We'll have to decode it
- # here, and hope it doesn't lead to breakage in Twisted. We'll
- # figure out what to do about non-ASCII characters in URLs
- # later.
- return CobblerXMLRPCProxy(self.url.encode('ascii'))
-
- def record_state(self):
- """Return a cookie representing the session's current state.
-
- The cookie will change whenever the session is reconnected or
- re-authenticated. The only valid use of this cookie is to compare
- it for equality with another one.
-
- If two calls return different cookies, that means that the session
- has broken in some way and been re-established between the two calls.
- """
- return (self.connection_count, self.token)
-
- @inlineCallbacks
- def _authenticate(self, previous_state=None):
- """Log in asynchronously.
-
- 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.
-
- :param previous_state: The state of the session as recorded by
- `record_state` before the failed request was issued. If the
- session has had to reconnect or re-authenticate since then, the
- method will assume that a concurrent authentication request has
- completed and the failed request can be retried without logging
- in again.
- If no `previous_state` is given, authentication will happen
- regardless.
- :return: A `Deferred`.
- """
- if previous_state is None:
- previous_state = self.record_state()
-
- yield self.authentication_lock.acquire()
- try:
- if self.record_state() == previous_state:
- # If we're here, nobody else is authenticating this
- # session. Clear the stale token as a hint to
- # subsequent calls on the session. If they see that the
- # session is unauthenticated they won't issue and fail,
- # but rather block for this authentication attempt to
- # complete.
- self.token = None
-
- # Now initiate our new authentication.
- self.token = yield self._issue_call(
- 'login', self.user, self.password)
- finally:
- self.authentication_lock.release()
-
- def _substitute_token(self, arg):
- """Return `arg`, or the current auth token for `token_placeholder`."""
- if arg is self.token_placeholder:
- return self.token
- else:
- return arg
-
- def _with_timeout(self, d, timeout=DEFAULT_TIMEOUT, reactor=None):
- """Wrap the xmlrpc call that returns "d" so that it is cancelled if
- it exceeds a timeout.
-
- :param d: The Deferred to cancel.
- :param timeout: timeout in seconds, defaults to 30.
- :param reactor: override the default reactor, useful for testing.
- """
- if reactor is None:
- reactor = default_reactor
- delayed_call = reactor.callLater(timeout, d.cancel)
-
- def cancel_timeout(passthrough):
- if not delayed_call.called:
- delayed_call.cancel()
- return passthrough
- return d.addBoth(cancel_timeout)
-
- def _log_call(self, method, args):
- """Log the call.
-
- If the authentication token placeholder is included in `args`, the
- string ``<token>`` will be printed in its place; the token itself will
- not be included.
-
- :param method: The name of the method.
- :param args: A sequence of arguments.
- """
- args_repr = ", ".join(
- "<token>" if arg is self.token_placeholder else repr(arg)
- for arg in args)
- message = "%s(%s)" % (method, args_repr)
- msg(message, system=__name__)
-
- @inlineCallbacks
- def _issue_call(self, method, *args):
- """Initiate call to XMLRPC method.
-
- :param method: Name of XMLRPC method to invoke.
- :param *args: Arguments for the call. If any of them is
- `token_placeholder`, the current security token will be
- substituted in its place.
- :return: `Deferred`.
- """
- # Twisted XMLRPC does not encode the method name, but breaks if
- # we give it in Unicode. Encode it here; thankfully we're
- # dealing with ASCII only in method names.
- method = method.encode('ascii')
- self._log_call(method, args)
- args = map(self._substitute_token, args)
- try:
- result = yield self._with_timeout(
- self.proxy.callRemote(method, *args))
- except xmlrpclib.Fault as e:
- raise convert_cobbler_exception(e)
- except DNSLookupError as e:
- hostname = urlparse(self.url).hostname
- raise ProvisioningError(
- faultCode=PSERV_FAULT.COBBLER_DNS_LOOKUP_ERROR,
- faultString=hostname)
- returnValue(result)
-
- @inlineCallbacks
- def call(self, method, *args):
- """Initiate call to XMLRPC `method` by name, through Twisted.
-
- This is for use by the Cobbler wrapper API only. Don't call it
- directly; instead, ensure that the wrapper API supports the
- method you want to call.
-
- Initiates XMLRPC call, yields back to the reactor until it's ready
- with a response, then returns the response. Use this as if it were
- a synchronous XMLRPC call; but be aware that it lets the reactor run
- other code in the meantime.
-
- :param method: Name of XMLRPC method to call.
- :param *args: Positional arguments for the XMLRPC call.
- :return: A `Deferred` representing the call.
- """
- original_state = self.record_state()
- 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 e.faultCode == PSERV_FAULT.COBBLER_AUTH_ERROR:
- need_auth = True
- else:
- raise
-
- if need_auth:
- # We weren't authenticated when we started, but we should be
- # now. Make the final attempt.
- yield self._authenticate(original_state)
- result = yield self._issue_call(method, *args)
- returnValue(result)
-
-
-class CobblerObjectType(type):
- """Metaclass of Cobbler objects."""
-
- def __new__(mtype, name, bases, attributes):
- """Build a new `CobblerObject` class.
-
- Ensure that `known_attributes` and `required_attributes` are both
- frozensets. This indicates that they should not be modified at
- run-time, and it also improves performance of several methods, most
- notably `_trim_attributes`.
- """
- if "known_attributes" in attributes:
- attributes["known_attributes"] = frozenset(
- attributes["known_attributes"])
- if "required_attributes" in attributes:
- attributes["required_attributes"] = frozenset(
- attributes["required_attributes"])
- if "modification_attributes" in attributes:
- attributes["modification_attributes"] = frozenset(
- attributes["modification_attributes"])
- return super(CobblerObjectType, mtype).__new__(
- mtype, name, bases, attributes)
-
-
-class CobblerObject:
- """Abstract base class: a type of object in Cobbler's XMLRPC API.
-
- This defines the common interface to cobbler's distros, profiles,
- systems, and other objects it stores in its database and exposes
- through its API. Implement a new type by inheriting from this class.
- """
-
- __metaclass__ = CobblerObjectType
-
- # What are objects of this type called in the Cobbler API?
- object_type = None
-
- # What's the plural of object_type, if not object_type + "s"?
- object_type_plural = None
-
- # 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.
- # Some attributes in Cobbler uses dashes as separators, others use
- # underscores. In MAAS, use only underscores.
- known_attributes = []
-
- # What attributes does Cobbler require for this type of object?
- required_attributes = []
-
- # What attributes do we expect to support for modifications to this
- # object? Cobbler's API accepts some pseudo-attributes, only valid for
- # making modifications.
- modification_attributes = []
-
- def __init__(self, session, name):
- """Reference an object in Cobbler.
-
- :param session: A `CobblerSession`.
- :param name: Name for this object.
- """
- self.session = session
- self.name = name
-
- def _get_handle(self):
- """Retrieve the object's handle."""
- 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):
- """Interpolate object_type into a method name template.
-
- For example, on `CobblerSystem`, "get_%s_handle" would be
- interpolated into "get_system_handle" and "get_%s" with plural=True
- becomes "get_systems".
- """
- if plural:
- type_name = (cls.object_type_plural or '%ss' % cls.object_type)
- else:
- type_name = cls.object_type
- return name_template % type_name
-
- @classmethod
- def _normalize_attribute(cls, attribute_name, attributes=None):
- """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` by default, but `attributes` can be passed to
- override that.
-
- :param attribute_name: An attribute name, possibly using underscores
- where Cobbler expects dashes.
- :param attributes: None, or a set of attributes against which to
- match.
- :return: A Cobbler-style attribute name, using either dashes or
- underscores as used by Cobbler.
- """
- if attributes is None:
- attributes = cls.known_attributes
-
- if attribute_name in attributes:
- # Already spelled the way Cobbler likes it.
- return attribute_name
-
- attribute_name = attribute_name.replace('_', '-')
- if attribute_name in attributes:
- # Cobbler wants this one with dashes.
- return attribute_name
-
- attribute_name = attribute_name.replace('-', '_')
- if attribute_name not in attributes:
- raise AssertionError(
- "Unknown attribute for %s: %s." % (
- cls.object_type, attribute_name))
- return attribute_name
-
- @classmethod
- @inlineCallbacks
- def find(cls, session, **kwargs):
- """Find objects in the database.
-
- :param session: The `CobblerSession` to operate in. No authentication
- is required.
- :param **kwargs: Optional search criteria, e.g.
- hostname="*.maas3.example.com" to limit the search to items with
- a hostname attribute that ends in ".maas3.example.com".
- :return: A list of matching `cls` objects.
- """
- 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)
- returnValue([cls(session, name) for name in result])
-
- @classmethod
- def _trim_attributes(cls, attributes):
- """Return a dict containing only keys from `known_attributes`.
-
- If `attributes` is `None` - which can happen when querying a
- non-existent object - this returns `None`.
- """
- if attributes is None:
- return None
- else:
- return {
- name: value
- for name, value in attributes.items()
- if name in cls.known_attributes
- }
-
- @classmethod
- @inlineCallbacks
- def get_all_values(cls, session):
- """Load the attributes for all objects of this type.
-
- :return: A `Deferred` that delivers a dict, mapping objects' names
- to dicts containing their respective attributes.
- """
- method = cls._name_method("get_%s", plural=True)
- results = yield session.call(method)
- results = (cls._trim_attributes(result) for result in results)
- returnValue(dict((obj['name'], obj) for obj in results))
-
- def get_values(self):
- """Load the object's attributes as a dict.
-
- :return: A `Deferred` that delivers a dict containing the object's
- attribute names and values.
- """
- d = self.session.call(self._name_method("get_%s"), self.name)
- d.addCallback(self._trim_attributes)
- return d
-
- @classmethod
- @inlineCallbacks
- 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.
- """
- if 'name' in attributes:
- assert attributes['name'] == name, (
- "Creating %s called '%s', but 'name' attribute is '%s'."
- % (cls.object_type, name, attributes['name']))
- else:
- attributes['name'] = name
- missing_attributes = (
- set(cls.required_attributes).difference(attributes))
- assert len(missing_attributes) == 0, (
- "Required attributes for %s missing: %s"
- % (cls.object_type, missing_attributes))
-
- args = dict(
- (cls._normalize_attribute(key), value)
- for key, value in attributes.items())
-
- # Do not clobber under any circumstances.
- if "clobber" in args:
- del args["clobber"]
-
- try:
- # Attempt to edit an existing object.
- success = yield session.call(
- 'xapi_object_edit', cls.object_type, name, 'edit', args,
- session.token_placeholder)
- except xmlrpclib.Fault as e:
- # If it was not found, add the object.
- if looks_like_object_not_found(e):
- success = yield session.call(
- 'xapi_object_edit', cls.object_type, name, 'add', args,
- session.token_placeholder)
- else:
- raise
-
- if not success:
- raise RuntimeError(
- "Cobbler refused to create %s '%s'. Attributes: %s"
- % (cls.object_type, name, args))
- returnValue(cls(session, name))
-
- @inlineCallbacks
- def modify(self, delta):
- """Modify an object in Cobbler.
-
- :param name: Identifying name for the existing object.
- :param attributes: Dict mapping attribute names to values.
- """
- normalize = partial(
- self._normalize_attribute,
- attributes=self.modification_attributes)
- args = {
- normalize(key): value
- for key, value in delta.items()
- }
- success = yield self.session.call(
- 'xapi_object_edit', self.object_type, self.name, 'edit', args,
- self.session.token_placeholder)
- if not success:
- raise RuntimeError(
- "Cobbler refused to modify %s '%s'. Attributes: %s"
- % (self.object_type, self.name, args))
-
- @inlineCallbacks
- def delete(self, recurse=True):
- """Delete this object. Its name must be known.
-
- :param recurse: Delete dependent objects recursively?
- """
- assert self.name is not None, (
- "Can't delete %s; don't know its name." % self.object_type)
- method = self._name_method('remove_%s')
- yield self.session.call(
- method, self.name, self.session.token_placeholder, recurse)
-
-
-class CobblerProfile(CobblerObject):
- """A profile.
-
- See `CobblerObject` for common object-management properties.
- """
- 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',
- ]
- required_attributes = [
- 'name',
- 'distro',
- ]
- modification_attributes = [
- 'distro',
- ]
-
-
-class CobblerImage(CobblerObject):
- """An operating system image.
-
- See `CobblerObject` for common object-management properties.
- """
- 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',
- ]
-
-
-class CobblerDistro(CobblerObject):
- """A distribution.
-
- See `CobblerObject` for common object-management properties.
- """
- 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',
- ]
- required_attributes = [
- 'initrd',
- 'kernel',
- 'name',
- ]
- modification_attributes = [
- 'initrd',
- 'kernel',
- ]
-
-
-class CobblerRepo(CobblerObject):
- """A repository.
-
- See `CobblerObject` for common object-management properties.
- """
- object_type = 'repo'
- known_attributes = [
- 'arch',
- 'comment',
- 'createrepo_flags',
- 'environment',
- 'keep_updated',
- 'mirror',
- 'mirror_locally',
- 'name',
- 'owners',
- 'priority',
- ]
- required_attributes = [
- 'name',
- 'mirror',
- ]
-
-
-class CobblerSystem(CobblerObject):
- """A computer (node) on the network.
-
- See `CobblerObject` for common object-management properties.
- """
- object_type = 'system'
- known_attributes = [
- 'boot_files',
- 'comment',
- 'fetchable_files',
- 'gateway',
- # FQDN:
- 'hostname',
- 'interfaces',
- # Space-separated key=value pairs:
- 'kernel_options'
- 'kickstart',
- 'kopts',
- 'kopts_post',
- # Space-separated key=value pairs for preseed:
- 'ks_meta',
- 'mgmt_classes',
- # 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',
- ]
- required_attributes = [
- 'name',
- 'profile',
- ]
- modification_attributes = [
- 'hostname',
- 'delete_interface',
- 'interface', # Required for mac_address and delete_interface.
- 'mac_address',
- 'profile',
- 'dns_name',
- 'netboot_enabled',
- ]
-
- @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."""
- handle = yield self._get_handle()
- yield self.session.call(
- 'power_system', handle, operation,
- self.session.token_placeholder)
-
- def powerOn(self):
- """Turn system on.
-
- :return: Deferred.
- """
- return self._callPower('on')
-
- def powerOff(self):
- """Turn system on.
-
- :return: Deferred.
- """
- return self._callPower('off')
-
- def reboot(self):
- """Turn system on.
-
- :return: Deferred.
- """
- return self._callPower('reboot')
-
-
-class CobblerPreseeds:
- """Manage preseeds."""
-
- def __init__(self, session):
- self.session = session
-
- 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_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 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.
-
- :return: Deferred.
- """
- return self.session.call('sync', self.session.token_placeholder)
=== modified file 'src/provisioningserver/plugin.py'
--- src/provisioningserver/plugin.py 2012-07-18 16:26:29 +0000
+++ src/provisioningserver/plugin.py 2012-08-16 06:46:19 +0000
@@ -13,7 +13,6 @@
__all__ = []
from provisioningserver.amqpclient import AMQFactory
-from provisioningserver.cobblerclient import CobblerSession
from provisioningserver.config import Config
from provisioningserver.remote import ProvisioningAPI_XMLRPC
from provisioningserver.services import (
@@ -125,12 +124,6 @@
oops_reporter = oops_config["reporter"]
return OOPSService(log_service, oops_dir, oops_reporter)
- def _makeCobblerSession(self, cobbler_config):
- """Create a :class:`CobblerSession`."""
- return CobblerSession(
- cobbler_config["url"], cobbler_config["username"],
- cobbler_config["password"])
-
def _makeProvisioningAPI(self, cobbler_session, config):
"""Construct an :class:`IResource` for the Provisioning API."""
papi_xmlrpc = ProvisioningAPI_XMLRPC(cobbler_session)
@@ -199,14 +192,6 @@
client_service = self._makeBroker(broker_config)
client_service.setServiceParent(services)
- cobbler_config = config["cobbler"]
- cobbler_session = self._makeCobblerSession(cobbler_config)
-
- papi_root = self._makeProvisioningAPI(cobbler_session, config)
-
- site_service = self._makeSiteService(papi_root, config)
- site_service.setServiceParent(services)
-
tftp_service = self._makeTFTPService(config["tftp"])
tftp_service.setServiceParent(services)
=== modified file 'src/provisioningserver/testing/factory.py'
--- src/provisioningserver/testing/factory.py 2012-04-20 14:36:16 +0000
+++ src/provisioningserver/testing/factory.py 2012-08-16 06:46:19 +0000
@@ -21,10 +21,6 @@
from time import time
from xmlrpclib import Fault
-from provisioningserver.cobblerclient import (
- CobblerDistro,
- CobblerProfile,
- )
from provisioningserver.enum import POWER_TYPE
from twisted.internet.defer import (
inlineCallbacks,
@@ -184,12 +180,6 @@
attributes, 'kernel', object_class.required_attributes),
self.default_to_file(
attributes, 'initrd', object_class.required_attributes),
- yield self.default_to_object(
- attributes, 'profile', object_class.required_attributes, session,
- CobblerProfile)
- yield self.default_to_object(
- attributes, 'distro', object_class.required_attributes, session,
- CobblerDistro)
for attr in object_class.required_attributes:
if attr not in attributes:
attributes[attr] = '%s-%d' % (attr, unique_int)
=== removed file 'src/provisioningserver/testing/fakecobbler.py'
--- src/provisioningserver/testing/fakecobbler.py 2012-04-20 09:58:23 +0000
+++ src/provisioningserver/testing/fakecobbler.py 1970-01-01 00:00:00 +0000
@@ -1,594 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-"""Fake, synchronous Cobbler XMLRPC service for testing."""
-
-__metaclass__ = type
-__all__ = [
- 'fake_auth_failure_string',
- 'fake_object_not_found_string',
- 'fake_token',
- 'FakeCobbler',
- 'FakeTwistedProxy',
- 'log_in_to_fake_cobbler',
- 'make_fake_cobbler_session',
- ]
-
-from copy import deepcopy
-from fnmatch import fnmatch
-from itertools import count
-from random import randint
-from xmlrpclib import (
- dumps,
- Fault,
- )
-
-from provisioningserver.cobblerclient import CobblerSession
-from twisted.internet.defer import (
- inlineCallbacks,
- returnValue,
- )
-
-
-unique_ints = count(randint(0, 99999))
-
-
-def fake_auth_failure_string(token):
- """Fake a Cobbler authentication failure fault string for `token`."""
- return "<class 'cobbler.cexceptions.CX'>:'invalid token: %s'" % token
-
-
-def fake_object_not_found_string(object_type, object_name):
- """Fake a Cobbler unknown object fault string."""
- return (
- "<class 'cobbler.cexceptions.CX'>:'internal error, "
- "unknown %s name %s'" % (object_type, object_name))
-
-
-def fake_token(user=None, custom_id=None):
- """Make up a fake auth token.
-
- :param user: Optional user name to embed in the token id.
- :param custom_id: Optional custom id element to embed in the token id,
- for ease of debugging.
- """
- elements = ['token', '%s' % next(unique_ints), user, custom_id]
- return '-'.join(element for element in elements if bool(element))
-
-
-class FakeTwistedProxy:
- """Fake Twisted XMLRPC proxy that forwards calls to a `FakeCobbler`."""
-
- def __init__(self, fake_cobbler=None):
- if fake_cobbler is None:
- fake_cobbler = FakeCobbler()
- self.fake_cobbler = fake_cobbler
-
- @inlineCallbacks
- def callRemote(self, method, *args):
- # Dump the call information as an XML-RPC request to ensure that it
- # will travel over the wire. Cobbler does not allow None so we forbid
- # it here too.
- dumps(args, method, allow_none=False)
- # Continue to forward the call to fake_cobbler.
- callee = getattr(self.fake_cobbler, method, None)
- assert callee is not None, "Unknown Cobbler method: %s" % method
- result = yield callee(*args)
- returnValue(result)
-
-
-class FakeCobbler:
- """Fake implementation of the Cobbler XMLRPC API.
-
- This does nothing useful, but tries to be internally consistent 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:
- self.passwords = passwords
-
- self.tokens = {}
-
- # 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, fake_auth_failure_string(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,
- 'comment': (
- "Cobbler stores lots of things we're not interested in; "
- "this comment is here to break tests that let Cobbler's "
- "data leak out of the Provisioning Server."
- ),
- }
- 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, fake_object_not_found_string(object_type, name))
- return handle
-
- def _api_find_objects(self, object_type, criteria):
- """Find names of objects in the saved store that match `criteria`.
-
- :return: A list of object names.
- """
- # Assumption: these operations look only at saved objects.
- location = self.store[None].get(object_type, {})
- return [
- candidate['name']
- for candidate in location.values()
- if self._matches(candidate, criteria)]
-
- def _api_get_object(self, object_type, name):
- """Get object's attributes by name."""
- location = self.store[None].get(object_type, {})
- matches = [obj for obj in location.values() if obj['name'] == name]
- assert len(matches) <= 1, (
- "Multiple %s objects are called '%s'." % (object_type, name))
- if len(matches) == 0:
- return None
- else:
- return deepcopy(matches[0])
-
- def _api_get_objects(self, object_type):
- """Return all saved objects of type `object_type`.
-
- :return: A list of object dicts. The dicts are copied from the
- saved store; they are not references to the originals in the
- store.
- """
- # Assumption: these operations look only at saved objects.
- location = self.store[None].get(object_type, {})
- return [deepcopy(obj) for obj in location.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 = deepcopy(saved_obj)
- self._add_object_to_session(
- token, object_type, handle, session_obj)
-
- session_obj[key] = value
-
- def _validate(self, token, object_type, attributes):
- """Check object for validity. Raise :class:`Fault` if it's bad."""
- # Add more checks here as needed for testing realism.
- if object_type == 'system':
- profile_name = attributes.get('profile')
- profile = self._api_get_object('profile', profile_name)
- if profile is None:
- raise Fault(1, "invalid profile name: '%s'" % profile_name)
-
- 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
-
- self._validate(token, object_type, session_obj)
-
- 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 Fault(1, "login failed (%s)" % user)
- token = fake_token(user)
- self.tokens[token] = user
- return token
-
- def _xapi_edit_system_interfaces(self, token, handle, name, attrs):
- """Edit system interfaces with Cobbler's crazy protocol."""
- obj_state = self._api_get_object('system', name)
- interface_name = attrs.pop("interface")
- interfaces = obj_state.get("interfaces", {})
- if "mac_address" in attrs:
- interface = interfaces.setdefault(interface_name, {})
- interface["mac_address"] = attrs.pop("mac_address")
- elif "dns_name" in attrs:
- interface = interfaces.setdefault(interface_name, {})
- interface["dns_name"] = attrs.pop("dns_name")
- elif "delete_interface" in attrs:
- if interface_name in interfaces:
- del interfaces[interface_name]
- else:
- raise AssertionError(
- "Edit operation defined interface but not "
- "mac_address, dns_name, or delete_interface. "
- "Got: %r" % (attrs,))
- self._api_modify_object(
- token, 'system', handle, "interfaces", interfaces)
-
- def xapi_object_edit(self, object_type, name, operation, attrs, token):
- """Swiss-Army-Knife API: create/rename/copy/edit object."""
- self._check_token(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
- return self._api_save_object(token, object_type, handle)
- elif operation == 'edit':
- handle = self._api_get_handle(token, object_type, name)
- if object_type == "system" and "interface" in attrs:
- self._xapi_edit_system_interfaces(token, handle, name, attrs)
- for key, value in attrs.items():
- self._api_modify_object(token, object_type, handle, key, value)
- return self._api_save_object(token, object_type, handle)
- else:
- raise NotImplemented(
- "xapi_object_edit(%s, ..., %s, ...)"
- % (object_type, operation))
-
- def new_distro(self, token):
- return self._api_new_object(token, 'distro')
-
- def remove_distro(self, name, token, recurse=True):
- self._api_remove_object(token, 'distro', name)
-
- def get_distro_handle(self, name, token):
- return self._api_get_handle(token, 'distro', name)
-
- def find_distro(self, criteria):
- return self._api_find_objects('distro', criteria)
-
- def get_distro(self, name):
- return self._api_get_object('distro', name)
-
- def get_distros(self):
- return self._api_get_objects('distro')
-
- def modify_distro(self, handle, key, value, token):
- self._api_modify_object(token, 'distro', handle, key, value)
-
- def save_distro(self, handle, token):
- return self._api_save_object(token, 'distro', handle)
-
- def new_image(self, token):
- return self._api_new_object(token, 'image')
-
- def remove_image(self, name, token, recurse=True):
- self._api_remove_object(token, 'image', name)
-
- def get_image_handle(self, name, token):
- return self._api_get_handle(token, 'image', name)
-
- def find_image(self, criteria):
- return self._api_find_objects('image', criteria)
-
- def get_image(self, name):
- return self._api_get_object('image', name)
-
- def get_images(self):
- return self._api_get_objects('image')
-
- def modify_image(self, handle, key, value, token):
- self._api_modify_object(token, 'image', handle, key, value)
-
- def save_image(self, handle, token):
- return self._api_save_object(token, 'image', handle)
-
- def new_profile(self, token):
- return self._api_new_object(token, 'profile')
-
- def remove_profile(self, name, token, recurse=True):
- self._api_remove_object(token, 'profile', name)
-
- def get_profile_handle(self, name, token):
- return self._api_get_handle(token, 'profile', name)
-
- def find_profile(self, criteria):
- return self._api_find_objects('profile', criteria)
-
- def get_profile(self, name):
- return self._api_get_object('profile', name)
-
- def get_profiles(self):
- return self._api_get_objects('profile')
-
- def modify_profile(self, handle, key, value, token):
- self._api_modify_object(token, 'profile', handle, key, value)
-
- def save_profile(self, handle, token):
- 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_repo(self, name):
- return self._api_get_object('repo', name)
-
- 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):
- return self._api_new_object(token, 'system')
-
- def remove_system(self, name, token, recurse=True):
- self._api_remove_object(token, 'system', name)
-
- def background_power_system(self, args, token):
- """Asynchronous power on/off/reboot. No notification."""
- self._check_token(token)
- operation = args['power']
- # This version takes system names. The regular power_system
- # takes a system handle.
- 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):
- return self._api_get_handle(token, 'system', name)
-
- def find_system(self, criteria):
- return self._api_find_objects('system', criteria)
-
- def get_system(self, name):
- return self._api_get_object('system', name)
-
- def get_systems(self):
- return self._api_get_objects('system')
-
- def modify_system(self, handle, key, value, token):
- self._api_modify_object(token, 'system', handle, key, value)
-
- def save_system(self, handle, token):
- 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):
- return self._api_access_preseed(
- token, read, self.preseed_templates, path, contents)
-
- def get_kickstart_templates(self, token=None):
- return list(self.preseed_templates)
-
- 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 list(self.preseed_snippets)
-
- def sync(self, token):
- self._check_token(token)
-
-
-unique_ints = count(randint(0, 9999))
-
-
-class FakeCobblerSession(CobblerSession):
- """A `CobblerSession` instrumented not to use real XMLRPC."""
-
- def __init__(self, url, user, password, fake_cobbler=None):
- self.fake_proxy = FakeTwistedProxy(fake_cobbler=fake_cobbler)
- super(FakeCobblerSession, self).__init__(url, user, password)
-
- def _make_twisted_proxy(self):
- """Override for CobblerSession's proxy factory."""
- return self.fake_proxy
-
-
-def make_fake_cobbler_session(url=None, user=None, password=None,
- fake_cobbler=None):
- """Return a :class:`CobblerSession` wired up to a :class:`FakeCobbler`."""
- unique_number = next(unique_ints)
- if url is None:
- url = "http://localhost/url-%d/"; % unique_number
- if user is None:
- user = "user%s" % unique_number
- if password is None:
- password = "password%d" % unique_number
- if fake_cobbler is None:
- fake_cobbler = FakeCobbler(passwords={user: password})
-
- return FakeCobblerSession(url, user, password, fake_cobbler=fake_cobbler)
-
-
-@inlineCallbacks
-def log_in_to_fake_cobbler(*args, **kwargs):
- """Set up a fake Cobbler session, and log in."""
- session = make_fake_cobbler_session(*args, **kwargs)
- yield session._authenticate()
- returnValue(session)
=== removed file 'src/provisioningserver/testing/realcobbler.py'
--- src/provisioningserver/testing/realcobbler.py 2012-06-25 13:59:23 +0000
+++ src/provisioningserver/testing/realcobbler.py 1970-01-01 00:00:00 +0000
@@ -1,105 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Set up test sessions against real Cobblers."""
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-__metaclass__ = type
-__all__ = [
- 'RealCobbler',
- ]
-
-from os import environ
-from textwrap import dedent
-from unittest import skipIf
-from urlparse import urlparse
-
-from provisioningserver.cobblerclient import CobblerSession
-
-
-class RealCobbler:
- """Factory for test sessions on a real Cobbler instance, if available.
-
- To enable tests, set the PSERV_TEST_COBBLER_URL environment variable to
- point to the real Cobbler instance you want to test against. The URL
- should include username and password if required.
-
- Warning: this will mess with your Cobbler database. Never do this with
- a production machine.
- """
-
- env_var = 'PSERV_TEST_COBBLER_URL'
-
- help_text_available = dedent("""\
- Set %s to the URL for a Cobbler instance to test against,
- e.g. http://username:password@xxxxxxxxxxx/cobbler_api.
- WARNING: this will modify your Cobbler database.
- """ % env_var)
-
- help_text_local = dedent("""\
- Set %s to the URL for a *local* Cobbler instance to test against,
- e.g. http://username:password@localhost/cobbler_api.
- WARNING: this will modify your Cobbler database.
- """ % env_var)
-
- def __init__(self):
- self.url = environ.get(self.env_var)
- if self.url is not None:
- urlparts = urlparse(self.url)
- self.username = urlparts.username or 'cobbler'
- self.password = urlparts.password or ''
-
- @property
- def is_available(self):
- """Is a real Cobbler available for tests?"""
- return self.url is not None
-
- @property
- def skip_unless_available(self):
- """Decorator to disable tests if no real Cobbler is available.
-
- Annotate tests like so::
-
- @real_cobbler.skip_unless_available
- def test_something_that_requires_a_real_cobbler(self):
- ...
-
- """
- return skipIf(not self.is_available, self.help_text_available)
-
- @property
- def is_local(self):
- """Is a real Cobbler installed locally available for tests?"""
- if self.is_available:
- hostname = urlparse(self.url).hostname
- return hostname == "localhost" or hostname.startswith("127.")
- else:
- return False
-
- @property
- def skip_unless_local(self):
- """Decorator to disable tests if no real *local* Cobbler is available.
-
- Annotate tests like so::
-
- @real_cobbler.skip_unless_local
- def test_something_that_requires_a_real_local_cobbler(self):
- ...
-
- """
- return skipIf(not self.is_local, self.help_text_local)
-
- def get_session(self):
- """Obtain a session on the real Cobbler.
-
- Returns None if no real Cobbler is available.
- """
- if self.is_available:
- return CobblerSession(self.url, self.username, self.password)
- else:
- return None
=== modified file 'src/provisioningserver/tests/test_api.py'
--- src/provisioningserver/tests/test_api.py 2012-05-21 10:57:08 +0000
+++ src/provisioningserver/tests/test_api.py 2012-08-16 06:46:19 +0000
@@ -19,10 +19,7 @@
ABCMeta,
abstractmethod,
)
-from base64 import b64decode
-from contextlib import contextmanager
-from maasserver.utils import map_enum
from maastesting.factory import factory
from provisioningserver.api import (
cobbler_to_papi_distro,
@@ -30,22 +27,11 @@
cobbler_to_papi_profile,
gen_cobbler_interface_deltas,
postprocess_mapping,
- ProvisioningAPI,
)
-from provisioningserver.cobblerclient import CobblerSystem
-from provisioningserver.enum import POWER_TYPE
from provisioningserver.interfaces import IProvisioningAPI
from provisioningserver.testing.factory import ProvisioningFakeFactory
-from provisioningserver.testing.fakeapi import FakeAsynchronousProvisioningAPI
-from provisioningserver.testing.fakecobbler import make_fake_cobbler_session
-from provisioningserver.testing.realcobbler import RealCobbler
from testtools import TestCase
from testtools.deferredruntest import AsynchronousDeferredRunTest
-from testtools.matchers import (
- FileExists,
- Not,
- )
-from testtools.monkey import patch
from twisted.internet.defer import inlineCallbacks
from zope.interface.verify import verifyObject
@@ -581,218 +567,3 @@
yield papi.start_nodes([node_name])
# The test is that we get here without error.
pass
-
-
-class ProvisioningAPITestsWithCobbler:
- """Provisioning API tests that also access a real, or fake, Cobbler."""
-
- # Overridable: acceptable power types in the current test setup.
- valid_power_types = set(map_enum(POWER_TYPE).values())
-
- @inlineCallbacks
- def test_add_node_sets_power_type(self):
- papi = self.get_provisioning_api()
- power_types = self.valid_power_types.copy()
- # The DEFAULT value does not exist as far as the provisioning
- # server is concerned.
- power_types.discard(POWER_TYPE.DEFAULT)
- nodes = {}
- for power_type in power_types:
- nodes[power_type] = yield self.add_node(
- papi, power_type=power_type)
- cobbler_power_types = {}
- for power_type, node_id in nodes.items():
- attrs = yield CobblerSystem(papi.session, node_id).get_values()
- cobbler_power_types[power_type] = attrs['power_type']
- self.assertItemsEqual(
- dict(zip(power_types, power_types)), cobbler_power_types)
-
- @inlineCallbacks
- def test_add_node_provides_preseed(self):
- papi = self.get_provisioning_api()
- preseed_data = factory.getRandomString()
- node_name = yield self.add_node(papi, preseed_data=preseed_data)
- attrs = yield CobblerSystem(papi.session, node_name).get_values()
- self.assertEqual(
- preseed_data, b64decode(attrs['ks_meta']['MAAS_PRESEED']))
-
- @contextmanager
- def expected_sync(self, papi, times=1):
- """Context where # calls to `papi.sync` must match `times`."""
- sync_calls = []
- orig_sync = papi.sync
- fake_sync = lambda: orig_sync().addCallback(sync_calls.append)
- unpatch = patch(papi, "sync", fake_sync)
- try:
- yield
- finally:
- unpatch()
- self.assertEqual(times, len(sync_calls))
-
- @inlineCallbacks
- def test_add_distro_syncs(self):
- # add_distro ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield self.add_distro(papi)
-
- @inlineCallbacks
- def test_add_profile_syncs(self):
- # add_profile ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- distro_name = yield self.add_distro(papi)
- with self.expected_sync(papi):
- yield self.add_profile(papi, distro_name=distro_name)
-
- @inlineCallbacks
- def test_add_node_syncs(self):
- # add_node ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- profile_name = yield self.add_profile(papi)
- with self.expected_sync(papi):
- yield self.add_node(papi, profile_name=profile_name)
-
- @inlineCallbacks
- def test_modify_distros_syncs(self):
- # modify_distros ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.modify_distros({})
-
- @inlineCallbacks
- def test_modify_profiles_syncs(self):
- # modify_profiles ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.modify_profiles({})
-
- @inlineCallbacks
- def test_modify_nodes_syncs(self):
- # modify_nodes ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.modify_nodes({})
-
- @inlineCallbacks
- def test_delete_distros_by_name_syncs(self):
- # delete_distros_by_name ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.delete_distros_by_name([])
-
- @inlineCallbacks
- def test_delete_profiles_by_name_syncs(self):
- # delete_profiles_by_name ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.delete_profiles_by_name([])
-
- @inlineCallbacks
- def test_delete_nodes_by_name_syncs(self):
- # delete_nodes_by_name ensures that Cobbler syncs.
- papi = self.get_provisioning_api()
- with self.expected_sync(papi):
- yield papi.delete_nodes_by_name([])
-
-
-class TestFakeProvisioningAPI(ProvisioningAPITests, TestCase):
- """Test :class:`FakeAsynchronousProvisioningAPI`.
-
- Includes by inheritance all the tests in :class:`ProvisioningAPITests`.
- """
-
- def get_provisioning_api(self):
- """Return a fake ProvisioningAPI."""
- return FakeAsynchronousProvisioningAPI()
-
-
-class TestProvisioningAPIWithFakeCobbler(ProvisioningAPITests,
- ProvisioningAPITestsWithCobbler,
- TestCase):
- """Test :class:`ProvisioningAPI` with a fake Cobbler instance.
-
- Includes by inheritance all the tests in :class:`ProvisioningAPITests`.
- """
-
- def get_provisioning_api(self):
- """Return a real ProvisioningAPI, but using a fake Cobbler session."""
- return ProvisioningAPI(make_fake_cobbler_session())
-
- def test_sync(self):
- """`ProvisioningAPI.sync` issues an authenticated ``sync`` call.
-
- It is not exported - i.e. it is not part of :class:`IProvisioningAPI`
- - but is used heavily by other methods in `IProvisioningAPI`.
- """
- papi = self.get_provisioning_api()
- calls = []
- self.patch(
- papi.session, "call",
- lambda *args: calls.append(args))
- papi.sync()
- self.assertEqual(
- [("sync", papi.session.token_placeholder)],
- calls)
-
-
-class TestProvisioningAPIWithRealCobbler(ProvisioningAPITests,
- ProvisioningAPITestsWithCobbler,
- TestCase):
- """Test :class:`ProvisioningAPI` with a real Cobbler instance.
-
- The URL for the Cobbler instance must be provided in the
- `PSERV_TEST_COBBLER_URL` environment variable.
-
- Includes by inheritance all the tests in :class:`ProvisioningAPITests` and
- :class:`ProvisioningAPITestsWithCobbler`.
- """
-
- real_cobbler = RealCobbler()
-
- # Power methods that Cobbler accepts.
- cobbler_power_types = set([
- 'apc_snmp',
- 'bladecenter',
- 'bullpap',
- 'drac',
- 'ether_wake',
- 'ilo',
- 'integrity',
- 'ipmilan',
- 'ipmitool',
- 'lpar',
- 'rsa',
- 'sentryswitch_cdu',
- 'virsh',
- 'wti',
- ])
- # Only accept power methods that both MAAS and Cobbler know.
- valid_power_types = (
- ProvisioningAPITestsWithCobbler.valid_power_types.intersection(
- cobbler_power_types))
-
- @real_cobbler.skip_unless_available
- def get_provisioning_api(self):
- """Return a connected :class:`ProvisioningAPI`."""
- return ProvisioningAPI(self.real_cobbler.get_session())
-
- @real_cobbler.skip_unless_local
- @inlineCallbacks
- def test_sync_after_modify(self):
- # When MAAS modifies the MAC addresses of a node it triggers a sync of
- # Cobbler. This is to ensure that netboot files are up-to-date, or
- # removed as appropriate.
- papi = self.get_provisioning_api()
- node_name = yield self.add_node(papi)
- mac_address = factory.getRandomMACAddress()
- yield papi.modify_nodes(
- {node_name: {"mac_addresses": [mac_address]}})
- # The PXE file corresponding to the node's MAC address is present.
- pxe_filename = "/var/lib/tftpboot/pxelinux.cfg/01-%s" % (
- mac_address.replace(":", "-"),)
- self.assertThat(pxe_filename, FileExists())
- # Remove the MAC address again.
- yield papi.modify_nodes(
- {node_name: {"mac_addresses": []}})
- # The PXE file has been removed too.
- self.assertThat(pxe_filename, Not(FileExists()))
=== removed file 'src/provisioningserver/tests/test_cobblercatcher.py'
--- src/provisioningserver/tests/test_cobblercatcher.py 2012-04-20 11:00:49 +0000
+++ src/provisioningserver/tests/test_cobblercatcher.py 1970-01-01 00:00:00 +0000
@@ -1,186 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Tests for conversion of Cobbler exceptions to `ProvisioningError`."""
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-__metaclass__ = type
-__all__ = []
-
-from abc import (
- ABCMeta,
- abstractmethod,
- )
-from xmlrpclib import Fault
-
-from maastesting.factory import factory
-from provisioningserver.cobblercatcher import (
- convert_cobbler_exception,
- divine_fault_code,
- extract_text,
- ProvisioningError,
- )
-from provisioningserver.cobblerclient import CobblerSystem
-from provisioningserver.enum import PSERV_FAULT
-from provisioningserver.testing.fakecobbler import (
- fake_auth_failure_string,
- fake_token,
- make_fake_cobbler_session,
- )
-from provisioningserver.testing.realcobbler import RealCobbler
-from provisioningserver.utils import deferred
-from testtools import TestCase
-from testtools.deferredruntest import AsynchronousDeferredRunTest
-from twisted.internet.defer import inlineCallbacks
-
-
-class TestExceptionConversionWithFakes(TestCase):
- """Tests for handling of Cobbler errors by cobblercatcher.
-
- Can be targeted to real or fake Cobbler.
- """
-
- def test_recognizes_auth_expiry(self):
- original_fault = Fault(1, fake_auth_failure_string(fake_token()))
- converted_fault = convert_cobbler_exception(original_fault)
- self.assertEqual(
- PSERV_FAULT.COBBLER_AUTH_ERROR, converted_fault.faultCode)
-
- def test_extract_text_leaves_unrecognized_message_intact(self):
- text = factory.getRandomString()
- self.assertEqual(text, extract_text(text))
-
- def test_extract_text_strips_CX(self):
- text = factory.getRandomString()
- self.assertEqual(
- text,
- extract_text("<class 'cobbler.cexceptions.CX'>:'%s'" % text))
-
- def test_divine_fault_code_recognizes_errors(self):
- errors = {
- "login failed (%s)": PSERV_FAULT.COBBLER_AUTH_FAILED,
- "invalid token: %s": PSERV_FAULT.COBBLER_AUTH_ERROR,
- "invalid profile name: %s": PSERV_FAULT.NO_SUCH_PROFILE,
- "Huh? %s. Aaaaargh!": PSERV_FAULT.GENERIC_COBBLER_ERROR,
- }
- random = factory.getRandomString()
- self.assertEqual(
- errors, {
- text: divine_fault_code(text % random)
- for text in errors.keys()})
-
- def test_convert_cobbler_exception_passes_through_other_faults(self):
- original_fault = Fault(1234, "Error while talking to Cobbler")
- converted_fault = convert_cobbler_exception(original_fault)
- self.assertEqual(
- (PSERV_FAULT.NO_COBBLER, original_fault.faultString),
- (converted_fault.faultCode, converted_fault.faultString))
-
- def test_convert_cobbler_exception_converts_to_provisioning_error(self):
- self.assertIsInstance(
- convert_cobbler_exception(Fault(1, "Kaboom")), ProvisioningError)
-
- def test_convert_cobbler_exception_checks_against_double_conversion(self):
- self.assertRaises(
- AssertionError,
- convert_cobbler_exception,
- ProvisioningError(1, "Ayiieee!"))
-
-
-class FaultFinder:
- """Context manager: catch and store a :class:`Fault` exception.
-
- :ivar fault: The Fault this context manager caught. This attribute will
- not exist if no Fault occurred.
- """
-
- def __enter__(self):
- pass
-
- def __exit__(self, exc_type, exc_val, exc_tb):
- self.fault = exc_val
- return isinstance(exc_val, Fault)
-
-
-class ExceptionConversionTests:
- """Tests for exception handling; run against a Cobbler (real or fake)."""
-
- __metaclass__ = ABCMeta
-
- @abstractmethod
- def get_cobbler_session(self):
- """Override: provide a (real or fake) :class:`CobblerSession`.
-
- :return: A :class:`Deferred` which in turn will return a
- :class:`CobblerSession`.
- """
-
- @inlineCallbacks
- def test_bad_token_means_token_expired(self):
- session = yield self.get_cobbler_session()
- session.token = factory.getRandomString()
- arbitrary_id = factory.getRandomString()
-
- faultfinder = FaultFinder()
- with faultfinder:
- yield session._issue_call(
- 'xapi_object_edit', 'repo', 'repo-%s' % arbitrary_id,
- 'edit', {'mirror': 'on-the-wall'}, session.token)
-
- self.assertEqual(
- PSERV_FAULT.COBBLER_AUTH_ERROR, faultfinder.fault.faultCode)
-
- @inlineCallbacks
- def test_bad_profile_name_is_distinct_error(self):
- session = yield self.get_cobbler_session()
- arbitrary_id = factory.getRandomString()
-
- faultfinder = FaultFinder()
- with faultfinder:
- yield CobblerSystem.new(
- session, 'system-%s' % arbitrary_id,
- {'profile': 'profile-%s' % arbitrary_id})
-
- self.assertEqual(
- PSERV_FAULT.NO_SUCH_PROFILE, faultfinder.fault.faultCode)
-
-
-class TestExceptionConversionWithFakeCobbler(ExceptionConversionTests,
- TestCase):
- """Run `ExceptionConversionTests` against a fake Cobbler instance.
-
- All tests from `ExceptionConversionTests` are included here through
- inheritance.
- """
-
- run_tests_with = AsynchronousDeferredRunTest
-
- @deferred
- def get_cobbler_session(self):
- return make_fake_cobbler_session()
-
-
-class TestExceptionConversionWithRealCobbler(ExceptionConversionTests,
- TestCase):
- """Run `ExceptionConversionTests` against a real Cobbler instance.
-
- Activate this by setting the PSERV_TEST_COBBLER_URL environment variable
- to point to a real Cobbler instance.
-
- All tests from `ExceptionConversionTests` are included here through
- inheritance.
- """
-
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
-
- real_cobbler = RealCobbler()
-
- @real_cobbler.skip_unless_available
- @deferred
- def get_cobbler_session(self):
- return self.real_cobbler.get_session()
=== removed file 'src/provisioningserver/tests/test_cobblerclient.py'
--- src/provisioningserver/tests/test_cobblerclient.py 2012-04-20 13:45:22 +0000
+++ src/provisioningserver/tests/test_cobblerclient.py 1970-01-01 00:00:00 +0000
@@ -1,513 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Tests for `provisioningserver.cobblerclient`."""
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-__metaclass__ = type
-__all__ = []
-
-from itertools import count
-from random import randint
-
-from maastesting.testcase import TestCase
-from provisioningserver.cobblerclient import (
- CobblerDistro,
- CobblerImage,
- CobblerPreseeds,
- CobblerProfile,
- CobblerRepo,
- CobblerSystem,
- CobblerXMLRPCProxy,
- tilde_to_None,
- )
-from provisioningserver.testing.factory import CobblerFakeFactory
-from provisioningserver.testing.fakecobbler import log_in_to_fake_cobbler
-from testtools.deferredruntest import AsynchronousDeferredRunTest
-from testtools.testcase import ExpectedException
-from twisted.internet.defer import (
- inlineCallbacks,
- returnValue,
- )
-from twisted.test.proto_helpers import MemoryReactor
-
-
-class TestRepairingCobblerResponses(TestCase):
- """See `tilde_to_None`."""
-
- def test_tilde_to_None(self):
- self.assertIsNone(tilde_to_None("~"))
-
- def test_tilde_to_None_list(self):
- self.assertEqual(
- [1, 2, 3, None, 5],
- tilde_to_None([1, 2, 3, "~", 5]))
-
- def test_tilde_to_None_nested_list(self):
- self.assertEqual(
- [1, 2, [3, None], 5],
- tilde_to_None([1, 2, [3, "~"], 5]))
-
- def test_tilde_to_None_dict(self):
- self.assertEqual(
- {"one": 1, "two": None},
- tilde_to_None({"one": 1, "two": "~"}))
-
- def test_tilde_to_None_nested_dict(self):
- self.assertEqual(
- {"one": 1, "two": {"three": None}},
- tilde_to_None({"one": 1, "two": {"three": "~"}}))
-
- def test_tilde_to_None_nested_mixed(self):
- self.assertEqual(
- {"one": 1, "two": [3, 4, None]},
- tilde_to_None({"one": 1, "two": [3, 4, "~"]}))
-
- def test_CobblerXMLRPCProxy(self):
- reactor = MemoryReactor()
- proxy = CobblerXMLRPCProxy(
- "http://localhost:1234/nowhere";, reactor=reactor)
- d = proxy.callRemote("cobble", 1, 2, 3)
- # A connection has been initiated.
- self.assertEqual(1, len(reactor.tcpClients))
- [client] = reactor.tcpClients
- self.assertEqual("localhost", client[0])
- self.assertEqual(1234, client[1])
- # A "broken" response from Cobbler is "repaired".
- d.callback([1, 2, "~"])
- self.assertEqual([1, 2, None], d.result)
-
-
-unique_ints = count(randint(0, 9999))
-
-
-class CobblerObjectTestScenario(CobblerFakeFactory):
- """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')
-
- def test_normalize_attribute_alternative_attributes(self):
- # _normalize_attribute() can be passed a different set of attributes
- # against which to normalize.
- allowed_attributes = set(["some-unknown-attribute"])
- self.assertEqual(
- 'some-unknown-attribute',
- self.cobbler_class._normalize_attribute(
- 'some-unknown-attribute', allowed_attributes))
-
- @inlineCallbacks
- def test_create_object(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- self.assertEqual(name, obj.name)
-
- @inlineCallbacks
- def test_create_object_fails_if_cobbler_returns_False(self):
-
- def return_false(*args):
- return False
-
- session = yield log_in_to_fake_cobbler()
- session.fake_proxy.fake_cobbler.xapi_object_edit = return_false
- with ExpectedException(RuntimeError):
- yield self.fake_cobbler_object(session, self.cobbler_class)
-
- @inlineCallbacks
- def test_find_returns_empty_list_if_no_match(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- matches = yield self.cobbler_class.find(session, name=name)
- self.assertSequenceEqual([], matches)
-
- @inlineCallbacks
- def test_find_matches_name(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- yield self.fake_cobbler_object(session, self.cobbler_class, name)
- by_name = yield self.cobbler_class.find(session, name=name)
- self.assertSequenceEqual([name], [obj.name for obj in by_name])
-
- @inlineCallbacks
- def test_find_matches_attribute(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- comment = "This is comment #%d" % next(unique_ints)
- yield self.fake_cobbler_object(
- session, self.cobbler_class, name, {'comment': comment})
- by_comment = yield self.cobbler_class.find(session, comment=comment)
- self.assertSequenceEqual([name], [obj.name for obj in by_comment])
-
- @inlineCallbacks
- def test_find_without_args_finds_everything(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- yield self.fake_cobbler_object(session, self.cobbler_class, name)
- found_objects = yield self.cobbler_class.find(session)
- self.assertIn(name, [obj.name for obj in found_objects])
-
- @inlineCallbacks
- def test_get_object_retrieves_attributes(self):
- session = yield log_in_to_fake_cobbler()
- comment = "This is comment #%d" % next(unique_ints)
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name, {'comment': comment})
- attributes = yield obj.get_values()
- self.assertEqual(name, attributes['name'])
- self.assertEqual(comment, attributes['comment'])
-
- @inlineCallbacks
- def test_get_objects_retrieves_attributes_for_all_objects(self):
- session = yield log_in_to_fake_cobbler()
- comment = "This is comment #%d" % next(unique_ints)
- name = self.make_name()
- yield self.fake_cobbler_object(
- session, self.cobbler_class, name, {'comment': comment})
- all_objects = yield self.cobbler_class.get_all_values(session)
- found_obj = all_objects[name]
- self.assertEqual(name, found_obj['name'])
- self.assertEqual(comment, found_obj['comment'])
-
- @inlineCallbacks
- def test_get_values_returns_None_for_non_existent_object(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- values = yield self.cobbler_class(session, name).get_values()
- self.assertIsNone(values)
-
- @inlineCallbacks
- def test_get_handle_finds_handle(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- handle = yield obj._get_handle()
- self.assertNotEqual(None, handle)
-
- @inlineCallbacks
- def test_get_handle_distinguishes_objects(self):
- session = yield log_in_to_fake_cobbler()
- obj1 = yield self.fake_cobbler_object(session, self.cobbler_class)
- handle1 = yield obj1._get_handle()
- obj2 = yield self.fake_cobbler_object(session, self.cobbler_class)
- handle2 = yield obj2._get_handle()
- self.assertNotEqual(handle1, handle2)
-
- @inlineCallbacks
- def test_get_handle_is_consistent(self):
- session = yield log_in_to_fake_cobbler()
- obj = yield self.fake_cobbler_object(session, self.cobbler_class)
- handle1 = yield obj._get_handle()
- handle2 = yield obj._get_handle()
- self.assertEqual(handle1, handle2)
-
- @inlineCallbacks
- def test_delete_removes_object(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- yield obj.delete()
- matches = yield self.cobbler_class.find(session, name=name)
- self.assertSequenceEqual([], matches)
-
-
-class TestCobblerDistro(CobblerObjectTestScenario, TestCase):
- """Tests for `CobblerDistro`. Uses generic `CobblerObject` scenario."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
- 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.assertSequenceEqual(
- ['mgmt-classes'] * 2,
- map(self.cobbler_class._normalize_attribute, inputs))
-
-
-class TestCobblerImage(CobblerObjectTestScenario, TestCase):
- """Tests for `CobblerImage`. Uses generic `CobblerObject` scenario."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
- 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.assertSequenceEqual(
- ['image_type'] * 2,
- map(self.cobbler_class._normalize_attribute, inputs))
-
-
-class TestCobblerProfile(CobblerObjectTestScenario, TestCase):
- """Tests for `CobblerProfile`. Uses generic `CobblerObject` scenario."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
- 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.assertSequenceEqual(
- expected_outputs,
- map(self.cobbler_class._normalize_attribute, inputs))
-
-
-class TestCobblerRepo(CobblerObjectTestScenario, TestCase):
- """Tests for `CobblerRepo`. Uses generic `CobblerObject` scenario."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
- cobbler_class = CobblerRepo
-
-
-class TestCobblerSystem(CobblerObjectTestScenario, TestCase):
- """Tests for `CobblerSystem`. Uses generic `CobblerObject` scenario."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
- 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.fake_cobbler_object(
- session, self.cobbler_class, 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_interface_set_mac_address(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- yield obj.modify(
- {"interface": "eth0", "mac_address": "12:34:56:78:90:12"})
- state = yield obj.get_values()
- interfaces = state.get("interfaces", {})
- self.assertEqual(["eth0"], sorted(interfaces))
- state_eth0 = interfaces["eth0"]
- self.assertEqual("12:34:56:78:90:12", state_eth0["mac_address"])
-
- @inlineCallbacks
- def test_interface_set_dns_name(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- yield obj.modify(
- {"interface": "eth0", "dns_name": "epitaph"})
- state = yield obj.get_values()
- interfaces = state.get("interfaces", {})
- self.assertEqual(["eth0"], sorted(interfaces))
- state_eth0 = interfaces["eth0"]
- self.assertEqual("epitaph", state_eth0["dns_name"])
-
- @inlineCallbacks
- def test_interface_delete_interface(self):
- session = yield log_in_to_fake_cobbler()
- name = self.make_name()
- obj = yield self.fake_cobbler_object(
- session, self.cobbler_class, name)
- yield obj.modify(
- {"interface": "eth0", "mac_address": "12:34:56:78:90:12"})
- yield obj.modify(
- {"interface": "eth0", "delete_interface": "ignored"})
- state = yield obj.get_values()
- interfaces = state.get("interfaces", {})
- self.assertEqual([], sorted(interfaces))
-
- @inlineCallbacks
- def test_powerOnMultiple(self):
- session = yield log_in_to_fake_cobbler()
- 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 log_in_to_fake_cobbler()
- 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 log_in_to_fake_cobbler()
- 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 log_in_to_fake_cobbler()
- 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 log_in_to_fake_cobbler()
- 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 log_in_to_fake_cobbler()
- 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`."""
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
-
- @inlineCallbacks
- def make_preseeds_api(self):
- session = yield log_in_to_fake_cobbler()
- 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/kickstarts/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.assertIn(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/kickstarts/template-%d' % unique_int
- yield preseeds.write_template(path, "Text")
- snippets = yield preseeds.get_snippets()
- self.assertNotIn(path, 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.assertNotIn(path, templates)
=== removed file 'src/provisioningserver/tests/test_cobblersession.py'
--- src/provisioningserver/tests/test_cobblersession.py 2012-04-20 09:58:23 +0000
+++ src/provisioningserver/tests/test_cobblersession.py 1970-01-01 00:00:00 +0000
@@ -1,561 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-"""Tests for `CobblerSession`."""
-
-__metaclass__ = type
-__all__ = []
-
-from random import Random
-import re
-from xmlrpclib import Fault
-
-import fixtures
-from maastesting.factory import factory
-from provisioningserver import cobblerclient
-from provisioningserver.cobblercatcher import ProvisioningError
-from provisioningserver.enum import PSERV_FAULT
-from provisioningserver.testing.fakecobbler import (
- fake_auth_failure_string,
- fake_object_not_found_string,
- fake_token,
- )
-from testtools.content import text_content
-from testtools.deferredruntest import (
- assert_fails_with,
- AsynchronousDeferredRunTest,
- AsynchronousDeferredRunTestForBrokenTwisted,
- )
-from testtools.testcase import (
- ExpectedException,
- TestCase,
- )
-from twisted.internet import defer
-from twisted.internet.defer import inlineCallbacks
-from twisted.internet.error import DNSLookupError
-from twisted.internet.task import Clock
-
-
-randomizer = Random()
-
-
-def pick_number():
- """Pick an arbitrary number."""
- return randomizer.randint(0, 10 ** 9)
-
-
-def apply_to_all(function, args):
- """List the results of `function(arg)` for each `arg` in `args`."""
- return list(map(function, args))
-
-
-class FakeAuthFailure(Fault):
- """Imitated Cobbler authentication failure."""
-
- def __init__(self, token):
- super(FakeAuthFailure, self).__init__(
- 1, fake_auth_failure_string(token))
-
-
-def make_auth_failure():
- """Mimick a Cobbler authentication failure."""
- return FakeAuthFailure(fake_token())
-
-
-class RecordingFakeProxy:
- """Simple fake Twisted XMLRPC proxy.
-
- Records XMLRPC calls, and returns predetermined values.
- """
- def __init__(self, fake_token=None):
- self.fake_token = fake_token
- self.calls = []
- self.return_values = None
-
- def set_return_values(self, values):
- """Set predetermined value to return on following call(s).
-
- If any return value is an `Exception`, it will be raised instead.
- """
- self.return_values = values
-
- def callRemote(self, method, *args):
- if method == 'login':
- return defer.succeed(self.fake_token)
-
- self.calls.append((method, ) + args)
- if self.return_values:
- value = self.return_values.pop(0)
- else:
- value = None
- if isinstance(value, Exception):
- return defer.fail(value)
- else:
- return defer.succeed(value)
-
-
-class DeadProxy(RecordingFakeProxy):
- """Fake proxy that returns nothing. Useful for timeout testing."""
-
- def callRemote(self, method, *args):
- return defer.Deferred()
-
-
-class RecordingSession(cobblerclient.CobblerSession):
- """A `CobblerSession` instrumented to run against a `RecordingFakeProxy`.
-
- :ivar fake_token: Auth token that login will pretend to receive.
- """
-
- def __init__(self, *args, **kwargs):
- """Create and instrument a session.
-
- In addition to the arguments for `CobblerSession.__init__`, pass a
- keyword argument `fake_proxy` to set a test double that the session
- will use for its proxy; and `fake_token` to provide a login token
- that the session should pretend it gets from the server on login.
- """
- fake_token = kwargs.pop('fake_token')
- proxy_class = kwargs.pop('fake_proxy', RecordingFakeProxy)
- if proxy_class is None:
- proxy_class = RecordingFakeProxy
-
- self.fake_proxy = proxy_class(fake_token)
- super(RecordingSession, self).__init__(*args, **kwargs)
-
- def _make_twisted_proxy(self):
- """Override for CobblerSession's proxy factory."""
- return self.fake_proxy
-
-
-def make_url_user_password():
- """Produce arbitrary API URL, username, and password."""
- return (
- 'http://api.example.com/%d' % pick_number(),
- 'username%d' % pick_number(),
- 'password%d' % pick_number(),
- )
-
-
-def make_recording_session(session_args=None, token=None,
- fake_proxy=RecordingFakeProxy):
- """Create a `RecordingSession`."""
- if session_args is None:
- session_args = make_url_user_password()
- if token is None:
- token = fake_token()
- return RecordingSession(
- *session_args, fake_token=token, fake_proxy=fake_proxy)
-
-
-class TestCobblerSession(TestCase):
- """Test session management against a fake XMLRPC session."""
-
- # Use a slightly longer timeout so that we can run these tests
- # against a real Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
-
- def test_initializes_but_does_not_authenticate_on_creation(self):
- url, user, password = make_url_user_password()
- session = make_recording_session(
- token=fake_token(user, 'not-yet-authenticated'))
- self.assertEqual(None, session.token)
-
- @inlineCallbacks
- def test_authenticate_authenticates_initially(self):
- token = fake_token('authenticated')
- session = make_recording_session(token=token)
- self.assertEqual(None, session.token)
- yield session._authenticate()
- self.assertEqual(token, session.token)
-
- @inlineCallbacks
- def test_state_cookie_stays_constant_during_normal_use(self):
- session = make_recording_session()
- state = session.record_state()
- self.assertEqual(state, session.record_state())
- yield session.call("some_method")
- self.assertEqual(state, session.record_state())
-
- @inlineCallbacks
- def test_authentication_changes_state_cookie(self):
- session = make_recording_session()
- old_cookie = session.record_state()
- yield session._authenticate()
- self.assertNotEqual(old_cookie, session.record_state())
-
- @inlineCallbacks
- def test_authenticate_backs_off_from_overwriting_concurrent_auth(self):
- session = make_recording_session()
- # Two requests are made concurrently.
- cookie_before_request_1 = session.record_state()
- 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)
- 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)
- token_for_retrying_request_2 = session.token
-
- # The double authentication does not confuse the session; both
- # callbacks get the same auth token for their retries.
- self.assertEqual(
- token_for_retrying_request_1, token_for_retrying_request_2)
- # The token they get is a new token, not the one they started
- # with.
- self.assertNotEqual(cookie_before_request_1, session.token)
- self.assertNotEqual(cookie_before_request_2, session.token)
-
- @inlineCallbacks
- def test_substitute_token_substitutes_only_placeholder(self):
- session = make_recording_session(token=fake_token('for-subst'))
- yield session._authenticate()
- arbitrary_number = pick_number()
- arbitrary_string = 'string-%d' % pick_number()
- inputs = [
- arbitrary_number,
- cobblerclient.CobblerSession.token_placeholder,
- arbitrary_string,
- None,
- ]
- outputs = [
- arbitrary_number,
- session.token,
- arbitrary_string,
- None,
- ]
- self.assertEqual(
- outputs, apply_to_all(session._substitute_token, inputs))
-
- @inlineCallbacks
- def test_call_calls_xmlrpc(self):
- session = make_recording_session()
- return_value = 'returnval-%d' % pick_number()
- method = 'method_%d' % pick_number()
- arg = 'arg-%d' % pick_number()
- session.proxy.set_return_values([return_value])
- actual_return_value = yield session.call(method, arg)
- self.assertEqual(return_value, actual_return_value)
- self.assertEqual([(method, arg)], session.proxy.calls)
-
- def test_looks_like_object_not_found_for_regular_exception(self):
- self.assertFalse(
- cobblerclient.looks_like_object_not_found(RuntimeError("Error")))
-
- def test_looks_like_object_not_found_for_other_Fault(self):
- self.assertFalse(
- cobblerclient.looks_like_object_not_found(
- Fault(1, "Missing sprocket")))
-
- def test_looks_like_object_not_found_recognizes_object_not_found(self):
- error = Fault(1, fake_object_not_found_string("distro", "bob"))
- self.assertTrue(
- cobblerclient.looks_like_object_not_found(error))
-
- @inlineCallbacks
- def test_call_reauthenticates_and_retries_on_auth_failure(self):
- # If a call triggers an authentication error, call()
- # re-authenticates and then re-issues the call.
- session = make_recording_session()
- yield session._authenticate()
- successful_return_value = pick_number()
- session.proxy.set_return_values([
- make_auth_failure(),
- successful_return_value,
- ])
- session.proxy.calls = []
- old_token = session.token
- ultimate_return_value = yield session.call(
- "failing_method", cobblerclient.CobblerSession.token_placeholder)
- new_token = session.token
- self.assertEqual(
- [
- # Initial call to failing_method: auth failure.
- ('failing_method', old_token),
- # But call() re-authenticates, and retries.
- ('failing_method', new_token),
- ],
- session.proxy.calls)
- self.assertEqual(successful_return_value, ultimate_return_value)
-
- @inlineCallbacks
- def test_call_reauthentication_compares_against_original_cookie(self):
- # When a call triggers an authentication error, authenticate()
- # is called just once, with the state cookie from before the
- # call. This ensures that it will always notice a concurrent
- # re-authentication that it needs to back off from.
- session = make_recording_session()
- yield session._authenticate()
- authenticate_cookies = []
-
- def fake_authenticate(previous_state):
- authenticate_cookies.append(previous_state)
-
- session._authenticate = fake_authenticate
- session.proxy.set_return_values([make_auth_failure()])
- state_before_call = session.record_state()
- yield session.call(
- "fail", cobblerclient.CobblerSession.token_placeholder)
- self.assertEqual([state_before_call], authenticate_cookies)
-
- @inlineCallbacks
- def test_call_raises_repeated_auth_failure(self):
- session = make_recording_session()
- yield session._authenticate()
- failures = [
- # Initial operation fails: not authenticated.
- make_auth_failure(),
- # But retry still raises authentication failure.
- make_auth_failure(),
- ]
- session.proxy.set_return_values(failures)
- with ExpectedException(ProvisioningError, failures[-1].message):
- return_value = yield session.call(
- 'double_fail', cobblerclient.CobblerSession.token_placeholder)
- self.addDetail('return_value', text_content(repr(return_value)))
-
- @inlineCallbacks
- def test_call_raises_general_failure(self):
- session = make_recording_session()
- yield session._authenticate()
- failure = Exception("Memory error. Where did I put it?")
- session.proxy.set_return_values([failure])
- with ExpectedException(Exception, failure.message):
- return_value = yield session.call('failing_method')
- self.addDetail('return_value', text_content(repr(return_value)))
-
- @inlineCallbacks
- def test_call_authenticates_immediately_if_unauthenticated(self):
- # If there is no auth token, and authentication is required,
- # call() authenticates right away rather than waiting for the
- # first call attempt to fail.
- session = make_recording_session()
- session.token = None
- session.proxy.set_return_values([pick_number()])
- yield session.call(
- 'authenticate_me_first',
- cobblerclient.CobblerSession.token_placeholder)
- self.assertNotEqual(None, session.token)
- self.assertEqual(
- [('authenticate_me_first', session.token)], session.proxy.calls)
-
- @inlineCallbacks
- def test_dns_lookup_exception_handled(self):
- url = factory.getRandomString()
- session_args = (
- 'http://%s/%d' % (url, pick_number()),
- factory.getRandomString(), # username.
- factory.getRandomString(), # password.
- )
- session = make_recording_session(session_args=session_args)
- failure = DNSLookupError(factory.getRandomString())
- session.proxy.set_return_values([failure])
- expected_exception = ProvisioningError(
- faultCode=PSERV_FAULT.COBBLER_DNS_LOOKUP_ERROR,
- faultString=url.lower())
- expected_exception_re = re.escape(unicode(expected_exception))
- with ExpectedException(ProvisioningError, expected_exception_re):
- yield session.call('failing_method')
-
-
-class TestConnectionTimeouts(TestCase, fixtures.TestWithFixtures):
- """Tests for connection timeouts on `CobblerSession`."""
-
- run_tests_with = AsynchronousDeferredRunTestForBrokenTwisted
-
- def test__with_timeout_cancels(self):
- # Winding a clock reactor past the timeout value should cancel
- # the original Deferred.
- clock = Clock()
- session = make_recording_session()
- d = session._with_timeout(defer.Deferred(), 1, clock)
- clock.advance(2)
- return assert_fails_with(d, defer.CancelledError)
-
- def test__with_timeout_not_cancelled_with_success(self):
- # Winding a clock reactor past the timeout of a *called*
- # (defer.succeed() is pre-fired) Deferred should not trigger a
- # cancellation.
- clock = Clock()
- session = 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):
- # Winding a clock reactor forwards but not past the timeout
- # should result in no cancellation.
- clock = Clock()
- session = make_recording_session()
- d = session._with_timeout(defer.Deferred(), 5, clock)
- clock.advance(1)
- self.assertFalse(d.called)
-
- def test__issue_call_times_out(self):
- clock = Clock()
- patch = fixtures.MonkeyPatch(
- "provisioningserver.cobblerclient.default_reactor", clock)
- self.useFixture(patch)
-
- session = make_recording_session(fake_proxy=DeadProxy)
- d = session._issue_call("login", "foo")
- clock.advance(cobblerclient.DEFAULT_TIMEOUT + 1)
- return assert_fails_with(d, defer.CancelledError)
-
-
-class TestCobblerObject(TestCase):
- """Tests for the `CobblerObject` classes."""
-
- run_tests_with = AsynchronousDeferredRunTest
-
- def test_name_method_inserts_type_name(self):
- self.assertEqual(
- 'foo_system_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))
-
- def test_new_checks_required_attributes(self):
- # CobblerObject.new asserts that all required attributes for a
- # type of object are provided.
- session = make_recording_session()
- with ExpectedException(AssertionError):
- yield cobblerclient.CobblerSystem.new(
- session, 'incomplete_system', {})
-
- @inlineCallbacks
- def test_new_attempts_edit_before_creating_new(self):
- # CobblerObject.new always attempts an extended edit operation on the
- # given object first, following by an add should the object not yet
- # exist.
- session = make_recording_session()
- not_found_string = fake_object_not_found_string("system", "carcass")
- not_found = Fault(1, not_found_string)
- session.proxy.set_return_values([not_found, True])
- yield cobblerclient.CobblerSystem.new(
- session, "carcass", {"profile": "heartwork"})
- expected_calls = [
- # First an edit is attempted...
- ("xapi_object_edit", "system", "carcass", "edit",
- {"name": "carcass", "profile": "heartwork"},
- session.token),
- # Followed by an add.
- ("xapi_object_edit", "system", "carcass", "add",
- {"name": "carcass", "profile": "heartwork"},
- session.token),
- ]
- self.assertEqual(expected_calls, session.proxy.calls)
-
- @inlineCallbacks
- def test_modify(self):
- session = make_recording_session()
- session.proxy.set_return_values([True])
- distro = cobblerclient.CobblerDistro(session, "fred")
- yield distro.modify({"kernel": "sanders"})
- expected_call = (
- "xapi_object_edit", "distro", distro.name, "edit",
- {"kernel": "sanders"}, session.token)
- self.assertEqual([expected_call], session.proxy.calls)
-
- @inlineCallbacks
- def test_modify_only_permits_certain_attributes(self):
- session = make_recording_session()
- distro = cobblerclient.CobblerDistro(session, "fred")
- expected = ExpectedException(
- AssertionError, "Unknown attribute for distro: machine")
- with expected:
- yield distro.modify({"machine": "head"})
-
- @inlineCallbacks
- def test_get_values_returns_only_known_attributes(self):
- session = make_recording_session()
- # Create a new CobblerDistro. The True return value means the faked
- # call to xapi_object_edit was successful.
- session.proxy.set_return_values([True])
- distro = yield cobblerclient.CobblerDistro.new(
- session, name="fred", attributes={
- "initrd": "an_initrd", "kernel": "a_kernel"})
- # Fake that Cobbler holds the following attributes about the distro
- # just created.
- values_stored = {
- "initrd": "an_initrd",
- "kernel": "a_kernel",
- "likes": "cabbage",
- "name": "fred",
- }
- session.proxy.set_return_values([values_stored])
- # However, CobblerObject.get_values() only returns attributes that are
- # in known_attributes.
- values_observed = yield distro.get_values()
- self.assertIn("initrd", values_observed)
- self.assertNotIn("likes", values_observed)
-
- @inlineCallbacks
- def test_get_all_values_returns_only_known_attributes(self):
- session = make_recording_session()
- # Create some new CobblerDistros. The True return values mean the
- # faked calls to xapi_object_edit were successful.
- session.proxy.set_return_values([True])
- yield cobblerclient.CobblerDistro.new(
- session, name="alice", attributes={
- "initrd": "an_initrd", "kernel": "a_kernel"})
- # Fake that Cobbler holds the following attributes about the distros
- # just created.
- values_stored = [
- {"initrd": "an_initrd",
- "kernel": "a_kernel",
- "likes": "cabbage",
- "name": "alice"},
- ]
- session.proxy.set_return_values([values_stored])
- # However, CobblerObject.get_all_values() only returns attributes that
- # are in known_attributes.
- values_observed = yield (
- cobblerclient.CobblerDistro.get_all_values(session))
- values_observed_for_alice = values_observed["alice"]
- self.assertIn("initrd", values_observed_for_alice)
- self.assertNotIn("likes", values_observed_for_alice)
-
- def test_known_attributes(self):
- # known_attributes, a class attribute, is always a frozenset.
- self.assertIsInstance(
- cobblerclient.CobblerObject.known_attributes,
- frozenset)
- self.assertIsInstance(
- cobblerclient.CobblerProfile.known_attributes,
- frozenset)
-
- def test_required_attributes(self):
- # required_attributes, a class attribute, is always a frozenset.
- self.assertIsInstance(
- cobblerclient.CobblerObject.required_attributes,
- frozenset)
- self.assertIsInstance(
- cobblerclient.CobblerDistro.required_attributes,
- frozenset)
-
- def test_modification_attributes(self):
- # modification_attributes, a class attribute, is always a frozenset.
- self.assertIsInstance(
- cobblerclient.CobblerObject.modification_attributes,
- frozenset)
- self.assertIsInstance(
- cobblerclient.CobblerDistro.modification_attributes,
- frozenset)
=== removed file 'src/provisioningserver/tests/test_fakecobbler.py'
--- src/provisioningserver/tests/test_fakecobbler.py 2012-04-20 14:36:16 +0000
+++ src/provisioningserver/tests/test_fakecobbler.py 1970-01-01 00:00:00 +0000
@@ -1,66 +0,0 @@
-# Copyright 2012 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-from __future__ import (
- absolute_import,
- print_function,
- unicode_literals,
- )
-
-"""Tests for the fake Cobbler API."""
-
-__metaclass__ = type
-__all__ = []
-
-from maastesting.testcase import TestCase
-from provisioningserver.cobblercatcher import ProvisioningError
-from provisioningserver.cobblerclient import CobblerRepo
-from provisioningserver.testing.factory import CobblerFakeFactory
-from provisioningserver.testing.fakecobbler import (
- FakeCobbler,
- log_in_to_fake_cobbler,
- )
-from testtools.content import text_content
-from testtools.deferredruntest import AsynchronousDeferredRunTest
-from testtools.testcase import ExpectedException
-from twisted.internet.defer import inlineCallbacks
-
-
-class TestFakeCobbler(TestCase, CobblerFakeFactory):
- """Test `FakeCobbler`.
-
- These tests should also pass if run against a real (clean) Cobbler.
- """
- # Use a longer timeout so that we can run these tests against a real
- # Cobbler.
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
-
- @inlineCallbacks
- def test_login_failure_raises_failure(self):
- cobbler = FakeCobbler(passwords={'moi': 'potahto'})
- with ExpectedException(ProvisioningError):
- return_value = yield log_in_to_fake_cobbler(
- user='moi', password='potayto', fake_cobbler=cobbler)
- self.addDetail('return_value', text_content(repr(return_value)))
-
- @inlineCallbacks
- def test_expired_token_triggers_retry(self):
- session = yield log_in_to_fake_cobbler()
- # When an auth token expires, the server just forgets about it.
- old_token = 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 self.fake_cobbler_object(session, CobblerRepo)
-
- # 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):
- session = yield log_in_to_fake_cobbler()
- old_token = session.token
- yield self.fake_cobbler_object(session, CobblerRepo)
- self.assertEqual(old_token, session.token)
=== modified file 'src/provisioningserver/tests/test_plugin.py'
--- src/provisioningserver/tests/test_plugin.py 2012-07-18 16:26:29 +0000
+++ src/provisioningserver/tests/test_plugin.py 2012-08-16 06:46:19 +0000
@@ -12,12 +12,8 @@
__metaclass__ = type
__all__ = []
-from base64 import b64encode
from functools import partial
-import httplib
import os
-from StringIO import StringIO
-import xmlrpclib
from maastesting.factory import factory
from maastesting.testcase import TestCase
@@ -27,7 +23,6 @@
ProvisioningServiceMaker,
SingleUsernamePasswordChecker,
)
-from provisioningserver.testing.fakecobbler import make_fake_cobbler_session
from provisioningserver.tftp import TFTPBackend
from testtools.deferredruntest import (
assert_fails_with,
@@ -38,19 +33,13 @@
Raises,
)
from tftp.protocol import TFTP
-from twisted.application.internet import (
- TCPServer,
- UDPServer,
- )
+from twisted.application.internet import UDPServer
from twisted.application.service import MultiService
from twisted.cred.credentials import UsernamePassword
from twisted.cred.error import UnauthorizedLogin
from twisted.internet.defer import inlineCallbacks
from twisted.python.usage import UsageError
-from twisted.web.guard import HTTPAuthSessionWrapper
from twisted.web.resource import IResource
-from twisted.web.server import NOT_DONE_YET
-from twisted.web.test.test_web import DummyRequest
import yaml
@@ -106,7 +95,7 @@
service = service_maker.makeService(options)
self.assertIsInstance(service, MultiService)
self.assertSequenceEqual(
- ["log", "oops", "site", "tftp"],
+ ["log", "oops", "tftp"],
sorted(service.namedServices))
self.assertEqual(
len(service.namedServices), len(service.services),
@@ -124,80 +113,12 @@
service = service_maker.makeService(options)
self.assertIsInstance(service, MultiService)
self.assertSequenceEqual(
- ["amqp", "log", "oops", "site", "tftp"],
+ ["amqp", "log", "oops", "tftp"],
sorted(service.namedServices))
self.assertEqual(
len(service.namedServices), len(service.services),
"Not all services are named.")
- def test_makeService_api_requires_credentials(self):
- """
- The site service's /api resource requires credentials from clients.
- """
- options = Options()
- options["config-file"] = self.write_config({})
- service_maker = ProvisioningServiceMaker("Harry", "Hill")
- service = service_maker.makeService(options)
- self.assertIsInstance(service, MultiService)
- site_service = service.getServiceNamed("site")
- self.assertIsInstance(site_service, TCPServer)
- port, site = site_service.args
- self.assertIn("api", site.resource.listStaticNames())
- api = site.resource.getStaticEntity("api")
- # HTTPAuthSessionWrapper demands credentials from an HTTP request.
- self.assertIsInstance(api, HTTPAuthSessionWrapper)
-
- def exercise_api_credentials(self, config_file, username, password):
- """
- Create a new service with :class:`ProvisioningServiceMaker` and
- attempt to access the API with the given credentials.
- """
- options = Options()
- options["config-file"] = config_file
- service_maker = ProvisioningServiceMaker("Morecombe", "Wise")
- # Terminate the service in a fake Cobbler session.
- service_maker._makeCobblerSession = (
- lambda config: make_fake_cobbler_session())
- service = service_maker.makeService(options)
- port, site = service.getServiceNamed("site").args
- api = site.resource.getStaticEntity("api")
- # Create an XML-RPC request with valid credentials.
- request = DummyRequest([])
- request.method = "POST"
- request.content = StringIO(xmlrpclib.dumps((), "get_nodes"))
- request.prepath = ["api"]
- request.headers["authorization"] = (
- "Basic %s" % b64encode(b"%s:%s" % (username, password)))
- # The credential check and resource rendering is deferred, but
- # NOT_DONE_YET is returned from render(). The request signals
- # completion with the aid of notifyFinish().
- finished = request.notifyFinish()
- self.assertEqual(NOT_DONE_YET, api.render(request))
- return finished.addCallback(lambda ignored: request)
-
- @inlineCallbacks
- def test_makeService_api_accepts_valid_credentials(self):
- """
- The site service's /api resource accepts valid credentials.
- """
- config = {"username": "orange", "password": "goblin"}
- request = yield self.exercise_api_credentials(
- self.write_config(config), "orange", "goblin")
- # A valid XML-RPC response has been written.
- self.assertEqual(None, request.responseCode) # None implies 200.
- xmlrpclib.loads(b"".join(request.written))
-
- @inlineCallbacks
- def test_makeService_api_rejects_invalid_credentials(self):
- """
- The site service's /api resource rejects invalid credentials.
- """
- config = {"username": "orange", "password": "goblin"}
- request = yield self.exercise_api_credentials(
- self.write_config(config), "abigail", "williams")
- # The request has not been authorized.
- self.assertEqual(httplib.UNAUTHORIZED, request.responseCode)
-
def test_tftp_service(self):
# A TFTP service is configured and added to the top-level service.
config = {
=== modified file 'src/provisioningserver/tests/test_remote.py'
--- src/provisioningserver/tests/test_remote.py 2012-04-16 10:00:51 +0000
+++ src/provisioningserver/tests/test_remote.py 2012-08-16 06:46:19 +0000
@@ -17,7 +17,6 @@
IProvisioningAPI_XMLRPC,
)
from provisioningserver.remote import ProvisioningAPI_XMLRPC
-from provisioningserver.testing.fakecobbler import make_fake_cobbler_session
from testtools import TestCase
from testtools.deferredruntest import SynchronousDeferredRunTest
from zope.interface.verify import verifyObject
@@ -40,11 +39,3 @@
papi_xmlrpc = ProvisioningAPI_XMLRPC(dummy_session)
verifyObject(IProvisioningAPI, papi_xmlrpc)
verifyObject(IProvisioningAPI_XMLRPC, papi_xmlrpc)
-
- def test_ProvisioningAPI_invoke(self):
- # The xmlrpc_* methods can be invoked.
- session = make_fake_cobbler_session()
- papi_xmlrpc = ProvisioningAPI_XMLRPC(session)
- d = papi_xmlrpc.xmlrpc_add_distro("frank", "side", "bottom")
- d.addCallback(self.assertEqual, "frank")
- return d
