1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright (c) 2011 Zadara Storage Inc.
4
# Copyright (c) 2011 OpenStack LLC.
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
Unit Tests for volume types code
21
from nova import context
23
from nova import exception
24
from nova import flags
25
from nova import log as logging
27
from nova import utils
28
from nova.volume import volume_types
29
from nova.db.sqlalchemy.session import get_session
30
from nova.db.sqlalchemy import models
33
LOG = logging.getLogger('nova.tests.test_volume_types')
36
class VolumeTypeTestCase(test.TestCase):
37
"""Test cases for volume type code"""
39
super(VolumeTypeTestCase, self).setUp()
41
self.ctxt = context.get_admin_context()
42
self.vol_type1_name = str(int(time.time()))
43
self.vol_type1_specs = dict(
44
type="physical drive",
49
self.vol_type1 = dict(name=self.vol_type1_name,
50
extra_specs=self.vol_type1_specs)
52
def test_volume_type_create_then_destroy(self):
53
"""Ensure volume types can be created and deleted"""
54
prev_all_vtypes = volume_types.get_all_types(self.ctxt)
56
volume_types.create(self.ctxt,
59
new = volume_types.get_volume_type_by_name(self.ctxt,
62
LOG.info(_("Given data: %s"), self.vol_type1_specs)
63
LOG.info(_("Result data: %s"), new)
65
for k, v in self.vol_type1_specs.iteritems():
66
self.assertEqual(v, new['extra_specs'][k],
67
'one of fields doesnt match')
69
new_all_vtypes = volume_types.get_all_types(self.ctxt)
70
self.assertEqual(len(prev_all_vtypes) + 1,
72
'drive type was not created')
74
volume_types.destroy(self.ctxt, self.vol_type1_name)
75
new_all_vtypes = volume_types.get_all_types(self.ctxt)
76
self.assertEqual(prev_all_vtypes,
78
'drive type was not deleted')
80
def test_volume_type_create_then_purge(self):
81
"""Ensure volume types can be created and deleted"""
82
prev_all_vtypes = volume_types.get_all_types(self.ctxt, inactive=1)
84
volume_types.create(self.ctxt,
87
new = volume_types.get_volume_type_by_name(self.ctxt,
90
for k, v in self.vol_type1_specs.iteritems():
91
self.assertEqual(v, new['extra_specs'][k],
92
'one of fields doesnt match')
94
new_all_vtypes = volume_types.get_all_types(self.ctxt, inactive=1)
95
self.assertEqual(len(prev_all_vtypes) + 1,
97
'drive type was not created')
99
volume_types.destroy(self.ctxt, self.vol_type1_name)
100
new_all_vtypes2 = volume_types.get_all_types(self.ctxt, inactive=1)
101
self.assertEqual(len(new_all_vtypes),
102
len(new_all_vtypes2),
103
'drive type was incorrectly deleted')
105
volume_types.purge(self.ctxt, self.vol_type1_name)
106
new_all_vtypes2 = volume_types.get_all_types(self.ctxt, inactive=1)
107
self.assertEqual(len(new_all_vtypes) - 1,
108
len(new_all_vtypes2),
109
'drive type was not purged')
111
def test_get_all_volume_types(self):
112
"""Ensures that all volume types can be retrieved"""
113
session = get_session()
114
total_volume_types = session.query(models.VolumeTypes).\
116
vol_types = volume_types.get_all_types(self.ctxt)
117
self.assertEqual(total_volume_types, len(vol_types))
119
def test_non_existant_inst_type_shouldnt_delete(self):
120
"""Ensures that volume type creation fails with invalid args"""
121
self.assertRaises(exception.ApiError,
122
volume_types.destroy, self.ctxt, "sfsfsdfdfs")
124
def test_repeated_vol_types_should_raise_api_error(self):
125
"""Ensures that volume duplicates raises ApiError"""
126
new_name = self.vol_type1_name + "dup"
127
volume_types.create(self.ctxt, new_name)
128
volume_types.destroy(self.ctxt, new_name)
131
volume_types.create, self.ctxt, new_name)
133
def test_invalid_volume_types_params(self):
134
"""Ensures that volume type creation fails with invalid args"""
135
self.assertRaises(exception.InvalidVolumeType,
136
volume_types.destroy, self.ctxt, None)
137
self.assertRaises(exception.InvalidVolumeType,
138
volume_types.purge, self.ctxt, None)
139
self.assertRaises(exception.InvalidVolumeType,
140
volume_types.get_volume_type, self.ctxt, None)
141
self.assertRaises(exception.InvalidVolumeType,
142
volume_types.get_volume_type_by_name,
145
def test_volume_type_get_by_id_and_name(self):
146
"""Ensure volume types get returns same entry"""
147
volume_types.create(self.ctxt,
149
self.vol_type1_specs)
150
new = volume_types.get_volume_type_by_name(self.ctxt,
153
new2 = volume_types.get_volume_type(self.ctxt, new['id'])
154
self.assertEqual(new, new2)
156
def test_volume_type_search_by_extra_spec(self):
157
"""Ensure volume types get by extra spec returns correct type"""
158
volume_types.create(self.ctxt, "type1", {"key1": "val1",
160
volume_types.create(self.ctxt, "type2", {"key2": "val2",
162
volume_types.create(self.ctxt, "type3", {"key3": "another_value",
165
vol_types = volume_types.get_all_types(self.ctxt,
166
search_opts={'extra_specs': {"key1": "val1"}})
167
LOG.info("vol_types: %s" % vol_types)
168
self.assertEqual(len(vol_types), 1)
169
self.assertTrue("type1" in vol_types.keys())
170
self.assertEqual(vol_types['type1']['extra_specs'],
171
{"key1": "val1", "key2": "val2"})
173
vol_types = volume_types.get_all_types(self.ctxt,
174
search_opts={'extra_specs': {"key2": "val2"}})
175
LOG.info("vol_types: %s" % vol_types)
176
self.assertEqual(len(vol_types), 2)
177
self.assertTrue("type1" in vol_types.keys())
178
self.assertTrue("type2" in vol_types.keys())
180
vol_types = volume_types.get_all_types(self.ctxt,
181
search_opts={'extra_specs': {"key3": "val3"}})
182
LOG.info("vol_types: %s" % vol_types)
183
self.assertEqual(len(vol_types), 1)
184
self.assertTrue("type2" in vol_types.keys())
186
def test_volume_type_search_by_extra_spec_multiple(self):
187
"""Ensure volume types get by extra spec returns correct type"""
188
volume_types.create(self.ctxt, "type1", {"key1": "val1",
191
volume_types.create(self.ctxt, "type2", {"key2": "val2",
193
volume_types.create(self.ctxt, "type3", {"key1": "val1",
197
vol_types = volume_types.get_all_types(self.ctxt,
198
search_opts={'extra_specs': {"key1": "val1",
200
LOG.info("vol_types: %s" % vol_types)
201
self.assertEqual(len(vol_types), 2)
202
self.assertTrue("type1" in vol_types.keys())
203
self.assertTrue("type3" in vol_types.keys())
204
self.assertEqual(vol_types['type1']['extra_specs'],
205
{"key1": "val1", "key2": "val2", "key3": "val3"})
206
self.assertEqual(vol_types['type3']['extra_specs'],
207
{"key1": "val1", "key3": "val3", "key4": "val4"})
added
removed
nova/tests/test_volume_types.py
