← Back to branch summary

~lutostag/ubuntu/utopic/maas/1.5.2

« back to all changes in this revision

Viewing changes to src/provisioningserver/tests/test_plugin.py

  • Committer: Package Import Robot
  • Author(s): Andres Rodriguez
  • Date: 2012-04-12 16:46:22 UTC
  • mto: (20.1.1 quantal) (1.2.1)
  • mto: This revision was merged to the branch mainline in revision 14.
  • Revision ID: package-import@ubuntu.com-20120412164622-laz1qoxycfrddka0
Tags: upstream-0.1+bzr462+dfsg
Import upstream version 0.1+bzr462+dfsg

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/provisioningserver/tests/test_plugin.py
11
11
__metaclass__ = type
12
12
__all__ = []
13
13
 
 
14
from base64 import b64encode
14
15
from functools import partial
15
16
from getpass import getuser
 
17
import httplib
16
18
import os
 
19
from StringIO import StringIO
 
20
import xmlrpclib
17
21
 
18
22
from fixtures import TempDir
19
23
import formencode
 
24
from maastesting.factory import factory
20
25
from provisioningserver.plugin import (
21
26
    Config,
22
27
    Options,
 
28
    ProvisioningRealm,
23
29
    ProvisioningServiceMaker,
 
30
    SingleUsernamePasswordChecker,
24
31
    )
 
32
from provisioningserver.testing.fakecobbler import make_fake_cobbler_session
25
33
from testtools import TestCase
 
34
from testtools.deferredruntest import (
 
35
    assert_fails_with,
 
36
    AsynchronousDeferredRunTest,
 
37
    )
26
38
from testtools.matchers import (
27
39
    MatchesException,
28
40
    Raises,
29
41
    )
 
42
from twisted.application.internet import TCPServer
30
43
from twisted.application.service import MultiService
 
44
from twisted.cred.credentials import UsernamePassword
 
45
from twisted.cred.error import UnauthorizedLogin
 
46
from twisted.internet.defer import inlineCallbacks
31
47
from twisted.python.usage import UsageError
 
48
from twisted.web.guard import HTTPAuthSessionWrapper
 
49
from twisted.web.resource import IResource
 
50
from twisted.web.server import NOT_DONE_YET
 
51
from twisted.web.test.test_web import DummyRequest
32
52
import yaml
33
53
 
34
54
 
36
56
    """Tests for `provisioningserver.plugin.Config`."""
37
57
 
38
58
    def test_defaults(self):
 
59
        mandatory = {
 
60
            'password': 'killing_joke',
 
61
            }
39
62
        expected = {
40
63
            'broker': {
41
64
                'host': 'localhost',
54
77
                'directory': '',
55
78
                'reporter': '',
56
79
                },
 
80
            'interface': '127.0.0.1',
57
81
            'port': 5241,
 
82
            'username': getuser(),
58
83
            }
59
 
        observed = Config.to_python({})
 
84
        expected.update(mandatory)
 
85
        observed = Config.to_python(mandatory)
60
86
        self.assertEqual(expected, observed)
61
87
 
62
88
    def test_parse(self):
63
89
        # Configuration can be parsed from a snippet of YAML.
64
 
        observed = Config.parse(b'logfile: "/some/where.log"')
 
90
        observed = Config.parse(
 
91
            b'logfile: "/some/where.log"\n'
 
92
            b'password: "black_sabbath"\n'
 
93
            )
65
94
        self.assertEqual("/some/where.log", observed["logfile"])
66
95
 
67
96
    def test_load(self):
69
98
        filename = os.path.join(
70
99
            self.useFixture(TempDir()).path, "config.yaml")
71
100
        with open(filename, "wb") as stream:
72
 
            stream.write(b'logfile: "/some/where.log"')
 
101
            stream.write(b'logfile: "/some/where.log"\n')
 
102
            stream.write(b'password: "megadeth"\n')
73
103
        observed = Config.load(filename)
74
104
        self.assertEqual("/some/where.log", observed["logfile"])
75
105
 
117
147
class TestProvisioningServiceMaker(TestCase):
118
148
    """Tests for `provisioningserver.plugin.ProvisioningServiceMaker`."""
119
149
 
 
150
    run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
 
151
 
120
152
    def setUp(self):
121
153
        super(TestProvisioningServiceMaker, self).setUp()
122
154
        self.tempdir = self.useFixture(TempDir()).path
123
155
 
124
156
    def write_config(self, config):
 
157
        config.setdefault("password", factory.getRandomString())
125
158
        config_filename = os.path.join(self.tempdir, "config.yaml")
126
159
        with open(config_filename, "wb") as stream:
127
160
            yaml.dump(config, stream)
165
198
        self.assertEqual(
166
199
            len(service.namedServices), len(service.services),
167
200
            "Not all services are named.")
 
201
 
 
202
    def test_makeService_api_requires_credentials(self):
 
203
        """
 
204
        The site service's /api resource requires credentials from clients.
 
205
        """
 
206
        options = Options()
 
207
        options["config-file"] = self.write_config({})
 
