← Back to branch summary

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

« back to all changes in this revision

Viewing changes to cinder/tests/api/contrib/test_volume_encryption_metadata.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/api/contrib/test_volume_encryption_metadata.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright (c) 2013 The Johns Hopkins University/Applied Physics Laboratory
 
4
# All Rights Reserved.
 
5
#
 
6
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
7
#    not use this file except in compliance with the License. You may obtain
 
8
#    a copy of the License at
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
12
#    Unless required by applicable law or agreed to in writing, software
 
13
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
14
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
15
#    License for the specific language governing permissions and limitations
 
16
#    under the License.
 
17
 
 
18
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
19
 
 
20
import json
 
21
import webob
 
22
 
 
23
from cinder.api.contrib import volume_encryption_metadata
 
24
from cinder import context
 
25
from cinder import db
 
26
from cinder import test
 
27
from cinder.tests.api import fakes
 
28
from cinder.volume import volume_types
 
29
 
 
30
 
 
31
def return_volume_type_encryption_metadata(context, volume_type_id):
 
32
    return stub_volume_type_encryption()
 
33
 
 
34
 
 
35
def stub_volume_type_encryption():
 
36
    values = {
 
37
        'cipher': 'cipher',
 
38
        'key_size': 256,
 
39
        'provider': 'nova.volume.encryptors.base.VolumeEncryptor',
 
40
        'volume_type_id': 'volume_type',
 
41
        'control_location': 'front-end',
 
42
    }
 
43
    return values
 
44
 
 
45
 
 
46
class VolumeEncryptionMetadataTest(test.TestCase):
 
47
    @staticmethod
 
48
    def _create_volume(context,
 
49
                       display_name='test_volume',
 
50
                       display_description='this is a test volume',
 
51
                       status='creating',
 
52
                       availability_zone='fake_az',
 
53
                       host='fake_host',
 
54
                       size=1):
 
55
        """Create a volume object."""
 
56
        volume = {
 
57
            'size': size,
 
58
            'user_id': 'fake',
 
59
            'project_id': 'fake',
 
60
            'status': status,
 
61
            'display_name': display_name,
 
62
            'display_description': display_description,
 
63
            'attach_status': 'detached',
 
64
            'availability_zone': availability_zone,
 
65
            'host': host,
 
66
            'encryption_key_id': 'fake_key',
 
67
        }
 
68
        return db.volume_create(context, volume)['id']
 
69
 
 
70
    def setUp(self):
 
71
        super(VolumeEncryptionMetadataTest, self).setUp()
 
72
        self.controller = (volume_encryption_metadata.
 
73
                           VolumeEncryptionMetadataController())
 
74
        self.stubs.Set(db.sqlalchemy.api, 'volume_type_encryption_get',
 
75
                       return_volume_type_encryption_metadata)
 
76
 
 
77
        self.ctxt = context.RequestContext('fake', 'fake')
 
78
        self.volume_id = self._create_volume(self.ctxt)
 
79
 
 
80
    def tearDown(self):
 
81
        db.volume_destroy(self.ctxt.elevated(), self.volume_id)
 
82
        super(VolumeEncryptionMetadataTest, self).tearDown()
 
83
 
 
84
    def test_index(self):
 
85
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
86
 
 
87
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption'
 
88
                                  % self.volume_id)
 
89
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
90
        self.assertEqual(200, res.status_code)
 
91
        res_dict = json.loads(res.body)
 
92
 
 
93
        expected = {
 
94
            "encryption_key_id": "fake_key",
 
95
            "control_location": "front-end",
 
96
            "cipher": "cipher",
 
97
            "provider": "nova.volume.encryptors.base.VolumeEncryptor",
 
98
            "key_size": 256,
 
99
        }
 
100
        self.assertEqual(expected, res_dict)
 
101
 
 
102
    def test_index_bad_tenant_id(self):
 
103
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
104
 
 
105
        req = webob.Request.blank('/v2/%s/volumes/%s/encryption'
 
106
                                  % ('bad-tenant-id', self.volume_id))
 
107
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
108
        self.assertEqual(400, res.status_code)
 
109
 
 
110
        res_dict = json.loads(res.body)
 
