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