womax/python-glanceclient
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
323d32cc6d28434ad57f3ddd5253b9a8950b4c8a
python-glanceclient/tests/v2/test_shell_v2.py
T
David Koo 323d32cc6d Add support for location parameters in v2 commands 
Currently glanceclient's v2 commands don't support modification
operations on an image's location attribute - the argparse specification
for the location attribute of the image-update command causes the image
id argument to be included in list of locations and so the command
parsing fails (because it causes the image id to appear to be missing).

Furthermore even if the 'locations' argument were to be accepted by
argparse (e.g. by changing the argument specs and using --id to specify
the image id) the command would still fail because the arguments are
passed directly to the schema which expects the value of the 'locations'
argument to be a valid dictionary (there is nobody to convert the
argument string to a python dictionary that the schema expects).

This commit adds the following location related commands to
glanceclient:
    --location-add: Add a new location to the list of image locations.
    --location-delete: Remove an existing location from the list of
        image locations.
    --location-update: Update the metadata of existing location.

The glanceclient.v2.images.Controller class has been agumented with
three new methods to support the commands listed above:
    - add_location
    - delete_locations
    - update_location

The server has not been modified, i.e. all location related API requests
are passed to the server via HTTP PATCH requests and handled by the
server's image update function.

The v2 'image' and 'shell' related tests have also been supplemented.

