← Back to branch summary

~lutostag/ubuntu/utopic/maas/1.5.2

« back to all changes in this revision

Viewing changes to src/maasserver/tests/test_api.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/maasserver/tests/test_api.py
45
45
from maasserver.testing.factory import factory
46
46
from maasserver.testing.oauthclient import OAuthAuthenticatedClient
47
47
from maasserver.testing.testcase import (
 
48
    AdminLoggedInTestCase,
48
49
    LoggedInTestCase,
49
50
    TestCase,
50
51
    )
94
95
            extract_constraints(QueryDict('name=%s' % name)))
95
96
 
96
97
 
97
 
class AnonymousEnlistmentAPITest(APIv10TestMixin, TestCase):
98
 
    # Nodes can be enlisted anonymously.
 
98
class EnlistmentAPITest(APIv10TestMixin):
 
99
    # This is a mixin containing enlistement tests.  We will run this for:
 
100
    # an anonymous user, a simple (non-admin) user and an admin user.
 
101
    # XXX: rvb 2012-04-10 bug=978035: It would be better to use
 
102
    # testscenarios for this.
99
103
 
100
104
    def test_POST_new_creates_node(self):
101
105
        # The API allows a Node to be created.
116
120
        self.assertEqual('diane', parsed_result['hostname'])
117
121
        self.assertNotEqual(0, len(parsed_result.get('system_id')))
118
122
        [diane] = Node.objects.filter(hostname='diane')
119
 
        self.assertEqual(2, diane.after_commissioning_action)
 
123
        # XXX JeroenVermeulen 2012-04-12, bug=979539: re-enable.
 
124
        #self.assertEqual(2, diane.after_commissioning_action)
120
125
        self.assertEqual(architecture, diane.architecture)
121
126
 
122
 
    def test_POST_new_anonymous_creates_node_in_declared_state(self):
123
 
        # Upon anonymous enlistment, a node goes into the Declared
124
 
        # state.  Deliberate approval is required before we start
125
 
        # reinstalling the system, wiping its disks etc.
126
 
        response = self.client.post(
127
 
            self.get_uri('nodes/'),
128
 
            {
129
 
                'op': 'new',
130
 
                'hostname': factory.getRandomString(),
131
 
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
132
 
                'after_commissioning_action': '2',
133
 
                'mac_addresses': ['aa:bb:cc:dd:ee:ff'],
134
 
            })
135
 
        self.assertEqual(httplib.OK, response.status_code)
136
 
        system_id = json.loads(response.content)['system_id']
137
 
        self.assertEqual(
138
 
            NODE_STATUS.DECLARED,
139
 
            Node.objects.get(system_id=system_id).status)
140
 
 
141
127
    def test_POST_new_power_type_defaults_to_asking_config(self):
142
128
        architecture = factory.getRandomChoice(ARCHITECTURE_CHOICES)
143
129
        response = self.client.post(
194
180
            system_id=json.loads(response.content)['system_id'])
195
181
        self.assertEqual('node-aabbccddeeff.local', node.hostname)
196
182
 
197
 
    def test_POST_returns_limited_fields(self):
198
 
        architecture = factory.getRandomChoice(ARCHITECTURE_CHOICES)
199
 
        response = self.client.post(
200
 
            self.get_uri('nodes/'),
201
 
            {
202
 
                'op': 'new',
203
 
                'hostname': 'diane',
204
 
                'architecture': architecture,
205
 
                'after_commissioning_action': '2',
206
 
                'mac_addresses': ['aa:bb:cc:dd:ee:ff', '22:bb:cc:dd:ee:ff'],
207
 
            })
208
 
        parsed_result = json.loads(response.content)
209
 
        self.assertItemsEqual(
210
 
            [
211
 
                'hostname', 'system_id', 'macaddress_set', 'architecture',
212
 
                'status'
213
 
            ],
214
 
            list(parsed_result))
215
 
 
216
183
    def test_POST_fails_without_operation(self):
