1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright (c) 2010 Citrix Systems, Inc.
5
# Licensed under the Apache License, Version 2.0 (the "License"); you may
6
# not use this file except in compliance with the License. You may obtain
7
# a copy of the License at
9
# http://www.apache.org/licenses/LICENSE-2.0
11
# Unless required by applicable law or agreed to in writing, software
12
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14
# License for the specific language governing permissions and limitations
24
from nova import context
25
from nova import flags
27
from nova import utils
28
from nova.auth import manager
29
from nova.compute import instance_types
30
from nova.compute import power_state
31
from nova.virt import xenapi_conn
32
from nova.virt.xenapi import fake as xenapi_fake
33
from nova.virt.xenapi import volume_utils
34
from nova.virt.xenapi.vmops import SimpleDH
35
from nova.tests.db import fakes as db_fakes
36
from nova.tests.xenapi import stubs
37
from nova.tests.glance import stubs as glance_stubs
42
class XenAPIVolumeTestCase(test.TestCase):
44
Unit tests for Volume operations
47
super(XenAPIVolumeTestCase, self).setUp()
48
self.stubs = stubout.StubOutForTesting()
49
FLAGS.target_host = '127.0.0.1'
50
FLAGS.xenapi_connection_url = 'test_url'
51
FLAGS.xenapi_connection_password = 'test_pass'
52
db_fakes.stub_out_db_instance_api(self.stubs)
53
stubs.stub_out_get_target(self.stubs)
55
self.values = {'name': 1, 'id': 1,
61
'instance_type': 'm1.large',
62
'mac_address': 'aa:bb:cc:dd:ee:ff',
65
def _create_volume(self, size='0'):
66
"""Create a volume object."""
69
vol['user_id'] = 'fake'
70
vol['project_id'] = 'fake'
71
vol['host'] = 'localhost'
72
vol['availability_zone'] = FLAGS.storage_availability_zone
73
vol['status'] = "creating"
74
vol['attach_status'] = "detached"
75
return db.volume_create(context.get_admin_context(), vol)
77
def test_create_iscsi_storage(self):
78
""" This shows how to test helper classes' methods """
79
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
80
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
81
helper = volume_utils.VolumeHelper
82
helper.XenAPI = session.get_imported_xenapi()
83
vol = self._create_volume()
84
info = helper.parse_volume_info(vol['id'], '/dev/sdc')
85
label = 'SR-%s' % vol['id']
86
description = 'Test-SR'
87
sr_ref = helper.create_iscsi_storage(session, info, label, description)
88
srs = xenapi_fake.get_all('SR')
89
self.assertEqual(sr_ref, srs[0])
90
db.volume_destroy(context.get_admin_context(), vol['id'])
92
def test_parse_volume_info_raise_exception(self):
93
""" This shows how to test helper classes' methods """
94
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
95
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
96
helper = volume_utils.VolumeHelper
97
helper.XenAPI = session.get_imported_xenapi()
98
vol = self._create_volume()
99
# oops, wrong mount point!
100
self.assertRaises(volume_utils.StorageError,
101
helper.parse_volume_info,
104
db.volume_destroy(context.get_admin_context(), vol['id'])
106
def test_attach_volume(self):
107
""" This shows how to test Ops classes' methods """
108
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
109
conn = xenapi_conn.get_connection(False)
110
volume = self._create_volume()
111
instance = db.instance_create(self.values)
112
vm = xenapi_fake.create_vm(instance.name, 'Running')
113
result = conn.attach_volume(instance.name, volume['id'], '/dev/sdc')
116
# check that the VM has a VBD attached to it
117
# Get XenAPI record for VBD
118
vbds = xenapi_fake.get_all('VBD')
119
vbd = xenapi_fake.get_record('VBD', vbds[0])
121
self.assertEqual(vm_ref, vm)
125
def test_attach_volume_raise_exception(self):
126
""" This shows how to test when exceptions are raised """
127
stubs.stubout_session(self.stubs,
128
stubs.FakeSessionForVolumeFailedTests)
129
conn = xenapi_conn.get_connection(False)
130
volume = self._create_volume()
131
instance = db.instance_create(self.values)
132
xenapi_fake.create_vm(instance.name, 'Running')
133
self.assertRaises(Exception,
140
super(XenAPIVolumeTestCase, self).tearDown()
141
self.stubs.UnsetAll()
144
class XenAPIVMTestCase(test.TestCase):
146
Unit tests for VM operations
149
super(XenAPIVMTestCase, self).setUp()
150
self.manager = manager.AuthManager()
151
self.user = self.manager.create_user('fake', 'fake', 'fake',
153
self.project = self.manager.create_project('fake', 'fake', 'fake')
154
self.network = utils.import_object(FLAGS.network_manager)
155
self.stubs = stubout.StubOutForTesting()
156
FLAGS.xenapi_connection_url = 'test_url'
157
FLAGS.xenapi_connection_password = 'test_pass'
159
xenapi_fake.create_local_srs()
160
db_fakes.stub_out_db_instance_api(self.stubs)
161
xenapi_fake.create_network('fake', FLAGS.flat_network_bridge)
162
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
163
stubs.stubout_get_this_vm_uuid(self.stubs)
164
stubs.stubout_stream_disk(self.stubs)
165
glance_stubs.stubout_glance_client(self.stubs,
166
glance_stubs.FakeGlance)
167
self.conn = xenapi_conn.get_connection(False)
169
def test_list_instances_0(self):
170
instances = self.conn.list_instances()
171
self.assertEquals(instances, [])
173
def test_get_diagnostics(self):
174
instance = self._create_instance()
175
self.conn.get_diagnostics(instance)
177
def test_instance_snapshot(self):
178
stubs.stubout_instance_snapshot(self.stubs)
179
instance = self._create_instance()
182
template_vm_ref = self.conn.snapshot(instance, name)
184
def ensure_vm_was_torn_down():
186
for vm_ref in xenapi_fake.get_all('VM'):
187
vm_rec = xenapi_fake.get_record('VM', vm_ref)
188
if not vm_rec["is_control_domain"]:
189
vm_labels.append(vm_rec["name_label"])
191
self.assertEquals(vm_labels, [1])
193
def ensure_vbd_was_torn_down():
195
for vbd_ref in xenapi_fake.get_all('VBD'):
196
vbd_rec = xenapi_fake.get_record('VBD', vbd_ref)
197
vbd_labels.append(vbd_rec["vm_name_label"])
199
self.assertEquals(vbd_labels, [1])
201
def ensure_vdi_was_torn_down():
202
for vdi_ref in xenapi_fake.get_all('VDI'):
203
vdi_rec = xenapi_fake.get_record('VDI', vdi_ref)
204
name_label = vdi_rec["name_label"]
205
self.assert_(not name_label.endswith('snapshot'))
208
ensure_vm_was_torn_down()
209
ensure_vbd_was_torn_down()
210
ensure_vdi_was_torn_down()
214
def check_vm_record(self, conn):
215
instances = conn.list_instances()
216
self.assertEquals(instances, [1])
218
# Get Nova record for VM
219
vm_info = conn.get_info(1)
221
# Get XenAPI record for VM
222
vms = [rec for ref, rec
223
in xenapi_fake.get_all_records('VM').iteritems()
224
if not rec['is_control_domain']]
227
# Check that m1.large above turned into the right thing.
228
instance_type = instance_types.INSTANCE_TYPES['m1.large']
229
mem_kib = long(instance_type['memory_mb']) << 10
230
mem_bytes = str(mem_kib << 10)
231
vcpus = instance_type['vcpus']
232
self.assertEquals(vm_info['max_mem'], mem_kib)
233
self.assertEquals(vm_info['mem'], mem_kib)
234
self.assertEquals(vm['memory_static_max'], mem_bytes)
235
self.assertEquals(vm['memory_dynamic_max'], mem_bytes)
236
self.assertEquals(vm['memory_dynamic_min'], mem_bytes)
237
self.assertEquals(vm['VCPUs_max'], str(vcpus))
238
self.assertEquals(vm['VCPUs_at_startup'], str(vcpus))
240
# Check that the VM is running according to Nova
241
self.assertEquals(vm_info['state'], power_state.RUNNING)
243
# Check that the VM is running according to XenAPI.
244
self.assertEquals(vm['power_state'], 'Running')
246
def _test_spawn(self, image_id, kernel_id, ramdisk_id):
247
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
250
'project_id': self.project.id,
251
'user_id': self.user.id,
252
'image_id': image_id,
253
'kernel_id': kernel_id,
254
'ramdisk_id': ramdisk_id,
255
'instance_type': 'm1.large',
256
'mac_address': 'aa:bb:cc:dd:ee:ff',
258
conn = xenapi_conn.get_connection(False)
259
instance = db.instance_create(values)
261
self.check_vm_record(conn)
263
def test_spawn_raw_objectstore(self):
264
FLAGS.xenapi_image_service = 'objectstore'
265
self._test_spawn(1, None, None)
267
def test_spawn_objectstore(self):
268
FLAGS.xenapi_image_service = 'objectstore'
269
self._test_spawn(1, 2, 3)
271
def test_spawn_raw_glance(self):
272
FLAGS.xenapi_image_service = 'glance'
273
self._test_spawn(1, None, None)
275
def test_spawn_glance(self):
276
FLAGS.xenapi_image_service = 'glance'
277
self._test_spawn(1, 2, 3)
280
super(XenAPIVMTestCase, self).tearDown()
281
self.manager.delete_project(self.project)
282
self.manager.delete_user(self.user)
283
self.stubs.UnsetAll()
285
def _create_instance(self):
286
"""Creates and spawns a test instance"""
290
'project_id': self.project.id,
291
'user_id': self.user.id,
295
'instance_type': 'm1.large',
296
'mac_address': 'aa:bb:cc:dd:ee:ff'}
297
instance = db.instance_create(values)
298
self.conn.spawn(instance)
302
class XenAPIDiffieHellmanTestCase(test.TestCase):
304
Unit tests for Diffie-Hellman code
307
super(XenAPIDiffieHellmanTestCase, self).setUp()
308
self.alice = SimpleDH()
309
self.bob = SimpleDH()
311
def test_shared(self):
312
alice_pub = self.alice.get_public()
313
bob_pub = self.bob.get_public()
314
alice_shared = self.alice.compute_shared(bob_pub)
315
bob_shared = self.bob.compute_shared(alice_pub)
316
self.assertEquals(alice_shared, bob_shared)
318
def test_encryption(self):
319
msg = "This is a top-secret message"
320
enc = self.alice.encrypt(msg)
321
dec = self.bob.decrypt(enc)
322
self.assertEquals(dec, msg)
325
super(XenAPIDiffieHellmanTestCase, self).tearDown()
added
removed
nova/tests/test_xenapi.py
