Image create and set command updates and tests

Refactor image create and set commands to properly handle properties.
This is particularly tricky with exclusive booleans as in this case leaving
both choices off the command line should NOT assume a default value but
leave the existing value unchanged.

Properties were not being updated correctly in the 'image set' command.
Refactor it to use the same pattern as in other SetXxx commands.
Add tests for arg handling.

Change-Id: I123a64c9b4feecab25a3e2013cc047f55b1c9967
This commit is contained in:
Dean Troyer 2014-05-08 11:09:22 -05:00 committed by Steve Martinelli
parent aba1fb2268
commit 6380b8b959
3 changed files with 592 additions and 205 deletions

View File

@ -35,6 +35,10 @@ from openstackclient.common import parseractions
from openstackclient.common import utils from openstackclient.common import utils
DEFAULT_CONTAINER_FORMAT = 'bare'
DEFAULT_DISK_FORMAT = 'raw'
class CreateImage(show.ShowOne): class CreateImage(show.ShowOne):
"""Create/upload an image""" """Create/upload an image"""
@ -45,173 +49,205 @@ class CreateImage(show.ShowOne):
parser.add_argument( parser.add_argument(
"name", "name",
metavar="<name>", metavar="<name>",
help="Name of image", help="New image name",
)
parser.add_argument(
"--disk-format",
default="raw",
metavar="<disk-format>",
help="Disk format of image",
) )
parser.add_argument( parser.add_argument(
"--id", "--id",
metavar="<id>", metavar="<id>",
help="ID of image to reserve", help="Image ID to reserve",
) )
parser.add_argument( parser.add_argument(
"--store", "--store",
metavar="<store>", metavar="<store>",
help="Store to upload image to", help="Upload image to this store",
) )
parser.add_argument( parser.add_argument(
"--container-format", "--container-format",
default="bare", default=DEFAULT_CONTAINER_FORMAT,
metavar="<container-format>", metavar="<container-format>",
help="Container format of image", help="Image container format "
"(default: %s)" % DEFAULT_CONTAINER_FORMAT,
)
parser.add_argument(
"--disk-format",
default=DEFAULT_DISK_FORMAT,
metavar="<disk-format>",
help="Image disk format "
"(default: %s)" % DEFAULT_DISK_FORMAT,
) )
parser.add_argument( parser.add_argument(
"--owner", "--owner",
metavar="<project>", metavar="<project>",
help="Image owner (project name or ID)", help="Image owner project name or ID",
) )
parser.add_argument( parser.add_argument(
"--size", "--size",
metavar="<size>", metavar="<size>",
help="Size of image in bytes. Only used with --location and" help="Image size, in bytes (only used with --location and"
" --copy-from", " --copy-from)",
) )
parser.add_argument( parser.add_argument(
"--min-disk", "--min-disk",
metavar="<disk-gb>", metavar="<disk-gb>",
help="Minimum size of disk needed to boot image in gigabytes", type=int,
help="Minimum disk size needed to boot image, in gigabytes",
) )
parser.add_argument( parser.add_argument(
"--min-ram", "--min-ram",
metavar="<disk-ram>", metavar="<ram-mb>",
help="Minimum amount of ram needed to boot image in megabytes", type=int,
help="Minimum RAM size needed to boot image, in megabytes",
) )
parser.add_argument( parser.add_argument(
"--location", "--location",
metavar="<image-url>", metavar="<image-url>",
help="URL where the data for this image already resides", help="Download image from an existing URL",
)
parser.add_argument(
"--file",
metavar="<file>",
help="Local file that contains disk image",
)
parser.add_argument(
"--checksum",
metavar="<checksum>",
help="Hash of image data used for verification",
) )
parser.add_argument( parser.add_argument(
"--copy-from", "--copy-from",
metavar="<image-url>", metavar="<image-url>",
help="Similar to --location, but this indicates that the image" help="Copy image from the data store (similar to --location)",
" should immediately be copied from the data store", )
parser.add_argument(
"--file",
metavar="<file>",
help="Upload image from local file",
) )
parser.add_argument( parser.add_argument(
"--volume", "--volume",
metavar="<volume>", metavar="<volume>",
help="Create the image from the specified volume", help="Create image from a volume",
) )
parser.add_argument( parser.add_argument(
"--force", "--force",
dest='force', dest='force',
action='store_true', action='store_true',
default=False, default=False,
help="If the image is created from a volume, force creation of the" help="Force image creation if volume is in use "
" image even if volume is in use.", "(only meaningful with --volume)",
)
parser.add_argument(
"--checksum",
metavar="<checksum>",
help="Image hash used for verification",
)
protected_group = parser.add_mutually_exclusive_group()
protected_group.add_argument(
"--protected",
action="store_true",
help="Prevent image from being deleted",
)
protected_group.add_argument(
"--unprotected",
action="store_true",
help="Allow image to be deleted (default)",
)
public_group = parser.add_mutually_exclusive_group()
public_group.add_argument(
"--public",
action="store_true",
help="Image is accessible to the public",
)
public_group.add_argument(
"--private",
action="store_true",
help="Image is inaccessible to the public (default)",
) )
parser.add_argument( parser.add_argument(
"--property", "--property",
dest="properties", dest="properties",
metavar="<key=value>", metavar="<key=value>",
action=parseractions.KeyValueAction, action=parseractions.KeyValueAction,
help="Set property on this image " help="Set an image property "
'(repeat option to set multiple properties)', "(repeat option to set multiple properties)",
)
protected_group = parser.add_mutually_exclusive_group()
protected_group.add_argument(
"--protected",
dest="protected",
action="store_true",
help="Prevent image from being deleted (default: False)",
)
protected_group.add_argument(
"--unprotected",
dest="protected",
action="store_false",
default=False,
help="Allow images to be deleted (default: True)",
)
public_group = parser.add_mutually_exclusive_group()
public_group.add_argument(
"--public",
dest="is_public",
action="store_true",
default=True,
help="Image is accessible to the public (default)",
)
public_group.add_argument(
"--private",
dest="is_public",
action="store_false",
help="Image is inaccessible to the public",
) )
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
self.log.debug("take_action(%s)", parsed_args) self.log.debug("take_action(%s)", parsed_args)
image_client = self.app.client_manager.image
# NOTE(jk0): Since create() takes kwargs, it's easiest to just make a # Build an attribute dict from the parsed args, only include
# copy of parsed_args and remove what we don't need. # attributes that were actually set on the command line
args = vars(parsed_args) kwargs = {}
args = dict(filter(lambda x: x[1] is not None, args.items())) copy_attrs = ('name', 'id', 'store', 'container_format',
args.pop("columns") 'disk_format', 'owner', 'size', 'min_disk', 'min_ram',
args.pop("formatter") 'localtion', 'copy_from', 'volume', 'force',
args.pop("prefix") 'checksum', 'properties')
args.pop("variables") for attr in copy_attrs:
if attr in parsed_args:
val = getattr(parsed_args, attr, None)
if val:
# Only include a value in kwargs for attributes that are
# actually present on the command line
kwargs[attr] = val
# Handle exclusive booleans with care
# Avoid including attributes in kwargs if an option is not
# present on the command line. These exclusive booleans are not
# a single value for the pair of options because the default must be
# to do nothing when no options are present as opposed to always
# setting a default.
if parsed_args.protected:
kwargs['protected'] = True
if parsed_args.unprotected:
kwargs['protected'] = False
if parsed_args.public:
kwargs['is_public'] = True
if parsed_args.private:
kwargs['is_public'] = False
if "location" not in args and "copy_from" not in args: if not parsed_args.location and not parsed_args.copy_from:
if "volume" in args: if parsed_args.volume:
pass
elif "file" in args:
args["data"] = open(args.pop("file"), "rb")
else:
args["data"] = None
if sys.stdin.isatty() is not True:
if msvcrt:
msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
args["data"] = sys.stdin
if "volume" in args:
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.volume
source_volume = utils.find_resource(volume_client.volumes, source_volume = utils.find_resource(
parsed_args.volume) volume_client.volumes,
parsed_args.volume,
)
response, body = volume_client.volumes.upload_to_image( response, body = volume_client.volumes.upload_to_image(
source_volume, source_volume.id,
parsed_args.force, parsed_args.force,
parsed_args.name, parsed_args.name,
parsed_args.container_format, parsed_args.container_format,
parsed_args.disk_format) parsed_args.disk_format,
)
info = body['os-volume_upload_image'] info = body['os-volume_upload_image']
elif parsed_args.file:
# Send an open file handle to glanceclient so it will
# do a chunked transfer
kwargs["data"] = open(parsed_args.file, "rb")
else: else:
image_client = self.app.client_manager.image # Read file from stdin
kwargs["data"] = None
if sys.stdin.isatty() is not True:
if msvcrt:
msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
# Send an open file handle to glanceclient so it will
# do a chunked transfer
kwargs["data"] = sys.stdin
try: try:
image = utils.find_resource( image = utils.find_resource(
image_client.images, image_client.images,
parsed_args.name, parsed_args.name,
) )
# Preserve previous properties if any are being set now
if image.properties:
if parsed_args.properties:
image.properties.update(kwargs['properties'])
kwargs['properties'] = image.properties
except exceptions.CommandError: except exceptions.CommandError:
if not parsed_args.volume:
# This is normal for a create or reserve (create w/o an image) # This is normal for a create or reserve (create w/o an image)
image = image_client.images.create(**args) # But skip for create from volume
image = image_client.images.create(**kwargs)
else: else:
# It must be an update # Update an existing reservation
# If an image is specified via --file, --location or # If an image is specified via --file, --location or
# --copy-from let the API handle it # --copy-from let the API handle it
image = image_client.images.update(image, **args) image = image_client.images.update(image.id, **kwargs)
info = {} info = {}
info.update(image._info) info.update(image._info)
@ -314,88 +350,104 @@ class SetImage(show.ShowOne):
parser.add_argument( parser.add_argument(
"image", "image",
metavar="<image>", metavar="<image>",
help="Name or ID of image to change", help="Image name or ID to change",
) )
parser.add_argument( parser.add_argument(
"--name", "--name",
metavar="<name>", metavar="<name>",
help="Name of image", help="New image name",
) )
parser.add_argument( parser.add_argument(
"--owner", "--owner",
metavar="<project>", metavar="<project>",
help="Image owner (project name or ID)", help="New image owner project name or ID",
) )
parser.add_argument( parser.add_argument(
"--min-disk", "--min-disk",
metavar="<disk-gb>", metavar="<disk-gb>",
help="Minimum size of disk needed to boot image in gigabytes", type=int,
help="Minimum disk size needed to boot image, in gigabytes",
) )
parser.add_argument( parser.add_argument(
"--min-ram", "--min-ram",
metavar="<disk-ram>", metavar="<disk-ram>",
help="Minimum amount of ram needed to boot image in megabytes", type=int,
help="Minimum RAM size needed to boot image, in megabytes",
)
protected_group = parser.add_mutually_exclusive_group()
protected_group.add_argument(
"--protected",
action="store_true",
help="Prevent image from being deleted",
)
protected_group.add_argument(
"--unprotected",
action="store_true",
help="Allow image to be deleted (default)",
)
public_group = parser.add_mutually_exclusive_group()
public_group.add_argument(
"--public",
action="store_true",
help="Image is accessible to the public",
)
public_group.add_argument(
"--private",
action="store_true",
help="Image is inaccessible to the public (default)",
) )
parser.add_argument( parser.add_argument(
"--property", "--property",
dest="properties", dest="properties",
metavar="<key=value>", metavar="<key=value>",
default={},
action=parseractions.KeyValueAction, action=parseractions.KeyValueAction,
help="Set property on this image " help="Set an image property "
'(repeat option to set multiple properties)', "(repeat option to set multiple properties)",
)
protected_group = parser.add_mutually_exclusive_group()
protected_group.add_argument(
"--protected",
dest="protected",
action="store_true",
help="Prevent image from being deleted (default: False)",
)
protected_group.add_argument(
"--unprotected",
dest="protected",
action="store_false",
default=False,
help="Allow images to be deleted (default: True)",
)
public_group = parser.add_mutually_exclusive_group()
public_group.add_argument(
"--public",
dest="is_public",
action="store_true",
default=True,
help="Image is accessible to the public (default)",
)
public_group.add_argument(
"--private",
dest="is_public",
action="store_false",
help="Image is inaccessible to the public",
) )
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
self.log.debug("take_action(%s)", parsed_args) self.log.debug("take_action(%s)", parsed_args)
# NOTE(jk0): Since create() takes kwargs, it's easiest to just make a
# copy of parsed_args and remove what we don't need.
args = vars(parsed_args)
args = dict(filter(lambda x: x[1] is not None, args.items()))
args.pop("columns")
args.pop("formatter")
args.pop("prefix")
args.pop("variables")
image_arg = args.pop("image")
image_client = self.app.client_manager.image image_client = self.app.client_manager.image
kwargs = {}
copy_attrs = ('name', 'owner', 'min_disk', 'min_ram', 'properties')
for attr in copy_attrs:
if attr in parsed_args:
val = getattr(parsed_args, attr, None)
if val:
# Only include a value in kwargs for attributes that are
# actually present on the command line
kwargs[attr] = val
# Handle exclusive booleans with care
# Avoid including attributes in kwargs if an option is not
# present on the command line. These exclusive booleans are not
# a single value for the pair of options because the default must be
# to do nothing when no options are present as opposed to always
# setting a default.
if parsed_args.protected:
kwargs['protected'] = True
if parsed_args.unprotected:
kwargs['protected'] = False
if parsed_args.public:
kwargs['is_public'] = True
if parsed_args.private:
kwargs['is_public'] = False
if not kwargs:
self.log.warning('no arguments specified')
return {}, {}
image = utils.find_resource( image = utils.find_resource(
image_client.images, image_client.images,
image_arg, parsed_args.image,
) )
# Merge properties
args["properties"].update(image.properties) if image.properties and parsed_args.properties:
image = image_client.images.update(image, **args) image.properties.update(kwargs['properties'])
kwargs['properties'] = image.properties
image = image_client.images.update(image.id, **kwargs)
info = {} info = {}
info.update(image._info) info.update(image._info)

