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