217
184
        # If there is no operation ('op=operation_name') specified in the
218
185
        # request data, a 'Bad request' response is returned.
259
226
            self.fail("post_save should not have been called")
260
227
 
261
228
        post_save.connect(node_created, sender=Node)
 
229
        self.addCleanup(post_save.disconnect, node_created, sender=Node)
262
230
        self.client.post(
263
231
            self.get_uri('nodes/'),
264
232
            {
318
286
        self.assertIn('application/json', response['Content-Type'])
319
287
        self.assertItemsEqual(['architecture'], parsed_result)
320
288
 
 
289
 
 
290
class NonAdminEnlistmentAPITest(EnlistmentAPITest):
 
291
    # This is a mixin containing enlistement tests for non-admin users.
 
292
 
 
293
    def test_POST_non_admin_creates_node_in_declared_state(self):
 
294
        # Upon non-admin enlistment, a node goes into the Declared
 
295
        # state.  Deliberate approval is required before we start
 
296
        # reinstalling the system, wiping its disks etc.
 
297
        response = self.client.post(
 
298
            self.get_uri('nodes/'),
 
299
            {
 
300
                'op': 'new',
 
301
                'hostname': factory.getRandomString(),
 
302
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
 
303
                'after_commissioning_action': '2',
 
304
                'mac_addresses': ['aa:bb:cc:dd:ee:ff'],
 
305
            })
 
306
        self.assertEqual(httplib.OK, response.status_code)
 
307
        system_id = json.loads(response.content)['system_id']
 
308
        self.assertEqual(
 
309
            NODE_STATUS.DECLARED,
 
310
            Node.objects.get(system_id=system_id).status)
 
311
 
 
312
 
 
313
class AnonymousEnlistmentAPITest(NonAdminEnlistmentAPITest, TestCase):
 
314
    # This is an actual test case that uses the NonAdminEnlistmentAPITest
 
315
    # mixin and adds enlistement tests specific to anonymous users.
 
316
 
321
317
    def test_POST_accept_not_allowed(self):
322
318
        # An anonymous user is not allowed to accept an anonymously
323
319
        # enlisted node.  That would defeat the whole purpose of holding
329
325
            (httplib.UNAUTHORIZED, "You must be logged in to accept nodes."),
330
326
            (response.status_code, response.content))
331
327
 
 
328
    def test_POST_returns_limited_fields(self):
 
329
        response = self.client.post(
 
330
            self.get_uri('nodes/'),
 
331
            {
 
332
                'op': 'new',
 
333
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
 
334
                'hostname': factory.getRandomString(),
 
335
                'after_commissioning_action': '2',
 
336
                'mac_addresses': ['aa:bb:cc:dd:ee:ff', '22:bb:cc:dd:ee:ff'],
 
337
            })
 
338
        parsed_result = json.loads(response.content)
 
339
        self.assertItemsEqual(
 
340
            [
 
341
                'hostname', 'system_id', 'macaddress_set', 'architecture',
 
342
                'status',
 
343
            ],
 
344
            list(parsed_result))
 
345
 
 
346
 
 
347
class SimpleUserLoggedInEnlistmentAPITest(NonAdminEnlistmentAPITest,
 
348
                                          LoggedInTestCase):
 
349
    # This is an actual test case that uses the NonAdminEnlistmentAPITest
 
350
    # mixin plus enlistement tests specific to simple (non-admin) users.
 
351
 
 
352
    def test_POST_accept_not_allowed(self):
 
353
        # An non-admin user is not allowed to accept an anonymously
 
354
        # enlisted node.  That would defeat the whole purpose of holding
 
355
        # those nodes for approval.
 
356
        node_id = factory.make_node(status=NODE_STATUS.DECLARED).system_id
 
357
        response = self.client.post(
 
358
            self.get_uri('nodes/'), {'op': 'accept', 'nodes': [node_id]})
 
359
        self.assertEqual(
 
360
            (httplib.FORBIDDEN,
 
361
                "You don't have the required permission to accept the "
 
362
                "following node(s): %s." % node_id),
 
363
            (response.status_code, response.content))
 
364
 
 
365
    def test_POST_returns_limited_fields(self):
 
366
        response = self.client.post(
 
367
            self.get_uri('nodes/'),
 
368
            {
 
369
                'op': 'new',
 
370
                'hostname': factory.getRandomString(),
 
371
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
 
372
                'after_commissioning_action': '2',
 
373
                'mac_addresses': ['aa:bb:cc:dd:ee:ff', '22:bb:cc:dd:ee:ff'],
 
374
            })
 
375
        parsed_result = json.loads(response.content)
 
376
        self.assertItemsEqual(
 
377
            [
 
378
                'hostname', 'system_id', 'macaddress_set', 'architecture',
 
379
                'status', 'resource_uri',
 
380
            ],
 
381
            list(parsed_result))
 
382
 
 
383
 
 
384
class AdminLoggedInEnlistmentAPITest(EnlistmentAPITest,
 
385
                                     AdminLoggedInTestCase):
 
386
    # This is an actual test case that uses the EnlistmentAPITest mixin
 
387
    # and adds enlistement tests specific to admin users.
 
388
 
 
389
    def test_POST_admin_creates_node_in_commissioning_state(self):
 
390
        # When an admin user enlists a node, it goes into the
 
391
        # Commissioning state.
 
392
        response = self.client.post(
 
393
            self.get_uri('nodes/'),
 
394
            {
 
395
                'op': 'new',
 
396
                'hostname': factory.getRandomString(),
 
397
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
 
398
                'after_commissioning_action': '2',
 
399
                'mac_addresses': ['aa:bb:cc:dd:ee:ff'],
 
400
            })
 
401
        self.assertEqual(httplib.OK, response.status_code)
 
402
        system_id = json.loads(response.content)['system_id']
 
403
        self.assertEqual(
 
404
            NODE_STATUS.COMMISSIONING,
 
405
            Node.objects.get(system_id=system_id).status)
 
406
 
 
407
    def test_POST_returns_limited_fields(self):
 
408
        response = self.client.post(
 
409
            self.get_uri('nodes/'),
 
410
            {
 
411
                'op': 'new',
 
412
                'hostname': factory.getRandomString(),
 
413
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
 
414
                'after_commissioning_action': '2',
 
415
                'mac_addresses': ['aa:bb:cc:dd:ee:ff', '22:bb:cc:dd:ee:ff'],
 
416
            })
 
417
        parsed_result = json.loads(response.content)
 
418
        self.assertItemsEqual(
 
419
            [
 
420
                'hostname', 'system_id', 'macaddress_set', 'architecture',
 
421
                'status', 'resource_uri',
 
422
            ],
 
423
            list(parsed_result))
 
424
 
 
425
 
 
426
class AnonymousIsRegisteredAPITest(APIv10TestMixin, TestCase):
 
427
 
 
428
    def test_is_registered_returns_True_if_node_registered(self):
 
429
        mac_address = factory.getRandomMACAddress()
 
430
        factory.make_mac_address(mac_address)
 
431
        response = self.client.get(
 
432
            self.get_uri('nodes/'),
 
433
            {'op': 'is_registered', 'mac_address': mac_address})
 
434
        self.assertEqual(
 
435
            (httplib.OK, "true"),
 
436
            (response.status_code, response.content))
 
437
 
 
438
    def test_is_registered_returns_False_if_mac_registered_node_retired(self):
 
439
        mac_address = factory.getRandomMACAddress()
 
440
        mac = factory.make_mac_address(mac_address)
 
441
        mac.node.status = NODE_STATUS.RETIRED
 
442
        mac.node.save()
 
443
        response = self.client.get(
 
444
            self.get_uri('nodes/'),
 
445
            {'op': 'is_registered', 'mac_address': mac_address})
 
446
        self.assertEqual(
 
447
            (httplib.OK, "false"),
 
448
            (response.status_code, response.content))
 
449
 
 
450
    def test_is_registered_normalizes_mac_address(self):
 
451
        # These two non-normalized MAC Addresses are the same.
 
452
        non_normalized_mac_address = 'AA-bb-cc-dd-ee-ff'
 
453
        non_normalized_mac_address2 = 'aabbccddeeff'
 
454
        factory.make_mac_address(non_normalized_mac_address)
 
455
        response = self.client.get(
 
456
            self.get_uri('nodes/'),
 
457
            {
 
458
                'op': 'is_registered',
 
459
                'mac_address': non_normalized_mac_address2
 
460
            })
 
461
        self.assertEqual(
 
462
            (httplib.OK, "true"),
 
463
            (response.status_code, response.content))
 
464
 
 
465
    def test_is_registered_returns_False_if_node_not_registered(self):
 
466
        mac_address = factory.getRandomMACAddress()
 
467
        response = self.client.get(
 
468
            self.get_uri('nodes/'),
 
469
            {'op': 'is_registered', 'mac_address': mac_address})
 
470
        self.assertEqual(
 
471
            (httplib.OK, "false"),
 
472
            (response.status_code, response.content))
 
473
 
332
474
 
333
475
class NodeAnonAPITest(APIv10TestMixin, TestCase):
334
476
 
335
477
    def test_anon_nodes_GET(self):
336
 
        # Anonymous requests to the API are denied.
 
478
        # Anonymous requests to the API without a specified operation
 
479
        # get a "Bad Request" response.
337
480
        response = self.client.get(self.get_uri('nodes/'))
338
481
 
339
 
        self.assertEqual(httplib.UNAUTHORIZED, response.status_code)
 
482
        self.assertEqual(httplib.BAD_REQUEST, response.status_code)
340
483
 
341
484
    def test_anon_api_doc(self):
342
485
        # The documentation is accessible to anon users.
729
872
            NODE_STATUS.DECLARED,
730
873
            Node.objects.get(system_id=system_id).status)
