launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #07302
[Merge] lp:~jtv/maas/backport-987585 into lp:maas
Jeroen T. Vermeulen has proposed merging lp:~jtv/maas/backport-987585 into lp:maas.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
Related bugs:
Bug #987585 in MAAS: "Start Node button does not allocate node"
https://bugs.launchpad.net/maas/+bug/987585
For more details, see:
https://code.launchpad.net/~jtv/maas/backport-987585/+merge/103641
Francis asked for this to be backported to the 1.0 branch. What you see here not entirely a rewrite of the original fix in trunk, but neither is it a simple backport. Too many things had changed between trunk and 1.0. Some seemingly unrelated changes from trunk I introduced by hand (e.g. saving after acquire/release, from inside the model methods) but only minimally so (for example I didn't remove the now-obsolete saves after calls to acquire/release). I didn't even bother with the view changes.
The backport is not of the same quality and robustness as the trunk fix; to do so would have required importing tons more changes from trunk. We really have to get off the 1.0 branch soon. Functionally trunk is still very similar to 1.0, but it's miles ahead in quality and polish, and pulling away.
Jeroen
--
https://code.launchpad.net/~jtv/maas/backport-987585/+merge/103641
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jtv/maas/backport-987585 into lp:maas.
=== modified file 'src/maasserver/api.py'
--- src/maasserver/api.py 2012-04-24 16:46:33 +0000
+++ src/maasserver/api.py 2012-04-26 09:45:38 +0000
@@ -309,6 +309,7 @@
return None
+<<<<<<< TREE
def extract_oauth_key(request):
"""Extract the oauth key from a request's headers.
@@ -335,6 +336,21 @@
raise Unauthorized("Unknown OAuth token.")
+=======
+def get_oauth_token(request):
+ """Get the OAuth :class:`piston.models.Token` used for `request`.
+
+ Raises :class:`Unauthorized` if no key is found, or if the token is
+ unknown.
+ """
+ auth_header = request.META.get('HTTP_AUTHORIZATION')
+ try:
+ return Token.objects.get(key=extract_oauth_key(auth_header))
+ except Token.DoesNotExist:
+ raise Unauthorized("Unknown OAuth token.")
+
+
+>>>>>>> MERGE-SOURCE
NODE_FIELDS = (
'system_id',
'hostname',
=== modified file 'src/maasserver/forms.py'
--- src/maasserver/forms.py 2012-04-26 07:57:26 +0000
+++ src/maasserver/forms.py 2012-04-26 09:45:38 +0000
@@ -236,6 +236,7 @@
return node
+<<<<<<< TREE
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():
@@ -255,9 +256,18 @@
from maasserver.api import get_oauth_token
node.acquire(get_oauth_token(request))
+=======
+def acquire_and_start_node(node, user, request=None):
+ """Acquire and start a node from the UI. It will have no meta_data."""
+ # Avoid circular imports.
+ from maasserver.api import get_oauth_token
+
+ node.acquire(get_oauth_token(request))
+>>>>>>> MERGE-SOURCE
Node.objects.start_nodes([node.system_id], user)
+<<<<<<< TREE
def start_commissioning_node(node, user, request=None):
"""Start the commissioning process for a node."""
node.start_commissioning(user)
@@ -280,6 +290,13 @@
raise Redirect(reverse('node-delete', args=[node.system_id]))
+=======
+def start_commissioning_node(node, user, request=None):
+ """Start the commissioning process for a node."""
+ node.start_commissioning(user)
+
+
+>>>>>>> MERGE-SOURCE
# Node actions per status.
#
# This maps each NODE_STATUS to a list of actions applicable to a node
@@ -302,6 +319,7 @@
#
# # Permission required on the node to perform the action.
# 'permission': NODE_PERMISSION.EDIT,
+<<<<<<< TREE
#
# # Function that looks for reasons to disable the action.
# # If it returns None, the action is available; otherwise, it is
@@ -314,6 +332,14 @@
# # MAASAPIException if it wants to convey failure with a specific
# # http status code.
# 'execute': lambda node, user, request=None: paint_node(
+=======
+# # 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: paint_node(
+>>>>>>> MERGE-SOURCE
# node, favourite_colour(user)),
# }
#
@@ -332,6 +358,7 @@
{
'display': "Accept & commission",
'permission': NODE_PERMISSION.ADMIN,
+<<<<<<< TREE
'execute': start_commissioning_node,
'message': "Node commissioning started.",
},
@@ -350,11 +377,25 @@
],
NODE_STATUS.MISSING: [
DELETE_ACTION,
+=======
+ 'execute': start_commissioning_node,
+ 'message': "Node commissioning started.",
+ },
+ ],
+ NODE_STATUS.FAILED_TESTS: [
+ {
+ 'display': "Retry commissioning",
+ 'permission': NODE_PERMISSION.ADMIN,
+ 'execute': start_commissioning_node,
+ 'message': "Started a new attempt to commission this node.",
+ },
+>>>>>>> MERGE-SOURCE
],
NODE_STATUS.READY: [
DELETE_ACTION,
{
'display': "Start node",
+<<<<<<< TREE
'permission': NODE_PERMISSION.VIEW,
'execute': acquire_and_start_node,
'message': (
@@ -371,6 +412,14 @@
],
NODE_STATUS.RETIRED: [
DELETE_ACTION,
+=======
+ 'permission': NODE_PERMISSION.VIEW,
+ 'execute': acquire_and_start_node,
+ 'message': (
+ "This node is now allocated to you. "
+ "It has been asked to start up."),
+ },
+>>>>>>> MERGE-SOURCE
],
}
@@ -456,6 +505,7 @@
def save(self):
"""An action was requested. Perform it."""
action_name = self.data.get(self.input_name)
+<<<<<<< TREE
action = self.find_action(action_name)
if not self.have_permission(action):
@@ -467,6 +517,14 @@
action['execute'](
node=self.node, user=self.user, request=self.request)
self.display_message(action.get('message'))
+=======
+ permission, execute, message = (
+ self.action_dict.get(action_name, (None, None, None)))
+ if execute is None or not self.user.has_perm(permission, self.node):
+ raise PermissionDenied()
+ execute(node=self.node, user=self.user, request=self.request)
+ self.display_message(message)
+>>>>>>> MERGE-SOURCE
def get_action_form(user, request=None):
=== modified file 'src/maasserver/models/__init__.py'
=== modified file 'src/maasserver/tests/test_api.py'
--- src/maasserver/tests/test_api.py 2012-04-26 05:44:05 +0000
+++ src/maasserver/tests/test_api.py 2012-04-26 09:45:38 +0000
@@ -28,6 +28,7 @@
from maasserver.api import (
extract_constraints,
extract_oauth_key,
+<<<<<<< TREE
extract_oauth_key_from_auth_header,
get_oauth_token,
)
@@ -38,6 +39,11 @@
NODE_STATUS_CHOICES_DICT,
)
from maasserver.exceptions import Unauthorized
+=======
+ get_oauth_token,
+ )
+from maasserver.exceptions import Unauthorized
+>>>>>>> MERGE-SOURCE
from maasserver.models import (
Config,
create_auth_token,
@@ -79,6 +85,7 @@
class TestModuleHelpers(TestCase):
+<<<<<<< TREE
def make_fake_request(self, auth_header):
"""Create a very simple fake request, with just an auth header."""
FakeRequest = namedtuple('FakeRequest', ['META'])
@@ -126,6 +133,37 @@
self.assertRaises(
Unauthorized,
get_oauth_token, self.make_fake_request(header))
+=======
+ def make_fake_request(self, auth_header):
+ """Create a very simple fake request, with just an auth header."""
+ FakeRequest = namedtuple('FakeRequest', ['META'])
+ return FakeRequest(META={'HTTP_AUTHORIZATION': auth_header})
+
+ def test_extract_oauth_key_extracts_oauth_token_from_oauth_header(self):
+ token = factory.getRandomString(18)
+ self.assertEqual(
+ token,
+ extract_oauth_key(factory.make_oauth_header(oauth_token=token)))
+
+ def test_extract_oauth_key_returns_None_without_oauth_key(self):
+ self.assertIs(None, extract_oauth_key(''))
+>>>>>>> MERGE-SOURCE
+
+ def test_get_oauth_token_finds_token(self):
+ user = factory.make_user()
+ consumer, token = user.get_profile().create_authorisation_token()
+ self.assertEqual(
+ token,
+ get_oauth_token(
+ self.make_fake_request(
+ factory.make_oauth_header(oauth_token=token.key))))
+
+ def test_get_oauth_token_raises_Unauthorized_for_unknown_token(self):
+ fake_token = factory.getRandomString(18)
+ header = factory.make_oauth_header(oauth_token=fake_token)
+ self.assertRaises(
+ Unauthorized,
+ get_oauth_token, self.make_fake_request(header))
def test_extract_constraints_ignores_unknown_parameters(self):
unknown_parameter = "%s=%s" % (
=== 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 09:45:38 +0000
@@ -22,6 +22,7 @@
)
from django.core.urlresolvers import reverse
from django.http import QueryDict
+<<<<<<< TREE
import maasserver.api
from maasserver.enum import (
ARCHITECTURE,
@@ -31,6 +32,9 @@
NODE_STATUS_CHOICES_DICT,
)
from maasserver.exceptions import MAASAPIException
+=======
+import maasserver.api
+>>>>>>> MERGE-SOURCE
from maasserver.forms import (
ConfigForm,
delete_node,
@@ -423,6 +427,7 @@
self.assertRaises(PermissionDenied, form.save)
+<<<<<<< TREE
def test_save_double_checks_for_inhibitions(self):
admin = factory.make_admin()
node = factory.make_node(
@@ -473,6 +478,29 @@
node, {NodeActionForm.input_name: "Accept & commission"})
form.save()
self.assertEqual(NODE_STATUS.COMMISSIONING, node.status)
+=======
+ def test_start_action_acquires_and_starts_ready_node_for_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, {NodeActionForm.input_name: "Start node"})
+ form.save()
+
+ power_status = get_provisioning_api_proxy().power_status
+ self.assertEqual(NODE_STATUS.ALLOCATED, node.status)
+ self.assertEqual(user, node.owner)
+ self.assertEqual('start', power_status.get(node.system_id))
+
+ 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)
+>>>>>>> MERGE-SOURCE
class TestHostnameFormField(TestCase):
=== modified file 'src/maasserver/tests/test_views.py'
--- src/maasserver/tests/test_views.py 2012-04-23 14:02:51 +0000
+++ src/maasserver/tests/test_views.py 2012-04-26 09:45:38 +0000
@@ -104,6 +104,7 @@
'select#id_after_commissioning_action')
+<<<<<<< TREE
class FakeDeletableModel:
"""A fake model class, with a delete method."""
@@ -210,6 +211,602 @@
"%s deleted." % object_name.capitalize(),
view.compose_feedback_deleted(view.obj))
+=======
+class TestProxyView(LoggedInTestCase):
+ """Test the (dev) view used to proxy request to a txlongpoll server."""
+
+ def test_proxy_to_longpoll(self):
+ # Set LONGPOLL_SERVER_URL (to a random string).
+ longpoll_server_url = factory.getRandomString()
+ self.patch(settings, 'LONGPOLL_SERVER_URL', longpoll_server_url)
+
+ # Create content of the fake reponse.
+ query_string = factory.getRandomString()
+ mimetype = factory.getRandomString()
+ content = factory.getRandomString()
+ status_code = factory.getRandomStatusCode()
+
+ # Monkey patch urllib2.urlopen to make it return a (fake) response
+ # with status_code=code, headers.typeheader=mimetype and a
+ # 'read' method that will return 'content'.
+ def urlopen(url):
+ # Assert that urlopen is called on the longpoll url (plus
+ # additional parameters taken from the original request's
+ # query string).
+ self.assertEqual(
+ '%s?%s' % (longpoll_server_url, query_string), url)
+ FakeProxiedResponse = namedtuple(
+ 'FakeProxiedResponse', 'code headers read')
+ headers = namedtuple('Headers', 'typeheader')(mimetype)
+ return FakeProxiedResponse(status_code, headers, lambda: content)
+ self.patch(urllib2, 'urlopen', urlopen)
+
+ # Create a fake request.
+ request = namedtuple(
+ 'FakeRequest', ['META'])({'QUERY_STRING': query_string})
+ response = proxy_to_longpoll(request)
+
+ self.assertEqual(content, response.content)
+ self.assertEqual(mimetype, response['Content-Type'])
+ self.assertEqual(status_code, response.status_code)
+
+
+class TestGetLongpollenabled(TestCase):
+
+ def test_longpoll_not_included_if_LONGPOLL_SERVER_URL_None(self):
+ self.patch(settings, 'LONGPOLL_PATH', factory.getRandomString())
+ self.patch(settings, 'LONGPOLL_SERVER_URL', None)
+ self.assertFalse(get_proxy_longpoll_enabled())
+
+ def test_longpoll_not_included_if_LONGPOLL_PATH_None(self):
+ self.patch(settings, 'LONGPOLL_PATH', None)
+ self.patch(settings, 'LONGPOLL_SERVER_URL', factory.getRandomString())
+ self.assertFalse(get_proxy_longpoll_enabled())
+
+ def test_longpoll_included_if_LONGPOLL_PATH_and_LONGPOLL_SERVER_URL(self):
+ self.patch(settings, 'LONGPOLL_PATH', factory.getRandomString())
+ self.patch(settings, 'LONGPOLL_SERVER_URL', factory.getRandomString())
+ self.assertTrue(get_proxy_longpoll_enabled())
+
+
+class TestComboLoaderView(TestCase):
+ """Test combo loader view."""
+
+ def test_load_js(self):
+ requested_files = [
+ 'tests/build/oop/oop.js',
+ 'tests/build/event-custom-base/event-custom-base.js'
+ ]
+ response = self.client.get('/combo/?%s' % '&'.join(requested_files))
+ self.assertIn('text/javascript', response['Content-Type'])
+ for requested_file in requested_files:
+ self.assertIn(requested_file, response.content)
+ # No sign of a missing js file.
+ self.assertNotIn("/* [missing] */", response.content)
+ # The file contains a link to YUI's licence.
+ self.assertIn('http://yuilibrary.com/license/', response.content)
+
+ def test_load_css(self):
+ requested_files = [
+ 'tests/build/widget-base/assets/skins/sam/widget-base.css',
+ 'tests/build/widget-stack/assets/skins/sam/widget-stack.css',
+ ]
+ response = self.client.get('/combo/?%s' % '&'.join(requested_files))
+ self.assertIn('text/css', response['Content-Type'])
+ for requested_file in requested_files:
+ self.assertIn(requested_file, response.content)
+ # No sign of a missing css file.
+ self.assertNotIn("/* [missing] */", response.content)
+ # The file contains a link to YUI's licence.
+ self.assertIn('http://yuilibrary.com/license/', response.content)
+
+ def test_combo_no_file_returns_not_found(self):
+ response = self.client.get('/combo/')
+ self.assertEqual(httplib.NOT_FOUND, response.status_code)
+
+ def test_combo_wrong_file_extension_returns_bad_request(self):
+ response = self.client.get('/combo/?file.wrongextension')
+ self.assertEqual(httplib.BAD_REQUEST, response.status_code)
+ self.assertEqual("Invalid file type requested.", response.content)
+
+
+class TestUtilities(TestCase):
+
+ def test_get_yui_location_if_static_root_is_none(self):
+ self.patch(settings, 'STATIC_ROOT', None)
+ yui_location = os.path.join(
+ os.path.dirname(os.path.dirname(__file__)),
+ 'static', 'jslibs', 'yui')
+ self.assertEqual(yui_location, get_yui_location())
+
+ def test_get_yui_location(self):
+ static_root = factory.getRandomString()
+ self.patch(settings, 'STATIC_ROOT', static_root)
+ yui_location = os.path.join(static_root, 'jslibs', 'yui')
+ self.assertEqual(yui_location, get_yui_location())
+
+ def test_get_longpoll_context_empty_if_rabbitmq_publish_is_none(self):
+ self.patch(settings, 'RABBITMQ_PUBLISH', None)
+ self.patch(views, 'messaging', messages.get_messaging())
+ self.assertEqual({}, get_longpoll_context())
+
+ def test_get_longpoll_context_returns_empty_if_rabbit_not_running(self):
+
+ class FakeMessaging:
+ """Fake :class:`RabbitMessaging`: fail with `NoRabbit`."""
+
+ def getQueue(self, *args, **kwargs):
+ raise NoRabbit("Pretending not to have a rabbit.")
+
+ self.patch(messages, 'messaging', FakeMessaging())
+ self.assertEqual({}, get_longpoll_context())
+
+ def test_get_longpoll_context_empty_if_longpoll_url_is_None(self):
+ self.patch(settings, 'LONGPOLL_PATH', None)
+ self.patch(views, 'messaging', messages.get_messaging())
+ self.assertEqual({}, get_longpoll_context())
+
+ @uses_rabbit_fixture
+ def test_get_longpoll_context(self):
+ longpoll = factory.getRandomString()
+ self.patch(settings, 'LONGPOLL_PATH', longpoll)
+ self.patch(settings, 'RABBITMQ_PUBLISH', True)
+ self.patch(views, 'messaging', messages.get_messaging())
+ context = get_longpoll_context()
+ self.assertItemsEqual(
+ ['LONGPOLL_PATH', 'longpoll_queue'], list(context))
+ self.assertEqual(longpoll, context['LONGPOLL_PATH'])
+
+ def test_make_path_relative_if_prefix(self):
+ url_without_prefix = factory.getRandomString()
+ url = '/%s' % url_without_prefix
+ self.assertEqual(url_without_prefix, make_path_relative(url))
+
+ def test_make_path_relative_if_no_prefix(self):
+ url_without_prefix = factory.getRandomString()
+ self.assertEqual(
+ url_without_prefix, make_path_relative(url_without_prefix))
+
+
+class UserPrefsViewTest(LoggedInTestCase):
+
+ def test_prefs_GET_profile(self):
+ # The preferences page displays a form with the user's personal
+ # information.
+ user = self.logged_in_user
+ user.last_name = 'Steve Bam'
+ user.save()
+ response = self.client.get('/account/prefs/')
+ doc = fromstring(response.content)
+ self.assertSequenceEqual(
+ ['Steve Bam'],
+ [elem.value for elem in
+ doc.cssselect('input#id_profile-last_name')])
+
+ def test_prefs_GET_api(self):
+ # The preferences page displays the API access tokens.
+ user = self.logged_in_user
+ # Create a few tokens.
+ for i in range(3):
+ user.get_profile().create_authorisation_token()
+ response = self.client.get('/account/prefs/')
+ doc = fromstring(response.content)
+ # The OAuth tokens are displayed.
+ for token in user.get_profile().get_authorisation_tokens():
+ consumer = token.consumer
+ # The token string is a compact representation of the keys.
+ token_string = '%s:%s:%s' % (consumer.key, token.key, token.secret)
+ self.assertSequenceEqual(
+ [token_string],
+ [elem.value.strip() for elem in
+ doc.cssselect('input#%s' % token.key)])
+
+ def test_prefs_POST_profile(self):
+ # The preferences page allows the user the update its profile
+ # information.
+ params = {
+ 'last_name': 'John Doe',
+ 'email': 'jon@xxxxxxxxxxx',
+ }
+ response = self.client.post(
+ '/account/prefs/', get_prefixed_form_data('profile', params))
+
+ self.assertEqual(httplib.FOUND, response.status_code)
+ user = User.objects.get(id=self.logged_in_user.id)
+ self.assertAttributes(user, params)
+
+ def test_prefs_POST_password(self):
+ # The preferences page allows the user to change his password.
+ self.logged_in_user.set_password('password')
+ old_pw = self.logged_in_user.password
+ response = self.client.post(
+ '/account/prefs/',
+ get_prefixed_form_data(
+ 'password',
+ {
+ 'old_password': 'test',
+ 'new_password1': 'new',
+ 'new_password2': 'new',
+ }))
+
+ self.assertEqual(httplib.FOUND, response.status_code)
+ user = User.objects.get(id=self.logged_in_user.id)
+ # The password is SHA1ized, we just make sure that it has changed.
+ self.assertNotEqual(old_pw, user.password)
+
+ def test_prefs_displays_message_when_no_public_keys_are_configured(self):
+ response = self.client.get('/account/prefs/')
+ self.assertIn("No SSH key configured.", response.content)
+
+ def test_prefs_displays_add_ssh_key_button(self):
+ response = self.client.get('/account/prefs/')
+ add_key_link = reverse('prefs-add-sshkey')
+ self.assertIn(add_key_link, get_content_links(response))
+
+ def test_prefs_displays_compact_representation_of_users_keys(self):
+ _, keys = factory.make_user_with_keys(user=self.logged_in_user)
+ response = self.client.get('/account/prefs/')
+ for key in keys:
+ self.assertIn(key.display_html(), response.content)
+
+ def test_prefs_displays_link_to_delete_ssh_keys(self):
+ _, keys = factory.make_user_with_keys(user=self.logged_in_user)
+ response = self.client.get('/account/prefs/')
+ links = get_content_links(response)
+ for key in keys:
+ del_key_link = reverse('prefs-delete-sshkey', args=[key.id])
+ self.assertIn(del_key_link, links)
+
+
+class KeyManagementTest(LoggedInTestCase):
+
+ def test_add_key_GET(self):
+ # The 'Add key' page displays a form to add a key.
+ response = self.client.get(reverse('prefs-add-sshkey'))
+ doc = fromstring(response.content)
+
+ self.assertEqual(1, len(doc.cssselect('textarea#id_key')))
+ # The page features a form that submits to itself.
+ self.assertSequenceEqual(
+ ['.'],
+ [elem.get('action').strip() for elem in doc.cssselect(
+ '#content form')])
+
+ def test_add_key_POST_adds_key(self):
+ key_string = get_data('data/test_rsa0.pub')
+ response = self.client.post(
+ reverse('prefs-add-sshkey'), {'key': key_string})
+
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertTrue(SSHKey.objects.filter(key=key_string).exists())
+
+ def test_add_key_POST_fails_if_key_already_exists_for_the_user(self):
+ key_string = get_data('data/test_rsa0.pub')
+ key = SSHKey(user=self.logged_in_user, key=key_string)
+ key.save()
+ response = self.client.post(
+ reverse('prefs-add-sshkey'), {'key': key_string})
+
+ self.assertEqual(httplib.OK, response.status_code)
+ self.assertIn(
+ "This key has already been added for this user.",
+ response.content)
+ self.assertItemsEqual([key], SSHKey.objects.filter(key=key_string))
+
+ def test_key_can_be_added_if_same_key_already_setup_for_other_user(self):
+ key_string = get_data('data/test_rsa0.pub')
+ key = SSHKey(user=factory.make_user(), key=key_string)
+ key.save()
+ response = self.client.post(
+ reverse('prefs-add-sshkey'), {'key': key_string})
+ new_key = SSHKey.objects.get(key=key_string, user=self.logged_in_user)
+
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertItemsEqual(
+ [key, new_key], SSHKey.objects.filter(key=key_string))
+
+ def test_delete_key_GET(self):
+ # The 'Delete key' page displays a confirmation page with a form.
+ key = factory.make_sshkey(self.logged_in_user)
+ del_link = reverse('prefs-delete-sshkey', args=[key.id])
+ response = self.client.get(del_link)
+ doc = fromstring(response.content)
+
+ self.assertIn(
+ "Are you sure you want to delete the following key?",
+ response.content)
+ # The page features a form that submits to itself.
+ self.assertSequenceEqual(
+ ['.'],
+ [elem.get('action').strip() for elem in doc.cssselect(
+ '#content form')])
+
+ def test_delete_key_GET_cannot_access_someoneelses_key(self):
+ key = factory.make_sshkey(factory.make_user())
+ del_link = reverse('prefs-delete-sshkey', args=[key.id])
+ response = self.client.get(del_link)
+
+ self.assertEqual(httplib.NOT_FOUND, response.status_code)
+
+ def test_delete_key_POST(self):
+ # A POST request deletes the key.
+ key = factory.make_sshkey(self.logged_in_user)
+ del_link = reverse('prefs-delete-sshkey', args=[key.id])
+ response = self.client.post(del_link, {'post': 'yes'})
+
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertFalse(SSHKey.objects.filter(id=key.id).exists())
+
+
+class AdminLoggedInTestCase(LoggedInTestCase):
+
+ def setUp(self):
+ super(AdminLoggedInTestCase, self).setUp()
+ # Promote the logged-in user to admin.
+ self.logged_in_user.is_superuser = True
+ self.logged_in_user.save()
+
+
+def get_content_links(response, element='#content'):
+ """Extract links from :class:`HttpResponse` #content element."""
+ doc = fromstring(response.content)
+ [content_node] = doc.cssselect(element)
+ return [elem.get('href') for elem in content_node.cssselect('a')]
+
+
+class NodeViewsTest(LoggedInTestCase):
+
+ def test_node_list_contains_link_to_node_view(self):
+ node = factory.make_node()
+ response = self.client.get(reverse('node-list'))
+ node_link = reverse('node-view', args=[node.system_id])
+ self.assertIn(node_link, get_content_links(response))
+
+ def test_node_list_displays_sorted_list_of_nodes(self):
+ # Nodes are sorted on the node list page, newest first.
+ nodes = [factory.make_node() for i in range(3)]
+ nodes.reverse()
+ # Modify one node to make sure that the default db ordering
+ # (by modification date) is not used.
+ node = nodes[1]
+ node.hostname = factory.getRandomString()
+ node.save()
+ response = self.client.get(reverse('node-list'))
+ node_links = [
+ reverse('node-view', args=[node.system_id])
+ for node in nodes]
+ self.assertEqual(
+ node_links,
+ [link for link in get_content_links(response)
+ if link.startswith('/nodes')])
+
+ def test_view_node_displays_node_info(self):
+ # The node page features the basic information about the node.
+ node = factory.make_node(owner=self.logged_in_user)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ doc = fromstring(response.content)
+ content_text = doc.cssselect('#content')[0].text_content()
+ self.assertIn(node.hostname, content_text)
+ self.assertIn(node.display_status(), content_text)
+ self.assertIn(self.logged_in_user.username, content_text)
+
+ def test_view_node_displays_node_info_no_owner(self):
+ # If the node has no owner, the Owner 'slot' does not exist.
+ node = factory.make_node()
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ doc = fromstring(response.content)
+ content_text = doc.cssselect('#content')[0].text_content()
+ self.assertNotIn('Owner', content_text)
+
+ def test_view_node_displays_link_to_edit_if_user_owns_node(self):
+ node = factory.make_node(owner=self.logged_in_user)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ node_edit_link = reverse('node-edit', args=[node.system_id])
+ self.assertIn(node_edit_link, get_content_links(response))
+
+ def test_view_node_does_not_show_link_to_delete_node(self):
+ # Only admin users can delete nodes.
+ node = factory.make_node(owner=self.logged_in_user)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+ self.assertNotIn(node_delete_link, get_content_links(response))
+
+ def test_user_cannot_delete_node(self):
+ node = factory.make_node(owner=self.logged_in_user)
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+ response = self.client.get(node_delete_link)
+ self.assertEqual(httplib.FORBIDDEN, response.status_code)
+
+ def test_view_node_shows_message_for_commissioning_node(self):
+ statuses_with_message = (
+ NODE_STATUS.READY, NODE_STATUS.COMMISSIONING)
+ help_link = "https://wiki.ubuntu.com/ServerTeam/MAAS/AvahiBoot";
+ for status in map_enum(NODE_STATUS).values():
+ node = factory.make_node(status=status)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ links = get_content_links(response, '#flash-messages')
+ if status in statuses_with_message:
+ self.assertIn(help_link, links)
+ else:
+ self.assertNotIn(help_link, links)
+
+ def test_view_node_shows_link_to_delete_node_for_admin(self):
+ self.become_admin()
+ node = factory.make_node()
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+ self.assertIn(node_delete_link, get_content_links(response))
+
+ def test_admin_can_delete_nodes(self):
+ self.become_admin()
+ node = factory.make_node()
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+ response = self.client.post(node_delete_link, {'post': 'yes'})
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertFalse(Node.objects.filter(id=node.id).exists())
+
+ def test_allocated_node_view_page_says_node_cannot_be_deleted(self):
+ self.become_admin()
+ node = factory.make_node(
+ status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
+ node_view_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_view_link)
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+
+ self.assertEqual(httplib.OK, response.status_code)
+ self.assertNotIn(node_delete_link, get_content_links(response))
+ self.assertIn(
+ "You cannot delete this node because it's in use.",
+ response.content)
+
+ def test_allocated_node_cannot_be_deleted(self):
+ self.become_admin()
+ node = factory.make_node(
+ status=NODE_STATUS.ALLOCATED, owner=factory.make_user())
+ node_delete_link = reverse('node-delete', args=[node.system_id])
+ response = self.client.get(node_delete_link)
+
+ self.assertEqual(httplib.FORBIDDEN, response.status_code)
+
+ def test_user_cannot_view_someone_elses_node(self):
+ node = factory.make_node(owner=factory.make_user())
+ node_view_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_view_link)
+ self.assertEqual(httplib.FORBIDDEN, response.status_code)
+
+ def test_user_cannot_edit_someone_elses_node(self):
+ node = factory.make_node(owner=factory.make_user())
+ node_edit_link = reverse('node-edit', args=[node.system_id])
+ response = self.client.get(node_edit_link)
+ self.assertEqual(httplib.FORBIDDEN, response.status_code)
+
+ def test_admin_can_view_someonelses_node(self):
+ self.become_admin()
+ node = factory.make_node(owner=factory.make_user())
+ node_view_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_view_link)
+ self.assertEqual(httplib.OK, response.status_code)
+
+ def test_admin_can_edit_someonelses_node(self):
+ self.become_admin()
+ node = factory.make_node(owner=factory.make_user())
+ node_edit_link = reverse('node-edit', args=[node.system_id])
+ response = self.client.get(node_edit_link)
+ self.assertEqual(httplib.OK, response.status_code)
+
+ def test_user_can_access_the_edition_page_for_his_nodes(self):
+ node = factory.make_node(owner=self.logged_in_user)
+ node_edit_link = reverse('node-edit', args=[node.system_id])
+ response = self.client.get(node_edit_link)
+ self.assertEqual(httplib.OK, response.status_code)
+
+ def test_user_can_edit_his_nodes(self):
+ node = factory.make_node(owner=self.logged_in_user)
+ node_edit_link = reverse('node-edit', args=[node.system_id])
+ params = {
+ 'hostname': factory.getRandomString(),
+ # XXX JeroenVermeulen 2012-04-12, bug=979539: re-enable.
+ #'after_commissioning_action': factory.getRandomEnum(
+ # NODE_AFTER_COMMISSIONING_ACTION),
+ }
+ response = self.client.post(node_edit_link, params)
+
+ node = reload_object(node)
+ self.assertEqual(httplib.FOUND, response.status_code)
+ self.assertAttributes(node, params)
+
+ def test_view_node_has_button_to_accept_enlistement_for_user(self):
+ # A simple user can't see the button to enlist a declared node.
+ node = factory.make_node(status=NODE_STATUS.DECLARED)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ doc = fromstring(response.content)
+
+ self.assertEqual(0, len(doc.cssselect('form#node_actions input')))
+
+ def test_view_node_shows_console_output_if_error_set(self):
+ # When node.error is set but the node's status does not indicate an
+ # error condition, the contents of node.error are displayed as console
+ # output.
+ node = factory.make_node(
+ owner=self.logged_in_user, error=factory.getRandomString(),
+ status=NODE_STATUS.READY)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ console_output = fromstring(response.content).xpath(
+ '//h4[text()="Console output"]/following-sibling::span/text()')
+ self.assertEqual([node.error], console_output)
+
+ def test_view_node_shows_error_output_if_error_set(self):
+ # When node.error is set and the node's status indicates an error
+ # condition, the contents of node.error are displayed as error output.
+ node = factory.make_node(
+ owner=self.logged_in_user, error=factory.getRandomString(),
+ status=NODE_STATUS.FAILED_TESTS)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ error_output = fromstring(response.content).xpath(
+ '//h4[text()="Error output"]/following-sibling::span/text()')
+ self.assertEqual([node.error], error_output)
+
+ def test_view_node_shows_no_error_if_no_error_set(self):
+ node = factory.make_node(owner=self.logged_in_user)
+ node_link = reverse('node-view', args=[node.system_id])
+ response = self.client.get(node_link)
+ doc = fromstring(response.content)
+ content_text = doc.cssselect('#content')[0].text_content()
+ self.assertNotIn("Error output", content_text)
+
+ 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)
+
+ def perform_action_and_get_node_page(self, node, action_name):
+ node_link = reverse('node-view', args=[node.system_id])
+ self.client.post(
+ node_link,
+ data={
+ NodeActionForm.input_name: action_name,
+ })
+ response = self.client.get(node_link)
+ return response
+
+ def test_start_commisioning_displays_message(self):
+ self.become_admin()
+ node = factory.make_node(status=NODE_STATUS.DECLARED)
+ response = self.perform_action_and_get_node_page(
+ node, "Accept & commission")
+ self.assertIn(
+ "Node commissioning started.",
+ [message.message for message in response.context['messages']])
+
+>>>>>>> MERGE-SOURCE
class MAASExceptionHandledInView(LoggedInTestCase):