View File

@ -17,16 +17,38 @@ import mock
from openstackclient.tests import fakes from openstackclient.tests import fakes
from openstackclient.tests import utils from openstackclient.tests import utils
from openstackclient.tests.volume.v1 import fakes as volume_fakes
image_id = 'im1' image_id = 'im1'
image_name = 'graven' image_name = 'graven'
image_owner = 'baal'
image_protected = False
image_public = True
image_properties = {
'Alpha': 'a',
'Beta': 'b',
'Gamma': 'g',
}
image_properties_str = "{'Alpha': 'a', 'Beta': 'b', 'Gamma': 'g'}"
image_data = 'line 1\nline 2\n'
IMAGE = { IMAGE = {
'id': image_id, 'id': image_id,
'name': image_name 'name': image_name,
'container_format': '',
'disk_format': '',
'owner': image_owner,
'min_disk': 0,
'min_ram': 0,
'is_public': image_public,
'protected': image_protected,
'properties': image_properties,
} }
IMAGE_columns = tuple(sorted(IMAGE))
IMAGE_data = tuple((IMAGE[x] for x in sorted(IMAGE)))
class FakeImagev1Client(object): class FakeImagev1Client(object):
def __init__(self, **kwargs): def __init__(self, **kwargs):
@ -44,3 +66,7 @@ class TestImagev1(utils.TestCommand):
endpoint=fakes.AUTH_URL, endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN, token=fakes.AUTH_TOKEN,
) )
self.app.client_manager.volume = volume_fakes.FakeVolumev1Client(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,
)