731
874
 
732
 
    def test_POST_new_when_admin_creates_node_in_ready_state(self):
733
 
        # When an admin user enlists a node, it goes into the Ready state.
734
 
        # This will change once we start doing proper commissioning.
735
 
        self.become_admin()
736
 
        response = self.client.post(
737
 
            self.get_uri('nodes/'),
738
 
            {
739
 
                'op': 'new',
740
 
                'hostname': factory.getRandomString(),
741
 
                'architecture': factory.getRandomChoice(ARCHITECTURE_CHOICES),
742
 
                'after_commissioning_action': '2',
743
 
                'mac_addresses': ['aa:bb:cc:dd:ee:ff'],
744
 
            })
745
 
        self.assertEqual(httplib.OK, response.status_code)
746
 
        system_id = json.loads(response.content)['system_id']
747
 
        self.assertEqual(
748
 
            NODE_STATUS.READY, Node.objects.get(system_id=system_id).status)
749
 
 
750
875
    def test_GET_list_lists_nodes(self):
751
876
        # The api allows for fetching the list of Nodes.
752
877
        node1 = factory.make_node()
997
1122
        # This will change when we add provisioning.  Until then,
998
1123
        # acceptance gets a node straight to Ready state.
999
1124
        self.become_admin()
1000
 
        target_state = NODE_STATUS.READY
 
1125
        target_state = NODE_STATUS.COMMISSIONING
1001
1126
 
1002
1127
        node = factory.make_node(status=NODE_STATUS.DECLARED)
1003
1128
        response = self.client.post(
1075
1200
        # This will change when we add provisioning.  Until then,
1076
1201
        # acceptance gets a node straight to Ready state.
1077
1202
        self.become_admin()
1078
 
        target_state = NODE_STATUS.READY
 
1203
        target_state = NODE_STATUS.COMMISSIONING
1079
1204
 
1080
1205
        nodes = [
1081
1206
            factory.make_node(status=NODE_STATUS.DECLARED)