← Back to branch summary

~lutostag/ubuntu/trusty/maas/1.5.4+keystone

« 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: 2013-03-04 11:49:44 UTC
  • mto: This revision was merged to the branch mainline in revision 25.
  • Revision ID: package-import@ubuntu.com-20130304114944-azcvu9anlf8mizpa
Tags: upstream-1.3+bzr1452+dfsg
Import upstream version 1.3+bzr1452+dfsg

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/maasserver/tests/test_api.py
20
20
    b64decode,
21
21
    b64encode,
22
22
    )
23
 
from collections import namedtuple
24
23
from cStringIO import StringIO
25
24
from datetime import (
26
25
    datetime,
65
64
    describe,
66
65
    DISPLAYED_NODEGROUP_FIELDS,
67
66
    extract_constraints,
68
 
    extract_oauth_key,
69
 
    extract_oauth_key_from_auth_header,
70
67
    find_nodegroup_for_pxeconfig_request,
71
 
    get_oauth_token,
72
 
    get_overrided_query_dict,
73
68
    get_owned_file_or_404,
74
69
    store_node_power_parameters,
75
70
    )
84
79
    NODEGROUP_STATUS,
85
80
    NODEGROUPINTERFACE_MANAGEMENT,
86
81
    )
87
 
from maasserver.exceptions import (
88
 
    MAASAPIBadRequest,
89
 
    Unauthorized,
90
 
    )
 
82
from maasserver.exceptions import MAASAPIBadRequest
91
83
from maasserver.fields import mac_error_msg
92
84
from maasserver.forms import DEFAULT_ZONE_NAME
93
85
from maasserver.models import (
98
90
    MACAddress,
99
91
    Node,
100
92
    NodeGroup,
 
93
    nodegroup as nodegroup_module,
101
94
    NodeGroupInterface,
102
95
    Tag,
103
96
    )
135
128
from maastesting.matchers import ContainsAll
136
129
from maastesting.utils import sample_binary_data
137
130
from metadataserver.models import (
 
131
    CommissioningScript,
138
132
    NodeKey,
139
133
    NodeUserData,
140
134
    )
189
183
 
190
184
class TestModuleHelpers(TestCase):
191
185
 
192
 
    def make_fake_request(self, auth_header):
193
 
        """Create a very simple fake request, with just an auth header."""
194
 
        FakeRequest = namedtuple('FakeRequest', ['META'])
195
 
        return FakeRequest(META={'HTTP_AUTHORIZATION': auth_header})
196
 
 
197
 
    def test_extract_oauth_key_from_auth_header_returns_key(self):
198
 
        token = factory.getRandomString(18)
199
 
        self.assertEqual(
200
 
            token,
201
 
            extract_oauth_key_from_auth_header(
202
 
                factory.make_oauth_header(oauth_token=token)))
203
 
 
204
 
    def test_extract_oauth_key_from_auth_header_returns_None_if_missing(self):
205
 
        self.assertIs(None, extract_oauth_key_from_auth_header(''))
206
 
 
207
 
    def test_extract_oauth_key_raises_Unauthorized_if_no_auth_header(self):
208
 
        self.assertRaises(
209
 
            Unauthorized,
210
 
            extract_oauth_key, self.make_fake_request(None))
211
 
 
212
 
    def test_extract_oauth_key_raises_Unauthorized_if_no_key(self):
213
 
        self.assertRaises(
214
 
            Unauthorized,
215
 
            extract_oauth_key, self.make_fake_request(''))
216
 
 
217
 
    def test_extract_oauth_key_returns_key(self):
218
 
        token = factory.getRandomString(18)
219
 
        self.assertEqual(
220
 
            token,
221
 
            extract_oauth_key(self.make_fake_request(
222
 
                factory.make_oauth_header(oauth_token=token))))
223
 
 
224
 
    def test_get_oauth_token_finds_token(self):
225
 
        user = factory.make_user()
226
 
        consumer, token = user.get_profile().create_authorisation_token()
227
 
        self.assertEqual(
228
 
            token,
229
 
            get_oauth_token(
230
 
                self.make_fake_request(
231
 
                    factory.make_oauth_header(oauth_token=token.key))))
