ganeti_web/tests/views/virtual_machine/helpers.py

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