# 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 argparse from copy import deepcopy import io import json import os from unittest import mock import sys import tempfile import testtools from glanceclient.common import utils from glanceclient import exc from glanceclient import shell # NOTE(geguileo): This is very nasty, but I can't find a better way to set # command line arguments in glanceclient.v2.shell.do_image_create that are # set by decorator utils.schema_args while preserving the spirits of the test # Backup original decorator original_schema_args = utils.schema_args # Set our own decorator that calls the original but with simulated schema def schema_args(schema_getter, omit=None): global original_schema_args # We only add the 2 arguments that are required by image-create my_schema_getter = lambda: { 'properties': { 'container_format': { 'enum': [None, 'ami', 'ari', 'aki', 'bare', 'ovf', 'ova', 'docker'], 'type': 'string', 'description': 'Format of the container'}, 'disk_format': { 'enum': [None, 'ami', 'ari', 'aki', 'vhd', 'vhdx', 'vmdk', 'raw', 'qcow2', 'vdi', 'iso', 'ploop'], 'type': 'string', 'description': 'Format of the disk'}, 'location': {'type': 'string'}, 'locations': {'type': 'string'}, 'copy_from': {'type': 'string'}}} return original_schema_args(my_schema_getter, omit) utils.schema_args = schema_args from glanceclient.v2 import shell as test_shell # noqa # Return original decorator. utils.schema_args = original_schema_args class ShellV2Test(testtools.TestCase): def setUp(self): super(ShellV2Test, self).setUp() self._mock_utils() self.gc = self._mock_glance_client() self.shell = shell.OpenStackImagesShell() os.environ = { 'OS_USERNAME': 'username', 'OS_PASSWORD': 'password', 'OS_TENANT_ID': 'tenant_id', 'OS_TOKEN_ID': 'test', 'OS_AUTH_URL': 'http://127.0.0.1:5000/v2.0/', 'OS_AUTH_TOKEN': 'pass', 'OS_IMAGE_API_VERSION': '1', 'OS_REGION_NAME': 'test', 'OS_IMAGE_URL': 'http://is.invalid'} self.shell = shell.OpenStackImagesShell() self.patched = mock.patch('glanceclient.common.utils.get_data_file', autospec=True, return_value=None) self.mock_get_data_file = self.patched.start() def tearDown(self): super(ShellV2Test, self).tearDown() self.patched.stop() 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(object): def __init__(self, entries): self.store = None 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() utils.print_dict_list = mock.Mock() utils.print_cached_images = mock.Mock() def assert_exits_with_msg(self, func, func_args, err_msg=None): with mock.patch.object(utils, 'exit') as mocked_utils_exit: mocked_utils_exit.return_value = '%s' % err_msg func(self.gc, func_args) if err_msg: mocked_utils_exit.assert_called_once_with(err_msg) else: mocked_utils_exit.assert_called_once_with() def _run_command(self, cmd): self.shell.main(cmd.split()) stores_info_response = { "stores": [ { "default": "true", "id": "ceph1", "description": "RBD backend for glance." }, { "id": "file2", "description": "Filesystem backend for glance." }, { "id": "file1", "description": "Filesystem backend for gkance." }, { "id": "ceph2", "description": "RBD backend for glance." } ] } stores_info_detail_response = { "stores": [ { "default": "true", "id": "ceph1", "type": "rbd", "description": "RBD backend for glance.", "properties": { "pool": "pool1", "chunk_size": "4" } }, { "id": "file2", "type": "file", "description": "Filesystem backend for glance.", "properties": {} }, { "id": "file1", "type": "file", "description": "Filesystem backend for gkance.", "properties": {} }, { "id": "ceph2", "type": "rbd", "description": "RBD backend for glance.", "properties": { "pool": "pool2", "chunk_size": "4" } } ] } def test_do_stores_info(self): args = self._make_args({'detail': False}) with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_list: mocked_list.return_value = self.stores_info_response test_shell.do_stores_info(self.gc, args) mocked_list.assert_called_once_with() utils.print_dict.assert_called_once_with(self.stores_info_response) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_stores_info( self, mock_stdin, mock_utils_exit): expected_msg = ('Multi Backend support is not enabled') args = self._make_args({'detail': False}) mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_info: mocked_info.side_effect = exc.HTTPNotFound try: test_shell.do_stores_info(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) def test_do_stores_info_with_detail(self): args = self._make_args({'detail': True}) with mock.patch.object(self.gc.images, 'get_stores_info_detail') as mocked_list: mocked_list.return_value = self.stores_info_detail_response test_shell.do_stores_info(self.gc, args) mocked_list.assert_called_once_with() utils.print_dict.assert_called_once_with( self.stores_info_detail_response) @mock.patch('sys.stderr') def test_image_create_missing_disk_format(self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 image-create ' + '--file fake_src --container-format bare') self.assertEqual('error: Must provide --disk-format when using ' '--file.', e.message) @mock.patch('sys.stderr') def test_image_create_missing_container_format(self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 image-create ' + '--file fake_src --disk-format qcow2') self.assertEqual('error: Must provide --container-format when ' 'using --file.', e.message) @mock.patch('sys.stderr') def test_image_create_missing_container_format_stdin_data(self, __): # Fake that get_data_file method returns data self.mock_get_data_file.return_value = io.StringIO() e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 image-create' ' --disk-format qcow2') self.assertEqual('error: Must provide --container-format when ' 'using stdin.', e.message) @mock.patch('sys.stderr') def test_image_create_missing_disk_format_stdin_data(self, __): # Fake that get_data_file method returns data self.mock_get_data_file.return_value = io.StringIO() e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 image-create' ' --container-format bare') self.assertEqual('error: Must provide --disk-format when using stdin.', e.message) @mock.patch('sys.stderr') def test_create_via_import_glance_direct_missing_disk_format(self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 ' 'image-create-via-import ' '--file fake_src --container-format bare') self.assertEqual('error: Must provide --disk-format when using ' '--file.', e.message) @mock.patch('sys.stderr') def test_create_via_import_glance_direct_missing_container_format( self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 ' 'image-create-via-import ' '--file fake_src --disk-format qcow2') self.assertEqual('error: Must provide --container-format when ' 'using --file.', e.message) @mock.patch('sys.stderr') def test_create_via_import_web_download_missing_disk_format(self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 ' 'image-create-via-import ' + '--import-method web-download ' + '--uri fake_uri --container-format bare') self.assertEqual('error: Must provide --disk-format when using ' '--uri.', e.message) @mock.patch('sys.stderr') def test_create_via_import_web_download_missing_container_format( self, __): e = self.assertRaises(exc.CommandError, self._run_command, '--os-image-api-version 2 ' 'image-create-via-import ' '--import-method web-download ' '--uri fake_uri --disk-format qcow2') self.assertEqual('error: Must provide --container-format when ' 'using --uri.', e.message) def test_do_image_list(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name', 'id'], 'sort_dir': ['desc', 'asc'], 'sort': None, 'verbose': False, 'include_stores': False, 'os_hash_value': None, 'os_hidden': False } 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', 'os_hidden': False } mocked_list.assert_called_once_with(page_size=18, sort_key=['name', 'id'], sort_dir=['desc', 'asc'], filters=exp_img_filters) utils.print_list.assert_called_once_with({}, ['ID', 'Name']) def test_do_image_list_verbose(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name', 'id'], 'sort_dir': ['desc', 'asc'], 'sort': None, 'verbose': True, 'include_stores': False, 'os_hash_value': None, 'os_hidden': False } 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) utils.print_list.assert_called_once_with( {}, ['ID', 'Name', 'Disk_format', 'Container_format', 'Size', 'Status', 'Owner']) def test_do_image_list_verbose_cmd(self): self._run_command('--os-image-api-version 2 --verbose image-list') utils.print_list.assert_called_once_with( mock.ANY, ['ID', 'Name', 'Disk_format', 'Container_format', 'Size', 'Status', 'Owner']) def test_do_image_list_with_include_stores_true(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name', 'id'], 'sort_dir': ['desc', 'asc'], 'sort': None, 'verbose': False, 'include_stores': True, 'os_hash_value': None, 'os_hidden': False } 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) utils.print_list.assert_called_once_with( {}, ['ID', 'Name', 'Stores']) def test_do_image_list_verbose_with_include_stores_true(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name', 'id'], 'sort_dir': ['desc', 'asc'], 'sort': None, 'verbose': True, 'include_stores': True, 'os_hash_value': None, 'os_hidden': False } 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) utils.print_list.assert_called_once_with( {}, ['ID', 'Name', 'Disk_format', 'Container_format', 'Size', 'Status', 'Owner', 'Stores']) def test_do_image_list_with_hidden_true(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name', 'id'], 'sort_dir': ['desc', 'asc'], 'sort': None, 'verbose': False, 'include_stores': False, 'os_hash_value': None, 'os_hidden': True } 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', 'os_hidden': True } mocked_list.assert_called_once_with(page_size=18, sort_key=['name', 'id'], sort_dir=['desc', 'asc'], filters=exp_img_filters) utils.print_list.assert_called_once_with({}, ['ID', 'Name']) def test_do_image_list_with_single_sort_key(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort_key': ['name'], 'sort_dir': ['desc'], 'sort': None, 'verbose': False, 'include_stores': False, 'os_hash_value': None, 'os_hidden': False } 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', 'os_hidden': False } mocked_list.assert_called_once_with(page_size=18, sort_key=['name'], sort_dir=['desc'], filters=exp_img_filters) utils.print_list.assert_called_once_with({}, ['ID', 'Name']) def test_do_image_list_new_sorting_syntax(self): input = { 'limit': None, 'page_size': 18, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': [], 'sort': 'name:desc,size:asc', 'sort_key': [], 'sort_dir': [], 'verbose': False, 'include_stores': False, 'os_hash_value': None, 'os_hidden': False } 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', 'os_hidden': False } mocked_list.assert_called_once_with( page_size=18, sort='name:desc,size:asc', filters=exp_img_filters) utils.print_list.assert_called_once_with({}, ['ID', 'Name']) def test_do_image_list_with_property_filter(self): input = { 'limit': None, 'page_size': 1, 'visibility': True, 'member_status': 'Fake', 'owner': 'test', 'checksum': 'fake_checksum', 'tag': 'fake tag', 'properties': ['os_distro=NixOS', 'architecture=x86_64'], 'sort_key': ['name'], 'sort_dir': ['desc'], 'sort': None, 'verbose': False, 'include_stores': False, 'os_hash_value': None, 'os_hidden': False } 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', 'os_distro': 'NixOS', 'architecture': 'x86_64', 'os_hidden': False } mocked_list.assert_called_once_with(page_size=1, sort_key=['name'], sort_dir=['desc'], filters=exp_img_filters) utils.print_list.assert_called_once_with({}, ['ID', 'Name']) def test_do_image_show_human_readable(self): args = self._make_args({'id': 'pass', 'page_size': 18, 'human_readable': True, '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' expect_image['size'] = 1024 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', 'size': '1kB'}, max_column_width=120) def test_do_image_show(self): args = self._make_args({'id': 'pass', 'page_size': 18, 'human_readable': False, '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' expect_image['size'] = 1024 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', 'size': 1024}, max_column_width=120) def _test_do_image_tasks(self, verbose=False, supported=True): args = self._make_args({'id': 'pass', 'verbose': verbose}) expected_columns = ["Message", "Status", "Updated at"] expected_output = { "tasks": [ { "image_id": "pass", "id": "task_1", "user_id": "user_1", "request_id": "request_id_1", "message": "fake_message", "status": "status", } ] } if verbose: columns_to_prepend = ['Image Id', 'Task Id'] columns_to_extend = ['User Id', 'Request Id', 'Result', 'Owner', 'Input', 'Expires at'] expected_columns = (columns_to_prepend + expected_columns + columns_to_extend) expected_output["tasks"][0]["Result"] = "Fake Result" expected_output["tasks"][0]["Owner"] = "Fake Owner" expected_output["tasks"][0]["Input"] = "Fake Input" expected_output["tasks"][0]["Expires at"] = "Fake Expiry" with mock.patch.object(self.gc.images, 'get_associated_image_tasks') as mocked_tasks: if supported: mocked_tasks.return_value = expected_output else: mocked_tasks.side_effect = exc.HTTPNotImplemented test_shell.do_image_tasks(self.gc, args) mocked_tasks.assert_called_once_with('pass') if supported: utils.print_dict_list.assert_called_once_with( expected_output['tasks'], expected_columns) def test_do_image_tasks_without_verbose(self): self._test_do_image_tasks() def test_do_image_tasks_with_verbose(self): self._test_do_image_tasks(verbose=True) def test_do_image_tasks_unsupported(self): with mock.patch('glanceclient.common.utils.exit') as mock_exit: self._test_do_image_tasks(supported=False) mock_exit.assert_called_once_with( 'Server does not support image tasks API (v2.12)') def test_usage(self): with mock.patch.object(self.gc.info, 'get_usage') as mock_usage: mock_usage.return_value = {'quota1': {'limit': 10, 'usage': 0}, 'quota2': {'limit': 20, 'usage': 5}} test_shell.do_usage(self.gc, []) utils.print_dict_list.assert_called_once_with( [{'quota': 'quota1', 'limit': 10, 'usage': 0}, {'quota': 'quota2', 'limit': 20, 'usage': 5}], ['Quota', 'Limit', 'Usage']) def test_do_image_stage_size_match(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = 1024 args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=1024), \ mock.patch('glanceclient.v2.shell._validate_backend'), \ mock.patch.object(self.gc.images, 'stage') as mock_stage: test_shell.do_image_stage(self.gc, args) mock_stage.assert_called_once_with('IMG-01', 'fileobj', 1024) def test_do_image_stage_size_mismatch(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = 1024 args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=2048), \ mock.patch('glanceclient.v2.shell._validate_backend'): with self.assertRaisesRegex(ValueError, "Size mismatch: provided size 1024 " "does not match the size of the " "image 2048"): test_shell.do_image_stage(self.gc, args) def test_do_image_stage_no_size_in_args(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = None args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=1024), \ mock.patch('glanceclient.v2.shell._validate_backend'), \ mock.patch.object(self.gc.images, 'stage') as mock_upload: test_shell.do_image_stage(self.gc, args) mock_upload.assert_called_once_with('IMG-01', 'fileobj', 1024) def test_do_image_upload_size_match(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = 1024 args.store = None args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=1024), \ mock.patch('glanceclient.v2.shell._validate_backend'), \ mock.patch.object(self.gc.images, 'upload') as mock_upload: test_shell.do_image_upload(self.gc, args) mock_upload.assert_called_once_with('IMG-01', 'fileobj', 1024, backend=None) def test_do_image_upload_size_mismatch(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = 1024 args.store = None args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=2048), \ mock.patch('glanceclient.v2.shell._validate_backend'): with self.assertRaisesRegex(ValueError, "Size mismatch: provided size 1024 " "does not match the size of the " "image 2048"): test_shell.do_image_upload(self.gc, args) def test_do_image_upload_no_size_in_args(self): args = mock.Mock() args.id = 'IMG-01' args.file = 'testfile' args.size = None args.store = None args.progress = False with mock.patch('glanceclient.common.utils.get_data_file', return_value='fileobj'), \ mock.patch('glanceclient.common.utils.get_file_size', return_value=1024), \ mock.patch('glanceclient.v2.shell._validate_backend'), \ mock.patch.object(self.gc.images, 'upload') as mock_upload: test_shell.do_image_upload(self.gc, args) mock_upload.assert_called_once_with('IMG-01', 'fileobj', 1024, backend=None) @mock.patch('sys.stdin', autospec=True) def test_do_image_create_no_user_props(self, mock_stdin): args = self._make_args({'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'file': None}) 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 # Ensure that the test stdin is not considered # to be supplying image data mock_stdin.isatty = lambda: True 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'}) @mock.patch('sys.stdin', autospec=True) def test_do_image_create_for_none_multi_hash(self, mock_stdin): args = self._make_args({'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'file': None}) 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' expect_image['os_hash_algo'] = None expect_image['os_hash_value'] = None mocked_create.return_value = expect_image # Ensure that the test stdin is not considered # to be supplying image data mock_stdin.isatty = lambda: True 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', 'os_hash_algo': None, 'os_hash_value': None}) def test_do_image_create_with_multihash(self): self.mock_get_data_file.return_value = io.StringIO() try: with open(tempfile.mktemp(), 'w+') as f: f.write('Some data here') f.flush() f.seek(0) file_name = f.name temp_args = {'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'file': file_name, 'progress': False} args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: ignore_fields = ['self', 'access', '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' expect_image['checksum'] = 'fake-checksum' expect_image['os_hash_algo'] = 'fake-hash_algo' expect_image['os_hash_value'] = 'fake-hash_value' mocked_create.return_value = expect_image mocked_get.return_value = expect_image test_shell.do_image_create(self.gc, args) temp_args.pop('file', None) mocked_create.assert_called_once_with(**temp_args) mocked_get.assert_called_once_with('pass') utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'checksum': 'fake-checksum', 'os_hash_algo': 'fake-hash_algo', 'os_hash_value': 'fake-hash_value'}) finally: try: os.remove(f.name) except Exception: pass def _do_image_create(self, temp_args, expect_size=None): self.mock_get_data_file.return_value = io.StringIO() with open(tempfile.mktemp(), 'w+') as f: f.write('Some data here') f.flush() f.seek(0) file_name = f.name self.addCleanup(lambda: os.remove(f.name) if os.path.exists( f.name) else None) temp_args = temp_args.copy() temp_args['file'] = file_name args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create, \ mock.patch.object(self.gc.images, 'get') as mocked_get, \ mock.patch.object(utils, 'get_file_size') as mock_size: expected_size = len('Some data here') mock_size.return_value = expected_size ignore_fields = ['self', 'access', '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' expect_image['checksum'] = 'fake-checksum' expect_image['os_hash_algo'] = 'fake-hash_algo' expect_image['os_hash_value'] = 'fake-hash_value' if expect_size is not None: expect_image['size'] = expect_size mocked_create.return_value = expect_image mocked_get.return_value = expect_image test_shell.do_image_create(self.gc, args) temp_args.pop('file', None) mocked_create.assert_called_once_with(**temp_args) mocked_get.assert_called_once_with('pass') expected_dict = { 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'checksum': 'fake-checksum', 'os_hash_algo': 'fake-hash_algo', 'os_hash_value': 'fake-hash_value', } if expect_size is not None: expected_dict['size'] = expect_size utils.print_dict.assert_called_once_with(expected_dict) mock_size.assert_called() def test_do_image_create_without_size(self): self._do_image_create( temp_args={'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'progress': False}, expect_size=14) def test_do_image_create_with_size_exits(self): args = self._make_args({'name': 'test-image', 'size': 1234}) expected_msg = ("Setting 'size' during image creation is " "not supported. Please use --size only when " "uploading data.") with mock.patch('glanceclient.common.utils.exit') as mock_exit: mock_exit.side_effect = self._mock_utils_exit try: test_shell.do_image_create(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_exit.assert_called_once_with(expected_msg) @mock.patch('sys.stdin', autospec=True) def test_do_image_create_hidden_image(self, mock_stdin): args = self._make_args({'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'file': None, 'os_hidden': True}) 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' expect_image['os_hidden'] = True mocked_create.return_value = expect_image # Ensure that the test stdin is not considered # to be supplying image data mock_stdin.isatty = lambda: True test_shell.do_image_create(self.gc, args) mocked_create.assert_called_once_with(name='IMG-01', disk_format='vhd', container_format='bare', os_hidden=True) utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'os_hidden': True}) def test_do_image_create_with_file(self): self.mock_get_data_file.return_value = io.StringIO() try: file_name = None with open(tempfile.mktemp(), 'w+') as f: f.write('Some data here') f.flush() f.seek(0) file_name = f.name temp_args = {'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'file': file_name, 'progress': False} args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: ignore_fields = ['self', 'access', '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 mocked_get.return_value = expect_image test_shell.do_image_create(self.gc, args) temp_args.pop('file', None) mocked_create.assert_called_once_with(**temp_args) mocked_get.assert_called_once_with('pass') utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare'}) finally: try: os.remove(f.name) except Exception: pass @mock.patch('sys.stdin', autospec=True) def test_do_image_create_with_unicode(self, mock_stdin): name = '\u041f\u0420\u0418\u0412\u0415\u0422\u0418\u041a' args = self._make_args({'name': name, 'file': None}) 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'] = name mocked_create.return_value = expect_image mock_stdin.isatty = lambda: True test_shell.do_image_create(self.gc, args) mocked_create.assert_called_once_with(name=name) utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': name}) @mock.patch('sys.stdin', autospec=True) def test_do_image_create_with_user_props(self, mock_stdin): args = self._make_args({'name': 'IMG-01', 'property': ['myprop=myval'], 'file': None, 'container_format': 'bare', 'disk_format': 'qcow2'}) 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 # Ensure that the test stdin is not considered # to be supplying image data mock_stdin.isatty = lambda: True test_shell.do_image_create(self.gc, args) mocked_create.assert_called_once_with(name='IMG-01', myprop='myval', container_format='bare', disk_format='qcow2') utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'myprop': 'myval'}) @mock.patch('glanceclient.common.utils.exit') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_neg_do_image_create_no_file_and_stdin_with_store( self, mock_stdin, mock_access, mock_utils_exit): expected_msg = ('--store option should only be provided with --file ' 'option or stdin.') mock_utils_exit.side_effect = self._mock_utils_exit mock_stdin.isatty = lambda: True mock_access.return_value = False args = self._make_args({'name': 'IMG-01', 'property': ['myprop=myval'], 'file': None, 'store': 'file1', 'container_format': 'bare', 'disk_format': 'qcow2'}) try: test_shell.do_image_create(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_do_image_create_invalid_store( self, mock_utils_exit): expected_msg = ("Store 'dummy' is not valid for this cloud. " "Valid values can be retrieved with stores-info " "command.") mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args({'name': 'IMG-01', 'property': ['myprop=myval'], 'file': "somefile.txt", 'store': 'dummy', 'container_format': 'bare', 'disk_format': 'qcow2'}) with mock.patch.object(self.gc.images, 'get_stores_info') as mock_stores_info: mock_stores_info.return_value = self.stores_info_response try: test_shell.do_image_create(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) # NOTE(rosmaita): have to explicitly set to None the declared but unused # arguments (the configparser does that for us normally) base_args = {'name': 'Mortimer', 'disk_format': 'raw', 'container_format': 'bare', 'progress': False, 'file': None, 'uri': None, 'remote_region': None, 'import_method': None} import_info_response = {'import-methods': { 'type': 'array', 'description': 'Import methods available.', 'value': ['glance-direct', 'web-download', 'copy-image', 'glance-download']}} def _mock_utils_exit(self, msg): sys.exit(msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_no_method_with_file_and_stdin( self, mock_stdin, mock_access, mock_utils_exit): expected_msg = ('You cannot use both --file and stdin with the ' 'glance-direct import method.') my_args = self.base_args.copy() my_args['file'] = 'some.file' args = self._make_args(my_args) mock_stdin.isatty = lambda: False mock_access.return_value = True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_create_via_import_copy_image( self, mock_utils_exit): expected_msg = ("Import method 'copy-image' cannot be used " "while creating the image.") mock_utils_exit.side_effect = self._mock_utils_exit my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'copy-image'}) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_create_via_import_stores_all_stores_specified( self, mock_utils_exit): expected_msg = ('Only one of --store, --stores and --all-stores can ' 'be provided') mock_utils_exit.side_effect = self._mock_utils_exit my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'glance-direct', 'stores': 'file1,file2', 'os_all_stores': True, 'file': 'some.mufile', 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_stores_without_file( self, mock_stdin, mock_utils_exit): expected_msg = ('--stores option should only be provided with --file ' 'option or stdin for the glance-direct import method.') mock_utils_exit.side_effect = self._mock_utils_exit mock_stdin.isatty = lambda: True my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'glance-direct', 'stores': 'file1,file2', 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_stores_info: mocked_stores_info.return_value = self.stores_info_response mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_all_stores_without_file( self, mock_stdin, mock_utils_exit): expected_msg = ('--all-stores option should only be provided with ' '--file option or stdin for the glance-direct import ' 'method.') mock_utils_exit.side_effect = self._mock_utils_exit mock_stdin.isatty = lambda: True my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'glance-direct', 'os_all_stores': True, 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_no_file_and_stdin_with_store( self, mock_stdin, mock_access, mock_utils_exit): expected_msg = ('--store option should only be provided with --file ' 'option or stdin for the glance-direct import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' my_args['store'] = 'file1' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_access.return_value = False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_stores_info: mocked_stores_info.return_value = self.stores_info_response mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_no_uri_with_store( self, mock_stdin, mock_utils_exit): expected_msg = ('--store option should only be provided with --uri ' 'option for the web-download import method.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' my_args['store'] = 'file1' args = self._make_args(my_args) mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_stores_info: mocked_stores_info.return_value = self.stores_info_response mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_invalid_store( self, mock_stdin, mock_access, mock_utils_exit): expected_msg = ("Store 'dummy' is not valid for this cloud. " "Valid values can be retrieved with stores-info" " command.") my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' my_args['store'] = 'dummy' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_access.return_value = False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_stores_info: mocked_stores_info.return_value = self.stores_info_response mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_no_method_passing_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot use --uri without specifying an import ' 'method.') my_args = self.base_args.copy() my_args['uri'] = 'http://example.com/whatever' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_direct_no_data( self, mock_stdin, mock_utils_exit): expected_msg = ('You must specify a --file or provide data via stdin ' 'for the glance-direct import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_direct_with_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot specify a --uri with the glance-direct ' 'import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_direct_with_file_and_uri( self, mock_stdin, mock_access, mock_utils_exit): expected_msg = ('You cannot specify a --uri with the glance-direct ' 'import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' my_args['uri'] = 'https://example.com/some/stuff' my_args['file'] = 'my.browncow' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_access.return_value = True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_direct_with_data_and_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot specify a --uri with the glance-direct ' 'import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_stdin.isatty = lambda: False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_web_download_no_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('URI is required for web-download import method. ' 'Please use \'--uri \'.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_create_via_import_stores_without_uri( self, mock_utils_exit): expected_msg = ('--stores option should only be provided with --uri ' 'option for the web-download import method.') mock_utils_exit.side_effect = self._mock_utils_exit my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'web-download', 'stores': 'file1,file2', 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as mocked_stores_info: mocked_stores_info.return_value = self.stores_info_response mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_create_via_import_all_stores_without_uri( self, mock_utils_exit): expected_msg = ('--all-stores option should only be provided with ' '--uri option for the web-download import ' 'method.') mock_utils_exit.side_effect = self._mock_utils_exit my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'web-download', 'os_all_stores': True, 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_web_download_no_uri_with_file( self, mock_stdin, mock_utils_exit): expected_msg = ('URI is required for web-download import method. ' 'Please use \'--uri \'.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' my_args['file'] = 'my.browncow' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_web_download_no_uri_with_data( self, mock_stdin, mock_utils_exit): expected_msg = ('URI is required for web-download import method. ' 'Please use \'--uri \'.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' my_args['file'] = 'my.browncow' args = self._make_args(my_args) mock_stdin.isatty = lambda: False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_web_download_with_data_and_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot pass data via stdin with the web-download ' 'import method.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_stdin.isatty = lambda: False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_web_download_with_file_and_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot specify a --file with the web-download ' 'import method.') my_args = self.base_args.copy() my_args['import_method'] = 'web-download' my_args['uri'] = 'https://example.com/some/stuff' my_args['file'] = 'my.browncow' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_download_no_region_and_id( self, mock_stdin, mock_utils_exit): expected_msg = ('REMOTE GlANCE REGION and REMOTE IMAGE ID are ' 'required for glance-download import method. ' 'Please use --remote-region and ' '--remote-image-id .') my_args = self.base_args.copy() my_args['import_method'] = 'glance-download' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_download_with_uri( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot specify a --uri with the ' 'glance-download import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-download' my_args['remote_region'] = 'REGION2' my_args['remote_image_id'] = 'IMG2' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_download_with_file( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot specify a --file with the ' 'glance-download import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-download' my_args['remote_region'] = 'REGION2' my_args['remote_image_id'] = 'IMG2' my_args['file'] = 'my.browncow' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_glance_download_with_data( self, mock_stdin, mock_utils_exit): expected_msg = ('You cannot pass data via stdin with the ' 'glance-download import method.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-download' my_args['remote_region'] = 'REGION2' my_args['remote_image_id'] = 'IMG2' args = self._make_args(my_args) mock_stdin.isatty = lambda: False mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_bad_method( self, mock_stdin, mock_utils_exit): expected_msg = ('Import method \'swift-party-time\' is not valid ' 'for this cloud. Valid values can be retrieved with ' 'import-info command.') my_args = self.base_args.copy() my_args['import_method'] = 'swift-party-time' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_no_method_with_data_and_method_NA( self, mock_stdin, mock_utils_exit): expected_msg = ('Import method \'glance-direct\' is not valid ' 'for this cloud. Valid values can be retrieved with ' 'import-info command.') args = self._make_args(self.base_args) # need to fake some data, or this is "just like" a # create-image-record-only call mock_stdin.isatty = lambda: False mock_utils_exit.side_effect = self._mock_utils_exit my_import_info_response = deepcopy(self.import_info_response) my_import_info_response['import-methods']['value'] = ['web-download'] with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = my_import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') @mock.patch('sys.stdin', autospec=True) def test_neg_image_create_via_import_good_method_not_available( self, mock_stdin, mock_utils_exit): """Make sure the good method names aren't hard coded somewhere""" expected_msg = ('Import method \'glance-direct\' is not valid for ' 'this cloud. Valid values can be retrieved with ' 'import-info command.') my_args = self.base_args.copy() my_args['import_method'] = 'glance-direct' args = self._make_args(my_args) mock_stdin.isatty = lambda: True mock_utils_exit.side_effect = self._mock_utils_exit my_import_info_response = deepcopy(self.import_info_response) my_import_info_response['import-methods']['value'] = ['bad-bad-method'] with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = my_import_info_response try: test_shell.do_image_create_via_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('glanceclient.v2.shell.do_image_stage') @mock.patch('sys.stdin', autospec=True) def test_image_create_via_import_no_method_with_stdin( self, mock_stdin, mock_do_stage, mock_do_import): """Backward compat -> handle this like a glance-direct""" mock_stdin.isatty = lambda: False self.mock_get_data_file.return_value = io.StringIO() args = self._make_args(self.base_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: ignore_fields = ['self', 'access', 'schema'] expect_image = dict([(field, field) for field in ignore_fields]) expect_image['id'] = 'via-stdin' expect_image['name'] = 'Mortimer' expect_image['disk_format'] = 'raw' expect_image['container_format'] = 'bare' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response test_shell.do_image_create_via_import(self.gc, args) mocked_create.assert_called_once() mock_do_stage.assert_called_once() mock_do_import.assert_called_once() mocked_get.assert_called_with('via-stdin') utils.print_dict.assert_called_with({ 'id': 'via-stdin', 'name': 'Mortimer', 'disk_format': 'raw', 'container_format': 'bare'}) def _image_create_via_import_with_file_helper( self, with_access=True): """Helper for image create via import with file tests.""" @mock.patch('glanceclient.common.utils.get_file_size') @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def _test_with_access(mock_stdin, mock_access, mock_do_import, mock_size): mock_stdin.isatty = lambda: True self.mock_get_data_file.return_value = io.StringIO() mock_access.return_value = with_access mock_size.return_value = 14 with open(tempfile.mktemp(), 'w+') as f: f.write('Some data here') f.flush() f.seek(0) file_name = f.name self.addCleanup( lambda: os.remove(file_name) if os.path.exists( file_name) else None) my_args = self.base_args.copy() my_args.update({'file': file_name}) args = self._make_args(my_args) with mock.patch.object( self.gc.images, 'create') as mocked_create, \ mock.patch.object( self.gc.images, 'get') as mocked_get, \ mock.patch.object( self.gc.images, 'get_import_info') as mocked_info: ignore_fields = ['self', 'access', 'schema'] expect_image = dict( [(field, field) for field in ignore_fields]) expect_image['id'] = 'fake-image-id' expect_image['name'] = 'Mortimer' expect_image['disk_format'] = 'raw' expect_image['container_format'] = 'bare' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response test_shell.do_image_create_via_import(self.gc, args) mocked_create.assert_called_once() mock_do_import.assert_called_once() mocked_get.assert_called_with(expect_image['id']) mock_size.assert_called_once() utils.print_dict.assert_called_with({ 'id': expect_image['id'], 'name': 'Mortimer', 'disk_format': 'raw', 'container_format': 'bare' }) _test_with_access() def test_image_create_via_import_with_file(self): self._image_create_via_import_with_file_helper() def test_do_image_create_via_import_with_size_exits(self): args = self._make_args({'name': 'test-image', 'size': 1234}) expected_msg = ("Setting 'size' during image creation is not " "supported. Please use --size only when " "uploading data.") with mock.patch('glanceclient.common.utils.exit') as mock_exit: mock_exit.side_effect = self._mock_utils_exit try: test_shell.do_image_create(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('glanceclient.v2.shell.do_image_stage') @mock.patch('os.access') @mock.patch('sys.stdin', autospec=True) def test_image_create_via_import_no_method_passing_file( self, mock_stdin, mock_access, mock_do_stage, mock_do_import): """Backward compat -> handle this like a glance-direct""" mock_stdin.isatty = lambda: True self.mock_get_data_file.return_value = io.StringIO() mock_access.return_value = True my_args = self.base_args.copy() my_args['file'] = 'fake-image-file.browncow' args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: ignore_fields = ['self', 'access', 'schema'] expect_image = dict([(field, field) for field in ignore_fields]) expect_image['id'] = 'via-file' expect_image['name'] = 'Mortimer' expect_image['disk_format'] = 'raw' expect_image['container_format'] = 'bare' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response test_shell.do_image_create_via_import(self.gc, args) mocked_create.assert_called_once() mock_do_stage.assert_called_once() mock_do_import.assert_called_once() mocked_get.assert_called_with('via-file') utils.print_dict.assert_called_with({ 'id': 'via-file', 'name': 'Mortimer', 'disk_format': 'raw', 'container_format': 'bare'}) @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('glanceclient.v2.shell.do_image_stage') @mock.patch('sys.stdin', autospec=True) def test_do_image_create_via_import_with_no_method_no_data( self, mock_stdin, mock_do_image_stage, mock_do_image_import): """Create an image record without calling do_stage or do_import""" img_create_args = {'name': 'IMG-11', 'os_architecture': 'powerpc', 'id': 'watch-out-for-ossn-0075', 'progress': False} client_args = {'import_method': None, 'file': None, 'uri': None} temp_args = img_create_args.copy() temp_args.update(client_args) args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: ignore_fields = ['self', 'access', 'schema'] expect_image = dict([(field, field) for field in ignore_fields]) expect_image['name'] = 'IMG-11' expect_image['id'] = 'watch-out-for-ossn-0075' expect_image['os_architecture'] = 'powerpc' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response mock_stdin.isatty = lambda: True test_shell.do_image_create_via_import(self.gc, args) mocked_create.assert_called_once_with(**img_create_args) mocked_get.assert_called_with('watch-out-for-ossn-0075') mock_do_image_stage.assert_not_called() mock_do_image_import.assert_not_called() utils.print_dict.assert_called_with({ 'name': 'IMG-11', 'os_architecture': 'powerpc', 'id': 'watch-out-for-ossn-0075'}) @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('glanceclient.v2.shell.do_image_stage') @mock.patch('sys.stdin', autospec=True) def test_do_image_create_via_import_with_web_download( self, mock_stdin, mock_do_image_stage, mock_do_image_import): temp_args = {'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'uri': 'http://example.com/image.qcow', 'import_method': 'web-download', 'progress': False} args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: ignore_fields = ['self', 'access', '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' expect_image['status'] = 'queued' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response mock_stdin.isatty = lambda: True test_shell.do_image_create_via_import(self.gc, args) mock_do_image_stage.assert_not_called() mock_do_image_import.assert_called_once() mocked_create.assert_called_once_with(**temp_args) mocked_get.assert_called_with('pass') utils.print_dict.assert_called_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'status': 'queued'}) @mock.patch('glanceclient.v2.shell.do_image_import') @mock.patch('glanceclient.v2.shell.do_image_stage') @mock.patch('sys.stdin', autospec=True) def test_do_image_create_via_import_with_web_download_with_stores( self, mock_stdin, mock_do_image_stage, mock_do_image_import): temp_args = {'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'uri': 'http://example.com/image.qcow', 'import_method': 'web-download', 'progress': False, 'stores': 'file1,file2'} tmp2_args = {'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'uri': 'http://example.com/image.qcow', 'import_method': 'web-download', 'progress': False} args = self._make_args(temp_args) with mock.patch.object(self.gc.images, 'create') as mocked_create: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get_stores_info') as m_stores_info: ignore_fields = ['self', 'access', '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' expect_image['status'] = 'queued' mocked_create.return_value = expect_image mocked_get.return_value = expect_image mocked_info.return_value = self.import_info_response m_stores_info.return_value = self.stores_info_response mock_stdin.isatty = lambda: True test_shell.do_image_create_via_import(self.gc, args) mock_do_image_stage.assert_not_called() mock_do_image_import.assert_called_once() mocked_create.assert_called_once_with(**tmp2_args) mocked_get.assert_called_with('pass') utils.print_dict.assert_called_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'status': 'queued'}) 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_hide_image(self): args = self._make_args({'id': 'pass', 'os_hidden': 'true'}) 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' expect_image['os_hidden'] = True mocked_update.return_value = expect_image test_shell.do_image_update(self.gc, args) mocked_update.assert_called_once_with('pass', None, os_hidden='true') utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'os_hidden': True}) def test_do_image_update_revert_hide_image(self): args = self._make_args({'id': 'pass', 'os_hidden': 'false'}) 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' expect_image['os_hidden'] = False mocked_update.return_value = expect_image test_shell.do_image_update(self.gc, args) mocked_update.assert_called_once_with('pass', None, os_hidden='false') utils.print_dict.assert_called_once_with({ 'id': 'pass', 'name': 'IMG-01', 'disk_format': 'vhd', 'container_format': 'bare', 'os_hidden': False}) 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_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_do_location_add(self): gc = self.gc loc = {'url': 'http://foo.com/', 'metadata': {'foo': 'bar'}, 'validation_data': {'checksum': 'csum', 'os_hash_algo': 'algo', 'os_hash_value': 'value'}} args = {'id': 'pass', 'url': loc['url'], 'metadata': json.dumps(loc['metadata']), 'checksum': 'csum', 'hash_algo': 'algo', 'hash_value': 'value'} 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, self._make_args(args)) mocked_addloc.assert_called_once_with( 'pass', loc['url'], loc['metadata'], validation_data=loc['validation_data']) 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: 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_add_location(self): gc = self.gc url = 'http://foo.com/', validation_data = {'os_hash_algo': 'sha512', 'os_hash_value': 'value'} args = {'id': 'IMG-01', 'url': url, 'validation_data': json.dumps(validation_data)} with mock.patch.object(gc.images, 'add_image_location') as mock_addloc: expect_image = {'id': 'pass'} mock_addloc.return_value = expect_image test_shell.do_add_location(self.gc, self._make_args(args)) mock_addloc.assert_called_once_with( 'IMG-01', url, validation_data=validation_data) utils.print_dict.assert_called_once_with(expect_image) @mock.patch('glanceclient.common.utils.exit') def test_do_add_location_with_checksum_in_validation_data(self, mock_exit): validation_data = {'checksum': 'value', 'os_hash_algo': 'sha512', 'os_hash_value': 'value'} args = self._make_args( {'id': 'IMG-01', 'url': 'http://foo.com/', 'validation_data': json.dumps(validation_data)}) expected_msg = ('Validation Data should contain only os_hash_algo' ' and os_hash_value. `checksum` is not allowed') mock_exit.side_effect = self._mock_utils_exit try: test_shell.do_add_location(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_do_add_location_with_invalid_algo_in_validation_data(self, mock_exit): validation_data = {'os_hash_algo': 'algo', 'os_hash_value': 'value'} args = self._make_args( {'id': 'IMG-01', 'url': 'http://foo.com/', 'validation_data': json.dumps(validation_data)}) allowed_hash_algo = ['sha512', 'sha256', 'sha1', 'md5'] expected_msg = ('os_hash_algo: `%s` is incorrect, ' 'allowed hashing algorithms: %s' % (validation_data['os_hash_algo'], allowed_hash_algo)) mock_exit.side_effect = self._mock_utils_exit try: test_shell.do_add_location(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_exit.assert_called_once_with(expected_msg) def test_image_upload(self): args = self._make_args( {'id': 'IMG-01', 'file': 'test', 'size': 1024, 'progress': False}) with mock.patch.object(self.gc.images, 'upload') as mocked_upload: expected_data = '*' * 1024 utils.get_data_file = mock.Mock(return_value=expected_data) utils.get_file_size = mock.Mock(return_value=1024) mocked_upload.return_value = None test_shell.do_image_upload(self.gc, args) mocked_upload.assert_called_once_with('IMG-01', expected_data, 1024, backend=None) @mock.patch('glanceclient.common.utils.exit') def test_image_upload_invalid_store(self, mock_utils_exit): expected_msg = ("Store 'dummy' is not valid for this cloud. " "Valid values can be retrieved with stores-info " "command.") mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'file': 'test', 'size': 1024, 'progress': False, 'store': 'dummy'}) with mock.patch.object(self.gc.images, 'get_stores_info') as mock_stores_info: mock_stores_info.return_value = self.stores_info_response try: test_shell.do_image_upload(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_not_available(self, mock_utils_exit): expected_msg = 'Target Glance does not support Image Import workflow' mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'smarty-pants', 'uri': None}) with mock.patch.object(self.gc.images, 'import') as mocked_import: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.side_effect = exc.HTTPNotFound try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) mocked_import.assert_not_called() @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_bad_method(self, mock_utils_exit): expected_msg = ('Import method \'smarty-pants\' is not valid for this ' 'cloud. Valid values can be retrieved with ' 'import-info command.') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'smarty-pants', 'uri': None}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_no_methods_configured(self, mock_utils_exit): expected_msg = ('Import method \'glance-direct\' is not valid for ' 'this cloud. Valid values can be retrieved with ' 'import-info command.') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'glance-direct', 'uri': None}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = {"import-methods": {"value": []}} try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_glance_direct_image_not_uploading_status( self, mock_utils_exit): expected_msg = ('The \'glance-direct\' import method can only be ' 'applied to an image in status \'uploading\'') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'glance-direct', 'uri': None}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get') as mocked_get: mocked_get.return_value = {'status': 'queued', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_web_download_image_not_queued_status( self, mock_utils_exit): expected_msg = ('The \'web-download\' import method can only be ' 'applied to an image in status \'queued\'') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'web-download', 'uri': 'http://joes-image-shack.com/funky.qcow2'}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get') as mocked_get: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_image_no_container_format( self, mock_utils_exit): expected_msg = ('The \'container_format\' and \'disk_format\' ' 'properties must be set on an image before it can be ' 'imported.') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'web-download', 'uri': 'http://joes-image-shack.com/funky.qcow2'}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get') as mocked_get: mocked_get.return_value = {'status': 'uploading', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_image_no_disk_format( self, mock_utils_exit): expected_msg = ('The \'container_format\' and \'disk_format\' ' 'properties must be set on an image before it can be ' 'imported.') mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'web-download', 'uri': 'http://joes-image-shack.com/funky.qcow2'}) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: with mock.patch.object(self.gc.images, 'get') as mocked_get: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare'} mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_image_import_invalid_store(self, mock_utils_exit): expected_msg = ("Store 'dummy' is not valid for this cloud. " "Valid values can be retrieved with stores-info " "command.") mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-01', 'import_method': 'glance-direct', 'uri': None, 'store': 'dummy'}) with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} with mock.patch.object(self.gc.images, 'get_stores_info') as mock_stores_info: mocked_info.return_value = self.import_info_response mock_stores_info.return_value = self.stores_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) def test_image_import_glance_direct(self): args = self._make_args( {'id': 'IMG-01', 'import_method': 'glance-direct', 'uri': None}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-01', 'glance-direct', uri=None, remote_region=None, remote_image_id=None, remote_service_interface=None, backend=None, all_stores=None, allow_failure=True, stores=None) def test_image_import_web_download(self): args = self._make_args( {'id': 'IMG-01', 'uri': 'http://example.com/image.qcow', 'import_method': 'web-download'}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'queued', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-01', 'web-download', uri='http://example.com/image.qcow', remote_region=None, remote_image_id=None, remote_service_interface=None, all_stores=None, allow_failure=True, backend=None, stores=None) def test_image_import_glance_download(self): args = self._make_args( {'id': 'IMG-01', 'uri': None, 'remote_region': 'REGION2', 'remote_image_id': 'IMG-02', 'import_method': 'glance-download', 'remote_service_interface': 'public'}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'queued', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-01', 'glance-download', uri=None, remote_region='REGION2', remote_image_id='IMG-02', remote_service_interface='public', all_stores=None, allow_failure=True, backend=None, stores=None) @mock.patch('glanceclient.common.utils.exit') def test_image_import_neg_no_glance_download_with_remote_region( self, mock_utils_exit): expected_msg = ("Import method should be 'glance-download' if " "REMOTE REGION is provided.") my_args = self.base_args.copy() my_args['id'] = 'IMG-01' my_args['remote_region'] = 'REGION2' my_args['import_method'] = 'web-download' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_image_import_neg_no_glance_download_with_remote_id( self, mock_utils_exit): expected_msg = ("Import method should be 'glance-download' if " "REMOTE IMAGE ID is provided.") my_args = self.base_args.copy() my_args['id'] = 'IMG-01' my_args['remote_image_id'] = 'IMG-02' my_args['import_method'] = 'web-download' my_args['uri'] = 'https://example.com/some/stuff' args = self._make_args(my_args) mock_utils_exit.side_effect = self._mock_utils_exit with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) def test_image_import_glance_download_without_remote_service_interface( self): args = self._make_args( {'id': 'IMG-01', 'uri': None, 'remote_region': 'REGION2', 'remote_image_id': 'IMG-02', 'import_method': 'glance-download'}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'queued', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-01', 'glance-download', uri=None, remote_region='REGION2', remote_image_id='IMG-02', remote_service_interface=None, all_stores=None, allow_failure=True, backend=None, stores=None) @mock.patch('glanceclient.common.utils.print_image') def test_image_import_no_print_image(self, mocked_utils_print_image): args = self._make_args( {'id': 'IMG-02', 'uri': None, 'import_method': 'glance-direct', 'from_create': True}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-02', 'glance-direct', None, stores=None, all_stores=None, allow_failure=True, remote_region=None, remote_image_id=None, remote_service_interface=None, backend=None) mocked_utils_print_image.assert_not_called() @mock.patch('glanceclient.common.utils.print_image') @mock.patch('glanceclient.v2.shell._validate_backend') def test_image_import_multiple_stores(self, mocked_utils_print_image, msvb): args = self._make_args( {'id': 'IMG-02', 'uri': None, 'import_method': 'glance-direct', 'from_create': False, 'stores': 'site1,site2'}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-02', 'glance-direct', uri=None, remote_region=None, remote_image_id=None, remote_service_interface=None, all_stores=None, allow_failure=True, stores=['site1', 'site2'], backend=None) @mock.patch('glanceclient.common.utils.print_image') @mock.patch('glanceclient.v2.shell._validate_backend') def test_image_import_copy_image(self, mocked_utils_print_image, msvb): args = self._make_args( {'id': 'IMG-02', 'uri': None, 'import_method': 'copy-image', 'from_create': False, 'stores': 'file1,file2'}) with mock.patch.object(self.gc.images, 'image_import') as mock_import: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_get.return_value = {'status': 'active', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response mock_import.return_value = None test_shell.do_image_import(self.gc, args) mock_import.assert_called_once_with( 'IMG-02', 'copy-image', uri=None, remote_region=None, remote_image_id=None, remote_service_interface=None, all_stores=None, allow_failure=True, stores=['file1', 'file2'], backend=None) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_copy_image_not_active( self, mock_utils_exit): expected_msg = ("The 'copy-image' import method can only be used on " "an image with status 'active'.") mock_utils_exit.side_effect = self._mock_utils_exit args = self._make_args( {'id': 'IMG-02', 'uri': None, 'import_method': 'copy-image', 'disk_format': 'raw', 'container_format': 'bare', 'from_create': False, 'stores': 'file1,file2'}) with mock.patch.object( self.gc.images, 'get_stores_info') as mocked_stores_info: with mock.patch.object(self.gc.images, 'get') as mocked_get: with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_stores_info.return_value = self.stores_info_response mocked_get.return_value = {'status': 'uploading', 'container_format': 'bare', 'disk_format': 'raw'} mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) @mock.patch('glanceclient.common.utils.exit') def test_neg_image_import_stores_all_stores_not_specified( self, mock_utils_exit): expected_msg = ("Provide either --stores or --all-stores for " "'copy-image' import method.") mock_utils_exit.side_effect = self._mock_utils_exit my_args = self.base_args.copy() my_args.update( {'id': 'IMG-01', 'import_method': 'copy-image', 'disk_format': 'raw', 'container_format': 'bare', }) args = self._make_args(my_args) with mock.patch.object(self.gc.images, 'get_import_info') as mocked_info: mocked_info.return_value = self.import_info_response try: test_shell.do_image_import(self.gc, args) self.fail("utils.exit should have been called") except SystemExit: pass mock_utils_exit.assert_called_once_with(expected_msg) def test_image_download(self): args = self._make_args( {'id': 'IMG-01', 'file': 'test', 'progress': True, 'allow_md5_fallback': False}) with mock.patch.object(self.gc.images, 'data') as mocked_data, \ mock.patch.object(utils, '_extract_request_id'): mocked_data.return_value = utils.RequestIdProxy( [c for c in 'abcdef']) test_shell.do_image_download(self.gc, args) mocked_data.assert_called_once_with('IMG-01', allow_md5_fallback=False) # check that non-default value is being passed correctly args.allow_md5_fallback = True with mock.patch.object(self.gc.images, 'data') as mocked_data, \ mock.patch.object(utils, '_extract_request_id'): mocked_data.return_value = utils.RequestIdProxy( [c for c in 'abcdef']) test_shell.do_image_download(self.gc, args) mocked_data.assert_called_once_with('IMG-01', allow_md5_fallback=True) @mock.patch.object(utils, 'exit') @mock.patch('sys.stdout', autospec=True) def test_image_download_no_file_arg(self, mocked_stdout, mocked_utils_exit): # Indicate that no file name was given as command line argument args = self._make_args({'id': '1234', 'file': None, 'progress': False, 'allow_md5_fallback': False}) # Indicate that no file is specified for output redirection mocked_stdout.isatty = lambda: True test_shell.do_image_download(self.gc, args) mocked_utils_exit.assert_called_once_with( 'No redirection or local file specified for downloaded image' ' data. Please specify a local file with --file to save' ' downloaded image or redirect output to another source.') def test_do_image_delete(self): args = argparse.Namespace(id=['image1', 'image2']) with mock.patch.object(self.gc.images, 'delete') as mocked_delete: mocked_delete.return_value = 0 test_shell.do_image_delete(self.gc, args) self.assertEqual(2, mocked_delete.call_count) def test_do_image_deactivate(self): args = argparse.Namespace(id='image1') with mock.patch.object(self.gc.images, 'deactivate') as mocked_deactivate: mocked_deactivate.return_value = 0 test_shell.do_image_deactivate(self.gc, args) self.assertEqual(1, mocked_deactivate.call_count) def test_do_image_reactivate(self): args = argparse.Namespace(id='image1') with mock.patch.object(self.gc.images, 'reactivate') as mocked_reactivate: mocked_reactivate.return_value = 0 test_shell.do_image_reactivate(self.gc, args) self.assertEqual(1, mocked_reactivate.call_count) @mock.patch.object(utils, 'exit') @mock.patch.object(utils, 'print_err') def test_do_image_delete_with_invalid_ids(self, mocked_print_err, mocked_utils_exit): args = argparse.Namespace(id=['image1', 'image2']) with mock.patch.object(self.gc.images, 'delete') as mocked_delete: mocked_delete.side_effect = exc.HTTPNotFound test_shell.do_image_delete(self.gc, args) self.assertEqual(2, mocked_delete.call_count) self.assertEqual(2, mocked_print_err.call_count) mocked_utils_exit.assert_called_once_with() @mock.patch.object(utils, 'exit') def test_do_image_delete_from_store_not_found(self, mocked_utils_exit): args = argparse.Namespace(id='image1', store='store1') with mock.patch.object(self.gc.images, 'delete_from_store') as mocked_delete: mocked_delete.side_effect = exc.HTTPNotFound test_shell.do_stores_delete(self.gc, args) self.assertEqual(1, mocked_delete.call_count) mocked_utils_exit.assert_called_once_with('Multi Backend support ' 'is not enabled or ' 'Image/store not found.') def test_do_image_delete_from_store(self): args = argparse.Namespace(id='image1', store='store1') with mock.patch.object(self.gc.images, 'delete_from_store') as mocked_delete: test_shell.do_stores_delete(self.gc, args) mocked_delete.assert_called_once_with('store1', 'image1') @mock.patch.object(utils, 'exit') @mock.patch.object(utils, 'print_err') def test_do_image_delete_with_forbidden_ids(self, mocked_print_err, mocked_utils_exit): args = argparse.Namespace(id=['image1', 'image2']) with mock.patch.object(self.gc.images, 'delete') as mocked_delete: mocked_delete.side_effect = exc.HTTPForbidden test_shell.do_image_delete(self.gc, args) self.assertEqual(2, mocked_delete.call_count) self.assertEqual(2, mocked_print_err.call_count) mocked_utils_exit.assert_called_once_with() @mock.patch.object(utils, 'exit') @mock.patch.object(utils, 'print_err') def test_do_image_delete_with_image_in_use(self, mocked_print_err, mocked_utils_exit): args = argparse.Namespace(id=['image1', 'image2']) with mock.patch.object(self.gc.images, 'delete') as mocked_delete: mocked_delete.side_effect = exc.HTTPConflict test_shell.do_image_delete(self.gc, args) self.assertEqual(2, mocked_delete.call_count) self.assertEqual(2, mocked_print_err.call_count) mocked_utils_exit.assert_called_once_with() def test_do_image_delete_deleted(self): image_id = 'deleted-img' args = argparse.Namespace(id=[image_id]) with mock.patch.object(self.gc.images, 'delete') as mocked_delete: mocked_delete.side_effect = exc.HTTPNotFound self.assert_exits_with_msg(func=test_shell.do_image_delete, func_args=args) @mock.patch('sys.stdout', autospec=True) @mock.patch.object(utils, 'print_err') def test_do_image_download_with_forbidden_id(self, mocked_print_err, mocked_stdout): args = self._make_args({'id': 'IMG-01', 'file': None, 'progress': False, 'allow_md5_fallback': False}) mocked_stdout.isatty = lambda: False with mock.patch.object(self.gc.images, 'data') as mocked_data: mocked_data.side_effect = exc.HTTPForbidden try: test_shell.do_image_download(self.gc, args) self.fail('Exit not called') except SystemExit: pass self.assertEqual(1, mocked_data.call_count) self.assertEqual(1, mocked_print_err.call_count) @mock.patch('sys.stdout', autospec=True) @mock.patch.object(utils, 'print_err') def test_do_image_download_with_500(self, mocked_print_err, mocked_stdout): args = self._make_args({'id': 'IMG-01', 'file': None, 'progress': False, 'allow_md5_fallback': False}) mocked_stdout.isatty = lambda: False with mock.patch.object(self.gc.images, 'data') as mocked_data: mocked_data.side_effect = exc.HTTPInternalServerError try: test_shell.do_image_download(self.gc, args) self.fail('Exit not called') except SystemExit: pass self.assertEqual(1, mocked_data.call_count) self.assertEqual(1, mocked_print_err.call_count) 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_get(self): args = self._make_args({'image_id': 'IMG-01', 'member_id': 'MEM-01'}) with mock.patch.object(self.gc.image_members, 'get') as mock_get: mock_get.return_value = {} test_shell.do_member_get(self.gc, args) mock_get.assert_called_once_with('IMG-01', 'MEM-01') utils.print_dict.assert_called_once_with({}) 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) def test_do_md_namespace_create(self): args = self._make_args({'namespace': 'MyNamespace', 'protected': True}) with mock.patch.object(self.gc.metadefs_namespace, 'create') as mocked_create: expect_namespace = { 'namespace': 'MyNamespace', 'protected': True } mocked_create.return_value = expect_namespace test_shell.do_md_namespace_create(self.gc, args) mocked_create.assert_called_once_with(namespace='MyNamespace', protected=True) utils.print_dict.assert_called_once_with(expect_namespace) def test_do_md_namespace_import(self): args = self._make_args({'file': 'test'}) expect_namespace = { 'namespace': 'MyNamespace', 'protected': True } with mock.patch.object(self.gc.metadefs_namespace, 'create') as mocked_create: mock_read = mock.Mock(return_value=json.dumps(expect_namespace)) mock_file = mock.Mock(read=mock_read) utils.get_data_file = mock.Mock(return_value=mock_file) mocked_create.return_value = expect_namespace test_shell.do_md_namespace_import(self.gc, args) mocked_create.assert_called_once_with(**expect_namespace) utils.print_dict.assert_called_once_with(expect_namespace) def test_do_md_namespace_import_invalid_json(self): args = self._make_args({'file': 'test'}) mock_read = mock.Mock(return_value='Invalid') mock_file = mock.Mock(read=mock_read) utils.get_data_file = mock.Mock(return_value=mock_file) self.assertRaises(SystemExit, test_shell.do_md_namespace_import, self.gc, args) def test_do_md_namespace_import_no_input(self): args = self._make_args({'file': None}) utils.get_data_file = mock.Mock(return_value=None) self.assertRaises(SystemExit, test_shell.do_md_namespace_import, self.gc, args) def test_do_md_namespace_update(self): args = self._make_args({'id': 'MyNamespace', 'protected': True}) with mock.patch.object(self.gc.metadefs_namespace, 'update') as mocked_update: expect_namespace = { 'namespace': 'MyNamespace', 'protected': True } mocked_update.return_value = expect_namespace test_shell.do_md_namespace_update(self.gc, args) mocked_update.assert_called_once_with('MyNamespace', id='MyNamespace', protected=True) utils.print_dict.assert_called_once_with(expect_namespace) def test_do_md_namespace_show(self): args = self._make_args({'namespace': 'MyNamespace', 'max_column_width': 80, 'resource_type': None}) with mock.patch.object(self.gc.metadefs_namespace, 'get') as mocked_get: expect_namespace = {'namespace': 'MyNamespace'} mocked_get.return_value = expect_namespace test_shell.do_md_namespace_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace') utils.print_dict.assert_called_once_with(expect_namespace, 80) def test_do_md_namespace_show_resource_type(self): args = self._make_args({'namespace': 'MyNamespace', 'max_column_width': 80, 'resource_type': 'RESOURCE'}) with mock.patch.object(self.gc.metadefs_namespace, 'get') as mocked_get: expect_namespace = {'namespace': 'MyNamespace'} mocked_get.return_value = expect_namespace test_shell.do_md_namespace_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace', resource_type='RESOURCE') utils.print_dict.assert_called_once_with(expect_namespace, 80) def test_do_md_namespace_list(self): args = self._make_args({'resource_type': None, 'visibility': None, 'page_size': None}) with mock.patch.object(self.gc.metadefs_namespace, 'list') as mocked_list: expect_namespaces = [{'namespace': 'MyNamespace'}] mocked_list.return_value = expect_namespaces test_shell.do_md_namespace_list(self.gc, args) mocked_list.assert_called_once_with(filters={}) utils.print_list.assert_called_once_with(expect_namespaces, ['namespace']) def test_do_md_namespace_list_page_size(self): args = self._make_args({'resource_type': None, 'visibility': None, 'page_size': 2}) with mock.patch.object(self.gc.metadefs_namespace, 'list') as mocked_list: expect_namespaces = [{'namespace': 'MyNamespace'}] mocked_list.return_value = expect_namespaces test_shell.do_md_namespace_list(self.gc, args) mocked_list.assert_called_once_with(filters={}, page_size=2) utils.print_list.assert_called_once_with(expect_namespaces, ['namespace']) def test_do_md_namespace_list_one_filter(self): args = self._make_args({'resource_types': ['OS::Compute::Aggregate'], 'visibility': None, 'page_size': None}) with mock.patch.object(self.gc.metadefs_namespace, 'list') as \ mocked_list: expect_namespaces = [{'namespace': 'MyNamespace'}] mocked_list.return_value = expect_namespaces test_shell.do_md_namespace_list(self.gc, args) mocked_list.assert_called_once_with(filters={ 'resource_types': ['OS::Compute::Aggregate']}) utils.print_list.assert_called_once_with(expect_namespaces, ['namespace']) def test_do_md_namespace_list_all_filters(self): args = self._make_args({'resource_types': ['OS::Compute::Aggregate'], 'visibility': 'public', 'page_size': None}) with mock.patch.object(self.gc.metadefs_namespace, 'list') as mocked_list: expect_namespaces = [{'namespace': 'MyNamespace'}] mocked_list.return_value = expect_namespaces test_shell.do_md_namespace_list(self.gc, args) mocked_list.assert_called_once_with(filters={ 'resource_types': ['OS::Compute::Aggregate'], 'visibility': 'public'}) utils.print_list.assert_called_once_with(expect_namespaces, ['namespace']) def test_do_md_namespace_list_unknown_filter(self): args = self._make_args({'resource_type': None, 'visibility': None, 'some_arg': 'some_value', 'page_size': None}) with mock.patch.object(self.gc.metadefs_namespace, 'list') as mocked_list: expect_namespaces = [{'namespace': 'MyNamespace'}] mocked_list.return_value = expect_namespaces test_shell.do_md_namespace_list(self.gc, args) mocked_list.assert_called_once_with(filters={}) utils.print_list.assert_called_once_with(expect_namespaces, ['namespace']) def test_do_md_namespace_delete(self): args = self._make_args({'namespace': 'MyNamespace', 'content': False}) with mock.patch.object(self.gc.metadefs_namespace, 'delete') as \ mocked_delete: test_shell.do_md_namespace_delete(self.gc, args) mocked_delete.assert_called_once_with('MyNamespace') def test_do_md_resource_type_associate(self): args = self._make_args({'namespace': 'MyNamespace', 'name': 'MyResourceType', 'prefix': 'PREFIX:'}) with mock.patch.object(self.gc.metadefs_resource_type, 'associate') as mocked_associate: expect_rt = { 'namespace': 'MyNamespace', 'name': 'MyResourceType', 'prefix': 'PREFIX:' } mocked_associate.return_value = expect_rt test_shell.do_md_resource_type_associate(self.gc, args) mocked_associate.assert_called_once_with('MyNamespace', **expect_rt) utils.print_dict.assert_called_once_with(expect_rt) def test_do_md_resource_type_deassociate(self): args = self._make_args({'namespace': 'MyNamespace', 'resource_type': 'MyResourceType'}) with mock.patch.object(self.gc.metadefs_resource_type, 'deassociate') as mocked_deassociate: test_shell.do_md_resource_type_deassociate(self.gc, args) mocked_deassociate.assert_called_once_with('MyNamespace', 'MyResourceType') def test_do_md_resource_type_list(self): args = self._make_args({}) with mock.patch.object(self.gc.metadefs_resource_type, 'list') as mocked_list: expect_objects = ['MyResourceType1', 'MyResourceType2'] mocked_list.return_value = expect_objects test_shell.do_md_resource_type_list(self.gc, args) self.assertEqual(1, mocked_list.call_count) def test_do_md_namespace_resource_type_list(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_resource_type, 'get') as mocked_get: expect_objects = [{'namespace': 'MyNamespace', 'object': 'MyObject'}] mocked_get.return_value = expect_objects test_shell.do_md_namespace_resource_type_list(self.gc, args) mocked_get.assert_called_once_with('MyNamespace') utils.print_list.assert_called_once_with(expect_objects, ['name', 'prefix', 'properties_target']) def test_do_md_property_create(self): args = self._make_args({'namespace': 'MyNamespace', 'name': "MyProperty", 'title': "Title", 'type': 'boolean', 'schema': '{}'}) with mock.patch.object(self.gc.metadefs_property, 'create') as mocked_create: expect_property = { 'namespace': 'MyNamespace', 'name': 'MyProperty', 'title': 'Title', 'type': 'boolean', } mocked_create.return_value = expect_property test_shell.do_md_property_create(self.gc, args) mocked_create.assert_called_once_with('MyNamespace', name='MyProperty', title='Title', type='boolean') utils.print_dict.assert_called_once_with(expect_property) def test_do_md_property_create_invalid_schema(self): args = self._make_args({'namespace': 'MyNamespace', 'name': "MyProperty", 'title': "Title", 'type': "boolean", 'schema': 'Invalid'}) self.assertRaises(SystemExit, test_shell.do_md_property_create, self.gc, args) def test_do_md_property_update(self): args = self._make_args({'namespace': 'MyNamespace', 'property': 'MyProperty', 'name': 'NewName', 'title': "Title", 'schema': '{}'}) with mock.patch.object(self.gc.metadefs_property, 'update') as mocked_update: expect_property = { 'namespace': 'MyNamespace', 'name': 'MyProperty', 'title': 'Title' } mocked_update.return_value = expect_property test_shell.do_md_property_update(self.gc, args) mocked_update.assert_called_once_with('MyNamespace', 'MyProperty', name='NewName', title='Title') utils.print_dict.assert_called_once_with(expect_property) def test_do_md_property_update_invalid_schema(self): args = self._make_args({'namespace': 'MyNamespace', 'property': 'MyProperty', 'name': "MyObject", 'title': "Title", 'schema': 'Invalid'}) self.assertRaises(SystemExit, test_shell.do_md_property_update, self.gc, args) def test_do_md_property_show(self): args = self._make_args({'namespace': 'MyNamespace', 'property': 'MyProperty', 'max_column_width': 80}) with mock.patch.object(self.gc.metadefs_property, 'get') as mocked_get: expect_property = { 'namespace': 'MyNamespace', 'property': 'MyProperty', 'title': 'Title' } mocked_get.return_value = expect_property test_shell.do_md_property_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace', 'MyProperty') utils.print_dict.assert_called_once_with(expect_property, 80) def test_do_md_property_delete(self): args = self._make_args({'namespace': 'MyNamespace', 'property': 'MyProperty'}) with mock.patch.object(self.gc.metadefs_property, 'delete') as mocked_delete: test_shell.do_md_property_delete(self.gc, args) mocked_delete.assert_called_once_with('MyNamespace', 'MyProperty') def test_do_md_namespace_property_delete(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_property, 'delete_all') as mocked_delete_all: test_shell.do_md_namespace_properties_delete(self.gc, args) mocked_delete_all.assert_called_once_with('MyNamespace') def test_do_md_property_list(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_property, 'list') as mocked_list: expect_objects = [{'namespace': 'MyNamespace', 'property': 'MyProperty', 'title': 'MyTitle'}] mocked_list.return_value = expect_objects test_shell.do_md_property_list(self.gc, args) mocked_list.assert_called_once_with('MyNamespace') utils.print_list.assert_called_once_with(expect_objects, ['name', 'title', 'type']) def test_do_md_object_create(self): args = self._make_args({'namespace': 'MyNamespace', 'name': "MyObject", 'schema': '{}'}) with mock.patch.object(self.gc.metadefs_object, 'create') as mocked_create: expect_object = { 'namespace': 'MyNamespace', 'name': 'MyObject' } mocked_create.return_value = expect_object test_shell.do_md_object_create(self.gc, args) mocked_create.assert_called_once_with('MyNamespace', name='MyObject') utils.print_dict.assert_called_once_with(expect_object) def test_do_md_object_create_invalid_schema(self): args = self._make_args({'namespace': 'MyNamespace', 'name': "MyObject", 'schema': 'Invalid'}) self.assertRaises(SystemExit, test_shell.do_md_object_create, self.gc, args) def test_do_md_object_update(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject', 'name': 'NewName', 'schema': '{}'}) with mock.patch.object(self.gc.metadefs_object, 'update') as mocked_update: expect_object = { 'namespace': 'MyNamespace', 'name': 'MyObject' } mocked_update.return_value = expect_object test_shell.do_md_object_update(self.gc, args) mocked_update.assert_called_once_with('MyNamespace', 'MyObject', name='NewName') utils.print_dict.assert_called_once_with(expect_object) def test_do_md_object_update_invalid_schema(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject', 'name': "MyObject", 'schema': 'Invalid'}) self.assertRaises(SystemExit, test_shell.do_md_object_update, self.gc, args) def test_do_md_object_show(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject', 'max_column_width': 80}) with mock.patch.object(self.gc.metadefs_object, 'get') as mocked_get: expect_object = { 'namespace': 'MyNamespace', 'object': 'MyObject' } mocked_get.return_value = expect_object test_shell.do_md_object_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace', 'MyObject') utils.print_dict.assert_called_once_with(expect_object, 80) def test_do_md_object_property_show(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject', 'property': 'MyProperty', 'max_column_width': 80}) with mock.patch.object(self.gc.metadefs_object, 'get') as mocked_get: expect_object = {'name': 'MyObject', 'properties': { 'MyProperty': {'type': 'string'} }} mocked_get.return_value = expect_object test_shell.do_md_object_property_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace', 'MyObject') utils.print_dict.assert_called_once_with({'type': 'string', 'name': 'MyProperty'}, 80) def test_do_md_object_property_show_non_existing(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject', 'property': 'MyProperty', 'max_column_width': 80}) with mock.patch.object(self.gc.metadefs_object, 'get') as mocked_get: expect_object = {'name': 'MyObject', 'properties': {}} mocked_get.return_value = expect_object self.assertRaises(SystemExit, test_shell.do_md_object_property_show, self.gc, args) mocked_get.assert_called_once_with('MyNamespace', 'MyObject') def test_do_md_object_delete(self): args = self._make_args({'namespace': 'MyNamespace', 'object': 'MyObject'}) with mock.patch.object(self.gc.metadefs_object, 'delete') as mocked_delete: test_shell.do_md_object_delete(self.gc, args) mocked_delete.assert_called_once_with('MyNamespace', 'MyObject') def test_do_md_namespace_objects_delete(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_object, 'delete_all') as mocked_delete_all: test_shell.do_md_namespace_objects_delete(self.gc, args) mocked_delete_all.assert_called_once_with('MyNamespace') def test_do_md_object_list(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_object, 'list') as mocked_list: expect_objects = [{'namespace': 'MyNamespace', 'object': 'MyObject'}] mocked_list.return_value = expect_objects test_shell.do_md_object_list(self.gc, args) mocked_list.assert_called_once_with('MyNamespace') utils.print_list.assert_called_once_with( expect_objects, ['name', 'description'], field_settings={ 'description': {'align': 'l', 'max_width': 50}}) def test_do_md_tag_create(self): args = self._make_args({'namespace': 'MyNamespace', 'name': 'MyTag'}) with mock.patch.object(self.gc.metadefs_tag, 'create') as mocked_create: expect_tag = { 'namespace': 'MyNamespace', 'name': 'MyTag' } mocked_create.return_value = expect_tag test_shell.do_md_tag_create(self.gc, args) mocked_create.assert_called_once_with('MyNamespace', 'MyTag') utils.print_dict.assert_called_once_with(expect_tag) def test_do_md_tag_update(self): args = self._make_args({'namespace': 'MyNamespace', 'tag': 'MyTag', 'name': 'NewTag'}) with mock.patch.object(self.gc.metadefs_tag, 'update') as mocked_update: expect_tag = { 'namespace': 'MyNamespace', 'name': 'NewTag' } mocked_update.return_value = expect_tag test_shell.do_md_tag_update(self.gc, args) mocked_update.assert_called_once_with('MyNamespace', 'MyTag', name='NewTag') utils.print_dict.assert_called_once_with(expect_tag) def test_do_md_tag_show(self): args = self._make_args({'namespace': 'MyNamespace', 'tag': 'MyTag', 'sort_dir': 'desc'}) with mock.patch.object(self.gc.metadefs_tag, 'get') as mocked_get: expect_tag = { 'namespace': 'MyNamespace', 'tag': 'MyTag' } mocked_get.return_value = expect_tag test_shell.do_md_tag_show(self.gc, args) mocked_get.assert_called_once_with('MyNamespace', 'MyTag') utils.print_dict.assert_called_once_with(expect_tag) def test_do_md_tag_delete(self): args = self._make_args({'namespace': 'MyNamespace', 'tag': 'MyTag'}) with mock.patch.object(self.gc.metadefs_tag, 'delete') as mocked_delete: test_shell.do_md_tag_delete(self.gc, args) mocked_delete.assert_called_once_with('MyNamespace', 'MyTag') def test_do_md_namespace_tags_delete(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_tag, 'delete_all') as mocked_delete_all: test_shell.do_md_namespace_tags_delete(self.gc, args) mocked_delete_all.assert_called_once_with('MyNamespace') def test_do_md_tag_list(self): args = self._make_args({'namespace': 'MyNamespace'}) with mock.patch.object(self.gc.metadefs_tag, 'list') as mocked_list: expect_tags = [{'namespace': 'MyNamespace', 'tag': 'MyTag'}] mocked_list.return_value = expect_tags test_shell.do_md_tag_list(self.gc, args) mocked_list.assert_called_once_with('MyNamespace') utils.print_list.assert_called_once_with( expect_tags, ['name'], field_settings={ 'description': {'align': 'l', 'max_width': 50}}) def test_do_md_tag_create_multiple(self): args = self._make_args({'namespace': 'MyNamespace', 'delim': ',', 'names': 'MyTag1, MyTag2', 'append': False}) with mock.patch.object( self.gc.metadefs_tag, 'create_multiple') as mocked_create_tags: expect_tags = [{'tags': [{'name': 'MyTag1'}, {'name': 'MyTag2'}]}] mocked_create_tags.return_value = expect_tags test_shell.do_md_tag_create_multiple(self.gc, args) mocked_create_tags.assert_called_once_with( 'MyNamespace', tags=['MyTag1', 'MyTag2'], append=False) utils.print_list.assert_called_once_with( expect_tags, ['name'], field_settings={ 'description': {'align': 'l', 'max_width': 50}}) def test_do_md_tag_create_multiple_with_append(self): args = self._make_args({'namespace': 'MyNamespace', 'delim': ',', 'names': 'MyTag1, MyTag2', 'append': True}) with mock.patch.object( self.gc.metadefs_tag, 'create_multiple') as mocked_create_tags: expect_tags = [{'tags': [{'name': 'MyTag1'}, {'name': 'MyTag2'}]}] mocked_create_tags.return_value = expect_tags test_shell.do_md_tag_create_multiple(self.gc, args) mocked_create_tags.assert_called_once_with( 'MyNamespace', tags=['MyTag1', 'MyTag2'], append=True) utils.print_list.assert_called_once_with( expect_tags, ['name'], field_settings={ 'description': {'align': 'l', 'max_width': 50}}) def _test_do_cache_list(self, supported=True): args = self._make_args({}) expected_output = { "cached_images": [ { "image_id": "pass", "last_accessed": 0, "last_modified": 0, "size": "fake_size", "hits": "fake_hits", } ], "queued_images": ['fake_image'] } with mock.patch.object(self.gc.cache, 'list') as mocked_cache_list: if supported: mocked_cache_list.return_value = expected_output else: mocked_cache_list.side_effect = exc.HTTPNotImplemented test_shell.do_cache_list(self.gc, args) mocked_cache_list.assert_called() if supported: utils.print_cached_images.assert_called_once_with( expected_output) def test_do_cache_list(self): self._test_do_cache_list() def test_do_cache_list_unsupported(self): self.assertRaises(exc.HTTPNotImplemented, self._test_do_cache_list, supported=False) def test_do_cache_list_endpoint_not_provided(self): args = self._make_args({}) self.gc.endpoint_provided = False with mock.patch('glanceclient.common.utils.exit') as mock_exit: test_shell.do_cache_list(self.gc, args) mock_exit.assert_called_once_with( 'Direct server endpoint needs to be provided. Do ' 'not use loadbalanced or catalog endpoints.') def _test_cache_queue(self, supported=True, forbidden=False,): args = argparse.Namespace(id=['image1']) with mock.patch.object(self.gc.cache, 'queue') as mocked_cache_queue: if supported: mocked_cache_queue.return_value = None else: mocked_cache_queue.side_effect = exc.HTTPNotImplemented if forbidden: mocked_cache_queue.side_effect = exc.HTTPForbidden test_shell.do_cache_queue(self.gc, args) if supported: mocked_cache_queue.assert_called_once_with('image1') def test_do_cache_queue(self): self._test_cache_queue() def test_do_cache_queue_unsupported(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_queue(supported=False) mock_print_err.assert_called_once_with( "'HTTP HTTPNotImplemented': Unable to queue image " "'image1' for caching.") def test_do_cache_queue_forbidden(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_queue(forbidden=True) mock_print_err.assert_called_once_with( "You are not permitted to queue the image 'image1' for " "caching.") def test_do_cache_queue_endpoint_not_provided(self): args = argparse.Namespace(id=['image1']) self.gc.endpoint_provided = False with mock.patch('glanceclient.common.utils.exit') as mock_exit: test_shell.do_cache_queue(self.gc, args) mock_exit.assert_called_once_with( 'Direct server endpoint needs to be provided. Do ' 'not use loadbalanced or catalog endpoints.') def _test_cache_delete(self, supported=True, forbidden=False,): args = argparse.Namespace(id=['image1']) with mock.patch.object(self.gc.cache, 'delete') as mocked_cache_delete: if supported: mocked_cache_delete.return_value = None else: mocked_cache_delete.side_effect = exc.HTTPNotImplemented if forbidden: mocked_cache_delete.side_effect = exc.HTTPForbidden test_shell.do_cache_delete(self.gc, args) if supported: mocked_cache_delete.assert_called_once_with('image1') def test_do_cache_delete(self): self._test_cache_delete() def test_do_cache_delete_unsupported(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_delete(supported=False) mock_print_err.assert_called_once_with( "'HTTP HTTPNotImplemented': Unable to delete image " "'image1' from cache.") def test_do_cache_delete_forbidden(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_delete(forbidden=True) mock_print_err.assert_called_once_with( "You are not permitted to " "delete the image 'image1' from cache.") def test_do_cache_delete_endpoint_not_provided(self): args = argparse.Namespace(id=['image1']) self.gc.endpoint_provided = False with mock.patch('glanceclient.common.utils.exit') as mock_exit: test_shell.do_cache_delete(self.gc, args) mock_exit.assert_called_once_with( 'Direct server endpoint needs to be provided. Do ' 'not use loadbalanced or catalog endpoints.') def _test_cache_clear(self, target='both', supported=True, forbidden=False,): args = self._make_args({'target': target}) with mock.patch.object(self.gc.cache, 'clear') as mocked_cache_clear: if supported: mocked_cache_clear.return_value = None else: mocked_cache_clear.side_effect = exc.HTTPNotImplemented if forbidden: mocked_cache_clear.side_effect = exc.HTTPForbidden test_shell.do_cache_clear(self.gc, args) if supported: mocked_cache_clear.mocked_cache_clear(target) def test_do_cache_clear_all(self): self._test_cache_clear() def test_do_cache_clear_queued_only(self): self._test_cache_clear(target='queue') def test_do_cache_clear_cached_only(self): self._test_cache_clear(target='cache') def test_do_cache_clear_unsupported(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_clear(supported=False) mock_print_err.assert_called_once_with( "'HTTP HTTPNotImplemented': Unable to delete image(s) " "from cache.") def test_do_cache_clear_forbidden(self): with mock.patch( 'glanceclient.common.utils.print_err') as mock_print_err: self._test_cache_clear(forbidden=True) mock_print_err.assert_called_once_with( "You are not permitted to " "delete image(s) from cache.") def test_do_cache_clear_endpoint_not_provided(self): args = self._make_args({'target': 'both'}) self.gc.endpoint_provided = False with mock.patch('glanceclient.common.utils.exit') as mock_exit: test_shell.do_cache_clear(self.gc, args) mock_exit.assert_called_once_with( 'Direct server endpoint needs to be provided. Do ' 'not use loadbalanced or catalog endpoints.')