232
 
 
233
 
    def test_get_oauth_token_raises_Unauthorized_for_unknown_token(self):
234
 
        fake_token = factory.getRandomString(18)
235
 
        header = factory.make_oauth_header(oauth_token=fake_token)
236
 
        self.assertRaises(
237
 
            Unauthorized,
238
 
            get_oauth_token, self.make_fake_request(header))
239
 
 
240
186
    def test_extract_constraints_ignores_unknown_parameters(self):
241
187
        unknown_parameter = "%s=%s" % (
242
188
            factory.getRandomString(),
251
197
            {'hostname': name},
252
198
            extract_constraints(QueryDict('name=%s' % name)))
253
199
 
254
 
    def test_get_overrided_query_dict_returns_QueryDict(self):
255
 
        defaults = {factory.getRandomString(): factory.getRandomString()}
256
 
        results = get_overrided_query_dict(defaults, QueryDict(''))
257
 
        expected_results = QueryDict('').copy()
258
 
        expected_results.update(defaults)
259
 
        self.assertEqual(expected_results, results)
260
 
 
261
 
    def test_get_overrided_query_dict_values_in_data_replaces_defaults(self):
262
 
        key = factory.getRandomString()
263
 
        defaults = {key: factory.getRandomString()}
264
 
        data_value = factory.getRandomString()
265
 
        data = {key: data_value}
266
 
        results = get_overrided_query_dict(defaults, data)
267
 
        self.assertEqual([data_value], results.getlist(key))
268
 
 
269
200
 
270
201
class TestAuthentication(APIv10TestMixin, TestCase):
271
202
    """Tests for `maasserver.api_auth`."""
2368
2299
                })
2369
2300
        # Awkward parsing, but the order may vary and it's not JSON
2370
2301
        s = response.content
2371
 
        returned_ids = s[s.find(':')+2:s.rfind('.')].split(', ')
 
2302
        returned_ids = s[s.find(':') + 2:s.rfind('.')].split(', ')
2372
2303
        self.assertEqual(httplib.BAD_REQUEST, response.status_code)
2373
2304
        self.assertIn("Unknown node(s): ", response.content)
2374
2305
        self.assertItemsEqual(node_ids, returned_ids)
2384
2315
        # And one with no owner
2385
2316
        another_node = factory.make_node(status=NODE_STATUS.RESERVED)
2386
2317
        node_ids.add(another_node.system_id)
2387
 
        
2388
2318
        response = self.client.post(
2389
2319
            self.get_uri('nodes/'), {
2390
2320
                'op': 'release',
2395
2325
                "You don't have the required permission to release the "
2396
2326
                "following node(s): %s." % another_node.system_id),
2397
2327
            (response.status_code, response.content))
2398
 
        
 
2328
 
2399
2329
    def test_POST_release_rejects_impossible_state_changes(self):
2400
2330
        acceptable_states = {
2401
2331
            NODE_STATUS.ALLOCATED,
2419
2349
            for node in nodes
2420
2350
            if node.status not in acceptable_states]
2421
2351
        s = response.content
2422
 
        returned = s[s.rfind(':')+2:s.rfind('.')].split(', ') 
 
2352
        returned = s[s.rfind(':') + 2:s.rfind('.')].split(', ')
2423
2353
        self.assertEqual(httplib.CONFLICT, response.status_code)
