27
27
from nova.compute import api as compute_api
28
28
from nova.compute import instance_types
29
29
from nova.compute import power_state
30
from nova.compute import task_states
31
from nova.compute import vm_states
30
32
from nova import context
31
33
from nova import db
32
34
from nova import exception
33
from nova import flags
35
from nova.openstack.common import cfg
34
36
from nova.openstack.common import importutils
35
37
from nova.openstack.common import jsonutils
36
38
from nova.openstack.common import log as logging
37
from nova.openstack.common import timeutils
38
39
from nova import test
39
40
from nova.tests.db import fakes as db_fakes
40
41
from nova.tests import fake_network
41
42
from nova.tests import fake_utils
42
43
import nova.tests.image.fake as fake_image
44
from nova.tests import matchers
43
45
from nova.tests.xenapi import stubs
46
from nova.virt import fake
44
47
from nova.virt.xenapi import agent
45
48
from nova.virt.xenapi import driver as xenapi_conn
46
49
from nova.virt.xenapi import fake as xenapi_fake
50
from nova.virt.xenapi import host
47
51
from nova.virt.xenapi import pool
48
52
from nova.virt.xenapi import pool_states
49
53
from nova.virt.xenapi import vm_utils
170
178
vol['user_id'] = 'fake'
171
179
vol['project_id'] = 'fake'
172
180
vol['host'] = 'localhost'
173
vol['availability_zone'] = FLAGS.storage_availability_zone
181
vol['availability_zone'] = CONF.node_availability_zone
174
182
vol['status'] = "creating"
175
183
vol['attach_status'] = "detached"
176
184
return db.volume_create(self.context, vol)
187
def _make_connection_data():
190
'target_iqn': 'iqn.2010-10.org.openstack:volume-00000001',
191
'target_portal': '127.0.0.1:3260,fake',
193
'auth_method': 'CHAP',
194
'auth_username': 'username',
195
'auth_password': 'password',
199
def _make_connection_info(cls):
181
201
'driver_volume_type': 'iscsi',
184
'target_iqn': 'iqn.2010-10.org.openstack:volume-00000001',
185
'target_portal': '127.0.0.1:3260,fake',
187
'auth_method': 'CHAP',
188
'auth_method': 'fake',
189
'auth_method': 'fake',
202
'data': cls._make_connection_data()
193
205
def test_mountpoint_to_number(self):
211
223
self.assertEqual(actual, expected,
212
224
'%s yielded %s, not %s' % (input, actual, expected))
214
def test_parse_volume_info_raise_exception(self):
215
"""This shows how to test helper classes' methods."""
216
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
217
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
218
vol = self._create_volume()
219
# oops, wrong mount point!
220
self.assertRaises(volume_utils.StorageError,
221
volume_utils.parse_volume_info,
225
db.volume_destroy(context.get_admin_context(), vol['id'])
226
def test_parse_volume_info_parsing_auth_details(self):
227
result = volume_utils.parse_volume_info(
228
self._make_connection_data())
230
self.assertEquals('username', result['chapuser'])
231
self.assertEquals('password', result['chappassword'])
233
def test_get_device_number_raise_exception_on_wrong_mountpoint(self):
235
volume_utils.StorageError,
236
volume_utils.get_device_number,
227
239
def test_attach_volume(self):
228
240
"""This shows how to test Ops classes' methods."""
229
241
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
230
conn = xenapi_conn.XenAPIDriver(False)
231
volume = self._create_volume()
242
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
232
243
instance = db.instance_create(self.context, self.instance_values)
233
244
vm = xenapi_fake.create_vm(instance.name, 'Running')
234
result = conn.attach_volume(self._make_info(),
245
result = conn.attach_volume(self._make_connection_info(),
235
246
instance.name, '/dev/sdc')
237
248
# check that the VM has a VBD attached to it
504
517
self.assertEquals(self.vm['HVM_boot_policy'], '')
506
519
def _list_vdis(self):
507
url = FLAGS.xenapi_connection_url
508
username = FLAGS.xenapi_connection_username
509
password = FLAGS.xenapi_connection_password
510
session = xenapi_conn.XenAPISession(url, username, password)
520
url = CONF.xenapi_connection_url
521
username = CONF.xenapi_connection_username
522
password = CONF.xenapi_connection_password
523
session = xenapi_conn.XenAPISession(url, username, password,
511
525
return session.call_xenapi('VDI.get_all')
528
url = CONF.xenapi_connection_url
529
username = CONF.xenapi_connection_username
530
password = CONF.xenapi_connection_password
531
session = xenapi_conn.XenAPISession(url, username, password,
533
return session.call_xenapi('VM.get_all')
513
535
def _check_vdis(self, start_list, end_list):
514
536
for vdi_ref in end_list:
515
537
if not vdi_ref in start_list:
582
604
def test_spawn_fail_cleanup_1(self):
583
605
"""Simulates an error while downloading an image.
585
Verifies that VDIs created are properly cleaned up.
607
Verifies that the VM and VDIs created are properly cleaned up.
588
609
vdi_recs_start = self._list_vdis()
610
start_vms = self._list_vms()
589
611
stubs.stubout_fetch_disk_image(self.stubs, raise_failure=True)
590
612
self.assertRaises(xenapi_fake.Failure,
591
613
self._test_spawn, 1, 2, 3)
592
614
# No additional VDI should be found.
593
615
vdi_recs_end = self._list_vdis()
616
end_vms = self._list_vms()
594
617
self._check_vdis(vdi_recs_start, vdi_recs_end)
618
# No additional VMs should be found.
619
self.assertEqual(start_vms, end_vms)
596
621
def test_spawn_fail_cleanup_2(self):
597
622
"""Simulates an error while creating VM record.
599
It verifies that VDIs created are properly cleaned up.
624
Verifies that the VM and VDIs created are properly cleaned up.
602
626
vdi_recs_start = self._list_vdis()
627
start_vms = self._list_vms()
603
628
stubs.stubout_create_vm(self.stubs)
604
629
self.assertRaises(xenapi_fake.Failure,
605
630
self._test_spawn, 1, 2, 3)
606
631
# No additional VDI should be found.
607
632
vdi_recs_end = self._list_vdis()
608
self._check_vdis(vdi_recs_start, vdi_recs_end)
633
end_vms = self._list_vms()
634
self._check_vdis(vdi_recs_start, vdi_recs_end)
635
# No additional VMs should be found.
636
self.assertEqual(start_vms, end_vms)
638
def test_spawn_fail_cleanup_3(self):
639
"""Simulates an error while attaching disks.
641
Verifies that the VM and VDIs created are properly cleaned up.
643
stubs.stubout_attach_disks(self.stubs)
644
vdi_recs_start = self._list_vdis()
645
start_vms = self._list_vms()
646
self.assertRaises(xenapi_fake.Failure,
647
self._test_spawn, 1, 2, 3)
648
# No additional VDI should be found.
649
vdi_recs_end = self._list_vdis()
650
end_vms = self._list_vms()
651
self._check_vdis(vdi_recs_start, vdi_recs_end)
652
# No additional VMs should be found.
653
self.assertEqual(start_vms, end_vms)
610
655
@stub_vm_utils_with_vdi_attached_here
611
656
def test_spawn_raw_glance(self):
617
662
os_type="linux", architecture="x86-64")
618
663
self.check_vm_params_for_linux()
620
def test_spawn_vhd_glance_swapdisk(self):
621
# Change the default host_call_plugin to one that'll return
623
orig_func = stubs.FakeSessionForVMTests.host_call_plugin
624
_host_call_plugin = stubs.FakeSessionForVMTests.host_call_plugin_swap
625
stubs.FakeSessionForVMTests.host_call_plugin = _host_call_plugin
626
# Stubbing out firewall driver as previous stub sets a particular
627
# stub for async plugin calls
628
stubs.stubout_firewall_driver(self.stubs, self.conn)
630
# We'll steal the above glance linux test
631
self.test_spawn_vhd_glance_linux()
633
# Make sure to put this back
634
stubs.FakeSessionForVMTests.host_call_plugin = orig_func
636
# We should have 2 VBDs.
637
self.assertEqual(len(self.vm['VBDs']), 2)
638
# Now test that we have 1.
641
self.test_spawn_vhd_glance_linux()
642
self.assertEqual(len(self.vm['VBDs']), 1)
644
665
def test_spawn_vhd_glance_windows(self):
645
666
self._test_spawn(IMAGE_VHD, None, None,
646
667
os_type="windows", architecture="i386")
851
873
def finish_revert_migration(self, instance):
852
874
self.finish_revert_migration_called = True
854
conn = xenapi_conn.XenAPIDriver(False)
876
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
855
877
conn._vmops = VMOpsMock()
856
878
conn.finish_revert_migration(instance, None)
857
879
self.assertTrue(conn._vmops.finish_revert_migration_called)
859
881
def test_reboot_hard(self):
860
882
instance = self._create_instance()
861
conn = xenapi_conn.XenAPIDriver(False)
883
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
862
884
conn.reboot(instance, None, "HARD")
864
886
def test_reboot_soft(self):
865
887
instance = self._create_instance()
866
conn = xenapi_conn.XenAPIDriver(False)
888
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
867
889
conn.reboot(instance, None, "SOFT")
869
891
def test_reboot_halted(self):
870
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
892
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass',
871
894
instance = self._create_instance(spawn=False)
872
conn = xenapi_conn.XenAPIDriver(False)
895
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
873
896
xenapi_fake.create_vm(instance.name, 'Halted')
874
897
conn.reboot(instance, None, "SOFT")
875
898
vm_ref = vm_utils.lookup(session, instance.name)
879
902
def test_reboot_unknown_state(self):
880
903
instance = self._create_instance(spawn=False)
881
conn = xenapi_conn.XenAPIDriver(False)
904
conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
882
905
xenapi_fake.create_vm(instance.name, 'Unknown')
883
906
self.assertRaises(xenapi_fake.Failure, conn.reboot, instance,
909
def _test_maintenance_mode(self, find_host, find_aggregate):
910
real_call_xenapi = self.conn._session.call_xenapi
911
instance = self._create_instance(spawn=True)
914
# Record all the xenapi calls, and return a fake list of hosts
915
# for the host.get_all call
916
def fake_call_xenapi(method, *args):
917
api_calls[method] = args
918
if method == 'host.get_all':
919
return ['foo', 'bar', 'baz']
920
return real_call_xenapi(method, *args)
921
self.stubs.Set(self.conn._session, 'call_xenapi', fake_call_xenapi)
923
def fake_aggregate_get(context, host, key):
925
return [{'fake': 'aggregate'}]
928
self.stubs.Set(self.conn.virtapi, 'aggregate_get_by_host',
931
def fake_host_find(context, session, src, dst):
935
raise exception.NoValidHost("I saw this one coming...")
936
self.stubs.Set(host, '_host_find', fake_host_find)
938
result = self.conn.host_maintenance_mode('bar', 'on_maintenance')
939
self.assertEqual(result, 'on_maintenance')
941
# We expect the VM.pool_migrate call to have been called to
942
# migrate our instance to the 'bar' host
943
expected = (instance['uuid'], 'bar', {})
944
self.assertTrue(api_calls.get('VM.pool_migrate'), expected)
946
instance = db.instance_get_by_uuid(self.context, instance['uuid'])
947
self.assertTrue(instance['vm_state'], vm_states.ACTIVE)
948
self.assertTrue(instance['task_state'], task_states.MIGRATING)
950
def test_maintenance_mode(self):
951
self._test_maintenance_mode(True, True)
953
def test_maintenance_mode_no_host(self):
954
self.assertRaises(exception.NoValidHost,
955
self._test_maintenance_mode, False, True)
957
def test_maintenance_mode_no_aggregate(self):
958
self.assertRaises(exception.NotFound,
959
self._test_maintenance_mode, True, False)
961
def test_session_virtapi(self):
962
was = {'called': False}
964
def fake_aggregate_get_by_host(self, *args, **kwargs):
967
self.stubs.Set(self.conn._session._virtapi, "aggregate_get_by_host",
968
fake_aggregate_get_by_host)
970
self.stubs.Set(self.conn._session, "is_slave", True)
973
self.conn._session._get_host_uuid()
976
self.assertTrue(was['called'])
886
978
def _create_instance(self, instance_id=1, spawn=True):
887
979
"""Creates and spawns a test instance."""
888
980
instance_values = {
1172
1264
vm_utils.ImageType.to_string(vm_utils.ImageType.KERNEL),
1173
1265
vm_utils.ImageType.KERNEL_STR)
1175
def test_from_string(self):
1176
"""Can convert from string to type id."""
1267
def _assert_role(self, expected_role, image_type_id):
1177
1268
self.assertEquals(
1178
vm_utils.ImageType.from_string(vm_utils.ImageType.KERNEL_STR),
1179
vm_utils.ImageType.KERNEL)
1270
vm_utils.ImageType.get_role(image_type_id))
1272
def test_get_image_role_kernel(self):
1273
self._assert_role('kernel', vm_utils.ImageType.KERNEL)
1275
def test_get_image_role_ramdisk(self):
1276
self._assert_role('ramdisk', vm_utils.ImageType.RAMDISK)
1278
def test_get_image_role_disk(self):
1279
self._assert_role('root', vm_utils.ImageType.DISK)
1281
def test_get_image_role_disk_raw(self):
1282
self._assert_role('root', vm_utils.ImageType.DISK_RAW)
1284
def test_get_image_role_disk_vhd(self):
1285
self._assert_role('root', vm_utils.ImageType.DISK_VHD)
1182
1288
class XenAPIDetermineDiskImageTestCase(test.TestCase):
1275
1381
result = self.conn.get_host_uptime('host')
1276
1382
self.assertEqual(result, 'fake uptime')
1384
def test_supported_instances_is_included_in_host_state(self):
1385
stats = self.conn.get_host_stats()
1386
self.assertTrue('supported_instances' in stats)
1388
def test_supported_instances_is_calculated_by_to_supported_instances(self):
1390
def to_supported_instances(somedata):
1391
self.assertEquals(None, somedata)
1392
return "SOMERETURNVALUE"
1393
self.stubs.Set(host, 'to_supported_instances', to_supported_instances)
1395
stats = self.conn.get_host_stats()
1396
self.assertEquals("SOMERETURNVALUE", stats['supported_instances'])
1399
class ToSupportedInstancesTestCase(test.TestCase):
1400
def test_default_return_value(self):
1401
self.assertEquals([],
1402
host.to_supported_instances(None))
1404
def test_return_value(self):
1405
self.assertEquals([('x86_64', 'xapi', 'xen')],
1406
host.to_supported_instances([u'xen-3.0-x86_64']))
1408
def test_invalid_values_do_not_break(self):
1409
self.assertEquals([('x86_64', 'xapi', 'xen')],
1410
host.to_supported_instances([u'xen-3.0-x86_64', 'spam']))
1412
def test_multiple_values(self):
1415
('x86_64', 'xapi', 'xen'),
1416
('x86_32', 'xapi', 'hvm')
1418
host.to_supported_instances([u'xen-3.0-x86_64', 'hvm-3.0-x86_32'])
1279
1422
class XenAPIAutoDiskConfigTestCase(stubs.XenAPITestBase):
1280
1423
def setUp(self):
1396
1540
self.context = context.RequestContext(self.user_id, self.project_id)
1398
1542
def fake_create_vbd(session, vm_ref, vdi_ref, userdevice,
1399
vbd_type='disk', read_only=False, bootable=True):
1543
vbd_type='disk', read_only=False, bootable=True,
1402
1547
self.stubs.Set(vm_utils, 'create_vbd', fake_create_vbd)
1404
1549
def assertCalled(self, instance):
1405
1550
ctx = context.RequestContext(self.user_id, self.project_id)
1406
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
1551
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass',
1408
1554
disk_image_type = vm_utils.ImageType.DISK_VHD
1409
1555
vm_ref = xenapi_fake.create_vm(instance['name'], 'Halted')
1442
1598
self.assertCalled(instance)
1445
class XenAPIBWUsageTestCase(stubs.XenAPITestBase):
1601
class XenAPIBWCountersTestCase(stubs.XenAPITestBase):
1602
FAKE_VMS = {'test1:ref': dict(name_label='test1',
1603
other_config=dict(nova_uuid='hash'),
1605
_vifmap={'0': "a:b:c:d...",
1606
'1': "e:f:12:q..."}),
1607
'test2:ref': dict(name_label='test2',
1608
other_config=dict(nova_uuid='hash'),
1610
_vifmap={'0': "a:3:c:d...",
1611
'1': "e:f:42:q..."}),
1446
1614
def setUp(self):
1447
super(XenAPIBWUsageTestCase, self).setUp()
1448
self.stubs.Set(vm_utils, 'compile_metrics',
1449
XenAPIBWUsageTestCase._fake_compile_metrics)
1615
super(XenAPIBWCountersTestCase, self).setUp()
1616
self.stubs.Set(vm_utils, 'list_vms',
1617
XenAPIBWCountersTestCase._fake_list_vms)
1450
1618
self.flags(xenapi_connection_url='test_url',
1451
1619
xenapi_connection_password='test_pass',
1452
1620
firewall_driver='nova.virt.xenapi.firewall.'
1453
1621
'Dom0IptablesFirewallDriver')
1454
1622
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
1455
self.conn = xenapi_conn.XenAPIDriver(False)
1458
def _fake_compile_metrics(cls, start_time, stop_time=None):
1459
raise exception.CouldNotFetchMetrics()
1461
def test_get_all_bw_usage_in_failure_case(self):
1462
"""Test that get_all_bw_usage returns an empty list when metrics
1463
compilation failed. c.f. bug #910045.
1623
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
1625
def _fake_get_vif_device_map(vm_rec):
1626
return vm_rec['_vifmap']
1628
self.stubs.Set(self.conn._vmops, "_get_vif_device_map",
1629
_fake_get_vif_device_map)
1632
def _fake_list_vms(cls, session):
1633
return cls.FAKE_VMS.iteritems()
1636
def _fake_fetch_bandwidth_mt(cls, session):
1640
def _fake_fetch_bandwidth(cls, session):
1642
{'0': {'bw_in': 21024, 'bw_out': 22048},
1643
'1': {'bw_in': 231337, 'bw_out': 221212121}},
1645
{'0': {'bw_in': 1024, 'bw_out': 2048},
1646
'1': {'bw_in': 31337, 'bw_out': 21212121}},
1649
def test_get_all_bw_counters(self):
1650
class testinstance(object):
1651
def __init__(self, name, uuid):
1655
self.stubs.Set(vm_utils, 'fetch_bandwidth',
1656
XenAPIBWCountersTestCase._fake_fetch_bandwidth)
1657
result = self.conn.get_all_bw_counters([testinstance(
1663
self.assertEqual(len(result), 4)
1664
self.assertIn(dict(uuid='1-2-3',
1665
mac_address="a:b:c:d...",
1667
bw_out=2048), result)
1668
self.assertIn(dict(uuid='1-2-3',
1669
mac_address="e:f:12:q...",
1671
bw_out=21212121), result)
1673
self.assertIn(dict(uuid='4-5-6',
1674
mac_address="a:3:c:d...",
1676
bw_out=22048), result)
1677
self.assertIn(dict(uuid='4-5-6',
1678
mac_address="e:f:42:q...",
1680
bw_out=221212121), result)
1682
def test_get_all_bw_counters_in_failure_case(self):
1683
"""Test that get_all_bw_conters returns an empty list when
1684
no data returned from Xenserver. c.f. bug #910045.
1465
1686
class testinstance(object):
1466
1687
def __init__(self):
1467
1688
self.name = "instance-0001"
1468
1689
self.uuid = "1-2-3-4-5"
1470
result = self.conn.get_all_bw_usage([testinstance()],
1691
self.stubs.Set(vm_utils, 'fetch_bandwidth',
1692
XenAPIBWCountersTestCase._fake_fetch_bandwidth_mt)
1693
result = self.conn.get_all_bw_counters([testinstance()])
1472
1694
self.assertEqual(result, [])
1853
2079
firewall_driver='nova.virt.xenapi.firewall.'
1854
2080
'Dom0IptablesFirewallDriver',
1856
connection_type='xenapi',
1857
compute_driver='nova.virt.xenapi.driver.XenAPIDriver')
2082
compute_driver='xenapi.XenAPIDriver',
2083
node_availability_zone='avail_zone1')
1858
2084
host_ref = xenapi_fake.get_all('host')[0]
1859
2085
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
1860
2086
self.context = context.get_admin_context()
1861
self.conn = xenapi_conn.XenAPIDriver(False)
1862
self.compute = importutils.import_object(FLAGS.compute_manager)
2087
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2088
self.compute = importutils.import_object(CONF.compute_manager)
1863
2089
self.api = compute_api.AggregateAPI()
1864
2090
values = {'name': 'test_aggr',
1865
2091
'availability_zone': 'test_zone',
2118
2347
def __init__(self):
2119
2348
self._mock_calls = []
2121
def add_aggregate_host(self, ctxt, aggregate_id,
2350
def add_aggregate_host(self, ctxt, aggregate,
2122
2351
host_param, host, slave_info):
2123
2352
self._mock_calls.append((
2124
self.add_aggregate_host, ctxt, aggregate_id,
2353
self.add_aggregate_host, ctxt, aggregate,
2125
2354
host_param, host, slave_info))
2127
2356
def remove_aggregate_host(self, ctxt, aggregate_id, host_param,
2280
2507
def test_pre_live_migration(self):
2281
2508
# ensure method is present
2282
2509
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
2283
self.conn = xenapi_conn.XenAPIDriver(False)
2510
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2284
2511
self.conn.pre_live_migration(None, None, None, None)
2286
2513
def test_post_live_migration_at_destination(self):
2287
2514
# ensure method is present
2288
2515
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
2289
self.conn = xenapi_conn.XenAPIDriver(False)
2516
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2290
2517
self.conn.post_live_migration_at_destination(None, None, None, None)
2292
2519
def test_check_can_live_migrate_destination_with_block_migration(self):
2293
2520
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
2294
self.conn = xenapi_conn.XenAPIDriver(False)
2521
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2296
2523
self.stubs.Set(vm_utils, "safe_find_sr", lambda _x: "asdf")
2304
2531
result = self.conn.check_can_live_migrate_destination(self.context,
2305
{'host': 'host'}, True, False)
2306
2535
self.assertEqual(expected, result)
2308
2537
def test_check_can_live_migrate_destination_block_migration_fails(self):
2309
2538
stubs.stubout_session(self.stubs,
2310
2539
stubs.FakeSessionForFailedMigrateTests)
2311
self.conn = xenapi_conn.XenAPIDriver(False)
2540
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2312
2541
self.assertRaises(exception.MigrationError,
2313
2542
self.conn.check_can_live_migrate_destination,
2314
self.context, {'host': 'host'}, True, False)
2543
self.context, {'host': 'host'},
2316
2547
def test_check_can_live_migrate_source_with_block_migrate(self):
2317
2548
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
2318
self.conn = xenapi_conn.XenAPIDriver(False)
2549
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2320
2551
def fake_generate_vdi_map(destination_sr_ref, _vm_ref):
2333
2564
'destination_sr_ref': None,
2334
2565
'migrate_send_data': None
2336
self.assertNotRaises(None,
2337
self.conn.check_can_live_migrate_source,
2567
self.conn.check_can_live_migrate_source(self.context,
2342
2571
def test_check_can_live_migrate_source_with_block_migrate_fails(self):
2343
2572
stubs.stubout_session(self.stubs,
2344
2573
stubs.FakeSessionForFailedMigrateTests)
2345
self.conn = xenapi_conn.XenAPIDriver(False)
2574
self.conn = xenapi_conn.XenAPIDriver(fake.FakeVirtAPI(), False)
2347
2576
def fake_generate_vdi_map(destination_sr_ref, _vm_ref):
added
removed
nova/tests/test_xenapi.py