Note that in order to use these options the server must be first
configured to expose location related info to the clients (i.e.
'show_multiple_locations' must be set to 'True").

I also added a mailmap entry for myself.

DocImpact
Closes-bug: #1271452
Co-Author: David Koo (koofoss) <david.koo@huawei.com>

Change-Id: Id1f320af05d9344645836359758e4aa227aafc69
2014-06-16 21:56:41 +00:00

432 lines
18 KiB
Python
Raw Blame History

# Copyright 2013 OpenStack Foundation
# Copyright (C) 2013 Yahoo! Inc.
# 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.
import json
import mock
import six
import testtools
from glanceclient.common import http
from glanceclient.common import progressbar
from glanceclient.common import utils
from glanceclient.v2 import shell as test_shell
from tests import utils as test_utils
class ShellV2Test(testtools.TestCase):
def setUp(self):
super(ShellV2Test, self).setUp()
self._mock_utils()
self.gc = self._mock_glance_client()
def _make_args(self, args):
#NOTE(venkatesh): this conversion from a dict to an object
# is required because the test_shell.do_xxx(gc, args) methods
# expects the args to be attributes of an object. If passed as
# dict directly, it throws an AttributeError.
class Args():
def __init__(self, entries):
self.__dict__.update(entries)
return Args(args)
def _mock_glance_client(self):
my_mocked_gc = mock.Mock()
my_mocked_gc.schemas.return_value = 'test'
my_mocked_gc.get.return_value = {}
return my_mocked_gc
def _mock_utils(self):
utils.print_list = mock.Mock()
utils.print_dict = mock.Mock()
utils.save_image = mock.Mock()
def assert_exits_with_msg(self, func, func_args, err_msg):
with mock.patch.object(utils, 'exit') as mocked_utils_exit:
mocked_utils_exit.return_value = '%s' % err_msg
func(self.gc, func_args)
mocked_utils_exit.assert_called_once_with(err_msg)
def test_do_image_list(self):
input = {
'page_size': 18,
'visibility': True,
'member_status': 'Fake',
'owner': 'test',
'checksum': 'fake_checksum',
'tag': 'fake tag'
}
args = self._make_args(input)
with mock.patch.object(self.gc.images, 'list') as mocked_list:
mocked_list.return_value = {}
test_shell.do_image_list(self.gc, args)
exp_img_filters = {
'owner': 'test',
'member_status': 'Fake',
'visibility': True,
'checksum': 'fake_checksum',
'tag': 'fake tag'
}
mocked_list.assert_called_once_with(page_size=18,
filters=exp_img_filters)
utils.print_list.assert_called_once_with({}, ['ID', 'Name'])
def test_do_image_show(self):
args = self._make_args({'id': 'pass', 'page_size': 18,
'max_column_width': 120})
with mock.patch.object(self.gc.images, 'get') as mocked_list:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
mocked_list.return_value = expect_image
test_shell.do_image_show(self.gc, args)
mocked_list.assert_called_once_with('pass')
utils.print_dict.assert_called_once_with({'id': 'pass'},
max_column_width=120)
def test_do_image_create_no_user_props(self):
args = self._make_args({'name': 'IMG-01', 'disk_format': 'vhd',
'container_format': 'bare'})
with mock.patch.object(self.gc.images, 'create') as mocked_create:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
expect_image['name'] = 'IMG-01'
expect_image['disk_format'] = 'vhd'
expect_image['container_format'] = 'bare'
mocked_create.return_value = expect_image
test_shell.do_image_create(self.gc, args)
mocked_create.assert_called_once_with(name='IMG-01',
disk_format='vhd',
container_format='bare')
utils.print_dict.assert_called_once_with({
'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd',
'container_format': 'bare'})
def test_do_image_create_with_user_props(self):
args = self._make_args({'name': 'IMG-01',
'property': ['myprop=myval']})
with mock.patch.object(self.gc.images, 'create') as mocked_create:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
expect_image['name'] = 'IMG-01'
expect_image['myprop'] = 'myval'
mocked_create.return_value = expect_image
test_shell.do_image_create(self.gc, args)
mocked_create.assert_called_once_with(name='IMG-01',
myprop='myval')
utils.print_dict.assert_called_once_with({
'id': 'pass', 'name': 'IMG-01', 'myprop': 'myval'})
def test_do_image_update_no_user_props(self):
args = self._make_args({'id': 'pass', 'name': 'IMG-01',
'disk_format': 'vhd',
'container_format': 'bare'})
with mock.patch.object(self.gc.images, 'update') as mocked_update:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
expect_image['name'] = 'IMG-01'
expect_image['disk_format'] = 'vhd'
expect_image['container_format'] = 'bare'
mocked_update.return_value = expect_image
test_shell.do_image_update(self.gc, args)
mocked_update.assert_called_once_with('pass',
None,
name='IMG-01',
disk_format='vhd',
container_format='bare')
utils.print_dict.assert_called_once_with({
'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd',
'container_format': 'bare'})
def test_do_image_update_with_user_props(self):
args = self._make_args({'id': 'pass', 'name': 'IMG-01',
'property': ['myprop=myval']})
with mock.patch.object(self.gc.images, 'update') as mocked_update:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
expect_image['name'] = 'IMG-01'
expect_image['myprop'] = 'myval'
mocked_update.return_value = expect_image
test_shell.do_image_update(self.gc, args)
mocked_update.assert_called_once_with('pass',
None,
name='IMG-01',
myprop='myval')
utils.print_dict.assert_called_once_with({
'id': 'pass', 'name': 'IMG-01', 'myprop': 'myval'})
def test_do_image_update_with_remove_props(self):
args = self._make_args({'id': 'pass', 'name': 'IMG-01',
'disk_format': 'vhd',
'remove-property': ['container_format']})
with mock.patch.object(self.gc.images, 'update') as mocked_update:
ignore_fields = ['self', 'access', 'file', 'schema']
expect_image = dict([(field, field) for field in ignore_fields])
expect_image['id'] = 'pass'
expect_image['name'] = 'IMG-01'
expect_image['disk_format'] = 'vhd'
mocked_update.return_value = expect_image
test_shell.do_image_update(self.gc, args)
mocked_update.assert_called_once_with('pass',
['container_format'],
name='IMG-01',
disk_format='vhd')
utils.print_dict.assert_called_once_with({
'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd'})
def test_do_location_add_update_with_invalid_json_metadata(self):
args = self._make_args({'id': 'pass',
'url': 'http://foo/bar',
'metadata': '{1, 2, 3}'})
self.assert_exits_with_msg(test_shell.do_location_add,
args,
'Metadata is not a valid JSON object.')
self.assert_exits_with_msg(test_shell.do_location_update,
args,
'Metadata is not a valid JSON object.')
def test_do_location_add(self):
gc = self.gc
loc = {'url': 'http://foo.com/', 'metadata': {'foo': 'bar'}}
args = self._make_args({'id': 'pass',
'url': loc['url'],
'metadata': json.dumps(loc['metadata'])})
with mock.patch.object(gc.images, 'add_location') as mocked_addloc:
expect_image = {'id': 'pass', 'locations': [loc]}
mocked_addloc.return_value = expect_image
test_shell.do_location_add(self.gc, args)
mocked_addloc.assert_called_once_with('pass',
loc['url'],
loc['metadata'])
utils.print_dict.assert_called_once_with(expect_image)
def test_do_location_delete(self):
gc = self.gc
loc_set = set(['http://foo/bar', 'http://spam/ham'])
args = self._make_args({'id': 'pass', 'url': loc_set})
with mock.patch.object(gc.images, 'delete_locations') as mocked_rmloc:
expect_image = {'id': 'pass', 'locations': []}
test_shell.do_location_delete(self.gc, args)
mocked_rmloc.assert_called_once_with('pass', loc_set)
def test_do_location_update(self):
gc = self.gc
loc = {'url': 'http://foo.com/', 'metadata': {'foo': 'bar'}}
args = self._make_args({'id': 'pass',
'url': loc['url'],
'metadata': json.dumps(loc['metadata'])})
with mock.patch.object(gc.images, 'update_location') as mocked_modloc:
expect_image = {'id': 'pass', 'locations': [loc]}
mocked_modloc.return_value = expect_image
test_shell.do_location_update(self.gc, args)
mocked_modloc.assert_called_once_with('pass',
loc['url'],
loc['metadata'])
utils.print_dict.assert_called_once_with(expect_image)
def test_do_explain(self):
input = {
'page_size': 18,
'id': 'pass',
'schemas': 'test',
'model': 'test',
}
args = self._make_args(input)
with mock.patch.object(utils, 'print_list'):
test_shell.do_explain(self.gc, args)
self.gc.schemas.get.assert_called_once_with('test')
def test_image_upload(self):
args = self._make_args({'id': 'IMG-01', 'file': 'test', 'size': 1024})
with mock.patch.object(self.gc.images, 'upload') as mocked_upload:
utils.get_data_file = mock.Mock(return_value='testfile')
mocked_upload.return_value = None
test_shell.do_image_upload(self.gc, args)
mocked_upload.assert_called_once_with('IMG-01', 'testfile', 1024)
def test_image_download(self):
args = self._make_args(
{'id': 'pass', 'file': 'test', 'progress': False})
with mock.patch.object(self.gc.images, 'data') as mocked_data:
resp = test_utils.FakeResponse({}, six.StringIO('CCC'))
ret = mocked_data.return_value = http.ResponseBodyIterator(resp)
test_shell.do_image_download(self.gc, args)
mocked_data.assert_called_once_with('pass')
utils.save_image.assert_called_once_with(ret, 'test')
def test_image_download_with_progressbar(self):
args = self._make_args(
{'id': 'pass', 'file': 'test', 'progress': True})
with mock.patch.object(self.gc.images, 'data') as mocked_data:
resp = test_utils.FakeResponse({}, six.StringIO('CCC'))
mocked_data.return_value = http.ResponseBodyIterator(resp)
test_shell.do_image_download(self.gc, args)
mocked_data.assert_called_once_with('pass')
utils.save_image.assert_called_once_with(mock.ANY, 'test')
self.assertIsInstance(
utils.save_image.call_args[0][0],
progressbar.VerboseIteratorWrapper
)
def test_do_image_delete(self):
args = self._make_args({'id': 'pass', 'file': 'test'})
with mock.patch.object(self.gc.images, 'delete') as mocked_delete:
mocked_delete.return_value = 0
test_shell.do_image_delete(self.gc, args)
mocked_delete.assert_called_once_with('pass')
def test_do_member_list(self):
args = self._make_args({'image_id': 'IMG-01'})
with mock.patch.object(self.gc.image_members, 'list') as mocked_list:
mocked_list.return_value = {}
test_shell.do_member_list(self.gc, args)
mocked_list.assert_called_once_with('IMG-01')
columns = ['Image ID', 'Member ID', 'Status']
utils.print_list.assert_called_once_with({}, columns)
def test_do_member_create(self):
args = self._make_args({'image_id': 'IMG-01', 'member_id': 'MEM-01'})
with mock.patch.object(self.gc.image_members, 'create') as mock_create:
mock_create.return_value = {}
test_shell.do_member_create(self.gc, args)
mock_create.assert_called_once_with('IMG-01', 'MEM-01')
columns = ['Image ID', 'Member ID', 'Status']
utils.print_list.assert_called_once_with([{}], columns)
def test_do_member_create_with_few_arguments(self):
args = self._make_args({'image_id': None, 'member_id': 'MEM-01'})
msg = 'Unable to create member. Specify image_id and member_id'
self.assert_exits_with_msg(func=test_shell.do_member_create,
func_args=args,
err_msg=msg)
def test_do_member_update(self):
input = {
'image_id': 'IMG-01',
'member_id': 'MEM-01',
'member_status': 'status',
}
args = self._make_args(input)
with mock.patch.object(self.gc.image_members, 'update') as mock_update:
mock_update.return_value = {}
test_shell.do_member_update(self.gc, args)
mock_update.assert_called_once_with('IMG-01', 'MEM-01', 'status')
columns = ['Image ID', 'Member ID', 'Status']
utils.print_list.assert_called_once_with([{}], columns)
def test_do_member_update_with_few_arguments(self):
input = {
'image_id': 'IMG-01',
'member_id': 'MEM-01',
'member_status': None,
}
args = self._make_args(input)
msg = 'Unable to update member. Specify image_id, member_id' \
' and member_status'
self.assert_exits_with_msg(func=test_shell.do_member_update,
func_args=args,
err_msg=msg)
def test_do_member_delete(self):
args = self._make_args({'image_id': 'IMG-01', 'member_id': 'MEM-01'})
with mock.patch.object(self.gc.image_members, 'delete') as mock_delete:
test_shell.do_member_delete(self.gc, args)
mock_delete.assert_called_once_with('IMG-01', 'MEM-01')
def test_do_member_delete_with_few_arguments(self):
args = self._make_args({'image_id': None, 'member_id': 'MEM-01'})
msg = 'Unable to delete member. Specify image_id and member_id'
self.assert_exits_with_msg(func=test_shell.do_member_delete,
func_args=args,
err_msg=msg)
def test_image_tag_update(self):
args = self._make_args({'image_id': 'IMG-01', 'tag_value': 'tag01'})
with mock.patch.object(self.gc.image_tags, 'update') as mocked_update:
self.gc.images.get = mock.Mock(return_value={})
mocked_update.return_value = None
test_shell.do_image_tag_update(self.gc, args)
mocked_update.assert_called_once_with('IMG-01', 'tag01')
def test_image_tag_update_with_few_arguments(self):
args = self._make_args({'image_id': None, 'tag_value': 'tag01'})
msg = 'Unable to update tag. Specify image_id and tag_value'
self.assert_exits_with_msg(func=test_shell.do_image_tag_update,
func_args=args,
err_msg=msg)
def test_image_tag_delete(self):
args = self._make_args({'image_id': 'IMG-01', 'tag_value': 'tag01'})
with mock.patch.object(self.gc.image_tags, 'delete') as mocked_delete:
mocked_delete.return_value = None
test_shell.do_image_tag_delete(self.gc, args)
mocked_delete.assert_called_once_with('IMG-01', 'tag01')
def test_image_tag_delete_with_few_arguments(self):
args = self._make_args({'image_id': 'IMG-01', 'tag_value': None})
msg = 'Unable to delete tag. Specify image_id and tag_value'
self.assert_exits_with_msg(func=test_shell.do_image_tag_delete,
func_args=args,
err_msg=msg)
Reference in New Issue View Git Blame Copy Permalink