1# Copyright (c) 2011 Zadara Storage Inc. 2# Copyright (c) 2011 OpenStack Foundation 3# Copyright 2011 University of Southern California 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""" 16Unit Tests for volume types extra specs code 17""" 18 19from cinder import context 20from cinder import db 21from cinder import exception 22from cinder import objects 23from cinder import test 24from cinder.tests.unit import fake_constants as fake 25 26 27class VolumeGlanceMetadataTestCase(test.TestCase): 28 29 def setUp(self): 30 super(VolumeGlanceMetadataTestCase, self).setUp() 31 self.ctxt = context.get_admin_context() 32 objects.register_all() 33 34 def test_vol_glance_metadata_bad_vol_id(self): 35 ctxt = context.get_admin_context() 36 self.assertRaises(exception.VolumeNotFound, 37 db.volume_glance_metadata_create, 38 ctxt, fake.VOLUME_ID, 'key1', 'value1') 39 self.assertRaises(exception.VolumeNotFound, 40 db.volume_glance_metadata_get, ctxt, fake.VOLUME_ID) 41 db.volume_glance_metadata_delete_by_volume(ctxt, fake.VOLUME2_ID) 42 43 def test_vol_update_glance_metadata(self): 44 ctxt = context.get_admin_context() 45 db.volume_create(ctxt, {'id': fake.VOLUME_ID}) 46 db.volume_create(ctxt, {'id': fake.VOLUME2_ID}) 47 db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'key1', 48 'value1') 49 db.volume_glance_metadata_create(ctxt, fake.VOLUME2_ID, 'key1', 50 'value1') 51 db.volume_glance_metadata_create(ctxt, fake.VOLUME2_ID, 'key2', 52 'value2') 53 db.volume_glance_metadata_create(ctxt, fake.VOLUME2_ID, 'key3', 123) 54 55 expected_metadata_1 = {'volume_id': fake.VOLUME_ID, 56 'key': 'key1', 57 'value': 'value1'} 58 59 metadata = db.volume_glance_metadata_get(ctxt, fake.VOLUME_ID) 60 self.assertEqual(1, len(metadata)) 61 for key, value in expected_metadata_1.items(): 62 self.assertEqual(value, metadata[0][key]) 63 64 expected_metadata_2 = ({'volume_id': fake.VOLUME2_ID, 65 'key': 'key1', 66 'value': 'value1'}, 67 {'volume_id': fake.VOLUME2_ID, 68 'key': 'key2', 69 'value': 'value2'}, 70 {'volume_id': fake.VOLUME2_ID, 71 'key': 'key3', 72 'value': '123'}) 73 74 metadata = db.volume_glance_metadata_get(ctxt, fake.VOLUME2_ID) 75 self.assertEqual(3, len(metadata)) 76 for expected, meta in zip(expected_metadata_2, metadata): 77 for key, value in expected.items(): 78 self.assertEqual(value, meta[key]) 79 80 self.assertRaises(exception.GlanceMetadataExists, 81 db.volume_glance_metadata_create, 82 ctxt, fake.VOLUME_ID, 'key1', 'value1a') 83 84 metadata = db.volume_glance_metadata_get(ctxt, fake.VOLUME_ID) 85 self.assertEqual(1, len(metadata)) 86 for key, value in expected_metadata_1.items(): 87 self.assertEqual(value, metadata[0][key]) 88 89 def test_vols_get_glance_metadata(self): 90 ctxt = context.get_admin_context() 91 db.volume_create(ctxt, {'id': fake.VOLUME_ID}) 92 db.volume_create(ctxt, {'id': fake.VOLUME2_ID}) 93 db.volume_create(ctxt, {'id': '3'}) 94 db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'key1', 95 'value1') 96 db.volume_glance_metadata_create(ctxt, fake.VOLUME2_ID, 'key2', 97 'value2') 98 db.volume_glance_metadata_create(ctxt, fake.VOLUME2_ID, 'key22', 99 'value22') 100 101 metadata = db.volume_glance_metadata_get_all(ctxt) 102 self.assertEqual(3, len(metadata)) 103 self._assert_metadata_equals(fake.VOLUME_ID, 'key1', 'value1', 104 metadata[0]) 105 self._assert_metadata_equals(fake.VOLUME2_ID, 'key2', 'value2', 106 metadata[1]) 107 self._assert_metadata_equals(fake.VOLUME2_ID, 'key22', 'value22', 108 metadata[2]) 109 110 def _assert_metadata_equals(self, volume_id, key, value, observed): 111 self.assertEqual(volume_id, observed.volume_id) 112 self.assertEqual(key, observed.key) 113 self.assertEqual(value, observed.value) 114 115 def test_vol_delete_glance_metadata(self): 116 ctxt = context.get_admin_context() 117 db.volume_create(ctxt, {'id': fake.VOLUME_ID}) 118 db.volume_glance_metadata_delete_by_volume(ctxt, fake.VOLUME_ID) 119 db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'key1', 120 'value1') 121 db.volume_glance_metadata_delete_by_volume(ctxt, fake.VOLUME_ID) 122 self.assertRaises(exception.GlanceMetadataNotFound, 123 db.volume_glance_metadata_get, ctxt, fake.VOLUME_ID) 124 125 def test_vol_glance_metadata_copy_to_snapshot(self): 126 ctxt = context.get_admin_context() 127 db.volume_create(ctxt, {'id': fake.VOLUME_ID}) 128 snap = objects.Snapshot(ctxt, volume_id=fake.VOLUME_ID) 129 snap.create() 130 db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'key1', 131 'value1') 132 db.volume_glance_metadata_copy_to_snapshot(ctxt, snap.id, 133 fake.VOLUME_ID) 134 135 expected_meta = {'snapshot_id': snap.id, 136 'key': 'key1', 137 'value': 'value1'} 138 139 for meta in db.volume_snapshot_glance_metadata_get(ctxt, snap.id): 140 for (key, value) in expected_meta.items(): 141 self.assertEqual(value, meta[key]) 142 snap.destroy() 143 144 def test_vol_glance_metadata_copy_from_volume_to_volume(self): 145 ctxt = context.get_admin_context() 146 db.volume_create(ctxt, {'id': fake.VOLUME_ID}) 147 db.volume_create(ctxt, {'id': fake.VOLUME2_ID, 148 'source_volid': fake.VOLUME_ID}) 149 db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'key1', 150 'value1') 151 db.volume_glance_metadata_copy_from_volume_to_volume(ctxt, 152 fake.VOLUME_ID, 153 fake.VOLUME2_ID) 154 155 expected_meta = {'key': 'key1', 156 'value': 'value1'} 157 158 for meta in db.volume_glance_metadata_get(ctxt, fake.VOLUME2_ID): 159 for (key, value) in expected_meta.items(): 160 self.assertEqual(value, meta[key]) 161 162 def test_volume_glance_metadata_copy_to_volume(self): 163 vol1 = db.volume_create(self.ctxt, {}) 164 vol2 = db.volume_create(self.ctxt, {}) 165 db.volume_glance_metadata_create(self.ctxt, vol1['id'], 'm1', 'v1') 166 snapshot = objects.Snapshot(self.ctxt, volume_id=vol1['id']) 167 snapshot.create() 168 db.volume_glance_metadata_copy_to_snapshot(self.ctxt, snapshot.id, 169 vol1['id']) 170 db.volume_glance_metadata_copy_to_volume(self.ctxt, vol2['id'], 171 snapshot.id) 172 metadata = db.volume_glance_metadata_get(self.ctxt, vol2['id']) 173 metadata = {m['key']: m['value'] for m in metadata} 174 self.assertEqual({'m1': 'v1'}, metadata) 175 176 def test_volume_snapshot_glance_metadata_get_nonexistent(self): 177 vol = db.volume_create(self.ctxt, {}) 178 snapshot = objects.Snapshot(self.ctxt, volume_id=vol['id']) 179 snapshot.create() 180 self.assertRaises(exception.GlanceMetadataNotFound, 181 db.volume_snapshot_glance_metadata_get, 182 self.ctxt, snapshot.id) 183 snapshot.destroy() 184