111
        expected = {'badRequest': {'code': 400,
 
112
                                   'message': 'Malformed request url'}}
 
113
        self.assertEqual(expected, res_dict)
 
114
 
 
115
    def test_index_bad_volume_id(self):
 
116
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
117
 
 
118
        bad_volume_id = 'bad_volume_id'
 
119
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption'
 
120
                                  % bad_volume_id)
 
121
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
122
        self.assertEqual(404, res.status_code)
 
123
 
 
124
        res_dict = json.loads(res.body)
 
125
        expected = {'itemNotFound': {'code': 404,
 
126
                                     'message': 'VolumeNotFound: Volume '
 
127
                                                '%s could not be found.'
 
128
                                                % bad_volume_id}}
 
129
        self.assertEqual(expected, res_dict)
 
130
 
 
131
    def test_show_key(self):
 
132
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
133
 
 
134
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
135
                                  'encryption_key_id' % self.volume_id)
 
136
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
137
        self.assertEqual(200, res.status_code)
 
138
 
 
139
        self.assertEqual('fake_key', res.body)
 
140
 
 
141
    def test_show_control(self):
 
142
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
143
 
 
144
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
145
                                  'control_location' % self.volume_id)
 
146
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
147
        self.assertEqual(200, res.status_code)
 
148
 
 
149
        self.assertEqual('front-end', res.body)
 
150
 
 
151
    def test_show_provider(self):
 
152
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
153
 
 
154
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
155
                                  'provider' % self.volume_id)
 
156
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
157
        self.assertEqual(200, res.status_code)
 
158
 
 
159
        self.assertEqual('nova.volume.encryptors.base.VolumeEncryptor',
 
160
                         res.body)
 
161
 
 
162
    def test_show_bad_tenant_id(self):
 
163
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
164
 
 
165
        req = webob.Request.blank('/v2/%s/volumes/%s/encryption/'
 
166
                                  'encryption_key_id' % ('bad-tenant-id',
 
167
                                                         self.volume_id))
 
168
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
169
        self.assertEqual(400, res.status_code)
 
170
 
 
171
        res_dict = json.loads(res.body)
 
172
        expected = {'badRequest': {'code': 400,
 
173
                                   'message': 'Malformed request url'}}
 
174
        self.assertEqual(expected, res_dict)
 
175
 
 
176
    def test_show_bad_volume_id(self):
 
177
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
178
 
 
179
        bad_volume_id = 'bad_volume_id'
 
180
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
181
                                  'encryption_key_id' % bad_volume_id)
 
182
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
183
        self.assertEqual(404, res.status_code)
 
184
 
 
185
        res_dict = json.loads(res.body)
 
186
        expected = {'itemNotFound': {'code': 404,
 
187
                                     'message': 'VolumeNotFound: Volume '
 
188
                                                '%s could not be found.'
 
189
                                                % bad_volume_id}}
 
190
        self.assertEqual(expected, res_dict)
 
191
 
 
192
    def test_retrieve_key_admin(self):
 
193
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
 
194
 
 
195
        ctxt = context.RequestContext('fake', 'fake', is_admin=True)
 
196
 
 
197
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
198
                                  'encryption_key_id' % self.volume_id)
 
199
        res = req.get_response(fakes.wsgi_app(fake_auth_context=ctxt))
 
200
        self.assertEqual(200, res.status_code)
 
201
 
 
202
        self.assertEqual('fake_key', res.body)
 
203
 
 
204
    def test_show_volume_not_encrypted_type(self):
 
205
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: False)
 
206
 
 
207
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
 
208
                                  'encryption_key_id' % self.volume_id)
 
209
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
210
        self.assertEqual(200, res.status_code)
 
211
        self.assertEqual(0, len(res.body))
 
212
 
 
213
    def test_index_volume_not_encrypted_type(self):
 
214
        self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: False)
 
215
 
 
216
        req = webob.Request.blank('/v2/fake/volumes/%s/encryption'
 
217
                                  % self.volume_id)
 
218
        res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
 
219
 
 
220
        self.assertEqual(200, res.status_code)
 
221
        res_dict = json.loads(res.body)
 
222
 
 
223
        expected = {
 
224
            'encryption_key_id': None
 
225
        }
 
226
        self.assertEqual(expected, res_dict)