ganeti_web/tests/views/virtual_machine/helpers.py

   1 from django.contrib.auth.models import Group
   2 from django.test import TestCase
   3 # Per #6579, do not change this import without discussion.
   4 from django.utils import simplejson as json
   5
   6 from object_permissions import grant
   7
   8 from ganeti_web import models
   9 from ganeti_web.util.proxy.constants import INFO
  10 from ganeti_web.tests.views.virtual_machine.base import TestVirtualMachineViewsBase
  11 from ganeti_web.utilities import os_prettify
  12
  13 __all__ = ['TestVirtualMachineCreateHelpers', 'TestVirtualMachineHelpers']
  14
  15 Cluster = models.Cluster
  16
  17
  18 class TestVirtualMachineCreateHelpers(TestVirtualMachineViewsBase):
  19
  20     def test_view_cluster_choices(self):
  21         """
  22         Test retrieving list of clusters a user or usergroup has access to
  23         """
  24         url = '/vm/add/choices/'
  25         Cluster.objects.all().delete()
  26         cluster0 = Cluster(hostname='user.create_vm', slug='user_create_vm')
  27         cluster0.save()
  28         cluster1 = Cluster(hostname='user.admin', slug='user_admin')
  29         cluster1.save()
  30         cluster2 = Cluster(hostname='superuser', slug='superuser')
  31         cluster2.save()
  32         cluster3 = Cluster(hostname='group.create_vm', slug='group_create_vm')
  33         cluster3.save()
  34         cluster4 = Cluster(hostname='group.admin', slug='group_admin')
  35         cluster4.save()
  36         cluster5 = Cluster(hostname='no.perms.on.this.group', slug='no_perms')
  37         cluster5.save()
  38         # cluster ids are 1 through 6
  39
  40         # Set each cluster info and update
  41         for cluster in (cluster0, cluster1, cluster2, cluster3, cluster4, cluster5):
  42             cluster.username = self.user.username
  43             cluster.password = 'foobar'
  44             cluster.info = INFO
  45             cluster.save()
  46
  47         self.group.user_set.add(self.user)
  48         group1 = Group(id=43, name='testing_group2')
  49         group1.save()
  50         group1.grant('admin',cluster5)
  51
  52         # anonymous user
  53         response = self.c.get(url, follow=True)
  54         self.assertEqual(200, response.status_code)
  55         self.assertTemplateUsed(response, 'registration/login.html')
  56
  57         self.assertTrue(self.c.login(username=self.user.username, password='secret'))
  58
  59         # Invalid ClusterUser
  60         response = self.c.get(url, {'clusteruser_id':-1})
  61         self.assertEqual(404, response.status_code)
  62
  63         # create_vm permission through a group
  64         self.group.grant('create_vm', cluster3)
  65         response = self.c.get(url, {'clusteruser_id':
  66                                     self.group.organization.id})
  67         self.assertEqual(200, response.status_code)
  68         clusters = json.loads(response.content)
  69         self.assertTrue([cluster3.id,'group.create_vm'] in clusters)
  70         self.assertEqual(1, len(clusters))
  71
  72         # admin permission through a group
  73         self.group.grant('admin', cluster4)
  74         response = self.c.get(url, {'clusteruser_id':
  75                                     self.group.organization.id})
  76         self.assertEqual(200, response.status_code)
  77         clusters = json.loads(response.content)
  78         self.assertTrue([cluster3.id,'group.create_vm'] in clusters)
  79         self.assertTrue([cluster4.id,'group.admin'] in clusters)
  80         self.assertEqual(2, len(clusters))
  81
  82         # create_vm permission on the user
  83         self.user.grant('create_vm', cluster0)
  84         response = self.c.get(url)
  85         self.assertEqual(200, response.status_code)
  86         clusters = json.loads(response.content)
  87         self.assertTrue([cluster0.id,'user.create_vm'] in clusters)
  88         self.assertEqual(1, len(clusters), clusters)
  89
  90         # admin permission on the user
  91         self.user.grant('admin', cluster1)
  92         response = self.c.get(url)
  93         self.assertEqual(200, response.status_code)
  94         clusters = json.loads(response.content)
  95         self.assertTrue([cluster0.id,'user.create_vm'] in clusters)
  96         self.assertTrue([cluster1.id,'user.admin'] in clusters)
  97         self.assertEqual(2, len(clusters))
  98
  99         # Superusers see everything
 100         self.user.is_superuser = True
 101         self.user.save()
 102         response = self.c.get(url)
 103         self.assertEqual(200, response.status_code)
 104         self.assertEqual('application/json', response['content-type'])
 105         clusters = json.loads(response.content)
 106         self.assertTrue([cluster0.id,'user.create_vm'] in clusters)
 107         self.assertTrue([cluster1.id,'user.admin'] in clusters)
 108         self.assertTrue([cluster2.id,'superuser'] in clusters, clusters)
 109         self.assertTrue([cluster3.id,'group.create_vm'] in clusters)
 110         self.assertTrue([cluster4.id,'group.admin'] in clusters, clusters)
 111         self.assertTrue([cluster5.id,'no.perms.on.this.group'] in clusters)
 112         self.assertEqual(6, len(clusters))
 113
 114     def test_view_cluster_options(self):
 115         """
 116         Test retrieving list of options a cluster has for vms
 117         """
 118         url = '/vm/add/options/?cluster_id=%s'
 119         args = self.cluster.id
 120
 121         # anonymous user
 122         response = self.c.post(url % args, follow=True)
 123         self.assertEqual(200, response.status_code)
 124         self.assertTemplateUsed(response, 'registration/login.html')
 125
 126         # unauthorized user
 127         self.assertTrue(self.c.login(username=self.user.username,
 128                                      password='secret'))
 129         response = self.c.get(url % args)
 130         self.assertEqual(403, response.status_code)
 131
 132         # invalid cluster
 133         response = self.c.get(url % "-4")
 134         self.assertEqual(404, response.status_code)
 135
 136         # authorized (create_vm)
 137         grant(self.user, 'create_vm', self.cluster)
 138         response = self.c.get(url % args)
 139         self.assertEqual(200, response.status_code)
 140         self.assertEqual('application/json', response['content-type'])
 141         content = json.loads(response.content)
 142         self.assertEqual(set([u'gtest1.osuosl.bak', u'gtest2.osuosl.bak',
 143                               u'gtest3.osuosl.bak']), set(content['nodes']))
 144         self.assertEqual(content["os"],
 145             [[u'Image',
 146                 [[u'image+debian-osgeo', u'Debian Osgeo'],
 147                 [u'image+ubuntu-lucid', u'Ubuntu Lucid']]
 148             ]]
 149         )
 150         self.user.revoke_all(self.cluster)
 151
 152         # authorized (cluster admin)
 153         grant(self.user, 'admin', self.cluster)
 154         response = self.c.get(url % args)
 155         self.assertEqual(200, response.status_code)
 156         self.assertEqual('application/json', response['content-type'])
 157         content = json.loads(response.content)
 158
 159         self.assertEqual(set([u'gtest1.osuosl.bak', u'gtest2.osuosl.bak',
 160                               u'gtest3.osuosl.bak']), set(content['nodes']))
 161         self.assertEqual(content["os"],
 162             [[u'Image',
 163                 [[u'image+debian-osgeo', u'Debian Osgeo'],
 164                 [u'image+ubuntu-lucid', u'Ubuntu Lucid']]
 165             ]]
 166         )
 167         self.user.revoke_all(self.cluster)
 168
 169         # authorized (superuser)
 170         self.user.is_superuser = True
 171         self.user.save()
 172         response = self.c.get(url % args)
 173         self.assertEqual(200, response.status_code)
 174         self.assertEqual('application/json', response['content-type'])
 175         content = json.loads(response.content)
 176         self.assertEqual(set([u'gtest1.osuosl.bak', u'gtest2.osuosl.bak',
 177                               u'gtest3.osuosl.bak']), set(content['nodes']))
 178         self.assertEqual(content["os"],
 179             [[u'Image',
 180                 [[u'image+debian-osgeo', u'Debian Osgeo'],
 181                 [u'image+ubuntu-lucid', u'Ubuntu Lucid']]
 182             ]]
 183         )
 184
 185     def test_view_cluster_defaults(self):
 186         """
 187         Test retrieval of dict of default parameters set on cluster
 188         """
 189         url = '/vm/add/defaults/?cluster_id=%s'
 190         args = self.cluster.id
 191         """
 192         expected = dict(
 193             boot_order='disk',
 194             memory=512,
 195             nic_type='paravirtual',
 196             root_path='/dev/vda2',
 197             hypervisors=['kvm'],
 198             serial_console=True,
 199             cdrom_image_path='',
 200             disk_type ='paravirtual',
 201             nic_link ='br42',
 202             nic_mode='bridged',
 203             vcpus=2,
 204             iallocator='',
 205             kernel_path=''
 206         )
 207         """
 208         expected = dict(
 209             nic_type='paravirtual',
 210             use_chroot=False,
 211             migration_port=8102,
 212             vnc_bind_address='0.0.0.0',
 213             nic_mode='bridged',
 214             usb_mouse='',
 215             hypervisors=[['kvm', 'kvm']],
 216             migration_downtime=30,
 217             nic_types=[
 218                 ['', '---------'],
 219                 ['rtl8139', 'rtl8139'],
 220                 ['ne2k_isa', 'ne2k_isa'],
 221                 ['ne2k_pci', 'ne2k_pci'],
 222                 ['paravirtual', 'paravirtual'],
 223                 ['i82551', 'i82551'],
 224                 ['i82557b', 'i82557b'],
 225                 ['i82559er', 'i82559er'],
 226                 ['pcnet', 'pcnet'],
 227                 ['e1000', 'e1000']
 228             ],
 229             kernel_args='ro',
 230             cdrom_image_path='',
 231             boot_order='disk',
 232             vhost_net=False,
 233             disk_types=[
 234                 ['', '---------'],
 235                 ['paravirtual', 'paravirtual'],
 236                 ['ide', 'ide'],
 237                 ['scsi', 'scsi'],
 238                 ['sd', 'sd'],
 239                 ['mtd', 'mtd'],
 240                 ['pflash', 'pflash']
 241             ],
 242             initrd_path='',
 243             disk_cache='default',
 244             memory=512,
 245             kernel_path='',
 246             vnc_x509_path='',
 247             vnc_x509_verify=False,
 248             vnc_tls=False,
 249             use_localtime=False,
 250             boot_devices=[
 251                 ['disk', 'Hard Disk'],
 252                 ['cdrom', 'CD-ROM'],
 253                 ['network', 'Network']
 254             ],
 255             security_domain='',
 256             serial_console=True,
 257             kvm_flag='',
 258             vnc_password_file='',
 259             migration_bandwidth=32,
 260             disk_type='paravirtual',
 261             security_model='none',
 262             migration_mode='live',
 263             nic_link='br42',
 264             hypervisor='kvm',
 265             root_path='/dev/vda2',
 266             acpi=True,
 267             vcpus=2,
 268             iallocator='',
 269         )
 270
 271         #anonymous users
 272         response = self.c.post(url % args, follow=True)
 273         self.assertEqual(200, response.status_code)
 274         self.assertTemplateUsed(response, 'registration/login.html')
 275
 276         #unauthorized user
 277         self.assertTrue(self.c.login(username=self.user.username,
 278                                      password='secret'))
 279         response = self.c.get(url % args)
 280         self.assertEqual(403, response.status_code)
 281
 282         #invalid cluster
 283         response = self.c.get(url % "-2")
 284         self.assertEqual(404, response.status_code)
 285
 286         #authorized (create_vm)
 287         grant(self.user, 'create_vm', self.cluster)
 288         response = self.c.get(url % args)
 289         self.assertEqual(200, response.status_code)
 290         self.assertEqual('application/json', response['content-type'])
 291         content = json.loads(response.content)
 292         self.assertEqual(expected, content, msg=content)
 293         self.user.revoke_all(self.cluster)
 294
 295         #authorized (admin)
 296         grant(self.user, 'admin', self.cluster)
 297         response = self.c.get(url % args)
 298         self.assertEqual(200, response.status_code)
 299         self.assertEqual('application/json', response['content-type'])
 300         content = json.loads(response.content)
 301         self.assertEqual(expected, content)
 302         self.user.revoke_all(self.cluster)
 303
 304         #authorized (superuser)
 305         self.user.is_superuser = True
 306         self.user.save()
 307         response = self.c.get(url % args)
 308         self.assertEqual(200, response.status_code)
 309         self.assertEqual('application/json', response['content-type'])
 310         content = json.loads(response.content)
 311         self.assertEqual(expected, content)
 312         self.user.is_superuser = False
 313         self.user.save()
 314
 315
 316 class TestVirtualMachineHelpers(TestCase):
 317
 318     def test_os_prettify(self):
 319         """
 320         Test the os_prettify() helper function.
 321         """
 322
 323         # Test a single entry.
 324         self.assertEqual(os_prettify(["hurp+durp"]),
 325             [
 326                 ("Hurp",
 327                     [("hurp+durp", "Durp")]
 328                 )
 329             ])
 330
 331     def test_os_prettify_multiple(self):
 332         """
 333         Test os_prettify()'s ability to handle multiple entries, including two
 334         entries on the same category.
 335         """
 336
 337         self.assertEqual(
 338             os_prettify([
 339                 "image+obonto-hungry-hydralisk",
 340                 "image+fodoro-core",
 341                 "dobootstrop+dobion-lotso",
 342             ]), [
 343                 ("Dobootstrop", [
 344                     ("dobootstrop+dobion-lotso", "Dobion Lotso"),
 345                 ]),
 346                 ("Image", [
 347                     ("image+obonto-hungry-hydralisk",
 348                         "Obonto Hungry Hydralisk"),
 349                     ("image+fodoro-core", "Fodoro Core"),
 350                 ]),
 351             ])
 352
 353     def test_os_prettify_2517(self):
 354         """
 355         Test #2157 compliance.
 356
 357         This example should still parse, but in a weird way. Better than
 358         nothing, though.
 359         """
 360
 361         self.assertEqual(os_prettify(["debian-pressed+ia32"]),
 362             [('Debian-pressed', [('debian-pressed+ia32', 'Ia32')])])
 363
 364     def test_os_prettify_2517_unknown(self):
 365         """
 366         Test #2157 compliance.
 367
 368         This example wasn't part of the bug; it was constructed to show off
 369         the fix for #2157.
 370         """
 371
 372         self.assertEqual(os_prettify(["deb-ver1", "noop"]),
 373             [
 374                 ("Unknown", [
 375                     ("deb-ver1", "deb-ver1"),
 376                     ("noop", "noop"),
 377                 ]),
 378             ])