2424
2354
        self.assertIn(
2425
2355
            "Node(s) cannot be released in their current state:",
2697
2627
        self.tmpdir = os.path.join(media_root, "testing")
2698
2628
        os.mkdir(self.tmpdir)
2699
2629
 
2700
 
    def make_upload_file(self, name=None, contents=None):
2701
 
        """Make a temp upload file named `name` with contents `contents`.
2702
 
 
2703
 
        :return: The full file path of the file that was created.
2704
 
        """
2705
 
        return factory.make_file(
2706
 
            location=self.tmpdir, name=name, contents=contents)
2707
 
 
2708
2630
    def _create_API_params(self, op=None, filename=None, fileObj=None):
2709
2631
        params = {}
2710
2632
        if op is not None:
2811
2733
class FileStorageAPITest(FileStorageAPITestMixin, APITestCase):
2812
2734
 
2813
2735
    def test_add_file_succeeds(self):
2814
 
        filepath = self.make_upload_file()
2815
 
 
2816
 
        with open(filepath) as f:
2817
 
            response = self.make_API_POST_request("add", "foo", f)
2818
 
 
 
2736
        response = self.make_API_POST_request(
 
2737
            "add", factory.make_name('upload'), factory.make_file_upload())
2819
2738
        self.assertEqual(httplib.CREATED, response.status_code)
2820
2739
 
2821
2740
    def test_add_file_fails_with_no_filename(self):
2822
 
        filepath = self.make_upload_file()
2823
 
 
2824
 
        with open(filepath) as f:
2825
 
            response = self.make_API_POST_request("add", fileObj=f)
2826
 
 
 
2741
        response = self.make_API_POST_request(
 
2742
            "add", fileObj=factory.make_file_upload())
2827
2743
        self.assertEqual(httplib.BAD_REQUEST, response.status_code)
2828
2744
        self.assertIn('text/plain', response['Content-Type'])
2829
2745
        self.assertEqual("Filename not supplied", response.content)
2836
2752
        self.assertEqual("File not supplied", response.content)
2837
2753
 
2838
2754
    def test_add_file_fails_with_too_many_files(self):
2839
 
        filepath = self.make_upload_file(name="foo")
2840
 
        filepath2 = self.make_upload_file(name="foo2")
 
2755
        foo = factory.make_file_upload(name='foo')
 
2756
        foo2 = factory.make_file_upload(name='foo2')
2841
2757
 
2842
 
        with open(filepath) as f, open(filepath2) as f2:
2843
 
            response = self.client.post(
2844
 
                self.get_uri('files/'),
2845
 
                {
2846
 
                    "op": "add",
2847
 
                    "filename": "foo",
2848
 
                    "file": f,
2849
 
                    "file2": f2,
2850
 
                })
 
2758
        response = self.client.post(
 
2759
            self.get_uri('files/'),
 
2760
            {
 
2761
                "op": "add",
 
2762
                "filename": "foo",
 
2763
                "file": foo,
 
2764
                "file2": foo2,
 
2765
            })
2851
2766
 
2852
2767
        self.assertEqual(httplib.BAD_REQUEST, response.status_code)
2853
2768
        self.assertIn('text/plain', response['Content-Type'])
2855
2770
 
2856
2771
    def test_add_file_can_overwrite_existing_file_of_same_name(self):
2857
2772
        # Write file one.
2858
 
        filepath = self.make_upload_file(contents="file one")
2859
 
        with open(filepath) as f:
2860
 
            response = self.make_API_POST_request("add", "foo", f)
 
2773
        response = self.make_API_POST_request(
 
2774
            "add", "foo", factory.make_file_upload(content=b"file one"))
2861
2775
        self.assertEqual(httplib.CREATED, response.status_code)
2862
2776
 
2863
2777
        # Write file two with the same name but different contents.
2864
 
        filepath = self.make_upload_file(contents="file two")
2865
 
        with open(filepath) as f:
2866
 
            response = self.make_API_POST_request("add", "foo", f)
 
2778
        response = self.make_API_POST_request(
 
2779
            "add", "foo", factory.make_file_upload(content=b"file two"))
2867
2780
        self.assertEqual(httplib.CREATED, response.status_code)
2868
2781
 
2869
2782
        # Retrieve the file and check its contents are the new contents.
2870
2783
        response = self.make_API_GET_request("get", "foo")
2871
 
        self.assertEqual("file two", response.content)
 
2784
        self.assertEqual(b"file two", response.content)
2872
2785
 
2873
2786
    def test_get_file_succeeds(self):
2874
2787
        factory.make_file_storage(
3449
3362
            % (invalid,))
3450
3363
        self.assertFalse(Tag.objects.filter(name=invalid).exists())
3451
3364
 
 
3365
    def test_POST_new_kernel_opts(self):
 
3366
        self.become_admin()
 
3367
        name = factory.getRandomString()
 
3368
        definition = '//node'
 
3369
        comment = factory.getRandomString()
 
3370
        extra_kernel_opts = factory.getRandomString()
 
3371
        response = self.client.post(
 
3372
            self.get_uri('tags/'),
 
3373
            {
 
3374
                'op': 'new',
 
3375
                'name': name,
 
3376
                'comment': comment,
 
3377
                'definition': definition,
 
3378
                'kernel_opts': extra_kernel_opts,
 
3379
            })
 
3380
        self.assertEqual(httplib.OK, response.status_code)
 
3381
        parsed_result = json.loads(response.content)
 
3382
        self.assertEqual(name, parsed_result['name'])
 
3383
        self.assertEqual(comment, parsed_result['comment'])
 
3384
        self.assertEqual(definition, parsed_result['definition'])
 
3385
        self.assertEqual(extra_kernel_opts, parsed_result['kernel_opts'])
 
3386
        self.assertEqual(
 
3387
            extra_kernel_opts, Tag.objects.filter(name=name)[0].kernel_opts)
 
3388
 
3452
3389
    def test_POST_new_populates_nodes(self):
3453
3390
        self.become_admin()
3454
3391
        node1 = factory.make_node()
3894
3831
        kernel_params = KernelParameters(**self.get_pxeconfig(params))
3895
3832
        self.assertEqual(params["local"], kernel_params.fs_host)
3896
3833
 
 
3834
    def test_pxeconfig_returns_extra_kernel_options(self):
 
3835
        node = factory.make_node()
 
3836
        extra_kernel_opts = factory.getRandomString()
 
3837
        Config.objects.set_config('kernel_opts', extra_kernel_opts)
 
3838
        mac = factory.make_mac_address(node=node)
 
3839
        params = self.get_default_params()
 
3840
        params['mac'] = mac.mac_address
 
3841
        pxe_config = self.get_pxeconfig(params)
 
3842
        self.assertEqual(extra_kernel_opts, pxe_config['extra_opts'])
 
3843
 
 
3844
    def test_pxeconfig_returns_None_for_extra_kernel_opts(self):
 
3845
        mac = factory.make_mac_address()
 
3846
        params = self.get_default_params()
 
3847
        params['mac'] = mac.mac_address
 
3848
        pxe_config = self.get_pxeconfig(params)
 
3849
        self.assertEqual(None, pxe_config['extra_opts'])
 
3850
 
3897
3851
 
3898
3852
class TestNodeGroupsAPI(APIv10TestMixin, MultipleUsersScenarios, TestCase):
3899
3853
    scenarios = [
4565
4519
            })