View File

@ -16,6 +16,7 @@
import copy import copy
import mock import mock
from openstackclient.common import exceptions
from openstackclient.image.v1 import image from openstackclient.image.v1 import image
from openstackclient.tests import fakes from openstackclient.tests import fakes
from openstackclient.tests.image.v1 import fakes as image_fakes from openstackclient.tests.image.v1 import fakes as image_fakes
@ -35,75 +36,228 @@ class TestImageCreate(TestImage):
def setUp(self): def setUp(self):
super(TestImageCreate, self).setUp() super(TestImageCreate, self).setUp()
self.images_mock.create.return_value = fakes.FakeResource(
None,
copy.deepcopy(image_fakes.IMAGE),
loaded=True,
)
# This is the return value for utils.find_resource()
self.images_mock.get.return_value = fakes.FakeResource( self.images_mock.get.return_value = fakes.FakeResource(
None, None,
copy.deepcopy(image_fakes.IMAGE), copy.deepcopy(image_fakes.IMAGE),
loaded=True, loaded=True,
) )
self.images_mock.update.return_value = fakes.FakeResource(
None,
copy.deepcopy(image_fakes.IMAGE),
loaded=True,
)
# Get the command object to test
self.cmd = image.CreateImage(self.app, None) self.cmd = image.CreateImage(self.app, None)
def test_create_volume(self): def test_image_reserve_no_options(self):
mock_exception = {
'find.side_effect': exceptions.CommandError('x'),
'get.side_effect': exceptions.CommandError('x'),
}
self.images_mock.configure_mock(**mock_exception)
arglist = [ arglist = [
'--volume', 'volly',
image_fakes.image_name, image_fakes.image_name,
] ]
verifylist = [ verifylist = [
('volume', 'volly'), ('container_format', image.DEFAULT_CONTAINER_FORMAT),
('disk_format', image.DEFAULT_DISK_FORMAT),
('name', image_fakes.image_name), ('name', image_fakes.image_name),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.app.client_manager.volume = mock.Mock()
self.app.client_manager.volume.volumes = mock.Mock() # DisplayCommandBase.take_action() returns two tuples
volumes = self.app.client_manager.volume.volumes columns, data = self.cmd.take_action(parsed_args)
volumes.upload_to_image = mock.Mock()
response = {"id": 'volume_id', # ImageManager.create(name=, **)
self.images_mock.create.assert_called_with(
name=image_fakes.image_name,
container_format=image.DEFAULT_CONTAINER_FORMAT,
disk_format=image.DEFAULT_DISK_FORMAT,
data=mock.ANY,
)
# Verify update() was not called, if it was show the args
self.assertEqual(self.images_mock.update.call_args_list, [])
self.assertEqual(image_fakes.IMAGE_columns, columns)
self.assertEqual(image_fakes.IMAGE_data, data)
def test_image_reserve_options(self):
mock_exception = {
'find.side_effect': exceptions.CommandError('x'),
'get.side_effect': exceptions.CommandError('x'),
}
self.images_mock.configure_mock(**mock_exception)
arglist = [
'--container-format', 'ovf',
'--disk-format', 'fs',
'--min-disk', '10',
'--min-ram', '4',
'--protected',
'--private',
image_fakes.image_name,
]
verifylist = [
('container_format', 'ovf'),
('disk_format', 'fs'),
('min_disk', 10),
('min_ram', 4),
('protected', True),
('unprotected', False),
('public', False),
('private', True),
('name', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
# ImageManager.create(name=, **)
self.images_mock.create.assert_called_with(
name=image_fakes.image_name,
container_format='ovf',
disk_format='fs',
min_disk=10,
min_ram=4,
protected=True,
is_public=False,
data=mock.ANY,
)
# Verify update() was not called, if it was show the args
self.assertEqual(self.images_mock.update.call_args_list, [])
self.assertEqual(image_fakes.IMAGE_columns, columns)
self.assertEqual(image_fakes.IMAGE_data, data)
@mock.patch('__builtin__.open')
def test_image_create_file(self, open_mock):
mock_exception = {
'find.side_effect': exceptions.CommandError('x'),
'get.side_effect': exceptions.CommandError('x'),
}
self.images_mock.configure_mock(**mock_exception)
open_mock.return_value = image_fakes.image_data
arglist = [
'--file', 'filer',
'--unprotected',
'--public',
'--property', 'Alpha=1',
'--property', 'Beta=2',
image_fakes.image_name,
]
verifylist = [
('file', 'filer'),
('protected', False),
('unprotected', True),
('public', True),
('private', False),
('properties', {'Alpha': '1', 'Beta': '2'}),
('name', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
open_mock.assert_called_with('filer', 'rb')
# ImageManager.get(name)
self.images_mock.get.assert_called_with(image_fakes.image_name)
# ImageManager.create(name=, **)
self.images_mock.create.assert_called_with(
name=image_fakes.image_name,
container_format=image.DEFAULT_CONTAINER_FORMAT,
disk_format=image.DEFAULT_DISK_FORMAT,
protected=False,
is_public=True,
properties={
'Alpha': '1',
'Beta': '2',
},
data=image_fakes.image_data,
)
# Verify update() was not called, if it was show the args
self.assertEqual(self.images_mock.update.call_args_list, [])
self.assertEqual(image_fakes.IMAGE_columns, columns)
self.assertEqual(image_fakes.IMAGE_data, data)
def test_image_create_volume(self):
# Set up VolumeManager Mock
volumes_mock = self.app.client_manager.volume.volumes
volumes_mock.reset_mock()
volumes_mock.get.return_value = fakes.FakeResource(
None,
copy.deepcopy({'id': 'vol1', 'name': 'volly'}),
loaded=True,
)
response = {
"id": 'volume_id',
"updated_at": 'updated_at', "updated_at": 'updated_at',
"status": 'uploading', "status": 'uploading',
"display_description": 'desc', "display_description": 'desc',
"size": 'size', "size": 'size',
"volume_type": 'volume_type', "volume_type": 'volume_type',
"image_id": 'image1', "image_id": 'image1',
"container_format": parsed_args.container_format, "container_format": image.DEFAULT_CONTAINER_FORMAT,
"disk_format": parsed_args.disk_format, "disk_format": image.DEFAULT_DISK_FORMAT,
"image_name": parsed_args.name} "image_name": image_fakes.image_name,
}
full_response = {"os-volume_upload_image": response} full_response = {"os-volume_upload_image": response}
volumes.upload_to_image.return_value = (201, full_response) volumes_mock.upload_to_image.return_value = (201, full_response)
volume_resource = fakes.FakeResource(
None, arglist = [
copy.deepcopy({'id': 'vol1', 'name': 'volly'}), '--volume', 'volly',
loaded=True, image_fakes.image_name,
) ]
volumes.get.return_value = volume_resource verifylist = [
results = self.cmd.take_action(parsed_args) ('private', False),
volumes.upload_to_image.assert_called_with( ('protected', False),
volume_resource, ('public', False),
('unprotected', False),
('volume', 'volly'),
('force', False),
('name', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
# VolumeManager.upload_to_image(volume, force, image_name,
# container_format, disk_format)
volumes_mock.upload_to_image.assert_called_with(
'vol1',
False, False,
image_fakes.image_name, image_fakes.image_name,
'bare', 'bare',
'raw', 'raw',
) )
expects = [('container_format',
'disk_format', # ImageManager.update(image_id, remove_props=, **)
'display_description', self.images_mock.update.assert_called_with(
'id', image_fakes.image_id,
'image_id', name=image_fakes.image_name,
'image_name', container_format=image.DEFAULT_CONTAINER_FORMAT,
'size', disk_format=image.DEFAULT_DISK_FORMAT,
'status', properties=image_fakes.image_properties,
'updated_at', volume='volly',
'volume_type'), )
('bare',
'raw', self.assertEqual(image_fakes.IMAGE_columns, columns)
'desc', self.assertEqual(image_fakes.IMAGE_data, data)
'volume_id',
'image1',
'graven',
'size',
'uploading',
'updated_at',
'volume_type')]
for expected, result in zip(expects, results):
self.assertEqual(expected, result)
class TestImageDelete(TestImage): class TestImageDelete(TestImage):
@ -137,3 +291,158 @@ class TestImageDelete(TestImage):
self.images_mock.delete.assert_called_with( self.images_mock.delete.assert_called_with(
image_fakes.image_id, image_fakes.image_id,
) )
class TestImageSet(TestImage):
def setUp(self):
super(TestImageSet, self).setUp()
# This is the return value for utils.find_resource()
self.images_mock.get.return_value = fakes.FakeResource(
None,
copy.deepcopy(image_fakes.IMAGE),
loaded=True,
)
self.images_mock.update.return_value = fakes.FakeResource(
None,
copy.deepcopy(image_fakes.IMAGE),
loaded=True,
)
# Get the command object to test
self.cmd = image.SetImage(self.app, None)
def test_image_set_no_options(self):
arglist = [
image_fakes.image_name,
]
verifylist = [
('image', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
self.cmd.take_action(parsed_args)
# Verify update() was not called, if it was show the args
self.assertEqual(self.images_mock.update.call_args_list, [])
def test_image_set_options(self):
arglist = [
'--name', 'new-name',
'--owner', 'new-owner',
'--min-disk', '2',
'--min-ram', '4',
image_fakes.image_name,
]
verifylist = [
('name', 'new-name'),
('owner', 'new-owner'),
('min_disk', 2),
('min_ram', 4),
('image', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
kwargs = {
'name': 'new-name',
'owner': 'new-owner',
'min_disk': 2,
'min_ram': 4,
}
# ImageManager.update(image, **kwargs)
self.images_mock.update.assert_called_with(
image_fakes.image_id,
**kwargs
)
self.assertEqual(image_fakes.IMAGE_columns, columns)
self.assertEqual(image_fakes.IMAGE_data, data)
def test_image_set_bools1(self):
arglist = [
'--protected',
'--private',
image_fakes.image_name,
]
verifylist = [
('protected', True),
('unprotected', False),
('public', False),
('private', True),
('image', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
self.cmd.take_action(parsed_args)
kwargs = {
'protected': True,
'is_public': False,
}
# ImageManager.update(image, **kwargs)
self.images_mock.update.assert_called_with(
image_fakes.image_id,
**kwargs
)
def test_image_set_bools2(self):
arglist = [
'--unprotected',
'--public',
image_fakes.image_name,
]
verifylist = [
('protected', False),
('unprotected', True),
('public', True),
('private', False),
('image', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
self.cmd.take_action(parsed_args)
kwargs = {
'protected': False,
'is_public': True,
}
# ImageManager.update(image, **kwargs)
self.images_mock.update.assert_called_with(
image_fakes.image_id,
**kwargs
)
def test_image_set_properties(self):
arglist = [
'--property', 'Alpha=1',
'--property', 'Beta=2',
image_fakes.image_name,
]
verifylist = [
('properties', {'Alpha': '1', 'Beta': '2'}),
('image', image_fakes.image_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
self.cmd.take_action(parsed_args)
kwargs = {
'properties': {
'Alpha': '1',
'Beta': '2',
'Gamma': 'g',
},
}
# ImageManager.update(image, **kwargs)
self.images_mock.update.assert_called_with(
image_fakes.image_id,
**kwargs
)