ganeti_web/tests/importing.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, Group
  20 from django.test import TestCase
  21 from django.test.client import Client
  22
  23 from ganeti_web.util.proxy import RapiProxy, CallProxy
  24 from ganeti_web import models
  25 Cluster = models.Cluster
  26 VirtualMachine = models.VirtualMachine
  27 Organization = models.Organization
  28 Profile = models.Profile
  29
  30 __all__ = ('ImportViews', )
  31
  32
  33 class ImportViews(TestCase):
  34
  35     def setUp(self):
  36         self.tearDown()
  37
  38         models.client.GanetiRapiClient = RapiProxy
  39
  40         self.user = User(id=2, username='tester0')
  41         self.user.set_password('secret')
  42         self.user.save()
  43
  44         self.group = Group(name='testing_group')
  45         self.group.save()
  46
  47         self.cluster0 = Cluster(hostname='test0', slug='OSL_TEST0')
  48         self.cluster0.save()
  49         self.cluster1 = Cluster(hostname='test1', slug='OSL_TEST1')
  50         self.cluster1.save()
  51
  52         self.vm0 = VirtualMachine(hostname='vm0', cluster=self.cluster0)
  53         self.vm1 = VirtualMachine(hostname='vm1', cluster=self.cluster0,
  54                                   owner=self.user.get_profile())
  55         #self.vm2 = VirtualMachine(hostname='vm2', cluster=cluster0)
  56         self.vm3 = VirtualMachine(hostname='vm3', cluster=self.cluster1)
  57         self.vm4 = VirtualMachine(hostname='vm4', cluster=self.cluster1,
  58                                   owner=self.user.get_profile())
  59         #self.vm5 = VirtualMachine(hostname='vm5', cluster=cluster1)
  60         self.vm0.save()
  61         self.vm1.save()
  62         self.vm3.save()
  63         self.vm4.save()
  64
  65         self.c = Client()
  66         self.owner = self.user.get_profile()
  67
  68     def tearDown(self):
  69         VirtualMachine.objects.all().delete()
  70         Cluster.objects.all().delete()
  71         Organization.objects.all().delete()
  72         Profile.objects.all().delete()
  73         User.objects.all().delete()
  74         Group.objects.all().delete()
  75
  76     def test_orphans_view(self):
  77         """
  78         Test orphans view
  79         """
  80         url = '/import/orphans/'
  81
  82         # anonymous user
  83         response = self.c.get(url, follow=True)
  84         self.assertEqual(200, response.status_code)
  85         self.assertTemplateUsed(response, 'registration/login.html')
  86
  87         # unauthorized user
  88         self.assertTrue(self.c.login(username=self.user.username,
  89                                      password='secret'))
  90         response = self.c.get(url)
  91         self.assertEqual(403, response.status_code)
  92
  93         # authorized get (cluster admin perm)
  94         self.user.grant('admin', self.cluster0)
  95         response = self.c.get(url)
  96         self.assertEqual(200, response.status_code)
  97         self.assertEqual('text/html; charset=utf-8', response['content-type'])
  98         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
  99         self.assertEqual([(self.vm0.id, 'test0', 'vm0')],
 100                          response.context['vms'])
 101         self.user.revoke_all(self.cluster0)
 102
 103         # authorized get (superuser)
 104         self.user.is_superuser = True
 105         self.user.save()
 106         response = self.c.get(url)
 107         self.assertEqual(200, response.status_code)
 108         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 109         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
 110         self.assertEqual([(self.vm0.id, 'test0', 'vm0'),
 111                          (self.vm3.id, 'test1', 'vm3')],
 112                          response.context['vms'])
 113         self.user.is_superuser = False
 114         self.user.save()
 115
 116         # POST - invalid vm
 117         self.user.grant('admin', self.cluster0)
 118         data = {'virtual_machines': [-1], 'owner': self.owner.id}
 119         response = self.c.post(url, data)
 120         self.assertEqual(200, response.status_code)
 121         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 122         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
 123         self.assertTrue(response.context['form'].errors)
 124
 125         # POST - invalid owner
 126         data = {'virtual_machines': [self.vm0.id], 'owner': -1}
 127         response = self.c.post(url, data)
 128         self.assertEqual(200, response.status_code)
 129         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 130         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
 131         self.assertTrue(response.context['form'].errors)
 132
 133         # POST - user does not have perms for cluster
 134         data = {'virtual_machines': [self.vm3.id], 'owner': self.owner.id}
 135         response = self.c.post(url, data)
 136         self.assertEqual(200, response.status_code)
 137         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 138         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
 139         self.assertTrue(response.context['form'].errors)
 140
 141         # POST - success
 142         data = {'virtual_machines': [self.vm0.id], 'owner': self.owner.id}
 143         response = self.c.post(url, data)
 144         self.assertEqual(200, response.status_code)
 145         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 146         self.assertTemplateUsed(response, 'ganeti/importing/orphans.html')
 147         self.assertFalse(response.context['form'].errors)
 148         self.assertEqual([], response.context['vms'])
 149
 150     def test_missing_ganeti(self):
 151         """
 152         Tests view for Virtual Machines missing from ganeti
 153         """
 154         url = '/import/missing/'
 155         self.cluster0.rapi.GetInstances.response = ['vm0', 'vm2']
 156         self.cluster1.rapi.GetInstances.response = ['vm3', 'vm5']
 157
 158         # anonymous user
 159         response = self.c.get(url, follow=True)
 160         self.assertEqual(200, response.status_code)
 161         self.assertTemplateUsed(response, 'registration/login.html')
 162
 163         # unauthorized user
 164         self.assertTrue(self.c.login(username=self.user.username,
 165                                      password='secret'))
 166         response = self.c.get(url)
 167         self.assertEqual(403, response.status_code)
 168
 169         # authorized get (cluster admin perm)
 170         self.user.grant('admin', self.cluster0)
 171         response = self.c.get(url)
 172         self.assertEqual(200, response.status_code)
 173         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 174         self.assertTemplateUsed(response, 'ganeti/importing/missing.html')
 175         self.assertEqual([('vm1', 'test0', 'vm1')], response.context['vms'])
 176         self.user.revoke_all(self.cluster0)
 177
 178         # authorized get (superuser)
 179         self.user.is_superuser = True
 180         self.user.save()
 181         response = self.c.get(url)
 182         self.assertEqual(200, response.status_code)
 183         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 184         self.assertTemplateUsed(response, 'ganeti/importing/missing.html')
 185         self.assertEqual([('vm1', 'test0', 'vm1'),
 186                          ('vm4', 'test1', 'vm4')],
 187                          response.context['vms'])
 188         self.user.is_superuser = False
 189         self.user.save()
 190
 191         # POST - invalid vm
 192         self.user.grant('admin', self.cluster0)
 193         data = {'virtual_machines': [-1]}
 194         response = self.c.post(url, data)
 195         self.assertEqual(200, response.status_code)
 196         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 197         self.assertTemplateUsed(response, 'ganeti/importing/missing.html')
 198         self.assertTrue(response.context['form'].errors)
 199
 200         # POST - user does not have perms for cluster
 201         data = {'virtual_machines': [self.vm3.hostname]}
 202         response = self.c.post(url, data)
 203         self.assertEqual(200, response.status_code)
 204         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 205         self.assertTemplateUsed(response, 'ganeti/importing/missing.html')
 206         self.assertTrue(response.context['form'].errors)
 207
 208         # POST - success
 209         data = {'virtual_machines': ['vm1']}
 210         response = self.c.post(url, data)
 211         self.assertEqual(200, response.status_code)
 212         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 213         self.assertTemplateUsed(response, 'ganeti/importing/missing.html')
 214         self.assertFalse(response.context['form'].errors)
 215         self.assertEqual([], response.context['vms'])
 216
 217     def test_missing_db(self):
 218         """
 219         Tests view for Virtual Machines missing from database
 220         """
 221         url = '/import/missing_db/'
 222         self.cluster0.rapi.GetInstances.response = ['vm0', 'vm2']
 223         self.cluster1.rapi.GetInstances.response = ['vm3', 'vm5']
 224
 225         # anonymous user
 226         response = self.c.get(url, follow=True)
 227         self.assertEqual(200, response.status_code)
 228         self.assertTemplateUsed(response, 'registration/login.html')
 229
 230         # unauthorized user
 231         self.assertTrue(self.c.login(username=self.user.username,
 232                                      password='secret'))
 233         response = self.c.get(url)
 234         self.assertEqual(403, response.status_code)
 235
 236         # authorized get (cluster admin perm)
 237         self.user.grant('admin', self.cluster0)
 238         response = self.c.get(url)
 239         self.assertEqual(200, response.status_code)
 240         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 241         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 242         self.assertEqual([('1:vm2', 'test0', 'vm2')],
 243                          response.context['vms'])
 244         self.user.revoke_all(self.cluster0)
 245
 246         # authorized get (superuser)
 247         self.user.is_superuser = True
 248         self.user.save()
 249         response = self.c.get(url)
 250         self.assertEqual(200, response.status_code)
 251         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 252         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 253         self.assertEqual([('1:vm2', 'test0', 'vm2'),
 254                          ('2:vm5', 'test1', 'vm5')],
 255                          response.context['vms'])
 256         self.user.is_superuser = False
 257         self.user.save()
 258
 259         # POST - invalid vm
 260         self.user.grant('admin', self.cluster0)
 261         data = {'virtual_machines': [-1]}
 262         response = self.c.post(url, data)
 263         self.assertEqual(200, response.status_code)
 264         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 265         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 266         self.assertTrue(response.context['form'].errors)
 267
 268         # POST - invalid owner
 269         data = {'virtual_machines': [self.vm0.id], 'owner': -1}
 270         response = self.c.post(url, data)
 271         self.assertEqual(200, response.status_code)
 272         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 273         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 274         self.assertTrue(response.context['form'].errors)
 275
 276         # POST - user does not have perms for cluster
 277         data = {'virtual_machines': [self.vm3.hostname],
 278                 'owner': self.owner.id}
 279         response = self.c.post(url, data)
 280         self.assertEqual(200, response.status_code)
 281         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 282         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 283         self.assertTrue(response.context['form'].errors)
 284
 285         # POST - success
 286         data = {'virtual_machines': ['1:vm2'], 'owner': self.owner.id}
 287         response = self.c.post(url, data)
 288         self.assertEqual(200, response.status_code)
 289         self.assertEqual('text/html; charset=utf-8', response['content-type'])
 290         self.assertTemplateUsed(response, 'ganeti/importing/missing_db.html')
 291         self.assertFalse(response.context['form'].errors)
 292         self.assertEqual([], response.context['vms'])
 293         self.assertTrue(VirtualMachine.objects.filter(hostname='vm2').exists())