Files
python-cinderclient/cinderclient/tests/unit/v3/test_volumes.py
Vivek Agrawal 056cf5c059 Metadata based snapshop filtering
The snpapshot-list API for cinder gives a list of snapshots based
on certain criteria to the user. From microversion 3.22 onwards
the snapshot-list API has been enhanced to support snapshot list
filtering based on metadata of snapshots. The metadata is stored
as key-value pair for every snapshot.
With this commit cinder will be queried based on metadata key and
value specified in the API snaphot-list. All the snapshots which
match the key, value provided by the user along with any other
filter criteria will be returned.
Added the test cases for the CLI and web requests.

DocImpact: "Filters results by a metadata key and value pair.
Default=None." on cinder snapshot-list
APIImpact

Closes-bug: #1569554

Change-Id: Idec0d0d02e7956843f202508e32c023c3cafbb0f
2017-01-18 06:28:21 -08:00

96 lines
4.1 KiB
Python

# Copyright 2016 FUJITSU LIMITED
# Copyright (c) 2016 EMC Corporation
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from cinderclient import api_versions
from cinderclient.tests.unit import utils
from cinderclient.tests.unit.v3 import fakes
from cinderclient.v3 import volumes
from six.moves.urllib import parse
cs = fakes.FakeClient()
class VolumesTest(utils.TestCase):
def test_volume_manager_upload_to_image(self):
expected = {'os-volume_upload_image':
{'force': False,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'name',
'visibility': 'public',
'protected': True}}
api_version = api_versions.APIVersion('3.1')
cs = fakes.FakeClient(api_version)
manager = volumes.VolumeManager(cs)
fake_volume = volumes.Volume(manager,
{'id': 1234, 'name': 'sample-volume'},
loaded=True)
fake_volume.upload_to_image(False, 'name', 'bare', 'raw',
visibility='public', protected=True)
cs.assert_called_anytime('POST', '/volumes/1234/action', body=expected)
def test_create_volume(self):
vol = cs.volumes.create(1, group_id='1234', volume_type='5678')
expected = {'volume': {'status': 'creating',
'description': None,
'availability_zone': None,
'source_volid': None,
'snapshot_id': None,
'size': 1,
'user_id': None,
'name': None,
'imageRef': None,
'attach_status': 'detached',
'volume_type': '5678',
'project_id': None,
'metadata': {},
'source_replica': None,
'consistencygroup_id': None,
'multiattach': False,
'group_id': '1234'}}
cs.assert_called('POST', '/volumes', body=expected)
self._assert_request_id(vol)
def test_volume_list_manageable(self):
cs = fakes.FakeClient(api_versions.APIVersion('3.8'))
cs.volumes.list_manageable('host1', detailed=False)
cs.assert_called('GET', '/manageable_volumes?host=host1')
def test_volume_list_manageable_detailed(self):
cs = fakes.FakeClient(api_versions.APIVersion('3.8'))
cs.volumes.list_manageable('host1', detailed=True)
cs.assert_called('GET', '/manageable_volumes/detail?host=host1')
def test_snapshot_list_manageable(self):
cs = fakes.FakeClient(api_versions.APIVersion('3.8'))
cs.volume_snapshots.list_manageable('host1', detailed=False)
cs.assert_called('GET', '/manageable_snapshots?host=host1')
def test_snapshot_list_manageable_detailed(self):
cs = fakes.FakeClient(api_versions.APIVersion('3.8'))
cs.volume_snapshots.list_manageable('host1', detailed=True)
cs.assert_called('GET', '/manageable_snapshots/detail?host=host1')
def test_snapshot_list_with_metadata(self):
cs = fakes.FakeClient(api_versions.APIVersion('3.22'))
cs.volume_snapshots.list(search_opts={'metadata': {'key1': 'val1'}})
expected = ("/snapshots/detail?metadata=%s"
% parse.quote_plus("{'key1': 'val1'}"))
cs.assert_called('GET', expected)