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
17
"""Test suite for Xen Storage Manager Volume Driver."""
20
from nova import context
22
from nova import exception
23
from nova import flags
24
from nova.openstack.common import log as logging
25
from nova.tests.xenapi import stubs
26
from nova.virt.xenapi import fake as xenapi_fake
27
from nova.volume import xensm
29
LOG = logging.getLogger(__name__)
34
class XenSMTestCase(stubs.XenAPITestBase):
35
"""Unit tests for Xen Storage Manager Volume operations."""
37
def _get_sm_backend_params(self):
38
config_params = ("name_label=testsmbackend "
40
"serverpath=/tmp/nfspath")
41
params = dict(flavor_id=1,
44
config_params=config_params)
48
super(XenSMTestCase, self).setUp()
50
self.project_id = 'fake'
51
self.context = context.RequestContext(self.user_id, self.project_id)
52
self.flags(connection_type='xenapi',
53
xenapi_connection_url='http://test_url',
54
xenapi_connection_username='test_user',
55
xenapi_connection_password='test_pass')
56
stubs.stubout_session(self.stubs, xenapi_fake.SessionBase)
58
self.driver = xensm.XenSMDriver()
61
def _setup_step(self, ctxt):
62
# Create a fake backend conf
63
params = self._get_sm_backend_params()
64
beconf = db.sm_backend_conf_create(ctxt,
66
# Call setup, the one time operation that will create a backend SR
67
self.driver.do_setup(ctxt)
70
def test_do_setup(self):
71
ctxt = context.get_admin_context()
72
beconf = self._setup_step(ctxt)
73
beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
74
self.assertIsInstance(beconf['sr_uuid'], basestring)
76
def _create_volume(self, size=0):
77
"""Create a volume object."""
80
vol['user_id'] = 'fake'
81
vol['project_id'] = 'fake'
82
vol['host'] = 'localhost'
83
vol['availability_zone'] = FLAGS.storage_availability_zone
84
vol['status'] = "creating"
85
vol['attach_status'] = "detached"
86
return db.volume_create(self.context, vol)
88
def test_create_volume(self):
89
ctxt = context.get_admin_context()
90
beconf = self._setup_step(ctxt)
91
volume = self._create_volume()
92
self.assertNotRaises(None, self.driver.create_volume, volume)
93
self.assertNotRaises(None,
98
def test_local_path(self):
99
ctxt = context.get_admin_context()
100
volume = self._create_volume()
101
val = self.driver.local_path(volume)
102
self.assertIsInstance(val, basestring)
104
def test_delete_volume(self):
105
ctxt = context.get_admin_context()
106
beconf = self._setup_step(ctxt)
107
volume = self._create_volume()
108
self.driver.create_volume(volume)
109
self.assertNotRaises(None, self.driver.delete_volume, volume)
110
self.assertRaises(exception.NotFound,
115
def test_delete_volume_raises_notfound(self):
116
ctxt = context.get_admin_context()
117
beconf = self._setup_step(ctxt)
118
self.assertRaises(exception.NotFound,
119
self.driver.delete_volume,
122
def _get_expected_conn(self, beconf, vol):
124
expected['volume_id'] = unicode(vol['id'])
125
expected['flavor_id'] = beconf['flavor_id']
126
expected['sr_uuid'] = unicode(beconf['sr_uuid'])
127
expected['sr_type'] = unicode(beconf['sr_type'])
130
def test_initialize_connection(self):
131
ctxt = context.get_admin_context()
132
beconf = self._setup_step(ctxt)
133
beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
134
volume = self._create_volume()
135
self.driver.create_volume(volume)
136
expected = self._get_expected_conn(beconf, volume)
137
conn = self.driver.initialize_connection(volume, 'fakeConnector')
139
for key in ['volume_id', 'flavor_id', 'sr_uuid', 'sr_type']:
140
res[key] = conn['data'][key]
141
self.assertDictMatch(expected, res)
142
self.assertEqual(set(conn['data']['introduce_sr_keys']),
143
set([u'sr_type', u'server', u'serverpath']))