← Back to branch summary

~ubuntu-cloud-archive/ubuntu/precise/nova/trunk

« back to all changes in this revision

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

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Adam Gandelman
  • Date: 2012-06-22 12:39:57 UTC
  • mfrom: (1.1.57)
  • Revision ID: package-import@ubuntu.com-20120622123957-hbzwg84nt9rqwg8r
Tags: 2012.2~f2~20120621.14517-0ubuntu1
[ Chuck Short ]
* New upstream version.

[ Adam Gandelman ]
* debian/rules: Temporarily disable test suite while blocking
  tests are investigated. 
* debian/patches/kombu_tests_timeout.patch: Dropped.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/xenapi/stubs.py
15
15
 
16
16
"""Stubouts, mocks and fixtures for the test suite"""
17
17
 
18
 
import json
19
18
import random
20
19
 
21
20
from eventlet import tpool
22
21
 
 
22
from nova.openstack.common import jsonutils
23
23
from nova.virt.xenapi import connection as xenapi_conn
24
24
from nova.virt.xenapi import fake
25
25
from nova.virt.xenapi import vm_utils
37
37
 
38
38
 
39
39
def stubout_instance_snapshot(stubs):
40
 
    @classmethod
41
 
    def fake_fetch_image(cls, context, session, instance, image, type):
42
 
        return [dict(vdi_type='root', vdi_uuid=_make_fake_vdi())]
 
40
    def fake_fetch_image(context, session, instance, image, type):
 
41
        return {'root': dict(uuid=_make_fake_vdi(), file=None),
 
42
                'kernel': dict(uuid=_make_fake_vdi(), file=None),
 
43
                'ramdisk': dict(uuid=_make_fake_vdi(), file=None)}
43
44
 
44
 
    stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image)
 
45
    stubs.Set(vm_utils, 'fetch_image', fake_fetch_image)
45
46
 
46
47
    def fake_wait_for_vhd_coalesce(*args):
47
48
        #TODO(sirp): Should we actually fake out the data here
95
96
def stubout_determine_is_pv_objectstore(stubs):
96
97
    """Assumes VMs stu have PV kernels"""
97
98
 
98
 
    @classmethod
99
 
    def f(cls, *args):
 
99
    def f(*args):
100
100
        return False
101
 
    stubs.Set(vm_utils.VMHelper, '_determine_is_pv_objectstore', f)
 
101
    stubs.Set(vm_utils, '_determine_is_pv_objectstore', f)
102
102
 
103
103
 
104
104
def stubout_is_snapshot(stubs):
105
105
    """ Always returns true
106
106
        xenapi fake driver does not create vmrefs for snapshots """
107
107
 
108
 
    @classmethod
109
 
    def f(cls, *args):
 
108
    def f(*args):
110
109
        return True
111
 
    stubs.Set(vm_utils.VMHelper, 'is_snapshot', f)
 
110
    stubs.Set(vm_utils, 'is_snapshot', f)
112
111
 
113
112
 
114
113
def stubout_lookup_image(stubs):
121
120
def stubout_fetch_image_glance_disk(stubs, raise_failure=False):
122
121
    """Simulates a failure in fetch image_glance_disk."""
123
122
 
124
 
    @classmethod
