← Back to branch summary

~ubuntu-branches/ubuntu/precise/maas/precise-updates

« back to all changes in this revision

Viewing changes to src/maasserver/tests/test_node_action.py

Tags: 1.2+bzr1373+dfsg-0ubuntu1~12.04.4
https://launchpad.net/bugs/1039513
https://launchpad.net/bugs/1158425
* SECURITY UPDATE: failure to authenticate downloaded content (LP: #1039513)
  - debian/patches/CVE-2013-1058.patch: Authenticate downloaded files with
    GnuPG and MD5SUM files. Thanks to Julian Edwards.
  - CVE-2013-1058
* SECURITY UPDATE: configuration options may be loaded from current working
  directory (LP: #1158425)
  - debian/patches/CVE-2013-1057-1-2.patch: Do not load configuration
    options from the current working directory. Thanks to Julian Edwards.
  - CVE-2013-1057

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/maasserver/tests/test_node_action.py
 
1
# Copyright 2012 Canonical Ltd.  This software is licensed under the
 
2
# GNU Affero General Public License version 3 (see the file LICENSE).
 
3
 
 
4
"""Tests for node actions."""
 
5
 
 
6
from __future__ import (
 
7
    absolute_import,
 
8
    print_function,
 
9
    unicode_literals,
 
10
    )
 
11
 
 
12
__metaclass__ = type
 
13
__all__ = []
 
14
 
 
15
from urlparse import urlparse
 
16
 
 
17
from django.core.urlresolvers import reverse
 
18
from maasserver.enum import (
 
19
    NODE_PERMISSION,
 
20
    NODE_STATUS,
 
21
    NODE_STATUS_CHOICES_DICT,
 
22
    )
 
23
from maasserver.exceptions import Redirect
 
24
from maasserver.node_action import (
 
25
    AcceptAndCommission,
 
26
    compile_node_actions,
 
27
    Delete,
 
28
    NodeAction,
 
29
    RetryCommissioning,
 
30
    StartNode,
 
31
    )
 
32
from maasserver.testing.factory import factory
 
33
from maasserver.testing.testcase import TestCase
 
34
from provisioningserver.enum import POWER_TYPE
 
35
 
 
36
 
 
37
ALL_STATUSES = NODE_STATUS_CHOICES_DICT.keys()
 
38
 
 
39
 
 
40
class FakeNodeAction(NodeAction):
 
41
    display = "Action label"
 
42
    actionable_statuses = ALL_STATUSES
 
43
    permission = NODE_PERMISSION.VIEW
 
44
 
 
45
    # For testing: an inhibition for inhibit() to return.
 
46
    fake_inhibition = None
 
47
 
 
48
    def inhibit(self):
 
49
        return self.fake_inhibition
 
50
 
 
51
    def execute(self):
 
52
        pass
 
53
 
 
54
 
 
55
class TestNodeAction(TestCase):
 
56
 
 
57
    def test_compile_node_actions_returns_available_actions(self):
 
58
 
 
59
        class MyAction(FakeNodeAction):
 
60
            display = factory.getRandomString()
 
61
 
 
62
        actions = compile_node_actions(
 
63
            factory.make_node(), factory.make_admin(), classes=[MyAction])
 
64
        self.assertEqual([MyAction.display], actions.keys())
 
65
 
 
66
    def test_compile_node_actions_checks_node_status(self):
 
67
 
 
68
        class MyAction(FakeNodeAction):
 
69
            actionable_statuses = (NODE_STATUS.READY, )
 
70
 
 
71
        node = factory.make_node(status=NODE_STATUS.DECLARED)
 
72
        actions = compile_node_actions(
 
73
            node, factory.make_admin(), classes=[MyAction])
 
74
        self.assertEqual({}, actions)
 
75
 
 
76
    def test_compile_node_actions_checks_permission(self):
 
77
 
 
78
        class MyAction(FakeNodeAction):
 
79
            permission = NODE_PERMISSION.EDIT
 
80
 
 
81
        node = factory.make_node(status=NODE_STATUS.COMMISSIONING)
 
82
        actions = compile_node_actions(
 
83
            node, factory.make_user(), classes=[MyAction])
 
84
        self.assertEqual({}, actions)
 
85
 
 
86
    def test_compile_node_actions_includes_inhibited_actions(self):
 
87
 
 
88
        class MyAction(FakeNodeAction):
 
89
            fake_inhibition = factory.getRandomString()
 
90
 
 
91
        actions = compile_node_actions(
 
92
            factory.make_node(), factory.make_admin(), classes=[MyAction])
 
93
        self.assertEqual([MyAction.display], actions.keys())
 
94
 
 
95
    def test_compile_node_actions_maps_display_names(self):
 
96
 
 
97
        class Action1(FakeNodeAction):
 
98
            display = factory.getRandomString()
 
99
 
 
100
        class Action2(FakeNodeAction):
 
101
            display = factory.getRandomString()
 
102
 
 
103
        actions = compile_node_actions(
 
104
            factory.make_node(), factory.make_admin(),
 
105
            classes=[Action1, Action2])
 
106
        for label, action in actions.items():
 
107
            self.assertEqual(label, action.display)
 
108
 
 
109
    def test_compile_node_actions_maintains_order(self):
 
110
        labels = [factory.getRandomString() for counter in range(4)]
 
111
        classes = [
 
112
            type(b"Action%d" % counter, (FakeNodeAction,), {'display': label})
 
113
            for counter, label in enumerate(labels)]
 
114
        actions = compile_node_actions(
 
115
            factory.make_node(), factory.make_admin(), classes=classes)
 
116
        self.assertSequenceEqual(labels, actions.keys())
 
117
        self.assertSequenceEqual(
 
118
            labels, [action.display for action in actions.values()])
 
119
 
 
120
    def test_is_permitted_allows_if_user_has_permission(self):
 
121
 
 
122
        class MyAction(FakeNodeAction):
 
123
            permission = NODE_PERMISSION.EDIT
 
124
 
 
125
        node = factory.make_node(
 
126
            status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
 
127
        self.assertTrue(MyAction(node, node.owner).is_permitted())
 
128
 
 
129
    def test_is_permitted_disallows_if_user_lacks_permission(self):
 
130
 
 
131
        class MyAction(FakeNodeAction):
 
132
            permission = NODE_PERMISSION.EDIT
 
133
 
 
134
        node = factory.make_node(
 
135
            status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
 
136
        self.assertFalse(MyAction(node, factory.make_user()).is_permitted())
 
137
 
 
138
    def test_inhibition_wraps_inhibit(self):
 
139
        inhibition = factory.getRandomString()
 
140
        action = FakeNodeAction(factory.make_node(), factory.make_user())
 
141
        action.fake_inhibition = inhibition
 
142
        self.assertEqual(inhibition, action.inhibition)
 
143
 
 
144
    def test_inhibition_caches_inhibition(self):
 
145
        # The inhibition property will call inhibit() only once.  We can
 
146
        # prove this by changing the string inhibit() returns; it won't
 
147
        # affect the value of the property.
 
148
        inhibition = factory.getRandomString()
 
149
        action = FakeNodeAction(factory.make_node(), factory.make_user())
 
150
        action.fake_inhibition = inhibition
 
151
        self.assertEqual(inhibition, action.inhibition)
 
152
        action.fake_inhibition = factory.getRandomString()
 
153
        self.assertEqual(inhibition, action.inhibition)
 
154
 
 
155
    def test_inhibition_caches_None(self):
 
156
        # An inhibition of None is also faithfully cached.  In other
 
157
        # words, it doesn't get mistaken for an uninitialized cache or
 
158
        # anything.
 
159
        action = FakeNodeAction(factory.make_node(), factory.make_user())
 
160
        action.fake_inhibition = None
 
161
        self.assertIsNone(action.inhibition)
 
162
        action.fake_inhibition = factory.getRandomString()
 
163
        self.assertIsNone(action.inhibition)
 
164
 
 
165
    def test_Delete_inhibit_when_node_is_allocated(self):
 
166
        node = factory.make_node(status=NODE_STATUS.ALLOCATED)
 
167
        action = Delete(node, factory.make_admin())
 
168
        inhibition = action.inhibit()
 
169
        self.assertEqual(
 
170
            "You cannot delete this node because it's in use.", inhibition)
 
171
 
 
172
    def test_Delete_does_not_inhibit_otherwise(self):
 
173
        node = factory.make_node(status=NODE_STATUS.FAILED_TESTS)
 
174
        action = Delete(node, factory.make_admin())
 
175
        inhibition = action.inhibit()
 
176
        self.assertIsNone(inhibition)
 
177
 
 
178
    def test_Delete_redirects_to_node_delete_view(self):
 
179
        node = factory.make_node()
 
180
        action = Delete(node, factory.make_admin())
 
181
        try:
 
182
            action.execute()
 
183
        except Redirect as e:
 
184
            pass
 
185
        self.assertEqual(
 
186
            reverse('node-delete', args=[node.system_id]),
 
187
            urlparse(unicode(e)).path)
 
188
 
 
189
    def test_AcceptAndCommission_starts_commissioning(self):
 
190
        node = factory.make_node(
 
191
            mac=True, status=NODE_STATUS.DECLARED,
 
192
            power_type=POWER_TYPE.WAKE_ON_LAN)
 
193
        action = AcceptAndCommission(node, factory.make_admin())
 
194
        action.execute()
 
195
        self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
 
196
        self.assertEqual(
 
197
            'provisioningserver.tasks.power_on',
 
198
            self.celery.tasks[0]['task'].name)
 
199
 
 
200
    def test_RetryCommissioning_starts_commissioning(self):
 
201
        node = factory.make_node(
 
202
            mac=True, status=NODE_STATUS.FAILED_TESTS,
 
203
            power_type=POWER_TYPE.WAKE_ON_LAN)
 
204
        action = RetryCommissioning(node, factory.make_admin())
 
205
        action.execute()
 
206
        self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
 
207
        self.assertEqual(
 
208
            'provisioningserver.tasks.power_on',
 
209
            self.celery.tasks[0]['task'].name)
 
210
 
 
211
    def test_StartNode_inhibit_allows_user_with_SSH_key(self):
 
212
        user_with_key = factory.make_user()
 
213
        factory.make_sshkey(user_with_key)
 
214
        self.assertIsNone(
 
215
            StartNode(factory.make_node(), user_with_key).inhibit())
 
216
 
 
217
    def test_StartNode_inhibit_disallows_user_without_SSH_key(self):
 
218
        user_without_key = factory.make_user()
 
219
        action = StartNode(factory.make_node(), user_without_key)
 
220
        inhibition = action.inhibit()
 
221
        self.assertIsNotNone(inhibition)
 
222
        self.assertIn("SSH key", inhibition)
 
223
 
 
224
    def test_StartNode_acquires_and_starts_node(self):
 
225
        node = factory.make_node(
 
226
            mac=True, status=NODE_STATUS.READY,
 
227
            power_type=POWER_TYPE.WAKE_ON_LAN)
 
228
        user = factory.make_user()
 
229
        StartNode(node, user).execute()
 
230
        self.assertEqual(NODE_STATUS.ALLOCATED, node.status)
 
231
        self.assertEqual(user, node.owner)
 
232
        self.assertEqual(
 
233
            'provisioningserver.tasks.power_on',
 
234
            self.celery.tasks[0]['task'].name)