← Back to branch summary

~ubuntu-branches/ubuntu/raring/nova/raring-proposed

« back to all changes in this revision

Viewing changes to nova/tests/xenapi/test_vm_utils.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Adam Gandelman, Chuck Short
  • Date: 2012-11-23 09:04:58 UTC
  • mfrom: (1.1.66)
  • Revision ID: package-import@ubuntu.com-20121123090458-91565o7aev1i1h71
Tags: 2013.1~g1-0ubuntu1
https://launchpad.net/bugs/1073289
https://launchpad.net/bugs/1076442
[ Adam Gandelman ]
* debian/control: Ensure novaclient is upgraded with nova,
  require python-keystoneclient >= 1:2.9.0. (LP: #1073289)
* debian/patches/{ubuntu/*, rbd-security.patch}: Dropped, applied
  upstream.
* debian/control: Add python-testtools to Build-Depends.

[ Chuck Short ]
* New upstream version.
* Refreshed debian/patches/avoid_setuptools_git_dependency.patch.
* debian/rules: FTBFS if missing binaries.
* debian/nova-scheudler.install: Add missing rabbit-queues and
  nova-rpc-zmq-receiver.
* Remove nova-volume since it doesnt exist anymore, transition to cinder-*.
* debian/rules: install apport hook in the right place.
* debian/patches/ubuntu-show-tests.patch: Display test failures.
* debian/control: Add depends on genisoimage
* debian/control: Suggest guestmount.
* debian/control: Suggest websockify. (LP: #1076442)
* debian/nova.conf: Disable nova-volume service.
* debian/control: Depend on xen-system-* rather than the hypervisor.
* debian/control, debian/mans/nova-conductor.8, debian/nova-conductor.init,
  debian/nova-conductor.install, debian/nova-conductor.logrotate
  debian/nova-conductor.manpages, debian/nova-conductor.postrm
  debian/nova-conductor.upstart.in: Add nova-conductor service.
* debian/control: Add python-fixtures as a build deps.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/xenapi/test_vm_utils.py
 
1
import mox
 
2
from nova import context
 
3
from nova import db
 
4
from nova import exception
1
5
from nova.tests.xenapi import stubs
2
6
from nova.virt.xenapi import driver as xenapi_conn
3
7
from nova.virt.xenapi import fake
4
8
from nova.virt.xenapi import vm_utils
 
9
from nova.virt.xenapi import volume_utils
 
10
import unittest
5
11
 
6
12
 
7
13
class GetInstanceForVdisForSrTestCase(stubs.XenAPITestBase):
14
20
                   xenapi_connection_url='test_url',
15
21
                   xenapi_connection_password='test_pass',)
16
22
 
17
 
    def tearDown(self):
18
 
        super(GetInstanceForVdisForSrTestCase, self).tearDown()
19
 
 
20
23
    def test_get_instance_vdis_for_sr(self):
21
24
        vm_ref = fake.create_vm("foo", "Running")
22
25
        sr_ref = fake.create_sr()
46
49
            driver._session, vm_ref, sr_ref))
47
50
 
48
51
        self.assertEquals([], result)
 
52
 
 
53
    def test_get_vdis_for_boot_from_vol(self):
 
54
        dev_params = {'sr_uuid': 'falseSR',
 
55
                      'name_label': 'fake_storage',
 
56
                      'name_description': 'test purposes',
 
57
                      'server': 'myserver',
 
58
                      'serverpath': '/local/scratch/myname',
 
59
                      'sr_type': 'nfs',
 
60
                      'introduce_sr_keys': ['server', 'serverpath', 'sr_type'],
 
61
                      'vdi_uuid': 'falseVDI'}
 
62
        stubs.stubout_session(self.stubs, fake.SessionBase)
 
63
        driver = xenapi_conn.XenAPIDriver(False)
 
64
 
 
65
        result = vm_utils.get_vdis_for_boot_from_vol(driver._session,
 
66
                                                     dev_params)
 
67
        self.assertEquals(result['root']['uuid'], 'falseVDI')
 
68
 
 
69
    def test_get_vdis_for_boot_from_vol_failure(self):
 
70
        stubs.stubout_session(self.stubs, fake.SessionBase)
 
71
        driver = xenapi_conn.XenAPIDriver(False)
 
72
 
 
73
        def bad_introduce_sr(session, sr_uuid, label, sr_params):
 
74
            return None
 
75
 
 
76
        self.stubs.Set(volume_utils, 'introduce_sr', bad_introduce_sr)
 
77
        dev_params = {'sr_uuid': 'falseSR',
 
78
                      'name_label': 'fake_storage',
 
79
                      'name_description': 'test purposes',
 
80
                      'server': 'myserver',
 
81
                      'serverpath': '/local/scratch/myname',
 
82
                      'sr_type': 'nfs',
 
83
                      'introduce_sr_keys': ['server', 'serverpath', 'sr_type'],
 
84
                      'vdi_uuid': 'falseVDI'}
 
85
        self.assertRaises(exception.NovaException,
 
86
                          vm_utils.get_vdis_for_boot_from_vol,
 
87
                          driver._session, dev_params)
 
88
 
 
89
 
 
90
class VMRefOrRaiseVMFoundTestCase(unittest.TestCase):
 
91
 
 
92
    def test_lookup_call(self):
 
93
        mock = mox.Mox()
 
94
        mock.StubOutWithMock(vm_utils, 'lookup')
 
95
 
 
96
        vm_utils.lookup('session', 'somename').AndReturn('ignored')
 
97
 
 
98
        mock.ReplayAll()
 
99
        vm_utils.vm_ref_or_raise('session', 'somename')
 
100
        mock.VerifyAll()
 
101
 
 
102
    def test_return_value(self):
 
103
        mock = mox.Mox()
 
104
        mock.StubOutWithMock(vm_utils, 'lookup')
 
105
 
 
106
        vm_utils.lookup(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn('vmref')
 
107
 
 
108
        mock.ReplayAll()
 
109
        self.assertEquals(
 
110
            'vmref', vm_utils.vm_ref_or_raise('session', 'somename'))
 
111
        mock.VerifyAll()
 
112
 
 
113
 
 
114
class VMRefOrRaiseVMNotFoundTestCase(unittest.TestCase):
 
115
 
 
116
    def test_exception_raised(self):
 
117
        mock = mox.Mox()
 
118
        mock.StubOutWithMock(vm_utils, 'lookup')
 
119
 
 
120
        vm_utils.lookup('session', 'somename').AndReturn(None)
 
121
 
 
122
        mock.ReplayAll()
 
123
        self.assertRaises(
 
124
            exception.InstanceNotFound,
 
125
            lambda: vm_utils.vm_ref_or_raise('session', 'somename')
 
126
        )
 
127
        mock.VerifyAll()
 
128
 
 
129
    def test_exception_msg_contains_vm_name(self):
 
130
        mock = mox.Mox()
 
131
        mock.StubOutWithMock(vm_utils, 'lookup')
 
132
 
 
133
        vm_utils.lookup('session', 'somename').AndReturn(None)
 
134
 
 
135
        mock.ReplayAll()
 
136
        try:
 
137
            vm_utils.vm_ref_or_raise('session', 'somename')
 
138
        except exception.InstanceNotFound as e:
 
139
            self.assertTrue(
 
140
                'somename' in str(e))
 
141
        mock.VerifyAll()
 
142
 
 
143
 
 
144
class BittorrentTestCase(stubs.XenAPITestBase):
 
145
    def setUp(self):
 
146
        super(BittorrentTestCase, self).setUp()
 
147
        self.context = context.get_admin_context()
 
148
 
 
149
    def test_image_uses_bittorrent(self):
 
150
        sys_meta = {'image_bittorrent': True}
 
151
        instance = db.instance_create(self.context,
 
152
                                      {'system_metadata': sys_meta})
 
153
        instance = db.instance_get_by_uuid(self.context, instance['uuid'])
 
154
        self.flags(xenapi_torrent_images='some')
 
155
        self.assertTrue(vm_utils._image_uses_bittorrent(self.context,
 
156
                                                        instance))
 
157
 
 
158
    def _test_create_image(self, cache_type):
 
159
        sys_meta = {'image_cache_in_nova': True}
 
160
        instance = db.instance_create(self.context,
 
161
                                      {'system_metadata': sys_meta})
 
162
        instance = db.instance_get_by_uuid(self.context, instance['uuid'])
 
163
        self.flags(cache_images=cache_type)
 
164
 
 
165
        was = {'called': None}
 
166
 
 
167
        def fake_create_cached_image(*args):
 
168
            was['called'] = 'some'
 
169
            return {}
 
170
        self.stubs.Set(vm_utils, '_create_cached_image',
 
171
                       fake_create_cached_image)
 
172
 
 
173
        def fake_fetch_image(*args):
 
174
            was['called'] = 'none'
 
175
            return {}
 
176
        self.stubs.Set(vm_utils, '_fetch_image',
 
177
                       fake_fetch_image)
 
178
 
 
179
        vm_utils._create_image(self.context, None, instance,
 
180
                               'foo', 'bar', 'baz')
 
181
 
 
182
        self.assertEqual(was['called'], cache_type)
 
183
 
 
184
    def test_create_image_cached(self):
 
185
        self._test_create_image('some')
 
186
 
 
187
    def test_create_image_uncached(self):
 
188
        self._test_create_image('none')