← Back to branch summary

~ubuntu-branches/ubuntu/raring/maas/raring-updates

« back to all changes in this revision

Viewing changes to src/metadataserver/tests/test_api.py

  • Committer: Package Import Robot
  • Author(s): Andres Rodriguez
  • Date: 2012-07-03 17:42:37 UTC
  • mfrom: (1.1.13)
  • Revision ID: package-import@ubuntu.com-20120703174237-p8l0keuuznfg721k
Tags: 0.1+bzr709+dfsg-0ubuntu1
* New Upstream release
* debian/control:
  - Depends on python-celery, python-tempita, libjs-yui3-{full,min},
    libjs-raphael
* debian/maas.install:
  - Install apiclient, celeryconfig.py, maas-import-pxe-files, preseeds_v2.
  - Update to install various files from chroot, rather tha manually copy
    them from the source.
* debian/maas.links: symlink celeryconfig.py
* debian/maas.maas-celery.upstart: Add job.
* debian/rules:
  - Install celery upstart job.
  - Do not install jslibs as packages are now used.
  - Drop copying of maas_local_settings_sample.py as source now ships
    a maas_local_settings.py
* debian/patches:
  - 04-maas-http-fix.patch: Drop. Merged upstream.
  - 01-fix-database-settings.patch: Refreshed.
  - 99_enums_js.patch: Added until creation of enum.js / build process
    is fixed.
* debian/maas.postinst: Update bzr version to correctly handle upgrades.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/metadataserver/tests/test_api.py
16
16
import httplib
17
17
from io import BytesIO
18
18
 
19
 
from maasserver.exceptions import Unauthorized
20
 
