← Back to branch summary

~ubuntu-branches/ubuntu/quantal/nova/quantal-proposed

« back to all changes in this revision

Viewing changes to nova/tests/test_volume.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-08-16 14:04:11 UTC
  • mto: This revision was merged to the branch mainline in revision 84.
  • Revision ID: package-import@ubuntu.com-20120816140411-0mr4n241wmk30t9l
Tags: upstream-2012.2~f3
Import upstream version 2012.2~f3

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_volume.py
23
23
import cStringIO
24
24
 
25
25
import mox
 
26
import shutil
 
27
import tempfile
26
28
 
27
29
from nova import context
28
30
from nova import db
29
31
from nova import exception
30
32
from nova import flags
31
 
from nova.notifier import test_notifier
32
33
from nova.openstack.common import importutils
33
34
from nova.openstack.common import log as logging
 
35
from nova.openstack.common.notifier import api as notifier_api
 
36
from nova.openstack.common.notifier import test_notifier
34
37
from nova.openstack.common import rpc
35
38
import nova.policy
36
39
from nova import quota
48
51
    def setUp(self):
49
52
        super(VolumeTestCase, self).setUp()
50
53
        self.compute = importutils.import_object(FLAGS.compute_manager)
51
 
        self.flags(compute_driver='nova.virt.fake.FakeDriver')
 
54
        vol_tmpdir = tempfile.mkdtemp()
 
55
        self.flags(compute_driver='nova.virt.fake.FakeDriver',
 
56
                   volumes_dir=vol_tmpdir)
52
57
        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
53
 
                'nova.notifier.test_notifier')
 
58
                ['nova.openstack.common.notifier.test_notifier'])
54
59
        self.volume = importutils.import_object(FLAGS.volume_manager)
55
60
        self.context = context.get_admin_context()
56
61
        instance = db.instance_create(self.context, {})
59
64
        test_notifier.NOTIFICATIONS = []
60
65
 
61
66
    def tearDown(self):
 
67
        try:
 
68
            shutil.rmtree(FLAGS.volumes_dir)
 
69
        except OSError, e:
 
70
            pass
62
71
        db.instance_destroy(self.context, self.instance_uuid)
 
72
        notifier_api._reset_drivers()
63
73
        super(VolumeTestCase, self).tearDown()
64
74
 
65
75
    @staticmethod
121
131
                          self.context,
122
132
                          volume_id)
123
133
 
 
134
    def _do_test_create_over_quota(self, resource, expected):
 
135
        """Test volume creation over quota."""
 
136
 
 
137
        def fake_reserve(context, **deltas):
 
138
            kwargs = dict(overs=[resource],
 
139
                          quotas=dict(gigabytes=1000, volumes=10),
 
140
                          usages=dict(gigabytes=dict(reserved=1, in_use=999),
 
141
                                      volumes=dict(reserved=1, in_use=9)))
 
142
            raise exception.OverQuota(**kwargs)
 
143
 
 
144
        def fake_commit(context, reservations):
 
145
            self.fail('should not commit over quota')
 
146
 
 
147
        self.stubs.Set(QUOTAS, 'reserve', fake_reserve)
 
148
        self.stubs.Set(QUOTAS, 'commit', fake_commit)
 
149
 
 
150
        volume_api = nova.volume.api.API()
 
151
 
 
152
        self.assertRaises(expected,
 
153
                          volume_api.create,
 
154
                          self.context,
 
155
                          2,
 
156
                          'name',
 
157
                          'description')
 
158
 
 
159
    def test_create_volumes_over_quota(self):
 
160
        self._do_test_create_over_quota('volumes',
 
161
                                        exception.VolumeLimitExceeded)
 
162
 
 
163
    def test_create_gigabytes_over_quota(self):
 
164
        self._do_test_create_over_quota('gigabytes',
 
165
                                        exception.VolumeSizeTooLarge)
 
166
 
124
167
    def test_delete_busy_volume(self):
125
168
        """Test volume survives deletion if driver reports it as busy."""
126
169
        volume = self._create_volume()
419
462
 
420
463
    def setUp(self):
421
464
        super(DriverTestCase, self).setUp()
 
465
        vol_tmpdir = tempfile.mkdtemp()
422
466
        self.flags(volume_driver=self.driver_name,
 
467
                   volumes_dir=vol_tmpdir,
423
468
                   logging_default_format_string="%(message)s")
424
469
        self.volume = importutils.import_object(FLAGS.volume_manager)
425
470
        self.context = context.get_admin_context()
439
484
        self.instance_id = instance['id']
440
485
        self.instance_uuid = instance['uuid']
441
486
 
 
487
    def tearDown(self):
 
488
        try:
 
489
            shutil.rmtree(FLAGS.volumes_dir)
 
490
        except OSError, e:
 
491
            pass
 
492
        super(DriverTestCase, self).tearDown()
 
493
 
442
494
    def _attach_volume(self):
443
495
        """Attach volumes to an instance. This function also sets
444
496
           a fake log message."""