4566
4520
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
4567
4521
 
 
4522
    def test_import_boot_images_calls_script_for_all_accepted_clusters(self):
 
4523
        recorder = self.patch(nodegroup_module, 'import_boot_images')
 
4524
        proxy = factory.make_name('proxy')
 
4525
        Config.objects.set_config('http_proxy', proxy)
 
4526
        accepted_nodegroups = [
 
4527
            factory.make_node_group(status=NODEGROUP_STATUS.ACCEPTED),
 
4528
            factory.make_node_group(status=NODEGROUP_STATUS.ACCEPTED),
 
4529
        ]
 
4530
        factory.make_node_group(status=NODEGROUP_STATUS.REJECTED)
 
4531
        factory.make_node_group(status=NODEGROUP_STATUS.PENDING)
 
4532
        admin = factory.make_admin()
 
4533
        client = OAuthAuthenticatedClient(admin)
 
4534
        response = client.post(
 
4535
            reverse('nodegroups_handler'), {'op': 'import_boot_images'})
 
4536
        self.assertEqual(
 
4537
            httplib.OK, response.status_code,
 
4538
            explain_unexpected_response(httplib.OK, response))
 
4539
        queues = [
 
4540
            kwargs['queue']
 
4541
            for args, kwargs in recorder.apply_async.call_args_list]
 
4542
        self.assertItemsEqual(
 
4543
            [nodegroup.work_queue for nodegroup in accepted_nodegroups],
 
4544
            queues)
 
4545
 
 
4546
    def test_import_boot_images_denied_if_not_admin(self):
 
4547
        user = factory.make_user()
 
