ganeti_web/tests/importing_nodes.py

   1 # Copyright (C) 2010 Oregon State University et al.
   2 #
   3 # This program is free software; you can redistribute it and/or
   4 # modify it under the terms of the GNU General Public License
   5 # as published by the Free Software Foundation; either version 2
   6 # of the License, or (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
  16 # USA.
  17
  18
  19 from django.contrib.auth.models import User
  20 from django.test import TestCase
  21 from django.test.client import Client
  22 from ganeti_web.models import Node
  23
  24 from ganeti_web.tests.rapi_proxy import RapiProxy, NODES
  25 from ganeti_web import models
  26 Cluster = models.Cluster
  27 VirtualMachine = models.VirtualMachine
  28 Organization = models.Organization
  29 Profile = models.Profile
  30
  31 __all__ = ['NodeMissingDBTests', 'NodeMissingTests']
  32
  33
  34 class NodeImportBase(TestCase):
  35     url = ''
  36     c = None
  37     cluster0 = None
  38     cluster1 = None
  39
  40     def setUp(self):
  41         self.tearDown()
  42
  43         models.client.GanetiRapiClient = RapiProxy
  44
  45         self.unauthorized = User(id=2, username='tester0')
  46         self.authorized = User(id=3, username='tester1')
  47         self.superuser = User(id=4, username='tester2', is_superuser=True)
  48
  49         self.unauthorized.set_password('secret')
  50         self.authorized.set_password('secret')
  51         self.superuser.set_password('secret')
  52
  53         self.unauthorized.save()
  54         self.authorized.save()
  55         self.superuser.save()
  56
  57         self.cluster0 = Cluster.objects.create(hostname='test0', slug='OSL_TEST0')
  58         self.cluster1 = Cluster.objects.create(hostname='test1', slug='OSL_TEST1')
  59
  60         self.authorized.grant('admin', self.cluster0)
  61
  62         self.cluster0.rapi.GetNodes.response = ['node0','node2']
  63         self.cluster1.rapi.GetNodes.response = ['node3','node5']
  64
  65         self.vm = VirtualMachine.objects.create(hostname='gimager.osuosl.bak', cluster=self.cluster0)
  66
  67         self.node0 = Node.objects.create(hostname='node0', cluster=self.cluster0)
  68         self.node1 = Node.objects.create(hostname='node1', cluster=self.cluster0)
  69         self.node3 = Node.objects.create(hostname='node3', cluster=self.cluster1)
  70         self.node4 = Node.objects.create(hostname='node4', cluster=self.cluster1)
  71
  72         self.c = Client()
  73
  74     def tearDown(self):
  75         # reset proxy object default values, could cause collisions in other tests
  76         if self.cluster0 is not None:
  77             self.cluster0.rapi.GetNodes.response = NODES
  78         if self.cluster1 is not None:
  79             self.cluster1.rapi.GetNodes.response = NODES
  80
  81         if self.c is not None:
  82             self.c.logout()
  83         Node.objects.all().delete()
  84         Cluster.objects.all().delete()
  85         Profile.objects.all().delete()
  86         User.objects.all().delete()
  87
  88     def test_anonymous(self):
  89         """ anonymous user """
  90         response = self.c.get(self.url, follow=True)
  91         self.assertEqual(200, response.status_code)
  92         self.assertTemplateUsed(response, 'registration/login.html')
  93
  94     def test_unauthorized(self):
  95         """ unauthorized user """
  96         self.assert_(self.c.login(username=self.unauthorized.username, password='secret'))
  97         response = self.c.get(self.url)
  98         self.assertEqual(403, response.status_code)
  99
 100
 101 class NodeMissingDBTests(NodeImportBase):
 102
 103     url = '/import/node/missing_db/'
 104
 105     def test_get_form(self):
 106         """ authorized get (cluster admin perm) """
 107         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 108         response = self.c.get(self.url)
 109         self.assertEqual(200, response.status_code)
 110         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 111         self.assertTemplateUsed(response, 'importing/nodes/import.html')
 112         self.assertEqual([('%s:node2'%self.cluster0.pk,'test0','node2')], response.context['nodes'])
 113
 114     def test_get_form_superuser(self):
 115         """ authorized get (superuser) """
 116         self.assertTrue(self.c.login(username=self.superuser.username, password='secret'))
 117         response = self.c.get(self.url)
 118         self.assertEqual(200, response.status_code)
 119         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 120         self.assertTemplateUsed(response, 'importing/nodes/import.html')
 121         self.assertEqual([('%s:node2'%self.cluster0.pk,'test0','node2'), ('%s:node5'%self.cluster1.pk,'test1','node5')], response.context['nodes'])
 122
 123     def test_invalid_node(self):
 124         """ POST - invalid node """
 125         self.assertTrue(self.c.login(username=self.superuser.username, password='secret'))
 126         data = {'nodes':[-1]}
 127         response = self.c.post(self.url, data)
 128         self.assertEqual(200, response.status_code)
 129         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 130         self.assertTemplateUsed(response, 'importing/nodes/import.html')
 131         self.assertTrue(response.context['form'].errors)
 132
 133     def test_unauthorized_post(self):
 134         """ POST - user does not have perms for cluster """
 135         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 136         data = {'nodes':[self.node3.hostname]}
 137         response = self.c.post(self.url, data)
 138         self.assertEqual(200, response.status_code)
 139         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 140         self.assertTemplateUsed(response, 'importing/nodes/import.html')
 141         self.assertTrue(response.context['form'].errors)
 142
 143     def test_successful_import(self):
 144         """ POST - success """
 145         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 146         data = {'nodes':['%s:node2'%self.cluster0.pk]}
 147         response = self.c.post(self.url, data)
 148         self.assertEqual(200, response.status_code)
 149         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 150         self.assertTemplateUsed(response, 'importing/nodes/import.html')
 151         self.assertFalse(response.context['form'].errors)
 152         self.assertTrue(Node.objects.filter(hostname='node2').exists())
 153         self.assertEqual([], response.context['nodes'])
 154
 155         # check to see that vm nodes were updated
 156         vm = VirtualMachine.objects.filter(hostname='gimager.osuosl.bak') \
 157             .values_list('primary_node__hostname')[0][0]
 158         self.assertEqual('node2', vm)
 159
 160
 161 class NodeMissingTests(NodeImportBase):
 162
 163     url = '/import/node/missing/'
 164
 165     def test_get_form_authorized(self):
 166         # authorized get (cluster admin perm)
 167         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 168         response = self.c.get(self.url)
 169         self.assertEqual(200, response.status_code)
 170         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 171         self.assertTemplateUsed(response, 'importing/nodes/missing.html')
 172         self.assertEqual([('node1','test0','node1')], response.context['nodes'])
 173
 174     def test_get_form_superuser(self):
 175         """ authorized get (superuser) """
 176         self.assertTrue(self.c.login(username=self.superuser.username, password='secret'))
 177         response = self.c.get(self.url)
 178         self.assertEqual(200, response.status_code)
 179         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 180         self.assertTemplateUsed(response, 'importing/nodes/missing.html')
 181         self.assertEqual([('node1','test0','node1'), ('node4','test1','node4')], response.context['nodes'])
 182
 183     def test_invalid_node(self):
 184         """ POST - invalid vm """
 185         self.assertTrue(self.c.login(username=self.superuser.username, password='secret'))
 186         data = {'nodes':[-1]}
 187         response = self.c.post(self.url, data)
 188         self.assertEqual(200, response.status_code)
 189         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 190         self.assertTemplateUsed(response, 'importing/nodes/missing.html')
 191         self.assertTrue(response.context['form'].errors)
 192
 193     def test_post_unauthorized(self):
 194         """ POST - user does not have perms for cluster """
 195         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 196         data = {'nodes':[self.node3.hostname]}
 197         response = self.c.post(self.url, data)
 198         self.assertEqual(200, response.status_code)
 199         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 200         self.assertTemplateUsed(response, 'importing/nodes/missing.html')
 201         self.assertTrue(response.context['form'].errors)
 202
 203     def test_successful_deletion(self):
 204         """ POST - success """
 205         self.assertTrue(self.c.login(username=self.authorized.username, password='secret'))
 206         data = {'nodes':['node1']}
 207         response = self.c.post(self.url, data)
 208         self.assertEqual(200, response.status_code)
 209         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 210         self.assertTemplateUsed(response, 'importing/nodes/missing.html')
 211         self.assertFalse(response.context['form'].errors)
 212         self.assertFalse(Node.objects.filter(hostname='node1').exists())
 213         self.assertEqual([], response.context['nodes'])