1# Copyright 2010 OpenStack Foundation
2# All Rights Reserved.
3#
4#    Licensed under the Apache License, Version 2.0 (the "License"); you may
5#    not use this file except in compliance with the License. You may obtain
6#    a copy of the License at
7#
8#         http://www.apache.org/licenses/LICENSE-2.0
9#
10#    Unless required by applicable law or agreed to in writing, software
11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13#    License for the specific language governing permissions and limitations
14#    under the License.
15
16import datetime
17import iso8601
18
19from cinder import exception as exc
20from cinder import objects
21from cinder.objects import fields
22from cinder.tests.unit import fake_constants as fake
23from cinder.tests.unit import fake_volume
24from cinder import utils
25
26
27DEFAULT_VOL_NAME = "displayname"
28DEFAULT_VOL_DESCRIPTION = "displaydesc"
29DEFAULT_VOL_SIZE = 1
30DEFAULT_VOL_TYPE = "vol_type_name"
31DEFAULT_VOL_STATUS = "fakestatus"
32DEFAULT_VOL_ID = fake.VOLUME_ID
33
34# TODO(vbala): api.v1 tests use hard-coded "fakeaz" for verifying
35# post-conditions. Update value to "zone1:host1" once we remove
36# api.v1 tests and use it in api.v2 tests.
37DEFAULT_AZ = "fakeaz"
38
39
40def create_fake_volume(id, **kwargs):
41    volume = {
42        'id': id,
43        'user_id': fake.USER_ID,
44        'project_id': fake.PROJECT_ID,
45        'host': 'fakehost',
46        'size': DEFAULT_VOL_SIZE,
47        'availability_zone': DEFAULT_AZ,
48        'status': DEFAULT_VOL_STATUS,
49        'migration_status': None,
50        'attach_status': fields.VolumeAttachStatus.ATTACHED,
51        'name': 'vol name',
52        'display_name': DEFAULT_VOL_NAME,
53        'display_description': DEFAULT_VOL_DESCRIPTION,
54        'updated_at': datetime.datetime(1900, 1, 1, 1, 1, 1,
55                                        tzinfo=iso8601.UTC),
56        'created_at': datetime.datetime(1900, 1, 1, 1, 1, 1,
57                                        tzinfo=iso8601.UTC),
58        'snapshot_id': None,
59        'source_volid': None,
60        'volume_type_id': '3e196c20-3c06-11e2-81c1-0800200c9a66',
61        'encryption_key_id': None,
62        'volume_admin_metadata': [{'key': 'attached_mode', 'value': 'rw'},
63                                  {'key': 'readonly', 'value': 'False'}],
64        'bootable': False,
65        'launched_at': datetime.datetime(1900, 1, 1, 1, 1, 1,
66                                         tzinfo=iso8601.UTC),
67        'volume_type': fake_volume.fake_db_volume_type(name=DEFAULT_VOL_TYPE),
68        'replication_status': 'disabled',
69        'replication_extended_status': None,
70        'replication_driver_data': None,
71        'volume_attachment': [],
72        'multiattach': False,
73    }
74
75    volume.update(kwargs)
76    if kwargs.get('volume_glance_metadata', None):
77        volume['bootable'] = True
78    if kwargs.get('attach_status') == fields.VolumeAttachStatus.DETACHED:
79        del volume['volume_admin_metadata'][0]
80    return volume
81
82
83def fake_volume_create(self, context, size, name, description, snapshot=None,
84                       **param):
85    vol = create_fake_volume(DEFAULT_VOL_ID)
86    vol['size'] = size
87    vol['display_name'] = name
88    vol['display_description'] = description
89    source_volume = param.get('source_volume') or {}
90    vol['source_volid'] = source_volume.get('id')
91    vol['bootable'] = False
92    vol['volume_attachment'] = []
93    vol['multiattach'] = utils.get_bool_param('multiattach', param)
94    try:
95        vol['snapshot_id'] = snapshot['id']
96    except (KeyError, TypeError):
97        vol['snapshot_id'] = None
98    vol['availability_zone'] = param.get('availability_zone', 'fakeaz')
99    return vol
100
101
102def fake_volume_api_create(self, context, *args, **kwargs):
103    vol = fake_volume_create(self, context, *args, **kwargs)
104    return fake_volume.fake_volume_obj(context, **vol)
105
106
107def fake_image_service_detail(self, context, **kwargs):
108    filters = kwargs.get('filters', {'name': ''})
109    if filters['name'] == "Fedora-x86_64-20-20140618-sda":
110        return [{'id': "c905cedb-7281-47e4-8a62-f26bc5fc4c77"}]
111    elif filters['name'] == "multi":
112        return [{'id': "c905cedb-7281-47e4-8a62-f26bc5fc4c77"},
113                {'id': "c905cedb-abcd-47e4-8a62-f26bc5fc4c77"}]
114    return []
115
116
117def fake_volume_create_from_image(self, context, size, name, description,
118                                  snapshot, volume_type, metadata,
119                                  availability_zone):
120    vol = create_fake_volume(fake.VOLUME_ID)
121    vol['status'] = 'creating'
122    vol['size'] = size
123    vol['display_name'] = name
124    vol['display_description'] = description
125    vol['availability_zone'] = 'cinder'
126    vol['bootable'] = False
127    return vol
128
129
130def fake_volume_update(self, context, *args, **param):
131    pass
132
133
134def fake_volume_delete(self, context, *args, **param):
135    pass
136
137
138def fake_volume_get(self, context, volume_id, viewable_admin_meta=False):
139    if viewable_admin_meta:
140        return create_fake_volume(volume_id)
141    else:
142        volume = create_fake_volume(volume_id)
143        del volume['volume_admin_metadata']
144        return volume
145
146
147def fake_volume_get_notfound(self, context,
148                             volume_id, viewable_admin_meta=False):
149    raise exc.VolumeNotFound(volume_id)
150
151
152def fake_volume_get_db(context, volume_id):
153    if context.is_admin:
154        return create_fake_volume(volume_id)
155    else:
156        volume = create_fake_volume(volume_id)
157        del volume['volume_admin_metadata']
158        return volume
159
160
161def fake_volume_api_get(self, context, volume_id, viewable_admin_meta=False):
162    vol = create_fake_volume(volume_id)
163    return fake_volume.fake_volume_obj(context, **vol)
164
165
166def fake_volume_get_all(context, search_opts=None, marker=None, limit=None,
167                        sort_keys=None, sort_dirs=None, filters=None,
168                        viewable_admin_meta=False, offset=None):
169    return [create_fake_volume(fake.VOLUME_ID, project_id=fake.PROJECT_ID),
170            create_fake_volume(fake.VOLUME2_ID, project_id=fake.PROJECT2_ID),
171            create_fake_volume(fake.VOLUME3_ID, project_id=fake.PROJECT3_ID)]
172
173
174def fake_volume_get_all_by_project(self, context, marker, limit,
175                                   sort_keys=None, sort_dirs=None,
176                                   filters=None,
177                                   viewable_admin_meta=False, offset=None):
178    return [fake_volume_get(self, context, fake.VOLUME_ID,
179                            viewable_admin_meta=True)]
180
181
182def fake_volume_api_get_all_by_project(self, context, marker, limit,
183                                       sort_keys=None, sort_dirs=None,
184                                       filters=None,
185                                       viewable_admin_meta=False,
186                                       offset=None):
187    vol = fake_volume_get(self, context, fake.VOLUME_ID,
188                          viewable_admin_meta=viewable_admin_meta)
189    vol_obj = fake_volume.fake_volume_obj(context, **vol)
190    return objects.VolumeList(objects=[vol_obj])
191
192
193def fake_snapshot(id, **kwargs):
194    snapshot = {'id': id,
195                'volume_id': fake.VOLUME_ID,
196                'status': fields.SnapshotStatus.AVAILABLE,
197                'volume_size': 100,
198                'created_at': None,
199                'display_name': 'Default name',
200                'display_description': 'Default description',
201                'project_id': fake.PROJECT_ID,
202                'snapshot_metadata': []}
203
204    snapshot.update(kwargs)
205    return snapshot
206
207
208def fake_backup(id, **kwargs):
209    backup = {'id': fake.BACKUP_ID,
210              'volume_id': fake.VOLUME_ID,
211              'status': fields.BackupStatus.CREATING,
212              'size': 1,
213              'display_name': 'fake_name',
214              'display_description': 'fake_description',
215              'user_id': fake.USER_ID,
216              'project_id': fake.PROJECT_ID,
217              'temp_volume_id': None,
218              'temp_snapshot_id': None,
219              'snapshot_id': None,
220              'data_timestamp': None,
221              'restore_volume_id': None,
222              'backup_metadata': {}}
223
224    backup.update(kwargs)
225    return backup
226
227
228def fake_snapshot_get_all(context, filters=None, marker=None, limit=None,
229                          sort_keys=None, sort_dirs=None, offset=None):
230    return [fake_snapshot(fake.VOLUME_ID, project_id=fake.PROJECT_ID),
231            fake_snapshot(fake.VOLUME2_ID, project_id=fake.PROJECT2_ID),
232            fake_snapshot(fake.VOLUME3_ID, project_id=fake.PROJECT3_ID)]
233
234
235def fake_snapshot_get_all_by_project(context, project_id, filters=None,
236                                     marker=None, limit=None, sort_keys=None,
237                                     sort_dirs=None, offset=None):
238    return [fake_snapshot(fake.SNAPSHOT_ID)]
239
240
241def fake_snapshot_update(self, context, *args, **param):
242    pass
243
244
245def fake_service_get_all(*args, **kwargs):
246    return [{'availability_zone': "zone1:host1", "disabled": 0,
247             'uuid': 'a3a593da-7f8d-4bb7-8b4c-f2bc1e0b4824'}]
248
249
250def fake_service_get_all_by_topic(context, topic, disabled=None):
251    return [{'availability_zone': "zone1:host1", "disabled": 0,
252             'uuid': '4200b32b-0bf9-436c-86b2-0675f6ac218e'}]
253
254
255def fake_snapshot_get(self, context, snapshot_id):
256    if snapshot_id == fake.WILL_NOT_BE_FOUND_ID:
257        raise exc.SnapshotNotFound(snapshot_id=snapshot_id)
258
259    return fake_snapshot(snapshot_id)
260
261
262def fake_backup_get(self, context, backup_id):
263    if backup_id == fake.WILL_NOT_BE_FOUND_ID:
264        raise exc.BackupNotFound(backup_id=backup_id)
265
266    return fake_backup(backup_id)
267
268
269def fake_consistencygroup_get_notfound(self, context, cg_id):
270    raise exc.GroupNotFound(group_id=cg_id)
271
272
273def fake_volume_type_get(context, id, *args, **kwargs):
274    return {'id': id,
275            'name': 'vol_type_name',
276            'description': 'A fake volume type',
277            'is_public': True,
278            'projects': [],
279            'extra_specs': {},
280            'created_at': None,
281            'deleted_at': None,
282            'updated_at': None,
283            'qos_specs_id': fake.QOS_SPEC_ID,
284            'deleted': False}
285
286
287def fake_volume_admin_metadata_get(context, volume_id, **kwargs):
288    admin_meta = {'attached_mode': 'rw', 'readonly': 'False'}
289    if kwargs.get('attach_status') == fields.VolumeAttachStatus.DETACHED:
290        del admin_meta['attached_mode']
291
292    return admin_meta
293