4548
        client = OAuthAuthenticatedClient(user)
 
4549
        response = client.post(
 
4550
            reverse('nodegroups_handler'), {'op': 'import_boot_images'})
 
4551
        self.assertEqual(
 
4552
            httplib.FORBIDDEN, response.status_code,
 
4553
            explain_unexpected_response(httplib.FORBIDDEN, response))
 
4554
 
4568
4555
 
4569
4556
def log_in_as_normal_user(client):
4570
4557
    """Log `client` in as a normal user."""
4646
4633
 
4647
4634
    def test_nodegroup_list_nodes_works_for_admin(self):
4648
4635
        nodegroup = factory.make_node_group()
4649
 
        user = factory.make_user()
4650
 
        user.is_superuser = True
4651
 
        user.save()
4652
 
        client = OAuthAuthenticatedClient(user)
 
4636
        admin = factory.make_admin()
 
4637
        client = OAuthAuthenticatedClient(admin)
4653
4638
        node = factory.make_node(nodegroup=nodegroup)
4654
4639
        response = client.get(
4655
4640
            reverse('nodegroup_handler', args=[nodegroup.uuid]),
4660
4645
        parsed_result = json.loads(response.content)
4661
4646
        self.assertItemsEqual([node.system_id], parsed_result)
4662
4647
 
 
4648
    def test_nodegroup_import_boot_images_calls_script(self):
 
4649
        recorder = self.patch(tasks, 'check_call')
 
4650
        proxy = factory.getRandomString()
 
4651
        Config.objects.set_config('http_proxy', proxy)
 
4652
        nodegroup = factory.make_node_group()
 
4653
        admin = factory.make_admin()
 
4654
        client = OAuthAuthenticatedClient(admin)
 
4655
        response = client.post(
 
4656
            reverse('nodegroup_handler', args=[nodegroup.uuid]),
 
4657
            {'op': 'import_boot_images'})
 
4658
        self.assertEqual(
 
4659
            httplib.OK, response.status_code,
 
4660
            explain_unexpected_response(httplib.OK, response))
 
4661
        recorder.assert_called_once_with(
 
4662
            ['sudo', '-n', '-E', 'maas-import-pxe-files'], env=ANY)
 
4663
 
 
4664
    def test_nodegroup_import_boot_images_denied_if_not_admin(self):
 
4665
        nodegroup = factory.make_node_group()
 
4666
        user = factory.make_user()
 
4667
        client = OAuthAuthenticatedClient(user)
 
4668
        response = client.post(
 
4669
            reverse('nodegroup_handler', args=[nodegroup.uuid]),
 
4670
            {'op': 'import_boot_images'})
 
4671
        self.assertEqual(
 
4672
            httplib.FORBIDDEN, response.status_code,
 
4673
            explain_unexpected_response(httplib.FORBIDDEN, response))
 
4674
 
4663
4675
    def make_node_hardware_details_request(self, client, nodegroup=None):
4664
4676
        if nodegroup is None:
4665
4677
            nodegroup = factory.make_node_group()
4868
4880
            images=ANY, nodegroup=ANY)
4869
4881
 
4870
4882
 
 
4883
class AdminCommissioningScriptsAPITest(APIv10TestMixin, AdminLoggedInTestCase):
 
4884
    """Tests for `CommissioningScriptsHandler`."""
 
4885
 
 
4886
    def get_url(self):
 
4887
        return reverse('commissioning_scripts_handler')
 
4888
 
 
4889
    def test_GET_lists_commissioning_scripts(self):
 
4890
        # Use lower-case names.  The database and the test may use
 
4891
        # different collation orders with different ideas about case
 
4892
        # sensitivity.
 
4893
        names = {factory.make_name('script').lower() for counter in range(5)}
 
4894
        for name in names:
 
4895
            factory.make_commissioning_script(name=name)
 
4896
 
 
4897
        response = self.client.get(self.get_url())
 
4898
 
 
4899
        self.assertEqual(
 
4900
            (httplib.OK, sorted(names)),
 
4901
            (response.status_code, json.loads(response.content)))
 
4902
 
 
4903
    def test_POST_creates_commissioning_script(self):
 
