launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #07341
[Merge] lp:~jtv/maas/doris-please-spit-out-your-old-node-actions into lp:maas
Jeroen T. Vermeulen has proposed merging lp:~jtv/maas/doris-please-spit-out-your-old-node-actions into lp:maas with lp:~jtv/maas/doris-would-you-like-nicer-node-actions as a prerequisite.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~jtv/maas/doris-please-spit-out-your-old-node-actions/+merge/103743
This is all part of an undertaking I discussed with Julian: re-organized node actions. In my prerequisite branch I implemented a new node-actions structure. In this branch, I move the code over from the old structure to the new one.
You'll note that tests no longer need to spell out the full UI-visible text of the action buttons. Unit tests happen at a very low level on the new module, so the integration tests only need one example per behaviour. It also frees the unit tests of a lot of the tedium of simulating a request etc. The view code gets a bit simpler and the template, remarkably, does not need to change at all. (Actually there are a few attribute names that I'm tempted to change, but it's enough for one day).
Jeroen
--
https://code.launchpad.net/~jtv/maas/doris-please-spit-out-your-old-node-actions/+merge/103743
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jtv/maas/doris-please-spit-out-your-old-node-actions into lp:maas.
=== modified file 'src/maasserver/forms.py'
--- src/maasserver/forms.py 2012-04-26 07:57:26 +0000
+++ src/maasserver/forms.py 2012-04-26 18:00:32 +0000
@@ -23,8 +23,6 @@
"UINodeEditForm",
]
-from textwrap import dedent
-
from django import forms
from django.contrib import messages
from django.contrib.auth.forms import (
@@ -39,7 +37,6 @@
PermissionDenied,
ValidationError,
)
-from django.core.urlresolvers import reverse
from django.core.validators import URLValidator
from django.forms import (
CharField,
@@ -51,10 +48,7 @@
ARCHITECTURE_CHOICES,
NODE_AFTER_COMMISSIONING_ACTION,
NODE_AFTER_COMMISSIONING_ACTION_CHOICES,
- NODE_PERMISSION,
- NODE_STATUS,
)
-from maasserver.exceptions import Redirect
from maasserver.fields import MACAddressFormField
from maasserver.models import (
Config,
@@ -62,6 +56,7 @@
Node,
SSHKey,
)
+from maasserver.node_action import compile_node_actions
def compose_invalid_choice_text(choice_of_what, valid_choices):
@@ -236,145 +231,6 @@
return node
-def inhibit_acquisition(node, user, request=None):
- """Give me one reason why `user` can't acquire and start `node`."""
- if SSHKey.objects.get_keys_for_user(user).exists():
- return None
- else:
- return dedent("""
- You have no means of accessing the node after starting it.
- Register an SSH key first. Do this on your Preferences screen:
- click on the menu with your name at the top of the page, select
- Preferences, and look for the "SSH keys" section.
- """)
-
-
-def acquire_and_start_node(node, user, request=None):
- """Acquire and start a node from the UI. It will have no user_data."""
- # Avoid circular imports.
- from maasserver.api import get_oauth_token
-
- node.acquire(get_oauth_token(request))
- Node.objects.start_nodes([node.system_id], user)
-
-
-def start_commissioning_node(node, user, request=None):
- """Start the commissioning process for a node."""
- node.start_commissioning(user)
-
-
-def inhibit_delete(node, user, request=None):
- """Give me one reason why `user` can't delete `node`."""
- if node.owner is not None:
- return "You cannot delete this node because it's in use."
- else:
- return None
-
-
-def delete_node(node, user, request=None):
- """Delete-a-node action.
-
- Permissions and inhibitions have already been checked at this point. We
- redirect to the deletion view, which asks the user if they're sure.
- """
- raise Redirect(reverse('node-delete', args=[node.system_id]))
-
-
-# Node actions per status.
-#
-# This maps each NODE_STATUS to a list of actions applicable to a node
-# in that state:
-#
-# {
-# NODE_STATUS.<statusX>: [action1, action2],
-# NODE_STATUS.<statusY>: [action1],
-# NODE_STATUS.<statusZ>: [action1, action2, action3],
-# }
-#
-# The available actions (insofar as the user has privileges to use them)
-# show up in the user interface as buttons on the node page.
-#
-# Each action is a dict:
-#
-# {
-# # Action's display name; will be the button's label.
-# 'display': "Paint node",
-#
-# # Permission required on the node to perform the action.
-# 'permission': NODE_PERMISSION.EDIT,
-#
-# # Function that looks for reasons to disable the action.
-# # If it returns None, the action is available; otherwise, it is
-# # shown but disabled. A tooltip shows the inhibiting reason.
-# 'inhibit': lambda node, user, request=None: None,
-#
-# # Callable that performs action.
-# # Takes parameters (node, user, request).
-# # Even though this is not the API, the action may raise a
-# # MAASAPIException if it wants to convey failure with a specific
-# # http status code.
-# 'execute': lambda node, user, request=None: paint_node(
-# node, favourite_colour(user)),
-# }
-#
-DELETE_ACTION = {
- 'display': "Delete node",
- 'permission': NODE_PERMISSION.ADMIN,
- 'execute': delete_node,
- 'message': "Node deleted.",
- 'inhibit': inhibit_delete,
-}
-
-
-NODE_ACTIONS = {
- NODE_STATUS.DECLARED: [
- DELETE_ACTION,
- {
- 'display': "Accept & commission",
- 'permission': NODE_PERMISSION.ADMIN,
- 'execute': start_commissioning_node,
- 'message': "Node commissioning started.",
- },
- ],
- NODE_STATUS.FAILED_TESTS: [
- DELETE_ACTION,
- {
- 'display': "Retry commissioning",
- 'permission': NODE_PERMISSION.ADMIN,
- 'execute': start_commissioning_node,
- 'message': "Started a new attempt to commission this node.",
- },
- ],
- NODE_STATUS.COMMISSIONING: [
- DELETE_ACTION,
- ],
- NODE_STATUS.MISSING: [
- DELETE_ACTION,
- ],
- NODE_STATUS.READY: [
- DELETE_ACTION,
- {
- 'display': "Start node",
- 'permission': NODE_PERMISSION.VIEW,
- 'execute': acquire_and_start_node,
- 'message': (
- "This node is now allocated to you. "
- "It has been asked to start up."),
- 'inhibit': inhibit_acquisition,
- },
- ],
- NODE_STATUS.RESERVED: [
- DELETE_ACTION,
- ],
- NODE_STATUS.ALLOCATED: [
- DELETE_ACTION,
- ],
- NODE_STATUS.RETIRED: [
- DELETE_ACTION,
- ],
-}
-
-
class NodeActionForm(forms.Form):
"""Base form for performing a node action.
@@ -391,62 +247,8 @@
def __init__(self, instance, *args, **kwargs):
super(NodeActionForm, self).__init__(*args, **kwargs)
self.node = instance
- self.action_buttons = self.compile_actions()
-
- def have_permission(self, action=None):
- """Does the current user have the permission required by `action`?
-
- The permission is relative to `self.node`.
-
- :param action: Action to check. If no action is given, the answer is
- always False.
- :return: Does the user have the required permission to perform
- `action` on `self.node`?
- """
- if action is None:
- return False
- else:
- return self.user.has_perm(action['permission'], self.node)
-
- def compile_action(self, action):
- """Return a copy of `action`, but add "inhibition" entry.
-
- The key "inhibition" maps to a unicode string explaining why the
- action is not currently available for `self.node`, or None if the
- action can be executed.
- """
- compiled_action = action.copy()
- inhibit = action.get('inhibit', lambda node, user, request: None)
- compiled_action['inhibition'] = inhibit(
- node=self.node, user=self.user, request=self.request)
- return compiled_action
-
- def compile_actions(self):
- """Return the actions that this user is allowed to perform on a node.
-
- Each action will be "filled out," i.e. it will have an "inhibit"
- callable even if the original action in NODE_ACTIONS did not.
-
- :param node: The node for which the check should be performed.
- :type node: :class:`maasserver.models.Node`
- :param user: The user who would be performing the action. Only the
- actions available to this user will be returned.
- :type user: :class:`django.contrib.auth.models.User`
- :return: Any applicable action dicts, as compiled based on
- NODE_ACTIONS and the current user and node.
- :rtype: Sequence
- """
- return [
- self.compile_action(action)
- for action in NODE_ACTIONS.get(self.node.status, [])
- if self.have_permission(action)]
-
- def find_action(self, action_name):
- """Find the compiled action of the given name, or return None."""
- for action in self.action_buttons:
- if action['display'] == action_name:
- return action
- return None
+ self.actions = compile_node_actions(instance, self.user, self.request)
+ self.action_buttons = self.actions.values()
def display_message(self, message):
"""Show `message` as feedback after performing an action."""
@@ -456,17 +258,13 @@
def save(self):
"""An action was requested. Perform it."""
action_name = self.data.get(self.input_name)
- action = self.find_action(action_name)
-
- if not self.have_permission(action):
+ action = self.actions.get(action_name)
+ if action is None or not action.is_permitted():
raise PermissionDenied("Not a permitted action: %s" % action_name)
-
- if action['inhibition'] is not None:
- raise PermissionDenied(action['inhibition'])
-
- action['execute'](
- node=self.node, user=self.user, request=self.request)
- self.display_message(action.get('message'))
+ if action.inhibition is not None:
+ raise PermissionDenied(action.inhibition)
+ message = action.perform()
+ self.display_message(message)
def get_action_form(user, request=None):
=== added file 'src/maasserver/node_action.py'
--- src/maasserver/node_action.py 1970-01-01 00:00:00 +0000
+++ src/maasserver/node_action.py 2012-04-26 18:00:32 +0000
@@ -0,0 +1,225 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Node actions.
+
+These are actions that appear as buttons no the UI's Node page, depending
+on the node's state, the user's privileges etc.
+
+To define a new node action, derive a class for it from :class:`NodeAction`,
+provide the missing pieces documented in the class, and add it to
+`ACTION_CLASSES`. The actions will always appear on the page in the same
+order as they do in `ACTION_CLASSES`.
+"""
+
+from __future__ import (
+ absolute_import,
+ print_function,
+ unicode_literals,
+ )
+
+__metaclass__ = type
+__all__ = [
+ 'compile_node_actions',
+ ]
+
+from abc import (
+ ABCMeta,
+ abstractmethod,
+ abstractproperty,
+ )
+from collections import OrderedDict
+from textwrap import dedent
+
+from django.core.urlresolvers import reverse
+from maasserver.enum import (
+ NODE_PERMISSION,
+ NODE_STATUS,
+ NODE_STATUS_CHOICES_DICT,
+ )
+from maasserver.exceptions import Redirect
+from maasserver.models import (
+ Node,
+ SSHKey,
+ )
+
+# All node statuses.
+ALL_STATUSES = set(NODE_STATUS_CHOICES_DICT.keys())
+
+
+class NodeAction:
+ """Base class for node actions."""
+
+ __metaclass__ = ABCMeta
+
+ display = abstractproperty("""
+ Action name.
+
+ Will be used as the label for the action's button.
+ """)
+
+ statuses = abstractproperty("""
+ Node states that this action makes sense for.
+
+ A collection of NODE_STATUS values. The action will be available
+ only if `node.status in action.statuses`.
+ """)
+
+ permission = abstractproperty("""
+ Required permission.
+
+ A NODE_PERMISSION value. The action will be available only if the
+ user has this given permission on the subject node.
+ """)
+
+ def __init__(self, node, user, request=None):
+ """Initialize a node action.
+
+ All node actions' initializers must accept these same arguments,
+ without variations.
+ """
+ self.node = node
+ self.user = user
+ self.request = request
+
+ def inhibit(self):
+ """Overridable: is there any reason not to offer this action?
+
+ This property may return a reason to inhibit this action, in which
+ case its button may still be visible in the UI, but disabled. A
+ tooltip will provide the reason, as returned by this method.
+
+ :return: A human-readable reason to inhibit the action, or None if
+ the action is valid.
+ """
+ return None
+
+ @abstractmethod
+ def perform(self):
+ """Perform this action.
+
+ Even though this is not the API, the action may raise
+ :class:`MAASAPIException` exceptions. When this happens, the view
+ will return to the client an http response reflecting the exception.
+
+ :return: A human-readable message confirming that the action has been
+ performed. It will be shown as an informational notice on the
+ Node page.
+ """
+
+ def is_permitted(self):
+ """Does the current user have the permission required?"""
+ return self.user.has_perm(self.permission, self.node)
+
+ # Uninitialized inhibititions cache.
+ _cached_inhibition = object()
+
+ @property
+ def inhibition(self):
+ """Caching version of `inhibit`."""
+ if self._cached_inhibition == NodeAction._cached_inhibition:
+ self._cached_inhibition = self.inhibit()
+ return self._cached_inhibition
+
+
+class Delete(NodeAction):
+ """Delete a node."""
+ display = "Delete node"
+ statuses = ALL_STATUSES
+ permission = NODE_PERMISSION.ADMIN
+
+ def inhibit(self):
+ if self.node.owner is not None:
+ return "You cannot delete this node because it's in use."
+ return None
+
+ def perform(self):
+ """Redirect to the delete view's confirmation page.
+
+ The rest of deletion is handled by a specialized deletion view.
+ All that the action really does is get you to its are-you-sure
+ page.
+ """
+ raise Redirect(reverse('node-delete', args=[self.node.system_id]))
+
+
+class AcceptAndCommission(NodeAction):
+ """Accept a node into the MAAS, and start the commissioning process."""
+ display = "Accept & commission"
+ statuses = (NODE_STATUS.DECLARED, )
+ permission = NODE_PERMISSION.ADMIN
+
+ def perform(self):
+ self.node.start_commissioning(self.user)
+ return "Node commissioning started."
+
+
+class RetryCommissioning(NodeAction):
+ """Retry commissioning of a node that failed previously."""
+ display = "Retry commissioning"
+ statuses = (NODE_STATUS.FAILED_TESTS, )
+ permission = NODE_PERMISSION.ADMIN
+
+ def perform(self):
+ self.node.start_commissioning(self.user)
+ return "Started a new attempt to commission this node."
+
+
+class StartNode(NodeAction):
+ """Acquire and start a node."""
+ display = "Start node"
+ statuses = (NODE_STATUS.READY, )
+ permission = NODE_PERMISSION.VIEW
+
+ def inhibit(self):
+ """The user must have an SSH key, so that they access the node."""
+ if not SSHKey.objects.get_keys_for_user(self.user).exists():
+ return dedent("""
+ You have no means of accessing the node after starting it.
+ Register an SSH key first. Do this on your Preferences
+ screen: click on the menu with your name at the top of the
+ page, select Preferences, and look for the "SSH keys" section.
+ """)
+ return None
+
+ def perform(self):
+ # Avoid circular imports.
+ from maasserver.api import get_oauth_token
+
+ # Be sure to acquire before starting, or start_nodes will think
+ # the node ineligible based on its un-acquired status.
+ self.node.acquire(get_oauth_token(self.request))
+ Node.objects.start_nodes([self.node.system_id], self.user)
+ return dedent("""
+ This node is now allocated to you.
+ It has been asked to start up.
+ """)
+
+
+ACTION_CLASSES = (
+ Delete,
+ AcceptAndCommission,
+ RetryCommissioning,
+ StartNode,
+ )
+
+
+def compile_node_actions(node, user, request=None, classes=ACTION_CLASSES):
+ """Provide :class:`NodeAction` objects for given request.
+
+ :param node: The :class:`Node` that the request pertains to.
+ :param user: The :class:`User` making the request.
+ :param request: The :class:`HttpRequest` being serviced. It may be used
+ to obtain information about the OAuth token being used.
+ :return: An :class:`OrderedDict` mapping applicable actions' display names
+ to corresponding :class:`NodeAction` instances. The dict is ordered
+ for consistent display.
+ """
+ applicable_actions = (
+ action_class(node, user, request)
+ for action_class in classes
+ if node.status in action_class.statuses)
+ return OrderedDict(
+ (action.display, action)
+ for action in applicable_actions
+ if action.is_permitted())
=== modified file 'src/maasserver/tests/test_forms.py'
--- src/maasserver/tests/test_forms.py 2012-04-26 07:57:26 +0000
+++ src/maasserver/tests/test_forms.py 2012-04-26 18:00:32 +0000
@@ -12,34 +12,23 @@
__metaclass__ = type
__all__ = []
-import httplib
-from urlparse import urlparse
-
from django import forms
from django.core.exceptions import (
PermissionDenied,
ValidationError,
)
-from django.core.urlresolvers import reverse
from django.http import QueryDict
-import maasserver.api
from maasserver.enum import (
ARCHITECTURE,
NODE_AFTER_COMMISSIONING_ACTION_CHOICES,
- NODE_PERMISSION,
NODE_STATUS,
- NODE_STATUS_CHOICES_DICT,
)
-from maasserver.exceptions import MAASAPIException
from maasserver.forms import (
ConfigForm,
- delete_node,
EditUserForm,
get_action_form,
HostnameFormField,
- inhibit_delete,
NewUserCreationForm,
- NODE_ACTIONS,
NodeActionForm,
NodeWithMACAddressesForm,
ProfileForm,
@@ -51,7 +40,10 @@
Config,
DEFAULT_CONFIG,
)
-from maasserver.provisioning import get_provisioning_api_proxy
+from maasserver.node_action import (
+ AcceptAndCommission,
+ Delete,
+ )
from maasserver.testing.factory import factory
from maasserver.testing.testcase import TestCase
from provisioningserver.enum import POWER_TYPE_CHOICES
@@ -247,125 +239,8 @@
self.assertEqual(power_type, node.power_type)
-class NodeActionsTests(TestCase):
- """Test the structure of NODE_ACTIONS."""
-
- def test_NODE_ACTIONS_initial_states(self):
- allowed_states = set(NODE_STATUS_CHOICES_DICT.keys() + [None])
- self.assertTrue(set(NODE_ACTIONS.keys()) <= allowed_states)
-
- def test_NODE_ACTIONS_dict_contains_only_accepted_keys(self):
- actions = sum(NODE_ACTIONS.values(), [])
- accepted_keys = set([
- 'permission',
- 'inhibit',
- 'display',
- 'execute',
- 'message',
- ])
- actual_keys = set(sum([action.keys() for action in actions], []))
- unknown_keys = actual_keys - accepted_keys
- self.assertEqual(set(), unknown_keys)
-
-
class TestNodeActionForm(TestCase):
- def test_inhibit_delete_allows_deletion_of_unowned_node(self):
- admin = factory.make_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- self.assertIsNone(inhibit_delete(node, admin))
-
- def test_inhibit_delete_inhibits_deletion_of_owned_node(self):
- admin = factory.make_admin()
- node = factory.make_node(
- status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
- self.assertEqual(
- "You cannot delete this node because it's in use.",
- inhibit_delete(node, admin))
-
- def test_delete_node_redirects_to_confirmation_page(self):
- admin = factory.make_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- try:
- delete_node(node, admin)
- except MAASAPIException as e:
- pass
- else:
- self.fail("delete_node should have raised a Redirect.")
- response = e.make_http_response()
- self.assertEqual(httplib.FOUND, response.status_code)
- self.assertEqual(
- reverse('node-delete', args=[node.system_id]),
- urlparse(response['Location']).path)
-
- def have_permission_returns_False_if_action_is_None(self):
- admin = factory.make_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- self.assertFalse(get_action_form(admin)(node).have_permission(None))
-
- def have_permission_returns_False_if_lacking_permission(self):
- user = factory.make_user()
- node = factory.make_node(
- status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
- action = {
- 'permission': NODE_PERMISSION.EDIT,
- }
- self.assertFalse(get_action_form(user)(node).have_permission(action))
-
- def have_permission_returns_True_given_permission(self):
- node = factory.make_node(
- status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
- action = {
- 'permission': NODE_PERMISSION.EDIT,
- }
- form = get_action_form(node.owner)(node)
- self.assertTrue(form.have_permission(action))
-
- def test_compile_action_records_inhibition(self):
- node = factory.make_node()
- user = factory.make_user()
- form = get_action_form(user)(node)
- inhibition = factory.getRandomString()
-
- def inhibit(*args, **kwargs):
- return inhibition
-
- compiled_action = form.compile_action({'inhibit': inhibit})
- self.assertEqual(inhibition, compiled_action['inhibition'])
-
- def test_compile_action_records_None_if_no_inhibition(self):
- node = factory.make_node()
- user = factory.make_user()
- form = get_action_form(user)(node)
-
- def inhibit(*args, **kwargs):
- return None
-
- compiled_action = form.compile_action({'inhibit': inhibit})
- self.assertIsNone(compiled_action['inhibition'])
-
- def test_compile_actions_for_declared_node_admin(self):
- # Check which transitions are available for an admin on a
- # 'Declared' node.
- admin = factory.make_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- form = get_action_form(admin)(node)
- actions = form.compile_actions()
- self.assertItemsEqual(
- ["Accept & commission", "Delete node"],
- [action['display'] for action in actions])
- # All permissions should be ADMIN.
- self.assertEqual(
- [NODE_PERMISSION.ADMIN] * len(actions),
- [action['permission'] for actions in actions])
-
- def test_compile_actions_for_declared_node_simple_user(self):
- # A simple user sees no actions for a 'Declared' node.
- user = factory.make_user()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- form = get_action_form(user)(node)
- self.assertItemsEqual([], form.compile_actions())
-
def test_get_action_form_creates_form_class_with_attributes(self):
user = factory.make_admin()
form_class = get_action_form(user)
@@ -386,8 +261,8 @@
form = get_action_form(admin)(node)
self.assertItemsEqual(
- ["Accept & commission", "Delete node"],
- [action['display'] for action in form.action_buttons])
+ [AcceptAndCommission.display, Delete.display],
+ form.actions)
def test_get_action_form_for_user(self):
user = factory.make_user()
@@ -396,31 +271,28 @@
self.assertIsInstance(form, NodeActionForm)
self.assertEqual(node, form.node)
- self.assertItemsEqual({}, form.action_buttons)
+ self.assertItemsEqual({}, form.actions)
- def test_get_action_form_node_for_admin_save(self):
+ def test_save_performs_requested_action(self):
admin = factory.make_admin()
node = factory.make_node(status=NODE_STATUS.DECLARED)
form = get_action_form(admin)(
- node, {NodeActionForm.input_name: "Accept & commission"})
+ node, {NodeActionForm.input_name: AcceptAndCommission.display})
form.save()
-
self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
- def test_get_action_form_for_user_save(self):
+ def test_save_refuses_disallowed_action(self):
user = factory.make_user()
node = factory.make_node(status=NODE_STATUS.DECLARED)
form = get_action_form(user)(
- node, {NodeActionForm.input_name: "Enlist node"})
-
+ node, {NodeActionForm.input_name: AcceptAndCommission.display})
self.assertRaises(PermissionDenied, form.save)
- def test_get_action_form_for_user_save_unknown_trans(self):
+ def test_save_refuses_unknown_action(self):
user = factory.make_user()
node = factory.make_node(status=NODE_STATUS.DECLARED)
form = get_action_form(user)(
node, {NodeActionForm.input_name: factory.getRandomString()})
-
self.assertRaises(PermissionDenied, form.save)
def test_save_double_checks_for_inhibitions(self):
@@ -428,52 +300,10 @@
node = factory.make_node(
status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
form = get_action_form(admin)(
- node, {NodeActionForm.input_name: "Delete node"})
+ node, {NodeActionForm.input_name: Delete.display})
with ExpectedException(PermissionDenied, "You cannot delete.*"):
form.save()
- def test_start_action_acquires_and_starts_ready_node_for_user(self):
- node = factory.make_node(status=NODE_STATUS.READY)
- user = factory.make_user()
- factory.make_sshkey(user)
- consumer, token = user.get_profile().create_authorisation_token()
- self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
- form = get_action_form(user)(
- node, {NodeActionForm.input_name: "Start node"})
- form.save()
-
- power_status = get_provisioning_api_proxy().power_status
- self.assertEqual('start', power_status.get(node.system_id))
- self.assertEqual(NODE_STATUS.ALLOCATED, node.status)
- self.assertEqual(user, node.owner)
-
- def test_start_action_is_enabled_for_user_with_key(self):
- node = factory.make_node(status=NODE_STATUS.READY)
- user = factory.make_user()
- factory.make_sshkey(user)
- consumer, token = user.get_profile().create_authorisation_token()
- self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
- form = get_action_form(user)(node)
- # The user has an SSH key, so there is no inhibition to stop
- # them from starting a node.
- self.assertIsNone(form.find_action("Start node")['inhibition'])
-
- def test_start_action_is_disabled_for_keyless_user(self):
- node = factory.make_node(status=NODE_STATUS.READY)
- user = factory.make_user()
- consumer, token = user.get_profile().create_authorisation_token()
- self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
- form = get_action_form(user)(node)
- self.assertIn("SSH key", form.find_action("Start node")['inhibition'])
-
- def test_accept_and_commission_starts_commissioning(self):
- admin = factory.make_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- form = get_action_form(admin)(
- node, {NodeActionForm.input_name: "Accept & commission"})
- form.save()
- self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
-
class TestHostnameFormField(TestCase):
=== added file 'src/maasserver/tests/test_node_action.py'
--- src/maasserver/tests/test_node_action.py 1970-01-01 00:00:00 +0000
+++ src/maasserver/tests/test_node_action.py 2012-04-26 18:00:32 +0000
@@ -0,0 +1,232 @@
+# Copyright 2012 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for node actions."""
+
+from __future__ import (
+ absolute_import,
+ print_function,
+ unicode_literals,
+ )
+
+__metaclass__ = type
+__all__ = []
+
+from urlparse import urlparse
+
+from django.core.urlresolvers import reverse
+import maasserver.api
+from maasserver.enum import (
+ NODE_PERMISSION,
+ NODE_STATUS,
+ NODE_STATUS_CHOICES_DICT,
+ )
+from maasserver.exceptions import Redirect
+from maasserver.node_action import (
+ AcceptAndCommission,
+ compile_node_actions,
+ Delete,
+ NodeAction,
+ RetryCommissioning,
+ StartNode,
+ )
+from maasserver.provisioning import get_provisioning_api_proxy
+from maasserver.testing.factory import factory
+from maasserver.testing.testcase import TestCase
+
+
+ALL_STATUSES = NODE_STATUS_CHOICES_DICT.keys()
+
+
+class FakeNodeAction(NodeAction):
+ display = "Action label"
+ statuses = ALL_STATUSES
+ permission = NODE_PERMISSION.VIEW
+
+ # For testing: an inhibition for inhibit() to return.
+ fake_inhibition = None
+
+ def inhibit(self):
+ return self.fake_inhibition
+
+ def perform(self):
+ pass
+
+
+class TestNodeAction(TestCase):
+
+ def test_compile_node_actions_returns_available_actions(self):
+
+ class MyAction(FakeNodeAction):
+ display = factory.getRandomString()
+
+ actions = compile_node_actions(
+ factory.make_node(), factory.make_admin(), classes=[MyAction])
+ self.assertEqual([MyAction.display], actions.keys())
+
+ def test_compile_node_actions_checks_node_status(self):
+
+ class MyAction(FakeNodeAction):
+ statuses = (NODE_STATUS.READY, )
+
+ node = factory.make_node(status=NODE_STATUS.DECLARED)
+ actions = compile_node_actions(
+ node, factory.make_admin(), classes=[MyAction])
+ self.assertEqual({}, actions)
+
+ def test_compile_node_actions_checks_permission(self):
+
+ class MyAction(FakeNodeAction):
+ permission = NODE_PERMISSION.EDIT
+
+ node = factory.make_node(status=NODE_STATUS.COMMISSIONING)
+ actions = compile_node_actions(
+ node, factory.make_user(), classes=[MyAction])
+ self.assertEqual({}, actions)
+
+ def test_compile_node_actions_includes_inhibited_actions(self):
+
+ class MyAction(FakeNodeAction):
+ fake_inhibition = factory.getRandomString()
+
+ actions = compile_node_actions(
+ factory.make_node(), factory.make_admin(), classes=[MyAction])
+ self.assertEqual([MyAction.display], actions.keys())
+
+ def test_compile_node_actions_maps_display_names(self):
+
+ class Action1(FakeNodeAction):
+ display = factory.getRandomString()
+
+ class Action2(FakeNodeAction):
+ display = factory.getRandomString()
+
+ actions = compile_node_actions(
+ factory.make_node(), factory.make_admin(),
+ classes=[Action1, Action2])
+ for label, action in actions.items():
+ self.assertEqual(label, action.display)
+
+ def test_compile_node_actions_maintains_order(self):
+ labels = [factory.getRandomString() for counter in range(4)]
+ classes = [
+ type(b"Action%d" % counter, (FakeNodeAction,), {'display': label})
+ for counter, label in enumerate(labels)]
+ actions = compile_node_actions(
+ factory.make_node(), factory.make_admin(), classes=classes)
+ self.assertSequenceEqual(labels, actions.keys())
+ self.assertSequenceEqual(
+ labels, [action.display for action in actions.values()])
+
+ def test_is_permitted_allows_if_user_has_permission(self):
+
+ class MyAction(FakeNodeAction):
+ permission = NODE_PERMISSION.EDIT
+
+ node = factory.make_node(
+ status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
+ self.assertTrue(MyAction(node, node.owner).is_permitted())
+
+ def test_is_permitted_disallows_if_user_lacks_permission(self):
+
+ class MyAction(FakeNodeAction):
+ permission = NODE_PERMISSION.EDIT
+
+ node = factory.make_node(
+ status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
+ self.assertFalse(MyAction(node, factory.make_user()).is_permitted())
+
+ def test_inhibition_wraps_inhibit(self):
+ inhibition = factory.getRandomString()
+ action = FakeNodeAction(factory.make_node(), factory.make_user())
+ action.fake_inhibition = inhibition
+ self.assertEqual(inhibition, action.inhibition)
+
+ def test_inhibition_caches_inhibition(self):
+ # The inhibition property will call inhibit() only once. We can
+ # prove this by changing the string inhibit() returns; it won't
+ # affect the value of the property.
+ inhibition = factory.getRandomString()
+ action = FakeNodeAction(factory.make_node(), factory.make_user())
+ action.fake_inhibition = inhibition
+ self.assertEqual(inhibition, action.inhibition)
+ action.fake_inhibition = factory.getRandomString()
+ self.assertEqual(inhibition, action.inhibition)
+
+ def test_inhibition_caches_None(self):
+ # An inhibition of None is also faithfully cached. In other
+ # words, it doesn't get mistaken for an uninitialized cache or
+ # anything.
+ action = FakeNodeAction(factory.make_node(), factory.make_user())
+ action.fake_inhibition = None
+ self.assertIsNone(action.inhibition)
+ action.fake_inhibition = factory.getRandomString()
+ self.assertIsNone(action.inhibition)
+
+ def test_Delete_inhibit_allows_if_node_has_no_owner(self):
+ unowned_node = factory.make_node(status=NODE_STATUS.READY)
+ self.assertIsNone(
+ Delete(unowned_node, factory.make_admin()).inhibit())
+
+ def test_Delete_inhibit_disallows_if_node_has_owner(self):
+ owned_node = factory.make_node(
+ status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
+ action = Delete(owned_node, factory.make_admin())
+ inhibition = action.inhibit()
+ self.assertIsNotNone(inhibition)
+ self.assertEqual(
+ "You cannot delete this node because it's in use.", inhibition)
+
+ def test_Delete_redirects_to_node_delete_view(self):
+ node = factory.make_node()
+ action = Delete(node, factory.make_admin())
+ try:
+ action.perform()
+ except Redirect as e:
+ pass
+ self.assertEqual(
+ reverse('node-delete', args=[node.system_id]),
+ urlparse(unicode(e)).path)
+
+ def test_AcceptAndCommission_starts_commissioning(self):
+ node = factory.make_node(status=NODE_STATUS.DECLARED)
+ action = AcceptAndCommission(node, factory.make_admin())
+ action.perform()
+ self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
+ self.assertEqual(
+ 'start',
+ get_provisioning_api_proxy().power_status.get(node.system_id))
+
+ def test_RetryCommissioning_starts_commissioning(self):
+ node = factory.make_node(status=NODE_STATUS.FAILED_TESTS)
+ action = RetryCommissioning(node, factory.make_admin())
+ action.perform()
+ self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
+ self.assertEqual(
+ 'start',
+ get_provisioning_api_proxy().power_status.get(node.system_id))
+
+ def test_StartNode_inhibit_allows_user_with_SSH_key(self):
+ user_with_key = factory.make_user()
+ factory.make_sshkey(user_with_key)
+ self.assertIsNone(
+ StartNode(factory.make_node(), user_with_key).inhibit())
+
+ def test_StartNode_inhibit_disallows_user_without_SSH_key(self):
+ user_without_key = factory.make_user()
+ action = StartNode(factory.make_node(), user_without_key)
+ inhibition = action.inhibit()
+ self.assertIsNotNone(inhibition)
+ self.assertIn("SSH key", inhibition)
+
+ def test_StartNode_acquires_and_starts_node(self):
+ node = factory.make_node(status=NODE_STATUS.READY)
+ user = factory.make_user()
+ consumer, token = user.get_profile().create_authorisation_token()
+ self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
+ StartNode(node, user).perform()
+ self.assertEqual(NODE_STATUS.ALLOCATED, node.status)
+ self.assertEqual(user, node.owner)
+ self.assertEqual(
+ 'start',
+ get_provisioning_api_proxy().power_status.get(node.system_id))
=== modified file 'src/maasserver/tests/test_views_nodes.py'
--- src/maasserver/tests/test_views_nodes.py 2012-04-26 07:57:26 +0000
+++ src/maasserver/tests/test_views_nodes.py 2012-04-26 18:00:32 +0000
@@ -27,6 +27,7 @@
from maasserver.exceptions import NoRabbit
from maasserver.forms import NodeActionForm
from maasserver.models import Node
+from maasserver.node_action import StartNode
from maasserver.testing import (
get_content_links,
reload_object,
@@ -47,6 +48,12 @@
class NodeViewsTest(LoggedInTestCase):
+ def set_up_oauth_token(self):
+ """Set up an oauth token to be used for requests."""
+ profile = self.logged_in_user.get_profile()
+ consumer, token = profile.create_authorisation_token()
+ self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
+
def test_node_list_contains_link_to_node_view(self):
node = factory.make_node()
response = self.client.get(reverse('node-list'))
@@ -248,95 +255,44 @@
content_text = doc.cssselect('#content')[0].text_content()
self.assertNotIn("Error output", content_text)
- def test_view_node_POST_admin_can_delete_unused_node(self):
- self.become_admin()
+ def test_view_node_POST_performs_action(self):
+ factory.make_sshkey(self.logged_in_user)
+ self.set_up_oauth_token()
node = factory.make_node(status=NODE_STATUS.READY)
- response = self.client.post(
- reverse('node-view', args=[node.system_id]),
- data={NodeActionForm.input_name: "Delete node"})
- self.assertEqual(httplib.FOUND, response.status_code)
- self.assertEqual(
- reverse('node-delete', args=[node.system_id]),
- urlparse(response['Location']).path)
-
- def test_view_node_POST_admin_cannot_delete_used_node(self):
- self.become_admin()
- node = factory.make_node(
- status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
- response = self.client.post(
- reverse('node-view', args=[node.system_id]),
- data={NodeActionForm.input_name: "Delete node"})
- self.assertEqual(httplib.FORBIDDEN, response.status_code)
-
- def test_view_node_POST_admin_can_start_commissioning_node(self):
- self.become_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
- node_link = reverse('node-view', args=[node.system_id])
- response = self.client.post(
- node_link,
- data={
- NodeActionForm.input_name: "Accept & commission",
- })
- self.assertEqual(httplib.FOUND, response.status_code)
- self.assertEqual(
- NODE_STATUS.COMMISSIONING, reload_object(node).status)
-
- def test_view_node_POST_admin_can_retry_failed_commissioning(self):
- self.become_admin()
- node = factory.make_node(status=NODE_STATUS.FAILED_TESTS)
- node_link = reverse('node-view', args=[node.system_id])
- response = self.client.post(
- node_link,
- data={NodeActionForm.input_name: "Retry commissioning"})
- self.assertEqual(httplib.FOUND, response.status_code)
- self.assertEqual(
- NODE_STATUS.COMMISSIONING, reload_object(node).status)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.post(
+ node_link, data={NodeActionForm.input_name: StartNode.display})
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertEqual(NODE_STATUS.ALLOCATED, reload_object(node).status)
def perform_action_and_get_node_page(self, node, action_name):
"""POST to perform a node action, then load the resulting page."""
node_link = reverse('node-view', args=[node.system_id])
response = self.client.post(
- node_link,
- data={
- NodeActionForm.input_name: action_name,
- })
+ node_link, data={NodeActionForm.input_name: action_name})
if response.status_code != httplib.FOUND:
- self.fail(
- "POST failed with code %d: '%s'"
- % (response.status_code, response.content))
+ self.fail("%d: '%s'" % (response.status_code, response.content))
redirect = urlparse(response['Location']).path
if redirect != node_link:
- self.fail(
- "Odd: POST on %s redirected to %s." % (node_link, redirect))
+ self.fail("Odd: %s redirected to %s." % (node_link, redirect))
return self.client.get(redirect)
- def test_start_commisioning_displays_message(self):
- self.become_admin()
- node = factory.make_node(status=NODE_STATUS.DECLARED)
+ def test_view_node_POST_action_displays_message(self):
+ factory.make_sshkey(self.logged_in_user)
+ self.set_up_oauth_token()
+ node = factory.make_node(status=NODE_STATUS.READY)
response = self.perform_action_and_get_node_page(
- node, "Accept & commission")
+ node, StartNode.display)
self.assertIn(
- "Node commissioning started.",
- [message.message for message in response.context['messages']])
-
- def test_start_node_displays_message(self):
- factory.make_sshkey(self.logged_in_user)
- profile = self.logged_in_user.get_profile()
- consumer, token = profile.create_authorisation_token()
- self.patch(maasserver.api, 'get_oauth_token', lambda request: token)
- node = factory.make_node(status=NODE_STATUS.READY)
- response = self.perform_action_and_get_node_page(node, "Start node")
- notices = '\n'.join(
- message.message for message in response.context['messages'])
- self.assertIn("This node is now allocated to you.", notices)
- self.assertIn("asked to start up.", notices)
-
- def test_start_node_without_auth_returns_Unauthorized(self):
+ "This node is now allocated to you.",
+ '\n'.join(msg.message for msg in response.context['messages']))
+
+ def test_view_node_POST_without_oauth_returns_Unauthorized(self):
factory.make_sshkey(self.logged_in_user)
node = factory.make_node(status=NODE_STATUS.READY)
response = self.client.post(
reverse('node-view', args=[node.system_id]),
- data={NodeActionForm.input_name: "Start node"})
+ data={NodeActionForm.input_name: StartNode.display})
self.assertEqual(httplib.UNAUTHORIZED, response.status_code)
Follow ups
