launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #06277
[Merge] lp:~allenap/maas/pserv-xmlrpc-split into lp:maas
Gavin Panella has proposed merging lp:~allenap/maas/pserv-xmlrpc-split into lp:maas with lp:~allenap/maas/pserv-xmlrpc-stuff-by-name as a prerequisite.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~allenap/maas/pserv-xmlrpc-split/+merge/91950
This splits the Provisioning API from its XML-RPC incarnation, and provides interfaces for validation. Apart from the validating done here, these interfaces will be used later on to help validate mocks used when integrating the MaaS web component with the provisioning server.
--
https://code.launchpad.net/~allenap/maas/pserv-xmlrpc-split/+merge/91950
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~allenap/maas/pserv-xmlrpc-split into lp:maas.
=== added file 'src/provisioningserver/api.py'
--- src/provisioningserver/api.py 1970-01-01 00:00:00 +0000
+++ src/provisioningserver/api.py 2012-02-08 00:24:06 +0000
@@ -0,0 +1,133 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Provisioning API for external use."""
+
+from __future__ import (
+ print_function,
+ unicode_literals,
+ )
+
+__metaclass__ = type
+__all__ = [
+ "ProvisioningAPI",
+ ]
+
+from provisioningserver.cobblerclient import (
+ CobblerDistro,
+ CobblerProfile,
+ CobblerSystem,
+ )
+from provisioningserver.interfaces import IProvisioningAPI
+from twisted.internet.defer import (
+ inlineCallbacks,
+ returnValue,
+ )
+from zope.interface import implements
+
+
+class ProvisioningAPI:
+
+ implements(IProvisioningAPI)
+
+ def __init__(self, session):
+ super(ProvisioningAPI, self).__init__()
+ self.session = session
+
+ @inlineCallbacks
+ def add_distro(self, name, initrd, kernel):
+ assert isinstance(name, basestring)
+ assert isinstance(initrd, basestring)
+ assert isinstance(kernel, basestring)
+ distro = yield CobblerDistro.new(
+ self.session, name, {
+ "initrd": initrd,
+ "kernel": kernel,
+ })
+ 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})
+ returnValue(profile.name)
+
+ @inlineCallbacks
+ def add_node(self, name, profile):
+ assert isinstance(name, basestring)
+ assert isinstance(profile, basestring)
+ system = yield CobblerSystem.new(
+ self.session, name, {"profile": profile})
+ returnValue(system.name)
+
+ @inlineCallbacks
+ def get_objects_by_name(self, object_type, names):
+ """Get `object_type` objects by name.
+
+ :param object_type: The type of object to look for.
+ :type object_type: provisioningserver.objectclient.CobblerObjectType
+ :param names: A list of names to search for.
+ :type names: list
+ """
+ objects_by_name = {}
+ for name in names:
+ objects = yield object_type.find(self.session, name=name)
+ for obj in objects:
+ values = yield obj.get_values()
+ objects_by_name[obj.name] = values
+ returnValue(objects_by_name)
+
+ def get_distros_by_name(self, names):
+ return self.get_objects_by_name(CobblerDistro, names)
+
+ def get_profiles_by_name(self, names):
+ return self.get_objects_by_name(CobblerProfile, names)
+
+ def get_nodes_by_name(self, names):
+ return self.get_objects_by_name(CobblerSystem, names)
+
+ @inlineCallbacks
+ def delete_objects_by_name(self, object_type, names):
+ """Delete `object_type` objects by name.
+
+ :param object_type: The type of object to delete.
+ :type object_type: provisioningserver.objectclient.CobblerObjectType
+ :param names: A list of names to search for.
+ :type names: list
+ """
+ for name in names:
+ objects = yield object_type.find(self.session, name=name)
+ for obj in objects:
+ yield obj.delete()
+
+ def delete_distros_by_name(self, names):
+ return self.delete_objects_by_name(CobblerDistro, names)
+
+ def delete_profiles_by_name(self, names):
+ return self.delete_objects_by_name(CobblerProfile, names)
+
+ def delete_nodes_by_name(self, names):
+ return self.delete_objects_by_name(CobblerSystem, names)
+
+ @inlineCallbacks
+ def get_distros(self):
+ # WARNING: This could return a *huge* number of results. Consider
+ # adding filtering options to this function before using it in anger.
+ distros = yield CobblerDistro.get_all_values(self.session)
+ returnValue(distros)
+
+ @inlineCallbacks
+ def get_profiles(self):
+ # WARNING: This could return a *huge* number of results. Consider
+ # adding filtering options to this function before using it in anger.
+ profiles = yield CobblerProfile.get_all_values(self.session)
+ returnValue(profiles)
+
+ @inlineCallbacks
+ 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.
+ systems = yield CobblerSystem.get_all_values(self.session)
+ returnValue(systems)
=== added file 'src/provisioningserver/interfaces.py'
--- src/provisioningserver/interfaces.py 1970-01-01 00:00:00 +0000
+++ src/provisioningserver/interfaces.py 2012-02-08 00:24:06 +0000
@@ -0,0 +1,80 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Provisioning API interfaces."""
+
+from __future__ import (
+ print_function,
+ unicode_literals,
+ )
+
+__metaclass__ = type
+__all__ = [
+ "IProvisioningAPI",
+ "IProvisioningAPI_XMLRPC",
+ ]
+
+from inspect import getmembers
+from types import MethodType
+
+from zope.interface import Interface
+from zope.interface.interface import InterfaceClass
+
+
+class ProvisioningAPIBase:
+ # TODO: Flesh this out.
+
+ def add_distro(name, initrd, kernel):
+ """ """
+
+ def add_profile(name, distro):
+ """ """
+
+ def add_node(name, profile):
+ """ """
+
+ def get_distros_by_name(names):
+ """ """
+
+ def get_profiles_by_name(names):
+ """ """
+
+ def get_nodes_by_name(names):
+ """ """
+
+ def delete_distros_by_name(names):
+ """ """
+
+ def delete_profiles_by_name(names):
+ """ """
+
+ def delete_nodes_by_name(names):
+ """ """
+
+ def get_distros():
+ """ """
+
+ def get_profiles():
+ """ """
+
+ def get_nodes():
+ """ """
+
+
+PAPI_FUNCTIONS = {
+ name: value.im_func
+ for name, value in getmembers(ProvisioningAPIBase)
+ if not name.startswith("_") and isinstance(value, MethodType)
+ }
+
+IProvisioningAPI = InterfaceClass(
+ b"IProvisioningAPI", (Interface,), PAPI_FUNCTIONS)
+
+
+PAPI_XMLRPC_FUNCTIONS = {
+ "xmlrpc_%s" % name: value
+ for name, value in PAPI_FUNCTIONS.iteritems()
+ }
+
+IProvisioningAPI_XMLRPC = InterfaceClass(
+ b"IProvisioningAPI_XMLRPC", (Interface,), PAPI_XMLRPC_FUNCTIONS)
=== modified file 'src/provisioningserver/plugin.py'
--- src/provisioningserver/plugin.py 2012-02-05 14:54:48 +0000
+++ src/provisioningserver/plugin.py 2012-02-08 00:24:06 +0000
@@ -13,7 +13,7 @@
from amqpclient import AMQFactory
from provisioningserver.cobblerclient import CobblerSession
-from provisioningserver.remote import ProvisioningAPI
+from provisioningserver.remote import ProvisioningAPI_XMLRPC
from provisioningserver.services import (
LogService,
OOPSService,
@@ -133,7 +133,7 @@
session.proxy = fake_cobbler_proxy
site_root = Resource()
- site_root.putChild("api", ProvisioningAPI(session))
+ site_root.putChild("api", ProvisioningAPI_XMLRPC(session))
site = Site(site_root)
site_port = options["port"]
site_service = TCPServer(site_port, site)
=== modified file 'src/provisioningserver/remote.py'
--- src/provisioningserver/remote.py 2012-02-08 00:24:06 +0000
+++ src/provisioningserver/remote.py 2012-02-08 00:24:06 +0000
@@ -1,7 +1,7 @@
# Copyright 2012 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
-"""Provisioning API for use by the MaaS API server."""
+"""Provisioning API over XML-RPC."""
from __future__ import (
print_function,
@@ -10,121 +10,27 @@
__metaclass__ = type
__all__ = [
- "ProvisioningAPI",
+ "ProvisioningAPI_XMLRPC",
]
-from provisioningserver.cobblerclient import (
- CobblerDistro,
- CobblerProfile,
- CobblerSystem,
- )
-from twisted.internet.defer import (
- inlineCallbacks,
- returnValue,
+from provisioningserver.api import ProvisioningAPI
+from provisioningserver.interfaces import (
+ IProvisioningAPI,
+ IProvisioningAPI_XMLRPC,
)
from twisted.web.xmlrpc import XMLRPC
-
-
-class ProvisioningAPI(XMLRPC):
+from zope.interface import implements
+
+
+class ProvisioningAPI_XMLRPC(XMLRPC, ProvisioningAPI):
+
+ implements(IProvisioningAPI_XMLRPC)
def __init__(self, session):
XMLRPC.__init__(self, allowNone=True, useDateTime=True)
- self.session = session
-
- @inlineCallbacks
- def xmlrpc_add_distro(self, name, initrd, kernel):
- assert isinstance(name, basestring)
- assert isinstance(initrd, basestring)
- assert isinstance(kernel, basestring)
- distro = yield CobblerDistro.new(
- self.session, name, {
- "initrd": initrd,
- "kernel": kernel,
- })
- returnValue(distro.name)
-
- @inlineCallbacks
- def xmlrpc_add_profile(self, name, distro):
- assert isinstance(name, basestring)
- assert isinstance(distro, basestring)
- profile = yield CobblerProfile.new(
- self.session, name, {"distro": distro})
- returnValue(profile.name)
-
- @inlineCallbacks
- def xmlrpc_add_node(self, name, profile):
- assert isinstance(name, basestring)
- assert isinstance(profile, basestring)
- system = yield CobblerSystem.new(
- self.session, name, {"profile": profile})
- returnValue(system.name)
-
- @inlineCallbacks
- def get_objects_by_name(self, object_type, names):
- """Get `object_type` objects by name.
-
- :param object_type: The type of object to look for.
- :type object_type: provisioningserver.objectclient.CobblerObjectType
- :param names: A list of names to search for.
- :type names: list
- """
- objects_by_name = {}
- for name in names:
- objects = yield object_type.find(self.session, name=name)
- for obj in objects:
- values = yield obj.get_values()
- objects_by_name[obj.name] = values
- returnValue(objects_by_name)
-
- def xmlrpc_get_distros_by_name(self, names):
- return self.get_objects_by_name(CobblerDistro, names)
-
- def xmlrpc_get_profiles_by_name(self, names):
- return self.get_objects_by_name(CobblerProfile, names)
-
- def xmlrpc_get_nodes_by_name(self, names):
- return self.get_objects_by_name(CobblerSystem, names)
-
- @inlineCallbacks
- def delete_objects_by_name(self, object_type, names):
- """Delete `object_type` objects by name.
-
- :param object_type: The type of object to delete.
- :type object_type: provisioningserver.objectclient.CobblerObjectType
- :param names: A list of names to search for.
- :type names: list
- """
- for name in names:
- objects = yield object_type.find(self.session, name=name)
- for obj in objects:
- yield obj.delete()
-
- def xmlrpc_delete_distros_by_name(self, names):
- return self.delete_objects_by_name(CobblerDistro, names)
-
- def xmlrpc_delete_profiles_by_name(self, names):
- return self.delete_objects_by_name(CobblerProfile, names)
-
- def xmlrpc_delete_nodes_by_name(self, names):
- return self.delete_objects_by_name(CobblerSystem, names)
-
- @inlineCallbacks
- def xmlrpc_get_distros(self):
- # WARNING: This could return a *huge* number of results. Consider
- # adding filtering options to this function before using it in anger.
- distros = yield CobblerDistro.get_all_values(self.session)
- returnValue(distros)
-
- @inlineCallbacks
- def xmlrpc_get_profiles(self):
- # WARNING: This could return a *huge* number of results. Consider
- # adding filtering options to this function before using it in anger.
- profiles = yield CobblerProfile.get_all_values(self.session)
- returnValue(profiles)
-
- @inlineCallbacks
- def xmlrpc_get_nodes(self):
- # WARNING: This could return a *huge* number of results. Consider
- # adding filtering options to this function before using it in anger.
- systems = yield CobblerSystem.get_all_values(self.session)
- returnValue(systems)
+ ProvisioningAPI.__init__(self, session)
+
+# Add an xmlrpc_* method for each function defined in IProvisioningAPI.
+for name in IProvisioningAPI.names(all=True):
+ method = getattr(ProvisioningAPI, name)
+ setattr(ProvisioningAPI_XMLRPC, "xmlrpc_%s" % name, method)
=== added file 'src/provisioningserver/tests/test_api.py'
--- src/provisioningserver/tests/test_api.py 1970-01-01 00:00:00 +0000
+++ src/provisioningserver/tests/test_api.py 2012-02-08 00:24:06 +0000
@@ -0,0 +1,201 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for `provisioningserver.api`."""
+
+from __future__ import (
+ print_function,
+ unicode_literals,
+ )
+
+__metaclass__ = type
+__all__ = []
+
+from provisioningserver.api import ProvisioningAPI
+from provisioningserver.cobblerclient import CobblerSession
+from provisioningserver.interfaces import IProvisioningAPI
+from provisioningserver.testing.fakecobbler import (
+ FakeCobbler,
+ FakeTwistedProxy,
+ )
+from testtools import TestCase
+from testtools.deferredruntest import AsynchronousDeferredRunTest
+from twisted.internet.defer import inlineCallbacks
+from zope.interface.verify import verifyObject
+
+
+class TestProvisioningAPI(TestCase):
+ """Tests for `provisioningserver.api.ProvisioningAPI`."""
+
+ run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
+
+ def get_cobbler_session(self):
+ cobbler_session = CobblerSession(
+ "http://localhost/does/not/exist";, "user", "password")
+ cobbler_fake = FakeCobbler({"user": "password"})
+ cobbler_proxy = FakeTwistedProxy(cobbler_fake)
+ cobbler_session.proxy = cobbler_proxy
+ return cobbler_session
+
+ def get_provisioning_api(self):
+ cobbler_session = self.get_cobbler_session()
+ return ProvisioningAPI(cobbler_session)
+
+ def test_ProvisioningAPI_interfaces(self):
+ papi = self.get_provisioning_api()
+ verifyObject(IProvisioningAPI, papi)
+
+ @inlineCallbacks
+ def test_add_distro(self):
+ # Create a distro via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ self.assertEqual("distro", distro)
+
+ @inlineCallbacks
+ def test_add_profile(self):
+ # Create a profile via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profile = yield papi.add_profile("profile", distro)
+ self.assertEqual("profile", profile)
+
+ @inlineCallbacks
+ def test_add_node(self):
+ # Create a system/node via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profile = yield papi.add_profile("profile", distro)
+ node = yield papi.add_node("node", profile)
+ self.assertEqual("node", node)
+
+ @inlineCallbacks
+ def test_delete_distros_by_name(self):
+ # Create a distro via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ # Delete it again via the Provisioning API.
+ yield papi.delete_distros_by_name([distro])
+ # It has gone, checked via the Cobbler session.
+ distros = yield papi.get_distros_by_name([distro])
+ self.assertEqual({}, distros)
+
+ @inlineCallbacks
+ def test_delete_profiles_by_name(self):
+ # Create a profile via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profile = yield papi.add_profile("profile", distro)
+ # Delete it again via the Provisioning API.
+ yield papi.delete_profiles_by_name([profile])
+ # It has gone, checked via the Cobbler session.
+ profiles = yield papi.get_profiles_by_name([profile])
+ self.assertEqual({}, profiles)
+
+ @inlineCallbacks
+ def test_delete_nodes_by_name(self):
+ # Create a node via the Provisioning API.
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profile = yield papi.add_profile("profile", distro)
+ node = yield papi.add_node("node", profile)
+ # Delete it again via the Provisioning API.
+ yield papi.delete_nodes_by_name([node])
+ # It has gone, checked via the Cobbler session.
+ nodes = yield papi.get_nodes_by_name([node])
+ self.assertEqual({}, nodes)
+
+ @inlineCallbacks
+ def test_get_distros(self):
+ papi = self.get_provisioning_api()
+ distros = yield papi.get_distros()
+ self.assertEqual({}, distros)
+ # Create some distros via the Provisioning API.
+ expected = {}
+ for num in xrange(3):
+ initrd = self.getUniqueString()
+ kernel = self.getUniqueString()
+ name = self.getUniqueString()
+ yield papi.add_distro(name, initrd, kernel)
+ expected[name] = {
+ "initrd": initrd,
+ "kernel": kernel,
+ "name": name,
+ }
+ distros = yield papi.get_distros()
+ self.assertEqual(expected, distros)
+
+ @inlineCallbacks
+ def test_get_profiles(self):
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profiles = yield papi.get_profiles()
+ self.assertEqual({}, profiles)
+ # Create some profiles via the Provisioning API.
+ expected = {}
+ for num in xrange(3):
+ name = self.getUniqueString()
+ yield papi.add_profile(name, distro)
+ expected[name] = {u'distro': u'distro', u'name': name}
+ profiles = yield papi.get_profiles()
+ self.assertEqual(expected, profiles)
+
+ @inlineCallbacks
+ def test_get_nodes(self):
+ papi = self.get_provisioning_api()
+ distro = yield papi.add_distro(
+ "distro", "an_initrd", "a_kernel")
+ profile = yield papi.add_profile("profile", distro)
+ nodes = yield papi.get_nodes()
+ self.assertEqual({}, nodes)
+ # Create some nodes via the Provisioning API.
+ expected = {}
+ for num in xrange(3):
+ name = self.getUniqueString()
+ yield papi.add_node(name, profile)
+ expected[name] = {'name': name, 'profile': 'profile'}
+ nodes = yield papi.get_nodes()
+ self.assertEqual(expected, nodes)
+
+ @inlineCallbacks
+ def test_get_nodes_by_name(self):
+ papi = self.get_provisioning_api()
+ nodes = yield papi.get_nodes_by_name([])
+ self.assertEqual({}, nodes)
+ # Create a node via the Provisioning API.
+ distro = yield papi.add_distro("distro", "initrd", "kernel")
+ profile = yield papi.add_profile("profile", distro)
+ yield papi.add_node("alice", profile)
+ nodes = yield papi.get_nodes_by_name(["alice", "bob"])
+ # The response contains keys for all systems found.
+ self.assertSequenceEqual(["alice"], sorted(nodes))
+
+ @inlineCallbacks
+ def test_get_distros_by_name(self):
+ papi = self.get_provisioning_api()
+ distros = yield papi.get_distros_by_name([])
+ self.assertEqual({}, distros)
+ # Create a distro via the Provisioning API.
+ yield papi.add_distro("alice", "initrd", "kernel")
+ distros = yield papi.get_distros_by_name(["alice", "bob"])
+ # The response contains keys for all distributions found.
+ self.assertSequenceEqual(["alice"], sorted(distros))
+
+ @inlineCallbacks
+ def test_get_profiles_by_name(self):
+ papi = self.get_provisioning_api()
+ profiles = yield papi.get_profiles_by_name([])
+ self.assertEqual({}, profiles)
+ # Create a profile via the Provisioning API.
+ distro = yield papi.add_distro("distro", "initrd", "kernel")
+ yield papi.add_profile("alice", distro)
+ profiles = yield papi.get_profiles_by_name(["alice", "bob"])
+ # The response contains keys for all profiles found.
+ self.assertSequenceEqual(["alice"], sorted(profiles))
=== modified file 'src/provisioningserver/tests/test_remote.py'
--- src/provisioningserver/tests/test_remote.py 2012-02-08 00:24:06 +0000
+++ src/provisioningserver/tests/test_remote.py 2012-02-08 00:24:06 +0000
@@ -11,185 +11,27 @@
__metaclass__ = type
__all__ = []
-from provisioningserver.cobblerclient import CobblerSession
-from provisioningserver.remote import ProvisioningAPI
-from provisioningserver.testing.fakecobbler import (
- FakeCobbler,
- FakeTwistedProxy,
+from provisioningserver.interfaces import (
+ IProvisioningAPI,
+ IProvisioningAPI_XMLRPC,
)
+from provisioningserver.remote import ProvisioningAPI_XMLRPC
from testtools import TestCase
-from testtools.deferredruntest import AsynchronousDeferredRunTest
-from twisted.internet.defer import inlineCallbacks
-
-
-class TestProvisioningAPI(TestCase):
- """Tests for `provisioningserver.remote.ProvisioningAPI`."""
-
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
-
- def get_cobbler_session(self):
- cobbler_session = CobblerSession(
- "http://localhost/does/not/exist";, "user", "password")
- cobbler_fake = FakeCobbler({"user": "password"})
- cobbler_proxy = FakeTwistedProxy(cobbler_fake)
- cobbler_session.proxy = cobbler_proxy
- return cobbler_session
-
- def get_provisioning_api(self):
- cobbler_session = self.get_cobbler_session()
- return ProvisioningAPI(cobbler_session)
-
- @inlineCallbacks
- def test_add_distro(self):
- # Create a distro via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- self.assertEqual("distro", distro)
-
- @inlineCallbacks
- def test_add_profile(self):
- # Create a profile via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- self.assertEqual("profile", profile)
-
- @inlineCallbacks
- def test_add_node(self):
- # Create a system/node via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- node = yield prov.xmlrpc_add_node("node", profile)
- self.assertEqual("node", node)
-
- @inlineCallbacks
- def test_delete_distros_by_name(self):
- # Create a distro via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- # Delete it again via the Provisioning API.
- yield prov.xmlrpc_delete_distros_by_name([distro])
- # It has gone, checked via the Cobbler session.
- distros = yield prov.xmlrpc_get_distros_by_name([distro])
- self.assertEqual({}, distros)
-
- @inlineCallbacks
- def test_delete_profiles_by_name(self):
- # Create a profile via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- # Delete it again via the Provisioning API.
- yield prov.xmlrpc_delete_profiles_by_name([profile])
- # It has gone, checked via the Cobbler session.
- profiles = yield prov.xmlrpc_get_profiles_by_name([profile])
- self.assertEqual({}, profiles)
-
- @inlineCallbacks
- def test_delete_nodes_by_name(self):
- # Create a node via the Provisioning API.
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- node = yield prov.xmlrpc_add_node("node", profile)
- # Delete it again via the Provisioning API.
- yield prov.xmlrpc_delete_nodes_by_name([node])
- # It has gone, checked via the Cobbler session.
- nodes = yield prov.xmlrpc_get_nodes_by_name([node])
- self.assertEqual({}, nodes)
-
- @inlineCallbacks
- def test_get_distros(self):
- prov = self.get_provisioning_api()
- distros = yield prov.xmlrpc_get_distros()
- self.assertEqual({}, distros)
- # Create some distros via the Provisioning API.
- expected = {}
- for num in xrange(3):
- initrd = self.getUniqueString()
- kernel = self.getUniqueString()
- name = self.getUniqueString()
- yield prov.xmlrpc_add_distro(name, initrd, kernel)
- expected[name] = {
- "initrd": initrd,
- "kernel": kernel,
- "name": name,
- }
- distros = yield prov.xmlrpc_get_distros()
- self.assertEqual(expected, distros)
-
- @inlineCallbacks
- def test_get_profiles(self):
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profiles = yield prov.xmlrpc_get_profiles()
- self.assertEqual({}, profiles)
- # Create some profiles via the Provisioning API.
- expected = {}
- for num in xrange(3):
- name = self.getUniqueString()
- yield prov.xmlrpc_add_profile(name, distro)
- expected[name] = {u'distro': u'distro', u'name': name}
- profiles = yield prov.xmlrpc_get_profiles()
- self.assertEqual(expected, profiles)
-
- @inlineCallbacks
- def test_get_nodes(self):
- prov = self.get_provisioning_api()
- distro = yield prov.xmlrpc_add_distro(
- "distro", "an_initrd", "a_kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- nodes = yield prov.xmlrpc_get_nodes()
- self.assertEqual({}, nodes)
- # Create some nodes via the Provisioning API.
- expected = {}
- for num in xrange(3):
- name = self.getUniqueString()
- yield prov.xmlrpc_add_node(name, profile)
- expected[name] = {'name': name, 'profile': 'profile'}
- nodes = yield prov.xmlrpc_get_nodes()
- self.assertEqual(expected, nodes)
-
- @inlineCallbacks
- def test_get_nodes_by_name(self):
- prov = self.get_provisioning_api()
- nodes = yield prov.xmlrpc_get_nodes_by_name([])
- self.assertEqual({}, nodes)
- # Create a node via the Provisioning API.
- distro = yield prov.xmlrpc_add_distro("distro", "initrd", "kernel")
- profile = yield prov.xmlrpc_add_profile("profile", distro)
- yield prov.xmlrpc_add_node("alice", profile)
- nodes = yield prov.xmlrpc_get_nodes_by_name(["alice", "bob"])
- # The response contains keys for all systems found.
- self.assertSequenceEqual(["alice"], sorted(nodes))
-
- @inlineCallbacks
- def test_get_distros_by_name(self):
- prov = self.get_provisioning_api()
- distros = yield prov.xmlrpc_get_distros_by_name([])
- self.assertEqual({}, distros)
- # Create a distro via the Provisioning API.
- yield prov.xmlrpc_add_distro("alice", "initrd", "kernel")
- distros = yield prov.xmlrpc_get_distros_by_name(["alice", "bob"])
- # The response contains keys for all distributions found.
- self.assertSequenceEqual(["alice"], sorted(distros))
-
- @inlineCallbacks
- def test_get_profiles_by_name(self):
- prov = self.get_provisioning_api()
- profiles = yield prov.xmlrpc_get_profiles_by_name([])
- self.assertEqual({}, profiles)
- # Create a profile via the Provisioning API.
- distro = yield prov.xmlrpc_add_distro("distro", "initrd", "kernel")
- yield prov.xmlrpc_add_profile("alice", distro)
- profiles = yield prov.xmlrpc_get_profiles_by_name(["alice", "bob"])
- # The response contains keys for all profiles found.
- self.assertSequenceEqual(["alice"], sorted(profiles))
+from zope.interface.verify import verifyObject
+
+
+class TestProvisioningAPI_XMLRPC(TestCase):
+ """Tests for `provisioningserver.remote.ProvisioningAPI_XMLRPC`."""
+
+ def test_ProvisioningAPI_init(self):
+ dummy_session = object()
+ papi_xmlrpc = ProvisioningAPI_XMLRPC(dummy_session)
+ self.assertIs(dummy_session, papi_xmlrpc.session)
+ self.assertTrue(papi_xmlrpc.allowNone)
+ self.assertTrue(papi_xmlrpc.useDateTime)
+
+ def test_ProvisioningAPI_interfaces(self):
+ dummy_session = object()
+ papi_xmlrpc = ProvisioningAPI_XMLRPC(dummy_session)
+ verifyObject(IProvisioningAPI, papi_xmlrpc)
+ verifyObject(IProvisioningAPI_XMLRPC, papi_xmlrpc)
Follow ups