4904
        # This uses Piston's built-in POST code, so there are no tests for
 
4905
        # corner cases (like "script already exists") here.
 
4906
        name = factory.make_name('script')
 
4907
        content = factory.getRandomString().encode('ascii')
 
4908
 
 
4909
        # Every uploaded file also has a name.  But this is completely
 
4910
        # unrelated to the name we give to the commissioning script.
 
4911
        response = self.client.post(
 
4912
            self.get_url(),
 
4913
            {
 
4914
                'name': name,
 
4915
                'content': factory.make_file_upload(content=content),
 
4916
            })
 
4917
        self.assertEqual(httplib.OK, response.status_code)
 
4918
 
 
4919
        returned_script = json.loads(response.content)
 
4920
        self.assertEqual(
 
4921
            (name, content),
 
4922
            (returned_script['name'], returned_script['content']))
 
4923
 
 
4924
        stored_script = CommissioningScript.objects.get(name=name)
 
4925
        self.assertEqual(content, stored_script.content)
 
4926
 
 
4927
 
 
4928
class CommissioningScriptsAPITest(APITestCase):
 
4929
 
 
4930
    def get_url(self):
 
4931
        return reverse('commissioning_scripts_handler')
 
4932
 
 
4933
    def test_GET_is_forbidden(self):
 
4934
        response = self.client.get(self.get_url())
 
4935
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
4936
 
 
4937
    def test_POST_is_forbidden(self):
 
4938
        response = self.client.post(
 
4939
            self.get_url(),
 
4940
            {'name': factory.make_name('script')})
 
4941
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
4942
 
 
4943
 
 
4944
class AdminCommissioningScriptAPITest(APIv10TestMixin, AdminLoggedInTestCase):
 
4945
    """Tests for `CommissioningScriptHandler`."""
 
4946
 
 
4947
    def get_url(self, script_name):
 
4948
        return reverse('commissioning_script_handler', args=[script_name])
 
4949
 
 
4950
    def test_GET_returns_script_contents(self):
 
4951
        script = factory.make_commissioning_script()
 
4952
        response = self.client.get(self.get_url(script.name))
 
4953
        self.assertEqual(httplib.OK, response.status_code)
 
4954
        self.assertEqual(script.content, response.content)
 
4955
 
 
4956
    def test_GET_preserves_binary_data(self):
 
4957
        script = factory.make_commissioning_script(content=sample_binary_data)
 
4958
        response = self.client.get(self.get_url(script.name))
 
4959
        self.assertEqual(httplib.OK, response.status_code)
 
4960
        self.assertEqual(sample_binary_data, response.content)
 
4961
 
 
4962
    def test_PUT_updates_contents(self):
 
4963
        old_content = b'old:%s' % factory.getRandomString().encode('ascii')
 
4964
        script = factory.make_commissioning_script(content=old_content)
 
4965
        new_content = b'new:%s' % factory.getRandomString().encode('ascii')
 
4966
 
 
4967
        response = self.client.put(
 
4968
            self.get_url(script.name),
 
4969
            {'content': factory.make_file_upload(content=new_content)})
 
4970
        self.assertEqual(httplib.OK, response.status_code)
 
4971
 
 
4972
        self.assertEqual(new_content, reload_object(script).content)
 
4973
 
 
4974
    def test_DELETE_deletes_script(self):
 
4975
        script = factory.make_commissioning_script()
 
4976
        self.client.delete(self.get_url(script.name))
 
4977
        self.assertItemsEqual(
 
4978
            [],
 
4979
            CommissioningScript.objects.filter(name=script.name))
 
4980
 
 
4981
 
 
4982
class CommissioningScriptAPITest(APITestCase):
 
4983
 
 
4984
    def get_url(self, script_name):
 
4985
        return reverse('commissioning_script_handler', args=[script_name])
 
4986
 
 
4987
    def test_GET_is_forbidden(self):
 
4988
        # It's not inconceivable that commissioning scripts contain
 
4989
        # credentials of some sort.  There is no need for regular users
 
4990
        # (consumers of the MAAS) to see these.
 
4991
        script = factory.make_commissioning_script()
 