125
 
    def _fake_fetch_image_glance_disk(cls, context, session, instance, image,
 
123
    def _fake_fetch_image_glance_disk(context, session, instance, image,
126
124
                                      image_type):
127
125
        if raise_failure:
128
126
            raise fake.Failure("Test Exception raised by "
134
132
        else:
135
133
            filename = "unknown"
136
134
 
137
 
        return [dict(vdi_type=vm_utils.ImageType.to_string(image_type),
138
 
                     vdi_uuid=None,
139
 
                     file=filename)]
 
135
        vdi_type = vm_utils.ImageType.to_string(image_type)
 
136
        return {vdi_type: dict(uuid=None, file=filename)}
140
137
 
141
 
    stubs.Set(vm_utils.VMHelper, '_fetch_image_glance_disk',
 
138
    stubs.Set(vm_utils, '_fetch_image_glance_disk',
142
139
              _fake_fetch_image_glance_disk)
143
140
 
144
141
 
145
142
def stubout_create_vm(stubs):
146
143
    """Simulates a failure in create_vm."""
147
144
 
148
 
    @classmethod
149
 
    def f(cls, *args):
 
145
    def f(*args):
150
146
        raise fake.Failure("Test Exception raised by " +
151
147
                           "fake create_vm")
152
 
    stubs.Set(vm_utils.VMHelper, 'create_vm', f)
 
148
    stubs.Set(vm_utils, 'create_vm', f)
153
149
 
154
150
 
155
151
def _make_fake_vdi():
252
248
        if plugin == "xenhost" and method == "iptables_config":
253
249
            # The command to execute is a json-encoded list
254
250
            cmd_args = args.get('cmd_args', None)
255
 
            cmd = json.loads(cmd_args)
 
251
            cmd = jsonutils.loads(cmd_args)
256
252
            if not cmd:
257
253
                ret_str = ''
258
254
            else:
285
281
    def fake_release_bootlock(self, vm):
286
282
        pass
287
283
 
288
 
    @classmethod
289
 
    def fake_generate_ephemeral(cls, *args):
 
284
    def fake_generate_ephemeral(*args):
290
285
        pass
291
286
 
292
287
    def fake_wait_for_device(dev):
294
289
 
295
290
    stubs.Set(vmops.VMOps, "_acquire_bootlock", fake_acquire_bootlock)
296
291
    stubs.Set(vmops.VMOps, "_release_bootlock", fake_release_bootlock)
297
 
    stubs.Set(vm_utils.VMHelper, 'generate_ephemeral', fake_generate_ephemeral)
 
292
    stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
298
293
    stubs.Set(vm_utils, '_wait_for_device', fake_wait_for_device)
299
294
 
300
295
 
338
333
        return 'vm_ref', dict(image='foo', snap='bar')
339
334
 
340
335
    def fake_move_disks(self, instance, disk_info):
341
 
        vdi_ref = fake.create_vdi('new', 'fake')
342
 
        return fake.get_record('VDI', vdi_ref)['uuid']
 
336
        vdi_ref = fake.create_vdi(instance['name'], 'fake')
 
337
        vdi_rec = fake.get_record('VDI', vdi_ref)
 
338
        vdi_rec['other_config']['nova_disk_type'] = 'root'
 
339
        return {'uuid': vdi_rec['uuid'], 'ref': vdi_ref}
343
340
 
344
 
    @classmethod
345
 
    def fake_get_vdi(cls, session, vm_ref):
 
341
    def fake_get_vdi(session, vm_ref):
346
342
        vdi_ref = fake.create_vdi('derp', 'herp')
347
343
        vdi_rec = session.call_xenapi("VDI.get_record", vdi_ref)
348
344
        return vdi_ref, {'uuid': vdi_rec['uuid'], }
349
345
 
350
 
    @classmethod
351
 
    def fake_sr(cls, session, *args):
 
346
    def fake_sr(session, *args):
352
347
        pass
353
348
 
354
 
    @classmethod
355
 
    def fake_get_sr_path(cls, *args):
 
349
    def fake_get_sr_path(*args):
356
350
        return "fake"
357
351
 
358
352
    def fake_destroy(*args, **kwargs):
359
353
        pass
360
354
 
361
 
    @classmethod
362
 
    def fake_generate_ephemeral(cls, *args):
 
355
    def fake_generate_ephemeral(*args):
363
356
        pass
364
357
 
365
358
    stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
366
359
    stubs.Set(vmops.VMOps, '_move_disks', fake_move_disks)
367
 
    stubs.Set(vm_utils.VMHelper, 'scan_default_sr', fake_sr)
368
 
    stubs.Set(vm_utils.VMHelper, 'scan_sr', fake_sr)
 
360
    stubs.Set(vm_utils, 'scan_default_sr', fake_sr)
 
361
    stubs.Set(vm_utils, 'scan_sr', fake_sr)
369
362
    stubs.Set(vmops.VMOps, '_create_snapshot', fake_create_snapshot)
370
 
    stubs.Set(vm_utils.VMHelper, 'get_vdi_for_vm_safely', fake_get_vdi)
371
 
    stubs.Set(vm_utils.VMHelper, 'get_sr_path', fake_get_sr_path)
372
 
    stubs.Set(vm_utils.VMHelper, 'generate_ephemeral', fake_generate_ephemeral)
 
363
    stubs.Set(vm_utils, 'get_vdi_for_vm_safely', fake_get_vdi)
 
364
    stubs.Set(vm_utils, 'get_sr_path', fake_get_sr_path)
 
365
    stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)