208
        service_maker = ProvisioningServiceMaker("Harry", "Hill")
 
209
        service = service_maker.makeService(options)
 
210
        self.assertIsInstance(service, MultiService)
 
211
        site_service = service.getServiceNamed("site")
 
212
        self.assertIsInstance(site_service, TCPServer)
 
213
        port, site = site_service.args
 
214
        self.assertIn("api", site.resource.listStaticNames())
 
215
        api = site.resource.getStaticEntity("api")
 
216
        # HTTPAuthSessionWrapper demands credentials from an HTTP request.
 
217
        self.assertIsInstance(api, HTTPAuthSessionWrapper)
 
218
 
 
219
    def exercise_api_credentials(self, config_file, username, password):
 
220
        """
 
221
        Create a new service with :class:`ProvisioningServiceMaker` and
 
222
        attempt to access the API with the given credentials.
 
223
        """
 
224
        options = Options()
 
225
        options["config-file"] = config_file
 
226
        service_maker = ProvisioningServiceMaker("Morecombe", "Wise")
 
227
        # Terminate the service in a fake Cobbler session.
 
228
        service_maker._makeCobblerSession = (
 
229
            lambda config: make_fake_cobbler_session())
 
230
        service = service_maker.makeService(options)
 
231
        port, site = service.getServiceNamed("site").args
 
232
        api = site.resource.getStaticEntity("api")
 
233
        # Create an XML-RPC request with valid credentials.
 
234
        request = DummyRequest([])
 
235
        request.method = "POST"
 
236
        request.content = StringIO(xmlrpclib.dumps((), "get_nodes"))
 
237
        request.prepath = ["api"]
 
238
        request.headers["authorization"] = (
 
239
            "Basic %s" % b64encode(b"%s:%s" % (username, password)))
 
240
        # The credential check and resource rendering is deferred, but
 
241
        # NOT_DONE_YET is returned from render(). The request signals
 
242
        # completion with the aid of notifyFinish().
 
243
        finished = request.notifyFinish()
 
244
        self.assertEqual(NOT_DONE_YET, api.render(request))
 
245
        return finished.addCallback(lambda ignored: request)
 
246
 
 
247
    @inlineCallbacks
 
248
    def test_makeService_api_accepts_valid_credentials(self):
 
249
        """
 
250
        The site service's /api resource accepts valid credentials.
 
251
        """
 
252
        config = {"username": "orange", "password": "goblin"}
 
253
        request = yield self.exercise_api_credentials(
 
254
            self.write_config(config), "orange", "goblin")
 
255
        # A valid XML-RPC response has been written.
 
256
        self.assertEqual(None, request.responseCode)  # None implies 200.
 
257
        xmlrpclib.loads(b"".join(request.written))
 
258
 
 
259
    @inlineCallbacks
 
260
    def test_makeService_api_rejects_invalid_credentials(self):
 
261
        """
 
262
        The site service's /api resource rejects invalid credentials.
 
263
        """
 
264
        config = {"username": "orange", "password": "goblin"}
 
265
        request = yield self.exercise_api_credentials(
 
266
            self.write_config(config), "abigail", "williams")
 
267
        # The request has not been authorized.
 
268
        self.assertEqual(httplib.UNAUTHORIZED, request.responseCode)
 
269
 
 
270
 
 
271
class TestSingleUsernamePasswordChecker(TestCase):
 
272
    """Tests for `SingleUsernamePasswordChecker`."""
 
273
 
 
274
    run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
 
275
 
 
276
    @inlineCallbacks
 
277
    def test_requestAvatarId_okay(self):
 
278
        credentials = UsernamePassword("frank", "zappa")
 
279
        checker = SingleUsernamePasswordChecker("frank", "zappa")
 
280
        avatar = yield checker.requestAvatarId(credentials)
 
281
        self.assertEqual("frank", avatar)
 
282
 
 
283
    def test_requestAvatarId_bad(self):
 
284
        credentials = UsernamePassword("frank", "zappa")
 
285
        checker = SingleUsernamePasswordChecker("zap", "franka")
 
286
        d = checker.requestAvatarId(credentials)
 
287
        return assert_fails_with(d, UnauthorizedLogin)
 
288
 
 
289
 
 
290
class TestProvisioningRealm(TestCase):
 
291
    """Tests for `ProvisioningRealm`."""
 
292
 
 
293
    def test_requestAvatar_okay(self):
 
294
        resource = object()
 
295
        realm = ProvisioningRealm(resource)
 
296
        avatar = realm.requestAvatar(
 
297
            "irrelevant", "also irrelevant", IResource)
 
298
        self.assertEqual((IResource, resource, realm.noop), avatar)
 
299
 
 
300
    def test_requestAvatar_bad(self):
 
301
        # If IResource is not amongst the interfaces passed to requestAvatar,
 
302
        # NotImplementedError is raised.
 
303
        resource = object()
 
304
        realm = ProvisioningRealm(resource)
 
305
        self.assertRaises(
 
306
            NotImplementedError, realm.requestAvatar,
 
307
            "irrelevant", "also irrelevant")