← Back to branch summary

~ubuntu-branches/ubuntu/utopic/cinder/utopic

« back to all changes in this revision

Viewing changes to cinder/tests/db/test_qos_specs.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, James Page, Adam Gandelman, Chuck Short
  • Date: 2013-09-08 21:09:46 UTC
  • mfrom: (1.1.18)
  • Revision ID: package-import@ubuntu.com-20130908210946-3dbzq1jy5uji4wad
Tags: 1:2013.2~b3-0ubuntu1
https://launchpad.net/bugs/1190791
[ James Page ]
* d/control: Switch ceph-common -> python-ceph inline with upstream
  refactoring of Ceph RBD driver, move to Suggests of python-cinder.
  (LP: #1190791). 

[ Adam Gandelman ]
* debian/patches/avoid_paramiko_vers_depends.patch: Dropped, no longer
  required.
* Add minimum requirement python-greenlet (>= 0.3.2).
* Add minimum requirement python-eventlet (>= 0.12.0).
* Add minimum requirement python-paramiko (>= 1.8).

[ Chuck Short ]
* New upstream release.
* debian/patches/skip-sqlachemy-failures.patch: Skip testfailures
  with sqlalchemy 0.8 until they are fixed upstream.
* debian/control: Add python-babel to build-depends.
* debian/control: Add python-novaclient to build-depends.

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/db/test_qos_specs.py
 
1
# Copyright (C) 2013 eBay Inc.
 
2
# Copyright (C) 2013 OpenStack LLC.
 
3
# All Rights Reserved.
 
4
#
 
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
 
8
#
 
9
#         http://www.apache.org/licenses/LICENSE-2.0
 
10
#
 
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
 
15
#    under the License.
 
16
 
 
17
"""Tests for qaulity_of_service_specs table."""
 
18
 
 
19
 
 
20
import time
 
21
 
 
22
from cinder import context
 
23
from cinder import db
 
24
from cinder import exception
 
25
from cinder.openstack.common import log as logging
 
26
from cinder import test
 
27
from cinder.volume import volume_types
 
28
 
 
29
 
 
30
LOG = logging.getLogger(__name__)
 
31
 
 
32
 
 
33
def fake_qos_specs_get_by_name(context, name, session=None, inactive=False):
 
34
    pass
 
35
 
 
36
 
 
37
class QualityOfServiceSpecsTableTestCase(test.TestCase):
 
38
    """Test case for QualityOfServiceSpecs model."""
 
39
 
 
40
    def setUp(self):
 
41
        super(QualityOfServiceSpecsTableTestCase, self).setUp()
 
42
        self.ctxt = context.RequestContext(user_id='user_id',
 
43
                                           project_id='project_id',
 
44
                                           is_admin=True)
 
45
 
 
46
    def tearDown(self):
 
47
        super(QualityOfServiceSpecsTableTestCase, self).tearDown()
 
48
 
 
49
    def _create_qos_specs(self, name, values=None):
 
50
        """Create a transfer object."""
 
51
        if values:
 
52
            specs = dict(name=name, qos_specs=values)
 
53
        else:
 
54
            specs = {'name': name,
 
55
                     'qos_specs': {
 
56
                         'consumer': 'back-end',
 
57
                         'key1': 'value1',
 
58
                         'key2': 'value2'}}
 
59
        return db.qos_specs_create(self.ctxt, specs)['id']
 
60
 
 
61
    def test_qos_specs_create(self):
 
62
        # If there is qos specs with the same name exists,
 
63
        # a QoSSpecsExists exception will be raised.
 
64
        name = 'QoSSpecsCreationTest'
 
65
        self._create_qos_specs(name)
 
66
        self.assertRaises(exception.QoSSpecsExists,
 
67
                          db.qos_specs_create, self.ctxt, dict(name=name))
 
68
 
 
69
        specs_id = self._create_qos_specs('NewName')
 
70
        query_id = db.qos_specs_get_by_name(
 
71
            self.ctxt, 'NewName')['id']
 
72
        self.assertEquals(specs_id, query_id)
 
73
 
 
74
    def test_qos_specs_get(self):
 
75
        value = dict(consumer='front-end',
 
76
                     key1='foo', key2='bar')
 
77
        specs_id = self._create_qos_specs('Name1', value)
 
78
 
 
79
        fake_id = 'fake-UUID'
 
80
        self.assertRaises(exception.QoSSpecsNotFound,
 
81
                          db.qos_specs_get, self.ctxt, fake_id)
 
82
 
 
83
        specs = db.qos_specs_get(self.ctxt, specs_id)
 
84
        expected = dict(name='Name1', id=specs_id, consumer='front-end')
 
85
        del value['consumer']
 
86
        expected.update(dict(specs=value))
 
87
        self.assertDictMatch(specs, expected)
 
88
 
 
89
    def test_qos_specs_get_all(self):
 
90
        value1 = dict(consumer='front-end',
 
91
                      key1='v1', key2='v2')
 
92
        value2 = dict(consumer='back-end',
 
93
                      key3='v3', key4='v4')
 
94
        value3 = dict(consumer='back-end',
 
95
                      key5='v5', key6='v6')
 
96
 
 
97
        spec_id1 = self._create_qos_specs('Name1', value1)
 
98
        spec_id2 = self._create_qos_specs('Name2', value2)
 
99
        spec_id3 = self._create_qos_specs('Name3', value3)
 
100
 
 
101
        specs = db.qos_specs_get_all(self.ctxt)
 
102
        self.assertEquals(len(specs), 3,
 
103
                          "Unexpected number of qos specs records")
 
104
 
 
105
        expected1 = dict(name='Name1', id=spec_id1, consumer='front-end')
 
106
        expected2 = dict(name='Name2', id=spec_id2, consumer='back-end')
 
107
        expected3 = dict(name='Name3', id=spec_id3, consumer='back-end')
 
108
        del value1['consumer']
 
109
        del value2['consumer']
 
110
        del value3['consumer']
 
111
        expected1.update(dict(specs=value1))
 
112
        expected2.update(dict(specs=value2))
 
113
        expected3.update(dict(specs=value3))
 
114
        self.assertIn(expected1, specs)
 
115
        self.assertIn(expected2, specs)
 
116
        self.assertIn(expected3, specs)
 
117
 
 
118
    def test_qos_specs_get_by_name(self):
 
119
        name = str(int(time.time()))
 
120
        value = dict(consumer='front-end',
 
121
                     foo='Foo', bar='Bar')
 
122
        specs_id = self._create_qos_specs(name, value)
 
123
        specs = db.qos_specs_get_by_name(self.ctxt, name)
 
124
        del value['consumer']
 
125
        expected = {'name': name,
 
126
                    'id': specs_id,
 
127
                    'consumer': 'front-end',
 
128
                    'specs': value}
 
129
        self.assertDictMatch(specs, expected)
 
130
 
 
131
    def test_qos_specs_delete(self):
 
132
        name = str(int(time.time()))
 
133
        specs_id = self._create_qos_specs(name)
 
134
 
 
135
        db.qos_specs_delete(self.ctxt, specs_id)
 
136
        self.assertRaises(exception.QoSSpecsNotFound, db.qos_specs_get,
 
137
                          self.ctxt, specs_id)
 
138
 
 
139
    def test_associate_type_with_qos(self):
 
140
        self.assertRaises(exception.VolumeTypeNotFound,
 
141
                          db.volume_type_qos_associate,
 
142
                          self.ctxt, 'Fake-VOLID', 'Fake-QOSID')
 
143
        type_id = volume_types.create(self.ctxt, 'TypeName')['id']
 
144
        specs_id = self._create_qos_specs('FakeQos')
 
145
        db.volume_type_qos_associate(self.ctxt, type_id, specs_id)
 
146
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
147
        self.assertEquals(len(res), 1)
 
148
        self.assertEquals(res[0]['id'], type_id)
 
149
        self.assertEquals(res[0]['qos_specs_id'], specs_id)
 
150
 
 
151
    def test_qos_associations_get(self):
 
152
        self.assertRaises(exception.QoSSpecsNotFound,
 
153
                          db.qos_specs_associations_get,
 
154
                          self.ctxt, 'Fake-UUID')
 
155
 
 
156
        type_id = volume_types.create(self.ctxt, 'TypeName')['id']
 
157
        specs_id = self._create_qos_specs('FakeQos')
 
158
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
159
        self.assertEquals(len(res), 0)
 
160
 
 
161
        db.volume_type_qos_associate(self.ctxt, type_id, specs_id)
 
162
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
163
        self.assertEquals(len(res), 1)
 
164
        self.assertEquals(res[0]['id'], type_id)
 
165
        self.assertEquals(res[0]['qos_specs_id'], specs_id)
 
166
 
 
167
        type0_id = volume_types.create(self.ctxt, 'Type0Name')['id']
 
168
        db.volume_type_qos_associate(self.ctxt, type0_id, specs_id)
 
169
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
170
        self.assertEquals(len(res), 2)
 
171
        self.assertEquals(res[0]['qos_specs_id'], specs_id)
 
172
        self.assertEquals(res[1]['qos_specs_id'], specs_id)
 
173
 
 
174
    def test_qos_specs_disassociate(self):
 
175
        type_id = volume_types.create(self.ctxt, 'TypeName')['id']
 
176
        specs_id = self._create_qos_specs('FakeQos')
 
177
        db.volume_type_qos_associate(self.ctxt, type_id, specs_id)
 
178
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
179
        self.assertEquals(res[0]['id'], type_id)
 
180
        self.assertEquals(res[0]['qos_specs_id'], specs_id)
 
181
 
 
182
        db.qos_specs_disassociate(self.ctxt, specs_id, type_id)
 
183
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
184
        self.assertEquals(len(res), 0)
 
185
        res = db.volume_type_get(self.ctxt, type_id)
 
186
        self.assertEquals(res['qos_specs_id'], None)
 
187
 
 
188
    def test_qos_specs_disassociate_all(self):
 
189
        specs_id = self._create_qos_specs('FakeQos')
 
190
        type1_id = volume_types.create(self.ctxt, 'Type1Name')['id']
 
191
        type2_id = volume_types.create(self.ctxt, 'Type2Name')['id']
 
192
        type3_id = volume_types.create(self.ctxt, 'Type3Name')['id']
 
193
        db.volume_type_qos_associate(self.ctxt, type1_id, specs_id)
 
194
        db.volume_type_qos_associate(self.ctxt, type2_id, specs_id)
 
195
        db.volume_type_qos_associate(self.ctxt, type3_id, specs_id)
 
196
 
 
197
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
198
        self.assertEquals(len(res), 3)
 
199
 
 
200
        db.qos_specs_disassociate_all(self.ctxt, specs_id)
 
201
        res = db.qos_specs_associations_get(self.ctxt, specs_id)
 
202
        self.assertEquals(len(res), 0)
 
203
 
 
204
    def test_qos_specs_update(self):
 
205
        name = 'FakeName'
 
206
        specs_id = self._create_qos_specs(name)
 
207
        value = dict(key2='new_value2', key3='value3')
 
208
 
 
209
        self.assertRaises(exception.QoSSpecsNotFound, db.qos_specs_update,
 
210
                          self.ctxt, 'Fake-UUID', value)
 
211
        db.qos_specs_update(self.ctxt, specs_id, value)
 
212
        specs = db.qos_specs_get(self.ctxt, specs_id)
 
213
        self.assertEqual(specs['specs']['key2'], 'new_value2')
 
214
        self.assertEqual(specs['specs']['key3'], 'value3')