4992
        response = self.client.get(self.get_url(script.name))
 
4993
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
4994
 
 
4995
    def test_PUT_is_forbidden(self):
 
4996
        script = factory.make_commissioning_script()
 
4997
        response = self.client.put(
 
4998
            self.get_url(script.name), {'content': factory.getRandomString()})
 
4999
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
5000
 
 
5001
    def test_DELETE_is_forbidden(self):
 
5002
        script = factory.make_commissioning_script()
 
5003
        response = self.client.put(self.get_url(script.name))
 
5004
        self.assertEqual(httplib.FORBIDDEN, response.status_code)
 
5005
 
 
5006
 
 
5007
class NodeCommissionResultHandlerAPITest(APITestCase):
 
5008
 
 
5009
    def test_list_returns_commissioning_results(self):
 
5010
        commissioning_results = [
 
5011
            factory.make_node_commission_result()
 
5012
            for counter in range(3)]
 
5013
        url = reverse('commissioning_results_handler')
 
5014
        response = self.client.get(url, {'op': 'list'})
 
5015
        self.assertEqual(httplib.OK, response.status_code, response.content)
 
5016
        parsed_results = json.loads(response.content)
 
5017
        self.assertItemsEqual(
 
5018
            [(
 
5019
                commissioning_result.name,
 
5020
                commissioning_result.script_result,
 
5021
                commissioning_result.data,
 
5022
                commissioning_result.node.system_id,
 
5023
            )
 
5024
            for commissioning_result in commissioning_results
 
5025
            ],
 
5026
            [(
 
5027
                result.get('name'),
 
5028
                result.get('script_result'),
 
5029
                result.get('data'),
 
5030
                result.get('node').get('system_id')
 
5031
            )
 
5032
            for result in parsed_results
 
5033
            ])
 
5034
 
 
5035
    def test_list_can_be_filtered_by_node(self):
 
5036
        commissioning_results = [
 
5037
            factory.make_node_commission_result()
 
5038
            for counter in range(3)]
 
5039
        url = reverse('commissioning_results_handler')
 
5040
        response = self.client.get(
 
5041
            url,
 
5042
            {
 
5043
                'op': 'list',
 
5044
                'system_id':
 
5045
                    [
 
5046
                        commissioning_results[0].node.system_id,
 
5047
                        commissioning_results[1].node.system_id,
 
5048
                    ]
 
5049
            }
 
5050
        )
 
5051
        self.assertEqual(httplib.OK, response.status_code, response.content)
 
5052
        parsed_results = json.loads(response.content)
 
5053
        self.assertItemsEqual(
 
5054
            [commissioning_results[0].data, commissioning_results[1].data],
 
5055
            [result.get('data') for result in parsed_results])
 
5056
 
 
5057
    def test_list_can_be_filtered_by_name(self):
 
5058
        commissioning_results = [
 
5059
            factory.make_node_commission_result()
 
5060
            for counter in range(3)]
 
5061
        url = reverse('commissioning_results_handler')
 
5062
        response = self.client.get(
 
5063
            url,
 
5064
            {
 
5065
                'op': 'list',
 
5066
                'name': commissioning_results[0].name
 
5067
            }
 
5068
        )
 
5069
        self.assertEqual(httplib.OK, response.status_code, response.content)
 
5070
        parsed_results = json.loads(response.content)
 
5071
        self.assertItemsEqual(
 
5072
            [commissioning_results[0].data],
 
5073
            [result.get('data') for result in parsed_results])
 
5074
 
 
5075
    def test_list_displays_only_visible_nodes(self):
 
5076
        node = factory.make_node(owner=factory.make_user())
 
5077
        factory.make_node_commission_result(node)
 
5078
        url = reverse('commissioning_results_handler')
 
5079
        response = self.client.get(url, {'op': 'list'})
 
5080
        self.assertEqual(httplib.OK, response.status_code, response.content)
 
5081
        parsed_results = json.loads(response.content)
 
5082
        self.assertEqual([], parsed_results)
 
5083
 
 
5084
 
4871
5085
class TestDescribe(AnonAPITestCase):
4872
5086
    """Tests for the `describe` view."""
4873
5087