from maasserver.models import (
21
 
    NODE_STATUS,
22
 
    SSHKey,
 
19
from django.conf import settings
 
20
from django.core.exceptions import PermissionDenied
 
21
from django.core.urlresolvers import reverse
 
22
from maasserver.enum import NODE_STATUS
 
23
from maasserver.exceptions import (
 
24
    MAASAPINotFound,
 
25
    Unauthorized,
23
26
    )
 
27
from maasserver.models import SSHKey
24
28
from maasserver.provisioning import get_provisioning_api_proxy
25
29
from maasserver.testing import reload_object
26
30
from maasserver.testing.factory import factory
27
31
from maasserver.testing.oauthclient import OAuthAuthenticatedClient
28
 
from maastesting.testcase import TestCase
 
32
from maastesting.djangotestcase import DjangoTestCase
 
33
from metadataserver import api
29
34
from metadataserver.api import (
30
35
    check_version,
 
36
    get_node_for_mac,
31
37
    get_node_for_request,
 
38
    get_queried_node,
32
39
    make_list_response,
33
40
    make_text_response,
34
41
    MetaDataHandler,
43
50
from provisioningserver.testing.factory import ProvisioningFakeFactory
44
51
 
45
52
 
46
 
class TestHelpers(TestCase):
 
53
class TestHelpers(DjangoTestCase):
47
54
    """Tests for the API helper functions."""
48
55
 
49
56
    def fake_request(self, **kwargs):
85
92
            Unauthorized,
86
93
            get_node_for_request, self.fake_request())
87
94
 
88
 
 
89
 
class TestViews(TestCase, ProvisioningFakeFactory):
 
95
    def test_get_node_for_mac_refuses_if_anonymous_access_disabled(self):
 
96
        self.patch(settings, 'ALLOW_UNSAFE_METADATA_ACCESS', False)
 
97
        self.assertRaises(
 
98
            PermissionDenied, get_node_for_mac, factory.getRandomMACAddress())
 
99
 
 
100
    def test_get_node_for_mac_raises_404_for_unknown_mac(self):
 
101
        self.assertRaises(
 
102
            MAASAPINotFound, get_node_for_mac, factory.getRandomMACAddress())
 
103
 
 
104
    def test_get_node_for_mac_finds_node_by_mac(self):
 
105
        mac = factory.make_mac_address()
 
106
        self.assertEqual(mac.node, get_node_for_mac(mac.mac_address))
 
107
 
 
108
    def test_get_queried_node_looks_up_by_mac_if_given(self):
 
109
        mac = factory.make_mac_address()
 
110
        self.assertEqual(
 
111
            mac.node,
 
112
            get_queried_node(object(), for_mac=mac.mac_address))
 
113
 
 
114
    def test_get_queried_node_looks_up_oauth_key_by_default(self):
 
115
        node = factory.make_node()
 
116
        token = NodeKey.objects.get_token_for_node(node)
 
117
        request = self.fake_request(
 
118
            HTTP_AUTHORIZATION=factory.make_oauth_header(
 
119
                oauth_token=token.key))
 
120
        self.assertEqual(node, get_queried_node(request))
 
121
 
 
122
 
 
123
class TestViews(DjangoTestCase, ProvisioningFakeFactory):
90
124
    """Tests for the API views."""
91
125
 
92
126
    def make_node_client(self, node=None):
96
130
        token = NodeKey.objects.get_token_for_node(node)
97
131
        return OAuthAuthenticatedClient(get_node_init_user(), token)
98
132
 
99
 
    def make_url(self, path):
100
 
        """Create an absolute URL for `path` on the metadata API.
101
 
 
102
 
        :param path: Path within the metadata API to access.  Should start
103
 
            with a slash.
104
 
        :return: An absolute URL for the given path within the metadata
105
 
            service tree.
106
 
        """
107
 
        assert path.startswith('/'), "Give absolute metadata API path."
108
 
        # Root of the metadata API service.
109
 
        metadata_root = "/metadata"
110
 
        return metadata_root + path
111
 
 
112
 
    def get(self, path, client=None):
113
 
        """GET a resource from the metadata API.
114
 
 
115
 
        :param path: Path within the metadata API to access.  Should start
116
 
            with a slash.
117
 
        :param token: If given, authenticate the request using this token.
118
 
        :type token: oauth.oauth.OAuthToken
119
 
        :return: A response to the GET request.
120
 
        :rtype: django.http.HttpResponse
121
 
        """
122
 
        if client is None:
123
 
            client = self.client
124
 
        return client.get(self.make_url(path))
125
 
 
126
133
    def call_signal(self, client=None, version='latest', files={}, **kwargs):
127
134
        """Call the API's signal method.
128
135
 
145
152
        for name, content in files.items():
146
153
            params[name] = BytesIO(content)
147
154
            params[name].name = name
148
 
        return client.post(self.make_url('/%s/' % version), params)
 
155
        url = reverse('metadata-version', args=[version])
 
156
        return client.post(url, params)
149
157
 
150
158
    def test_no_anonymous_access(self):
151
 
        self.assertEqual(httplib.UNAUTHORIZED, self.get('/').status_code)
 
159
        self.assertEqual(
 
160
            httplib.UNAUTHORIZED,
 
161
            self.client.get(reverse('metadata')).status_code)
152
162
 
153
163
    def test_metadata_index_shows_latest(self):
154
164
        client = self.make_node_client()
155
 
        self.assertIn('latest', self.get('/', client).content)
 
165
        self.assertIn('latest', client.get(reverse('metadata')).content)
156
166
 
157
167
    def test_metadata_index_shows_only_known_versions(self):
158
168
        client = self.make_node_client()
159
 
        for item in self.get('/', client).content.splitlines():
 
169
        for item in client.get(reverse('metadata')).content.splitlines():
160
170
            check_version(item)
161
171
        # The test is that we get here without exception.
162
172
        pass
163
173
 
164
174
    def test_version_index_shows_meta_data(self):
165
175
        client = self.make_node_client()
166
 
        items = self.get('/latest/', client).content.splitlines()
 
176
        url = reverse('metadata-version', args=['latest'])
 
177
        items = client.get(url).content.splitlines()
167
178
        self.assertIn('meta-data', items)
168
179
 
169
180
    def test_version_index_does_not_show_user_data_if_not_available(self):
170
181
        client = self.make_node_client()
171
 
        items = self.get('/latest/', client).content.splitlines()
 
182
        url = reverse('metadata-version', args=['latest'])
 
183
        items = client.get(url).content.splitlines()
172
184
        self.assertNotIn('user-data', items)
173
185
 
174
186
    def test_version_index_shows_user_data_if_available(self):
175
187
        node = factory.make_node()
176
188
        NodeUserData.objects.set_user_data(node, b"User data for node")
177
189
        client = self.make_node_client(node)
178
 
        items = self.get('/latest/', client).content.splitlines()
 
190
        url = reverse('metadata-version', args=['latest'])
 
191
        items = client.get(url).content.splitlines()
179
192
        self.assertIn('user-data', items)
180
193
 
181
194
    def test_meta_data_view_lists_fields(self):
183
196
        user, _ = factory.make_user_with_keys(n_keys=2, username='my-user')
184
197
        node = factory.make_node(owner=user)
185
198
        client = self.make_node_client(node=node)
186
 
        response = self.get('/latest/meta-data/', client)
 
199
        url = reverse('metadata-meta-data', args=['latest', ''])
 
200
        response = client.get(url)
187
201
        self.assertIn('text/plain', response['Content-Type'])
188
202
        self.assertItemsEqual(
189
203
            MetaDataHandler.fields, response.content.split())
190
204
 
191
205
    def test_meta_data_view_is_sorted(self):
192
206
        client = self.make_node_client()
193
 
        response = self.get('/latest/meta-data/', client)
 
207
        url = reverse('metadata-meta-data', args=['latest', ''])
 
208
        response = client.get(url)
194
209
        attributes = response.content.split()
195
210
        self.assertEqual(sorted(attributes), attributes)
196
211
 
197
212
    def test_meta_data_unknown_item_is_not_found(self):
198
213
        client = self.make_node_client()
199
 
        response = self.get('/latest/meta-data/UNKNOWN-ITEM-HA-HA-HA', client)
 
214
        url = reverse('metadata-meta-data', args=['latest', 'UNKNOWN-ITEM'])
 
215
        response = client.get(url)
200
216
        self.assertEqual(httplib.NOT_FOUND, response.status_code)
201
217
 
202
218
    def test_get_attribute_producer_supports_all_fields(self):
207
223
    def test_meta_data_local_hostname_returns_hostname(self):
208
224
        hostname = factory.getRandomString()
209
225
        client = self.make_node_client(factory.make_node(hostname=hostname))
210
 
        response = self.get('/latest/meta-data/local-hostname', client)
 
226
        url = reverse('metadata-meta-data', args=['latest', 'local-hostname'])
 
227
        response = client.get(url)
211
228
        self.assertEqual(
212
229
            (httplib.OK, hostname),
213
230
            (response.status_code, response.content.decode('ascii')))
216
233
    def test_meta_data_instance_id_returns_system_id(self):
217
234
        node = factory.make_node()
218
235
        client = self.make_node_client(node)
219
 
        response = self.get('/latest/meta-data/instance-id', client)
 
236
        url = reverse('metadata-meta-data', args=['latest', 'instance-id'])
 
237
        response = client.get(url)
220
238
        self.assertEqual(
221
239
            (httplib.OK, node.system_id),
222
240
            (response.status_code, response.content.decode('ascii')))
227
245
        node = factory.make_node()
228
246
        NodeUserData.objects.set_user_data(node, data)
229
247
        client = self.make_node_client(node)
230
 
        response = self.get('/latest/user-data', client)
 
248
        response = client.get(reverse('metadata-user-data', args=['latest']))
231
249
        self.assertEqual('application/octet-stream', response['Content-Type'])
232
250
        self.assertIsInstance(response.content, str)
233
251
        self.assertEqual(
234
252
            (httplib.OK, data), (response.status_code, response.content))
235
253
 
236
254
    def test_user_data_for_node_without_user_data_returns_not_found(self):
237
 
        response = self.get('/latest/user-data', self.make_node_client())
 
255
        client = self.make_node_client()
 
256
        response = client.get(reverse('metadata-user-data', args=['latest']))
238
257
        self.assertEqual(httplib.NOT_FOUND, response.status_code)
239
258
 
240
259
    def test_public_keys_not_listed_for_node_without_public_keys(self):
241
 
        response = self.get('/latest/meta-data/', self.make_node_client())
 
260
        url = reverse('metadata-meta-data', args=['latest', ''])
 
261
        client = self.make_node_client()
 
262
        response = client.get(url)
242
263
        self.assertNotIn(
243
264
            'public-keys', response.content.decode('ascii').split('\n'))
244
265
 
245
266
    def test_public_keys_listed_for_node_with_public_keys(self):
246
267
        user, _ = factory.make_user_with_keys(n_keys=2, username='my-user')
247
268
        node = factory.make_node(owner=user)
248
 
        response = self.get(
249
 
            '/latest/meta-data/', self.make_node_client(node=node))
 
269
        url = reverse('metadata-meta-data', args=['latest', ''])
 
270
        client = self.make_node_client(node=node)
 
271
        response = client.get(url)
250
272
        self.assertIn(
251
273
            'public-keys', response.content.decode('ascii').split('\n'))
252
274
 
253
275
    def test_public_keys_for_node_without_public_keys_returns_not_found(self):
254
 
        response = self.get(
255
 
            '/latest/meta-data/public-keys', self.make_node_client())
 
276
        url = reverse('metadata-meta-data', args=['latest', 'public-keys'])
 
277
        client = self.make_node_client()
 
278
        response = client.get(url)
256
279
        self.assertEqual(httplib.NOT_FOUND, response.status_code)
257
280
 
258
281
    def test_public_keys_for_node_returns_list_of_keys(self):
259
282
        user, _ = factory.make_user_with_keys(n_keys=2, username='my-user')
260
283
        node = factory.make_node(owner=user)
261
 
        response = self.get(
262
 
            '/latest/meta-data/public-keys', self.make_node_client(node=node))
 
284
        url = reverse('metadata-meta-data', args=['latest', 'public-keys'])
 
285
        client = self.make_node_client(node=node)
 
286
        response = client.get(url)
263
287
        self.assertEqual(httplib.OK, response.status_code)
264
288
        keys = SSHKey.objects.filter(user=user).values_list('key', flat=True)
265
289
        expected_response = '\n'.join(keys)
288
312
    def test_signaling_requires_status_code(self):
289
313
        node = factory.make_node(status=NODE_STATUS.COMMISSIONING)
290
314
        client = self.make_node_client(node=node)
291
 
        response = client.post(self.make_url('/latest/'), {'op': 'signal'})
 
315
        url = reverse('metadata-version', args=['latest'])
 
316
        response = client.post(url, {'op': 'signal'})
292
317
        self.assertEqual(httplib.BAD_REQUEST, response.status_code)
293
318
 
294
319
    def test_signaling_rejects_unknown_status_code(self):
470
495
        stored_data = NodeCommissionResult.objects.get_data(
471
496
            node, 'output.txt')
472
497
        self.assertEqual(size_limit, len(stored_data))
 
498
 
 
499
    def test_api_retrieves_node_metadata_by_mac(self):
 
500
        mac = factory.make_mac_address()
 
501
        url = reverse(
 
502
            'metadata-meta-data-by-mac',
 
503
            args=['latest', mac.mac_address, 'instance-id'])
 
504
        response = self.client.get(url)
 
505
        self.assertEqual(
 
506
            (httplib.OK, mac.node.system_id),
 
507
            (response.status_code, response.content))
 
508
 
 
509
    def test_api_retrieves_node_userdata_by_mac(self):
 
510
        mac = factory.make_mac_address()
 
511
        user_data = factory.getRandomString().encode('ascii')
 
512
        NodeUserData.objects.set_user_data(mac.node, user_data)
 
513
        url = reverse(
 
514
            'metadata-user-data-by-mac', args=['latest', mac.mac_address])
 
515
        response = self.client.get(url)
 
516
        self.assertEqual(
 
517
            (httplib.OK, user_data),
 
518
            (response.status_code, response.content))
 
519
 
 
520
    def test_api_normally_disallows_anonymous_node_metadata_access(self):
 
521
        self.patch(settings, 'ALLOW_UNSAFE_METADATA_ACCESS', False)
 
522
        mac = factory.make_mac_address()
 
523
        url = reverse(
 
524
            'metadata-meta-data-by-mac',
 
525
            args=['latest', mac.mac_address, 'instance-id'])
 
526
        response = self.client.get(url)
 
527
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
528
 
 
529
    def test_netboot_off(self):
 
530
        node = factory.make_node(netboot=True)
 
531
        client = self.make_node_client(node=node)
 
532
        url = reverse('metadata-version', args=['latest'])
 
533
        response = client.post(url, {'op': 'netboot_off'})
 
534
        node = reload_object(node)
 
535
        self.assertFalse(node.netboot, response)
 
536
 
 
537
    def test_netboot_on(self):
 
538
        node = factory.make_node(netboot=False)
 
539
        client = self.make_node_client(node=node)
 
540
        url = reverse('metadata-version', args=['latest'])
 
541
        response = client.post(url, {'op': 'netboot_on'})
 
542
        node = reload_object(node)
 
543
        self.assertTrue(node.netboot, response)
 
544
 
 
545
    def test_anonymous_netboot_off(self):
 
546
        node = factory.make_node(netboot=True)
 
547
        anon_netboot_off_url = reverse(
 
548
            'metadata-node-by-id', args=['latest', node.system_id])
 
549
        response = self.client.post(
 
550
            anon_netboot_off_url, {'op': 'netboot_off'})
 
551
        node = reload_object(node)
 
552
        self.assertEqual(
 
553
            (httplib.OK, False),
 
554
            (response.status_code, node.netboot),
 
555
            response)
 
556
 
 
557
    def test_anonymous_get_enlist_preseed(self):
 
558
        # The preseed for enlistment can be obtained anonymously.
 
559
        anon_enlist_preseed_url = reverse(
 
560
            'metadata-enlist-preseed', args=['latest'])
 
561
        # Fake the preseed so we're just exercising the view.
 
562
        fake_preseed = factory.getRandomString()
 
563
        self.patch(api, "get_enlist_preseed", lambda: fake_preseed)
 
564
        response = self.client.get(
 
565
            anon_enlist_preseed_url, {'op': 'get_enlist_preseed'})
 
566
        self.assertEqual(
 
567
            (httplib.OK,
 
568
             "text/plain",
 
569
             fake_preseed),
 
570
            (response.status_code,
 
571
             response["Content-Type"],
 
572
             response.content),
 
573
            response)
 
574
 
 
575
    def test_anonymous_get_preseed(self):
 
576
        # The preseed for a node can be obtained anonymously.
 
577
        node = factory.make_node()
 
578
        anon_node_url = reverse(
 
579
            'metadata-node-by-id',
 
580
            args=['latest', node.system_id])
 
581
        # Fake the preseed so we're just exercising the view.
 
582
        fake_preseed = factory.getRandomString()
 
583
        self.patch(api, "get_preseed", lambda node: fake_preseed)
 
584
        response = self.client.get(
 
585
            anon_node_url, {'op': 'get_preseed'})
 
586
        self.assertEqual(
 
587
            (httplib.OK,
 
588
             "text/plain",
 
589
             fake_preseed),
 
590
            (response.status_code,
 
591
             response["Content-Type"],
 
592
             response.content),
 
593
            response)