openstack/python-glanceclient
Code Issues Proposed changes

Support for Metadata Definitions Catalog API

Browse Source 
API calls and shell commands added in this patch:
- CRUD for metadefs namespaces;
- CRUD for metadefs objects;
- CRUD for metadefs properites;
- CRD for metadefs resource types and resource type associations.

Change-Id: I6d15f749038e8fd24fc651f0b314df5be7c673ef
Implements: blueprint metadata-schema-catalog-support
Co-Authored-By: Facundo Maldonado <facundo.n.maldonado@intel.com>
Co-Authored-By: Michal Dulko <michal.dulko@intel.com>
Co-Authored-By: Lakshmi N Sampath <lakshmi.sampath@hp.com>
Co-Authored-By: Pawel Koniszewski <pawel.koniszewski@intel.com>
This commit is contained in:
Pawel Koniszewski 2014-09-03 07:37:45 -04:00 committed by Travis Tripp
parent 16077d91dd
commit 33dcea81b2
12 changed files with 2714 additions and 41 deletions

14
glanceclient/common/utils.py

@ -17,6 +17,7 @@ from __future__ import print_function
import errno
import hashlib
import json
import os
import re
import sys
@ -106,14 +107,20 @@ def pretty_choice_list(l):
return ', '.join("'%s'" % i for i in l)
def print_list(objs, fields, formatters=None):
def print_list(objs, fields, formatters=None, field_settings=None):
formatters = formatters or {}
field_settings = field_settings or {}
pt = prettytable.PrettyTable([f for f in fields], caching=False)
pt.align = 'l'
for o in objs:
row = []
for field in fields:
if field in field_settings:
for setting, value in six.iteritems(field_settings[field]):
setting_dict = getattr(pt, setting)
setting_dict[field] = value
if field in formatters:
row.append(formatters[field](o))
else:
@ -129,7 +136,10 @@ def print_dict(d, max_column_width=80):
pt = prettytable.PrettyTable(['Property', 'Value'], caching=False)
pt.align = 'l'
pt.max_width = max_column_width
[pt.add_row(list(r)) for r in six.iteritems(d)]
for k, v in six.iteritems(d):
if isinstance(v, (dict, list)):
v = json.dumps(v)
pt.add_row([k, v])
print(strutils.safe_encode(pt.get_string(sortby='Property')))

33
glanceclient/shell.py

@ -516,25 +516,30 @@ class OpenStackImagesShell(object):
client = glanceclient.Client(api_version, endpoint, **kwargs)
return client
def _cache_schema(self, options, home_dir='~/.glanceclient'):
def _cache_schemas(self, options, home_dir='~/.glanceclient'):
homedir = expanduser(home_dir)
if not os.path.exists(homedir):
os.makedirs(homedir)
schema_file_path = homedir + os.sep + "image_schema.json"
resources = ['image', 'metadefs/namespace', 'metadefs/resource_type']
schema_file_paths = [homedir + os.sep + x + '_schema.json'
for x in ['image', 'namespace', 'resource_type']]
if (not os.path.exists(schema_file_path)) or options.get_schema:
try:
client = self._get_versioned_client('2', options,
force_auth=True)
schema = client.schemas.get("image")
client = None
for resource, schema_file_path in zip(resources, schema_file_paths):
if (not os.path.exists(schema_file_path)) or options.get_schema:
try:
if not client:
client = self._get_versioned_client('2', options,
force_auth=True)
schema = client.schemas.get(resource)
with open(schema_file_path, 'w') as f:
f.write(json.dumps(schema.raw()))
except Exception:
#NOTE(esheffield) do nothing here, we'll get a message later
#if the schema is missing
pass
with open(schema_file_path, 'w') as f:
f.write(json.dumps(schema.raw()))
except Exception:
#NOTE(esheffield) do nothing here, we'll get a message
#later if the schema is missing
pass
def main(self, argv):
# Parse args once to find version
@ -549,7 +554,7 @@ class OpenStackImagesShell(object):
api_version = options.os_image_api_version
if api_version == '2':
self._cache_schema(options)
self._cache_schemas(options)
subcommand_parser = self.get_subcommand_parser(api_version)
self.parser = subcommand_parser

34
glanceclient/v2/client.py

@ -20,6 +20,7 @@ from glanceclient.common import utils
from glanceclient.v2 import image_members
from glanceclient.v2 import image_tags
from glanceclient.v2 import images
from glanceclient.v2 import metadefs
from glanceclient.v2 import schemas
@ -44,6 +45,23 @@ class Client(object):
self.image_members = image_members.Controller(self.http_client,
self._get_member_model())
resource_type_model = self._get_metadefs_resource_type_model()
self.metadefs_resource_type = (
metadefs.ResourceTypeController(self.http_client,
resource_type_model))
property_model = self._get_metadefs_property_model()
self.metadefs_property = (
metadefs.PropertyController(self.http_client, property_model))
object_model = self._get_metadefs_object_model()
self.metadefs_object = (
metadefs.ObjectController(self.http_client, object_model))
namespace_model = self._get_metadefs_namespace_model()
self.metadefs_namespace = (
metadefs.NamespaceController(self.http_client, namespace_model))
def _get_image_model(self):
schema = self.schemas.get('image')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)
@ -51,3 +69,19 @@ class Client(object):
def _get_member_model(self):
schema = self.schemas.get('member')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)
def _get_metadefs_namespace_model(self):
schema = self.schemas.get('metadefs/namespace')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)
def _get_metadefs_resource_type_model(self):
schema = self.schemas.get('metadefs/resource_type')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)
def _get_metadefs_property_model(self):
schema = self.schemas.get('metadefs/property')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)
def _get_metadefs_object_model(self):
schema = self.schemas.get('metadefs/object')
return warlock.model_factory(schema.raw(), schemas.SchemaBasedModel)

336
glanceclient/v2/metadefs.py Normal file

@ -0,0 +1,336 @@
# Copyright 2014 OpenStack Foundation
# 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 six
from six.moves.urllib import parse
import warlock
from glanceclient.common import utils
from glanceclient.openstack.common import strutils
DEFAULT_PAGE_SIZE = 20
class NamespaceController(object):
def __init__(self, http_client, model):
self.http_client = http_client
self.model = model
def create(self, **kwargs):
"""Create a namespace.
:param kwargs: Unpacked namespace object.
"""
url = '/v2/metadefs/namespaces'
try:
namespace = self.model(kwargs)
except (warlock.InvalidOperation, ValueError) as e:
raise TypeError(utils.exception_to_str(e))
resp, body = self.http_client.post(url, data=namespace)
body.pop('self', None)
return self.model(**body)
def update(self, namespace_name, **kwargs):
"""Update a namespace.
:param namespace_name: Name of a namespace (old one).
:param kwargs: Unpacked namespace object.
"""
namespace = self.get(namespace_name)
for (key, value) in six.iteritems(kwargs):
try:
setattr(namespace, key, value)
except warlock.InvalidOperation as e:
raise TypeError(utils.exception_to_str(e))
# Remove read-only parameters.
read_only = ['schema', 'updated_at', 'created_at']
for elem in read_only:
if elem in namespace:
del namespace[elem]
url = '/v2/metadefs/namespaces/{0}'.format(namespace_name)
self.http_client.put(url, data=namespace)
return self.get(namespace.namespace)
def get(self, namespace, **kwargs):
"""Get one namespace."""
query_params = parse.urlencode(kwargs)
if kwargs:
query_params = '?%s' % query_params
url = '/v2/metadefs/namespaces/{0}{1}'.format(namespace, query_params)
resp, body = self.http_client.get(url)
#NOTE(bcwaldon): remove 'self' for now until we have an elegant
# way to pass it into the model constructor without conflict
body.pop('self', None)
return self.model(**body)
def list(self, **kwargs):
"""Retrieve a listing of Namespace objects
:param page_size: Number of namespaces to request in each request
:returns generator over list of Namespaces
"""
ori_validate_fun = self.model.validate
empty_fun = lambda *args, **kwargs: None
def paginate(url):
resp, body = self.http_client.get(url)
for namespace in body['namespaces']:
# NOTE(bcwaldon): remove 'self' for now until we have
# an elegant way to pass it into the model constructor
# without conflict.
namespace.pop('self', None)
yield self.model(**namespace)
# NOTE(zhiyan): In order to resolve the performance issue
# of JSON schema validation for image listing case, we
# don't validate each image entry but do it only on first
# image entry for each page.
self.model.validate = empty_fun
# NOTE(zhiyan); Reset validation function.
self.model.validate = ori_validate_fun
try:
next_url = body['next']
except KeyError:
return
else:
for namespace in paginate(next_url):
yield namespace
filters = kwargs.get('filters', {})
filters = {} if filters is None else filters
if not kwargs.get('page_size'):
filters['limit'] = DEFAULT_PAGE_SIZE
else:
filters['limit'] = kwargs['page_size']
for param, value in six.iteritems(filters):
if isinstance(value, list):
filters[param] = strutils.safe_encode(','.join(value))
elif isinstance(value, six.string_types):
filters[param] = strutils.safe_encode(value)
url = '/v2/metadefs/namespaces?%s' % parse.urlencode(filters)
for namespace in paginate(url):
yield namespace
def delete(self, namespace):
"""Delete a namespace."""
url = '/v2/metadefs/namespaces/{0}'.format(namespace)
self.http_client.delete(url)
class ResourceTypeController(object):
def __init__(self, http_client, model):
self.http_client = http_client
self.model = model
def associate(self, namespace, **kwargs):
"""Associate a resource type with a namespace."""
try:
res_type = self.model(kwargs)
except (warlock.InvalidOperation, ValueError) as e:
raise TypeError(utils.exception_to_str(e))
url = '/v2/metadefs/namespaces/{0}/resource_types'.format(namespace,
res_type)
resp, body = self.http_client.post(url, data=res_type)
body.pop('self', None)
return self.model(**body)
def deassociate(self, namespace, resource):
"""Deasociate a resource type with a namespace."""
url = '/v2/metadefs/namespaces/{0}/resource_types/{1}'. \
format(namespace, resource)
self.http_client.delete(url)
def list(self):
"""Retrieve a listing of available resource types
:returns generator over list of resource_types
"""
url = '/v2/metadefs/resource_types'
resp, body = self.http_client.get(url)
for resource_type in body['resource_types']:
yield self.model(**resource_type)
def get(self, namespace):
url = '/v2/metadefs/namespaces/{0}/resource_types'.format(namespace)
resp, body = self.http_client.get(url)
body.pop('self', None)
for resource_type in body['resource_type_associations']:
yield self.model(**resource_type)
class PropertyController(object):
def __init__(self, http_client, model):
self.http_client = http_client
self.model = model
def create(self, namespace, **kwargs):
"""Create a property.
:param namespace: Name of a namespace the property will belong.
:param kwargs: Unpacked property object.
"""
try:
prop = self.model(kwargs)
except (warlock.InvalidOperation, ValueError) as e:
raise TypeError(utils.exception_to_str(e))
url = '/v2/metadefs/namespaces/{0}/properties'.format(namespace)
resp, body = self.http_client.post(url, data=prop)
body.pop('self', None)
return self.model(**body)
def update(self, namespace, prop_name, **kwargs):
"""Update a property.
:param namespace: Name of a namespace the property belongs.
:param prop_name: Name of a property (old one).
:param kwargs: Unpacked property object.
"""
prop = self.get(namespace, prop_name)
for (key, value) in kwargs.items():
try:
setattr(prop, key, value)
except warlock.InvalidOperation as e:
raise TypeError(utils.exception_to_str(e))
url = '/v2/metadefs/namespaces/{0}/properties/{1}'.format(namespace,
prop_name)
self.http_client.put(url, data=prop)
return self.get(namespace, prop.name)
def get(self, namespace, prop_name):
url = '/v2/metadefs/namespaces/{0}/properties/{1}'.format(namespace,
prop_name)
resp, body = self.http_client.get(url)
body.pop('self', None)
body['name'] = prop_name
return self.model(**body)
def list(self, namespace, **kwargs):
"""Retrieve a listing of metadata properties
:returns generator over list of objects
"""
url = '/v2/metadefs/namespaces/{0}/properties'.format(namespace)
resp, body = self.http_client.get(url)
for key, value in body['properties'].items():
value['name'] = key
yield self.model(value)
def delete(self, namespace, prop_name):
"""Delete a property."""
url = '/v2/metadefs/namespaces/{0}/properties/{1}'.format(namespace,
prop_name)
self.http_client.delete(url)
def delete_all(self, namespace):
"""Delete all properties in a namespace."""
url = '/v2/metadefs/namespaces/{0}/properties'.format(namespace)
self.http_client.delete(url)
class ObjectController(object):
def __init__(self, http_client, model):
self.http_client = http_client
self.model = model
def create(self, namespace, **kwargs):
"""Create an object.
:param namespace: Name of a namespace the object belongs.
:param kwargs: Unpacked object.
"""
try:
obj = self.model(kwargs)
except (warlock.InvalidOperation, ValueError) as e:
raise TypeError(utils.exception_to_str(e))
url = '/v2/metadefs/namespaces/{0}/objects'.format(namespace)
resp, body = self.http_client.post(url, data=obj)
body.pop('self', None)
return self.model(**body)
def update(self, namespace, object_name, **kwargs):
"""Update an object.
:param namespace: Name of a namespace the object belongs.
:param prop_name: Name of an object (old one).
:param kwargs: Unpacked object.
"""
obj = self.get(namespace, object_name)
for (key, value) in kwargs.items():
try:
setattr(obj, key, value)
except warlock.InvalidOperation as e:
raise TypeError(utils.exception_to_str(e))
# Remove read-only parameters.
read_only = ['schema', 'updated_at', 'created_at']
for elem in read_only:
if elem in namespace:
del namespace[elem]
url = '/v2/metadefs/namespaces/{0}/objects/{1}'.format(namespace,
object_name)
self.http_client.put(url, data=obj)
return self.get(namespace, obj.name)
def get(self, namespace, object_name):
url = '/v2/metadefs/namespaces/{0}/objects/{1}'.format(namespace,
object_name)
resp, body = self.http_client.get(url)
body.pop('self', None)
return self.model(**body)
def list(self, namespace, **kwargs):
"""Retrieve a listing of metadata objects
:returns generator over list of objects
"""
url = '/v2/metadefs/namespaces/{0}/objects'.format(namespace,)
resp, body = self.http_client.get(url)
for obj in body['objects']:
yield self.model(obj)
def delete(self, namespace, object_name):
"""Delete an object."""
url = '/v2/metadefs/namespaces/{0}/objects/{1}'.format(namespace,
object_name)
self.http_client.delete(url)
def delete_all(self, namespace):
"""Delete all objects in a namespace."""
url = '/v2/metadefs/namespaces/{0}/objects'.format(namespace)
self.http_client.delete(url)

386
glanceclient/v2/shell.py

@ -303,3 +303,389 @@ def do_location_update(gc, args):
else:
image = gc.images.update_location(args.id, args.url, metadata)
utils.print_dict(image)
# Metadata - catalog
NAMESPACE_SCHEMA = None
def get_namespace_schema():
global NAMESPACE_SCHEMA
if NAMESPACE_SCHEMA is None:
schema_path = expanduser("~/.glanceclient/namespace_schema.json")
if os.path.exists(schema_path) and os.path.isfile(schema_path):
with open(schema_path, "r") as f:
schema_raw = f.read()
NAMESPACE_SCHEMA = json.loads(schema_raw)
return NAMESPACE_SCHEMA
def _namespace_show(namespace, max_column_width=None):
namespace = dict(namespace) # Warlock objects are compatible with dicts
# Flatten dicts for display
if 'properties' in namespace:
props = [k for k in namespace['properties']]
namespace['properties'] = props
if 'resource_type_associations' in namespace:
assocs = [assoc['name']
for assoc in namespace['resource_type_associations']]
namespace['resource_type_associations'] = assocs
if 'objects' in namespace:
objects = [obj['name'] for obj in namespace['objects']]
namespace['objects'] = objects
if max_column_width:
utils.print_dict(namespace, max_column_width)
else:
utils.print_dict(namespace)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of the namespace.')
@utils.schema_args(get_namespace_schema, omit=['namespace', 'property_count',
'properties', 'tag_count',
'tags', 'object_count',
'objects', 'resource_types'])
def do_md_namespace_create(gc, args):
"""Create a new metadata definitions namespace."""
schema = gc.schemas.get('metadefs/namespace')
_args = [(x[0].replace('-', '_'), x[1]) for x in vars(args).items()]
fields = dict(filter(lambda x: x[1] is not None and
(schema.is_core_property(x[0])),
_args))
namespace = gc.metadefs_namespace.create(**fields)
_namespace_show(namespace)
@utils.arg('--file', metavar='<FILEPATH>',
help='Path to file with namespace schema to import. Alternatively, '
'namespaces schema can be passed to the client via stdin.')
def do_md_namespace_import(gc, args):
"""Import a metadata definitions namespace from file or standard input."""
namespace_data = utils.get_data_file(args)
if not namespace_data:
utils.exit('No metadata definition namespace passed via stdin or '
'--file argument.')
try:
namespace_json = json.load(namespace_data)
except ValueError:
utils.exit('Schema is not a valid JSON object.')
else:
namespace = gc.metadefs_namespace.create(**namespace_json)
_namespace_show(namespace)
@utils.arg('id', metavar='<NAMESPACE>', help='Name of namespace to update.')
@utils.schema_args(get_namespace_schema, omit=['property_count', 'properties',
'tag_count', 'tags',
'object_count', 'objects',
'resource_type_associations',
'schema'])
def do_md_namespace_update(gc, args):
"""Update an existing metadata definitions namespace."""
schema = gc.schemas.get('metadefs/namespace')
_args = [(x[0].replace('-', '_'), x[1]) for x in vars(args).items()]
fields = dict(filter(lambda x: x[1] is not None and
(schema.is_core_property(x[0])),
_args))
namespace = gc.metadefs_namespace.update(args.id, **fields)
_namespace_show(namespace)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace to describe.')
@utils.arg('--resource-type', metavar='<RESOURCE_TYPE>',
help='Applies prefix of given resource type associated to a '
'namespace to all properties of a namespace.', default=None)
@utils.arg('--max-column-width', metavar='<integer>', default=80,
help='The max column width of the printed table.')
def do_md_namespace_show(gc, args):
"""Describe a specific metadata definitions namespace.
Lists also the namespace properties, objects and resource type
associations.
"""
kwargs = {}
if args.resource_type:
kwargs['resource_type'] = args.resource_type
namespace = gc.metadefs_namespace.get(args.namespace, **kwargs)
_namespace_show(namespace, int(args.max_column_width))
@utils.arg('--resource-types', metavar='<RESOURCE_TYPES>', action='append',
help='Resource type to filter namespaces.')
@utils.arg('--visibility', metavar='<VISIBILITY>',
help='Visibility parameter to filter namespaces.')
@utils.arg('--page-size', metavar='<SIZE>', default=None, type=int,
help='Number of namespaces to request in each paginated request.')
def do_md_namespace_list(gc, args):
"""List metadata definitions namespaces."""
filter_keys = ['resource_types', 'visibility']
filter_items = [(key, getattr(args, key, None)) for key in filter_keys]
filters = dict([item for item in filter_items if item[1] is not None])
kwargs = {'filters': filters}
if args.page_size is not None:
kwargs['page_size'] = args.page_size
namespaces = gc.metadefs_namespace.list(**kwargs)
columns = ['namespace']
utils.print_list(namespaces, columns)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace to delete.')
def do_md_namespace_delete(gc, args):
"""Delete specified metadata definitions namespace with its contents."""
gc.metadefs_namespace.delete(args.namespace)
# Metadata - catalog
RESOURCE_TYPE_SCHEMA = None
def get_resource_type_schema():
global RESOURCE_TYPE_SCHEMA
if RESOURCE_TYPE_SCHEMA is None:
schema_path = expanduser("~/.glanceclient/resource_type_schema.json")
if os.path.exists(schema_path) and os.path.isfile(schema_path):
with open(schema_path, "r") as f:
schema_raw = f.read()
RESOURCE_TYPE_SCHEMA = json.loads(schema_raw)
return RESOURCE_TYPE_SCHEMA
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
@utils.schema_args(get_resource_type_schema)
def do_md_resource_type_associate(gc, args):
"""Associate resource type with a metadata definitions namespace."""
schema = gc.schemas.get('metadefs/resource_type')
_args = [(x[0].replace('-', '_'), x[1]) for x in vars(args).items()]
fields = dict(filter(lambda x: x[1] is not None and
(schema.is_core_property(x[0])),
_args))
resource_type = gc.metadefs_resource_type.associate(args.namespace,
**fields)
utils.print_dict(resource_type)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
@utils.arg('resource_type', metavar='<RESOURCE_TYPE>',
help='Name of resource type.')
def do_md_resource_type_deassociate(gc, args):
"""Deassociate resource type with a metadata definitions namespace."""
gc.metadefs_resource_type.deassociate(args.namespace, args.resource_type)
def do_md_resource_type_list(gc, args):
"""List available resource type names."""
resource_types = gc.metadefs_resource_type.list()
utils.print_list(resource_types, ['name'])
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
def do_md_namespace_resource_type_list(gc, args):
"""List resource types associated to specific namespace."""
resource_types = gc.metadefs_resource_type.get(args.namespace)
utils.print_list(resource_types, ['name', 'prefix', 'properties_target'])
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the property will belong.')
@utils.arg('--name', metavar='<NAME>', required=True,
help='Internal name of a property.')
@utils.arg('--title', metavar='<TITLE>', required=True,
help='Property name displayed to the user.')
@utils.arg('--schema', metavar='<SCHEMA>', required=True,
help='Valid JSON schema of a property.')
def do_md_property_create(gc, args):
"""Create a new metadata definitions property inside a namespace."""
try:
schema = json.loads(args.schema)
except ValueError:
utils.exit('Schema is not a valid JSON object.')
else:
fields = {'name': args.name, 'title': args.title}
fields.update(schema)
new_property = gc.metadefs_property.create(args.namespace, **fields)
utils.print_dict(new_property)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the property belongs.')
@utils.arg('property', metavar='<PROPERTY>', help='Name of a property.')
@utils.arg('--name', metavar='<NAME>', default=None,
help='New name of a property.')
@utils.arg('--title', metavar='<TITLE>', default=None,
help='Property name displayed to the user.')
@utils.arg('--schema', metavar='<SCHEMA>', default=None,
help='Valid JSON schema of a property.')
def do_md_property_update(gc, args):
"""Update metadata definitions property inside a namespace."""
fields = {}
if args.name:
fields['name'] = args.name
if args.title:
fields['title'] = args.title
if args.schema:
try:
schema = json.loads(args.schema)
except ValueError:
utils.exit('Schema is not a valid JSON object.')
else:
fields.update(schema)
new_property = gc.metadefs_property.update(args.namespace, args.property,
**fields)
utils.print_dict(new_property)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the property belongs.')
@utils.arg('property', metavar='<PROPERTY>', help='Name of a property.')
@utils.arg('--max-column-width', metavar='<integer>', default=80,
help='The max column width of the printed table.')
def do_md_property_show(gc, args):
"""Describe a specific metadata definitions property inside a namespace."""
prop = gc.metadefs_property.get(args.namespace, args.property)
utils.print_dict(prop, int(args.max_column_width))
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the property belongs.')
@utils.arg('property', metavar='<PROPERTY>', help='Name of a property.')
def do_md_property_delete(gc, args):
"""Delete a specific metadata definitions property inside a namespace."""
gc.metadefs_property.delete(args.namespace, args.property)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
def do_md_namespace_properties_delete(gc, args):
"""Delete all metadata definitions property inside a specific namespace."""
gc.metadefs_property.delete_all(args.namespace)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
def do_md_property_list(gc, args):
"""List metadata definitions properties inside a specific namespace."""
properties = gc.metadefs_property.list(args.namespace)
columns = ['name', 'title', 'type']
utils.print_list(properties, columns)
def _object_show(obj, max_column_width=None):
obj = dict(obj) # Warlock objects are compatible with dicts
# Flatten dicts for display
if 'properties' in obj:
objects = [k for k in obj['properties']]
obj['properties'] = objects
if max_column_width:
utils.print_dict(obj, max_column_width)
else:
utils.print_dict(obj)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the object will belong.')
@utils.arg('--name', metavar='<NAME>', required=True,
help='Internal name of an object.')
@utils.arg('--schema', metavar='<SCHEMA>', required=True,
help='Valid JSON schema of an object.')
def do_md_object_create(gc, args):
"""Create a new metadata definitions object inside a namespace."""
try:
schema = json.loads(args.schema)
except ValueError:
utils.exit('Schema is not a valid JSON object.')
else:
fields = {'name': args.name}
fields.update(schema)
new_object = gc.metadefs_object.create(args.namespace, **fields)
_object_show(new_object)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the object belongs.')
@utils.arg('object', metavar='<OBJECT>', help='Name of an object.')
@utils.arg('--name', metavar='<NAME>', default=None,
help='New name of an object.')
@utils.arg('--schema', metavar='<SCHEMA>', default=None,
help='Valid JSON schema of an object.')
def do_md_object_update(gc, args):
"""Update metadata definitions object inside a namespace."""
fields = {}
if args.name:
fields['name'] = args.name
if args.schema:
try:
schema = json.loads(args.schema)
except ValueError:
utils.exit('Schema is not a valid JSON object.')
else:
fields.update(schema)
new_object = gc.metadefs_object.update(args.namespace, args.object,
**fields)
_object_show(new_object)
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the object belongs.')
@utils.arg('object', metavar='<OBJECT>', help='Name of an object.')
@utils.arg('--max-column-width', metavar='<integer>', default=80,
help='The max column width of the printed table.')
def do_md_object_show(gc, args):
"""Describe a specific metadata definitions object inside a namespace."""
obj = gc.metadefs_object.get(args.namespace, args.object)
_object_show(obj, int(args.max_column_width))
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the object belongs.')
@utils.arg('object', metavar='<OBJECT>', help='Name of an object.')
@utils.arg('property', metavar='<PROPERTY>', help='Name of a property.')
@utils.arg('--max-column-width', metavar='<integer>', default=80,
help='The max column width of the printed table.')
def do_md_object_property_show(gc, args):
"""Describe a specific metadata definitions property inside an object."""
obj = gc.metadefs_object.get(args.namespace, args.object)
try:
prop = obj['properties'][args.property]
prop['name'] = args.property
except KeyError:
utils.exit('Property %s not found in object %s.' % (args.property,
args.object))
utils.print_dict(prop, int(args.max_column_width))
@utils.arg('namespace', metavar='<NAMESPACE>',
help='Name of namespace the object belongs.')
@utils.arg('object', metavar='<OBJECT>', help='Name of an object.')
def do_md_object_delete(gc, args):
"""Delete a specific metadata definitions object inside a namespace."""
gc.metadefs_object.delete(args.namespace, args.object)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
def do_md_namespace_objects_delete(gc, args):
"""Delete all metadata definitions objects inside a specific namespace."""
gc.metadefs_object.delete_all(args.namespace)
@utils.arg('namespace', metavar='<NAMESPACE>', help='Name of namespace.')
def do_md_object_list(gc, args):
"""List metadata definitions objects inside a specific namespace."""
objects = gc.metadefs_object.list(args.namespace)
columns = ['name', 'description']
column_settings = {
"description": {
"max_width": 50,
"align": "l"
}
}
utils.print_list(objects, columns, field_settings=column_settings)

63
tests/test_shell.py

@ -141,9 +141,9 @@ class ShellTest(utils.TestCase):
self.assertEqual(kwargs['token'], 'mytoken')
self.assertEqual(args[0], 'https://image:1234/v1')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schema')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schemas')
def test_no_auth_with_token_and_image_url_with_v2(self,
cache_schema):
cache_schemas):
with mock.patch('glanceclient.v2.client.Client') as v2_client:
# test no authentication is required if both token and endpoint url
# are specified
@ -181,14 +181,14 @@ class ShellTest(utils.TestCase):
@mock.patch('glanceclient.v2.client.Client')
@mock.patch('keystoneclient.session.Session')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schema')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schemas')
@mock.patch.object(keystoneclient.discover.Discover, 'url_for',
side_effect=[keystone_client_fixtures.V2_URL, None])
def test_auth_plugin_invocation_with_v2(self,
v2_client,
ks_session,
url_for,
cache_schema):
cache_schemas):
with mock.patch(self.auth_plugin) as mock_auth_plugin:
args = '--os-image-api-version 2 image-list'
glance_shell = openstack_shell.OpenStackImagesShell()
@ -211,12 +211,12 @@ class ShellTest(utils.TestCase):
@mock.patch('glanceclient.v2.client.Client')
@mock.patch('keystoneclient.session.Session')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schema')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schemas')
@mock.patch.object(keystoneclient.discover.Discover, 'url_for',
side_effect=[keystone_client_fixtures.V2_URL,
keystone_client_fixtures.V3_URL])
def test_auth_plugin_invocation_with_unversioned_auth_url_with_v2(
self, v2_client, ks_session, cache_schema, url_for):
self, v2_client, ks_session, cache_schemas, url_for):
with mock.patch(self.auth_plugin) as mock_auth_plugin:
args = ('--os-auth-url %s --os-image-api-version 2 '
'image-list') % (keystone_client_fixtures.BASE_URL)
@ -260,14 +260,14 @@ class ShellTestWithKeystoneV3Auth(ShellTest):
@mock.patch('glanceclient.v2.client.Client')
@mock.patch('keystoneclient.session.Session')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schema')
@mock.patch.object(openstack_shell.OpenStackImagesShell, '_cache_schemas')
@mock.patch.object(keystoneclient.discover.Discover, 'url_for',
side_effect=[None, keystone_client_fixtures.V3_URL])
def test_auth_plugin_invocation_with_v2(self,
v2_client,
ks_session,
url_for,
cache_schema):
cache_schemas):
with mock.patch(self.auth_plugin) as mock_auth_plugin:
args = '--os-image-api-version 2 image-list'
glance_shell = openstack_shell.OpenStackImagesShell()
@ -292,7 +292,9 @@ class ShellCacheSchemaTest(utils.TestCase):
self._mock_client_setup()
self._mock_shell_setup()
self.cache_dir = '/dir_for_cached_schema'
self.cache_file = self.cache_dir + '/image_schema.json'
self.cache_files = [self.cache_dir + '/image_schema.json',
self.cache_dir + '/namespace_schema.json',
self.cache_dir + '/resource_type_schema.json']
def tearDown(self):
super(ShellCacheSchemaTest, self).tearDown()
@ -322,45 +324,56 @@ class ShellCacheSchemaTest(utils.TestCase):
@mock.patch('six.moves.builtins.open', new=mock.mock_open(), create=True)
@mock.patch('os.path.exists', return_value=True)
def test_cache_schema_gets_when_forced(self, exists_mock):
def test_cache_schemas_gets_when_forced(self, exists_mock):
options = {
'get_schema': True
}
self.shell._cache_schema(self._make_args(options),
home_dir=self.cache_dir)
self.shell._cache_schemas(self._make_args(options),
home_dir=self.cache_dir)
self.assertEqual(4, open.mock_calls.__len__())
self.assertEqual(mock.call(self.cache_file, 'w'), open.mock_calls[0])
self.assertEqual(12, open.mock_calls.__len__())
self.assertEqual(mock.call(self.cache_files[0], 'w'),
open.mock_calls[0])
self.assertEqual(mock.call(self.cache_files[1], 'w'),
open.mock_calls[4])
self.assertEqual(mock.call().write(json.dumps(self.schema_dict)),
open.mock_calls[2])
self.assertEqual(mock.call().write(json.dumps(self.schema_dict)),
open.mock_calls[6])
@mock.patch('six.moves.builtins.open', new=mock.mock_open(), create=True)
@mock.patch('os.path.exists', side_effect=[True, False])
def test_cache_schema_gets_when_not_exists(self, exists_mock):
@mock.patch('os.path.exists', side_effect=[True, False, False, False])
def test_cache_schemas_gets_when_not_exists(self, exists_mock):
options = {
'get_schema': False
}
self.shell._cache_schema(self._make_args(options),
home_dir=self.cache_dir)
self.shell._cache_schemas(self._make_args(options),
home_dir=self.cache_dir)
self.assertEqual(4, open.mock_calls.__len__())
self.assertEqual(mock.call(self.cache_file, 'w'), open.mock_calls[0])
self.assertEqual(12, open.mock_calls.__len__())
self.assertEqual(mock.call(self.cache_files[0], 'w'),
open.mock_calls[0])
self.assertEqual(mock.call(self.cache_files[1], 'w'),
open.mock_calls[4])
self.assertEqual(mock.call().write(json.dumps(self.schema_dict)),
open.mock_calls[2])
self.assertEqual(mock.call().write(json.dumps(self.schema_dict)),
open.mock_calls[6])
@mock.patch('six.moves.builtins.open', new=mock.mock_open(), create=True)
@mock.patch('os.path.exists', return_value=True)
def test_cache_schema_leaves_when_present_not_forced(self, exists_mock):
def test_cache_schemas_leaves_when_present_not_forced(self, exists_mock):
options = {
'get_schema': False
}
self.shell._cache_schema(self._make_args(options),
home_dir=self.cache_dir)
self.shell._cache_schemas(self._make_args(options),
home_dir=self.cache_dir)
os.path.exists.assert_any_call(self.cache_dir)
os.path.exists.assert_any_call(self.cache_file)
self.assertEqual(2, exists_mock.call_count)
os.path.exists.assert_any_call(self.cache_files[0])
os.path.exists.assert_any_call(self.cache_files[1])
self.assertEqual(4, exists_mock.call_count)
self.assertEqual(0, open.mock_calls.__len__())

8
tests/v2/test_client.py

@ -26,6 +26,14 @@ class ClientTest(testtools.TestCase):
self.mock = mox.Mox()
self.mock.StubOutWithMock(client.Client, '_get_image_model')
self.mock.StubOutWithMock(client.Client, '_get_member_model')
self.mock.StubOutWithMock(client.Client,
'_get_metadefs_namespace_model')
self.mock.StubOutWithMock(client.Client,
'_get_metadefs_resource_type_model')
self.mock.StubOutWithMock(client.Client,
'_get_metadefs_property_model')
self.mock.StubOutWithMock(client.Client,
'_get_metadefs_object_model')
def tearDown(self):
super(ClientTest, self).tearDown()

592
tests/v2/test_metadefs_namespaces.py Normal file

@ -0,0 +1,592 @@
# Copyright 2012 OpenStack Foundation.
# 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 testtools
import warlock
from glanceclient.v2 import metadefs
from tests import utils
NAMESPACE1 = 'Namespace1'
NAMESPACE2 = 'Namespace2'
NAMESPACE3 = 'Namespace3'
NAMESPACE4 = 'Namespace4'
NAMESPACE5 = 'Namespace5'
NAMESPACE6 = 'Namespace6'
NAMESPACE7 = 'Namespace7'
NAMESPACE8 = 'Namespace8'
NAMESPACENEW = 'NamespaceNew'
RESOURCE_TYPE1 = 'ResourceType1'
RESOURCE_TYPE2 = 'ResourceType2'
OBJECT1 = 'Object1'
PROPERTY1 = 'Property1'
PROPERTY2 = 'Property2'
def _get_namespace_fixture(ns_name, rt_name=RESOURCE_TYPE1, **kwargs):
ns = {
"display_name": "Flavor Quota",
"description": "DESCRIPTION1",
"self": "/v2/metadefs/namespaces/%s" % ns_name,
"namespace": ns_name,
"visibility": "public",
"protected": True,
"owner": "admin",
"resource_types": [
{
"name": rt_name
}
],
"schema": "/v2/schemas/metadefs/namespace",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
ns.update(kwargs)
return ns
fixtures = {
"/v2/metadefs/namespaces?limit=20": {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=20",
"namespaces": [
_get_namespace_fixture(NAMESPACE1),
_get_namespace_fixture(NAMESPACE2),
],
"schema": "/v2/schemas/metadefs/namespaces"
}
)
},
"/v2/metadefs/namespaces?limit=1": {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=1",
"namespaces": [
_get_namespace_fixture(NAMESPACE7),
],
"schema": "/v2/schemas/metadefs/namespaces",
"next": "/v2/metadefs/namespaces?marker=%s&limit=1"
% NAMESPACE7,
}
)
},
"/v2/metadefs/namespaces?marker=%s&limit=1" % NAMESPACE7: {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=1",
"namespaces": [
_get_namespace_fixture(NAMESPACE8),
],
"schema": "/v2/schemas/metadefs/namespaces"
}
)
},
"/v2/metadefs/namespaces?limit=20&resource_types=%s" % RESOURCE_TYPE1: {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=20",
"namespaces": [
_get_namespace_fixture(NAMESPACE3),
],
"schema": "/v2/schemas/metadefs/namespaces"
}
)
},
"/v2/metadefs/namespaces?limit=20&resource_types="
"%s%%2C%s" % (RESOURCE_TYPE1, RESOURCE_TYPE2): {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=20",
"namespaces": [
_get_namespace_fixture(NAMESPACE4),
],
"schema": "/v2/schemas/metadefs/namespaces"
}
)
},
"/v2/metadefs/namespaces?limit=20&visibility=private": {
"GET": (
{},
{
"first": "/v2/metadefs/namespaces?limit=20",
"namespaces": [
_get_namespace_fixture(NAMESPACE5),
],
"schema": "/v2/schemas/metadefs/namespaces"
}
)
},
"/v2/metadefs/namespaces": {
"POST": (
{},
{
"display_name": "Flavor Quota",
"description": "DESCRIPTION1",
"self": "/v2/metadefs/namespaces/%s" % 'NamespaceNew',
"namespace": 'NamespaceNew',
"visibility": "public",
"protected": True,
"owner": "admin",
"schema": "/v2/schemas/metadefs/namespace",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
)
},
"/v2/metadefs/namespaces/%s" % NAMESPACE1: {
"GET": (
{},
{
"display_name": "Flavor Quota",
"description": "DESCRIPTION1",
"objects": [
{
"description": "DESCRIPTION2",
"name": "OBJECT1",
"self": "/v2/metadefs/namespaces/%s/objects/" %
OBJECT1,
"required": [],
"properties": {
PROPERTY1: {
"type": "integer",
"description": "DESCRIPTION3",
"title": "Quota: CPU Shares"
},
PROPERTY2: {
"minimum": 1000,
"type": "integer",
"description": "DESCRIPTION4",
"maximum": 1000000,
"title": "Quota: CPU Period"
},
},
"schema": "/v2/schemas/metadefs/object"
}
],
"self": "/v2/metadefs/namespaces/%s" % NAMESPACE1,
"namespace": NAMESPACE1,
"visibility": "public",
"protected": True,
"owner": "admin",
"resource_types": [
{
"name": RESOURCE_TYPE1
}
],
"schema": "/v2/schemas/metadefs/namespace",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
),
"PUT": (
{},
{
"display_name": "Flavor Quota",
"description": "DESCRIPTION1",
"objects": [
{
"description": "DESCRIPTION2",
"name": "OBJECT1",
"self": "/v2/metadefs/namespaces/%s/objects/" %
OBJECT1,
"required": [],
"properties": {
PROPERTY1: {
"type": "integer",
"description": "DESCRIPTION3",
"title": "Quota: CPU Shares"
},
PROPERTY2: {
"minimum": 1000,
"type": "integer",
"description": "DESCRIPTION4",
"maximum": 1000000,
"title": "Quota: CPU Period"
},
},
"schema": "/v2/schemas/metadefs/object"
}
],
"self": "/v2/metadefs/namespaces/%s" % NAMESPACENEW,
"namespace": NAMESPACENEW,
"visibility": "public",
"protected": True,
"owner": "admin",
"resource_types": [
{
"name": RESOURCE_TYPE1
}
],
"schema": "/v2/schemas/metadefs/namespace",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
),
"DELETE": (
{},
{}
)
},
"/v2/metadefs/namespaces/%s?resource_type=%s" % (NAMESPACE6,
RESOURCE_TYPE1):
{
"GET": (
{},
{
"display_name": "Flavor Quota",
"description": "DESCRIPTION1",
"objects": [],
"self": "/v2/metadefs/namespaces/%s" % NAMESPACE1,
"namespace": NAMESPACE6,
"visibility": "public",
"protected": True,
"owner": "admin",
"resource_types": [
{
"name": RESOURCE_TYPE1
}
],
"schema": "/v2/schemas/metadefs/namespace",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
),
},
}
fake_namespace_schema = {
"additionalProperties": False,
"definitions": {
"property": {
"additionalProperties": {
"required": [
"title",
"type"
],
"type": "object",
"properties": {
"additionalItems": {
"type": "boolean"
},
"enum": {
"type": "array"
},
"description": {
"type": "string"
},
"title": {
"type": "string"
},
"default": {},
"minLength": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"required": {
"$ref": "#/definitions/stringArray"
},
"maximum": {
"type": "number"
},
"minItems": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"readonly": {
"type": "boolean"
},
"minimum": {
"type": "number"
},
"maxItems": {
"$ref": "#/definitions/positiveInteger"
},
"maxLength": {
"$ref": "#/definitions/positiveInteger"
},
"uniqueItems": {
"default": False,
"type": "boolean"
},
"pattern": {
"type": "string",
"format": "regex"
},
"items": {
"type": "object",
"properties": {
"enum": {
"type": "array"
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
},
"type": "object"
},
"positiveIntegerDefault0": {
"allOf": [
{
"$ref": "#/definitions/positiveInteger"
},
{
"default": 0
}
]
},
"stringArray": {
"uniqueItems": True,
"items": {
"type": "string"
},
"type": "array"
},
"positiveInteger": {
"minimum": 0,
"type": "integer"
}
},
"required": [
"namespace"
],
"name": "namespace",
"properties": {
"description": {
"type": "string",
"description": "Provides a user friendly description of the "
"namespace.",
"maxLength": 500
},
"updated_at": {
"type": "string",
"description": "Date and time of the last namespace modification "
"(READ-ONLY)",
"format": "date-time"
},
"visibility": {
"enum": [
"public",
"private"
],
"type": "string",
"description": "Scope of namespace accessibility."
},
"self": {
"type": "string"
},
"objects": {
"items": {
"type": "object",
"properties": {
"properties": {
"$ref": "#/definitions/property"
},
"required": {
"$ref": "#/definitions/stringArray"
},
"name": {
"type": "string"
},
"description": {
"type": "string"
}
}
},
"type": "array"
},
"owner": {
"type": "string",
"description": "Owner of the namespace.",
"maxLength": 255
},
"resource_types": {
"items": {
"type": "object",
"properties": {
"prefix": {
"type": "string"
},
"name": {
"type": "string"
},
"metadata_type": {
"type": "string"
}
}
},
"type": "array"
},
"properties": {
"$ref": "#/definitions/property"
},
"display_name": {
"type": "string",
"description": "The user friendly name for the namespace. Used by"
" UI if available.",
"maxLength": 80
},
"created_at": {
"type": "string",
"description": "Date and time of namespace creation (READ-ONLY)",
"format": "date-time"
},
"namespace": {
"type": "string",
"description": "The unique namespace text.",
"maxLength": 80
},
"protected": {
"type": "boolean",
"description": "If true, namespace will not be deletable."
},
"schema": {
"type": "string"
}
}
}
FakeNamespaceModel = warlock.model_factory(fake_namespace_schema)
class TestNamespaceController(testtools.TestCase):
def setUp(self):
super(TestNamespaceController, self).setUp()
self.api = utils.FakeAPI(fixtures)
self.controller = metadefs.NamespaceController(self.api,
FakeNamespaceModel)
def test_list_namespaces(self):
namespaces = list(self.controller.list())
self.assertEqual(len(namespaces), 2)
self.assertEqual(NAMESPACE1, namespaces[0]['namespace'])
self.assertEqual(NAMESPACE2, namespaces[1]['namespace'])
def test_list_namespaces_paginate(self):
namespaces = list(self.controller.list(page_size=1))
self.assertEqual(len(namespaces), 2)
self.assertEqual(NAMESPACE7, namespaces[0]['namespace'])
self.assertEqual(NAMESPACE8, namespaces[1]['namespace'])
def test_list_namespaces_with_one_resource_type_filter(self):
namespaces = list(self.controller.list(
filters={
'resource_types': [RESOURCE_TYPE1]
}
))
self.assertEqual(len(namespaces), 1)
self.assertEqual(NAMESPACE3, namespaces[0]['namespace'])
def test_list_namespaces_with_multiple_resource_types_filter(self):
namespaces = list(self.controller.list(
filters={
'resource_types': [RESOURCE_TYPE1, RESOURCE_TYPE2]
}
))
self.assertEqual(len(namespaces), 1)
self.assertEqual(NAMESPACE4, namespaces[0]['namespace'])
def test_list_namespaces_with_visibility_filter(self):
namespaces = list(self.controller.list(
filters={
'visibility': 'private'
}
))
self.assertEqual(len(namespaces), 1)
self.assertEqual(NAMESPACE5, namespaces[0]['namespace'])
def test_get_namespace(self):
namespace = self.controller.get(NAMESPACE1)
self.assertEqual(NAMESPACE1, namespace.namespace)
self.assertTrue(namespace.protected)
def test_get_namespace_with_resource_type(self):
namespace = self.controller.get(NAMESPACE6,
resource_type=RESOURCE_TYPE1)
self.assertEqual(NAMESPACE6, namespace.namespace)
self.assertTrue(namespace.protected)
def test_create_namespace(self):
properties = {
'namespace': NAMESPACENEW
}
namespace = self.controller.create(**properties)
self.assertEqual(NAMESPACENEW, namespace.namespace)
self.assertTrue(namespace.protected)
def test_create_namespace_invalid_data(self):
properties = {}
self.assertRaises(TypeError, self.controller.create, **properties)
def test_create_namespace_invalid_property(self):
properties = {'namespace': 'NewNamespace', 'protected': '123'}
self.assertRaises(TypeError, self.controller.create, **properties)
def test_update_namespace(self):
properties = {'display_name': 'My Updated Name'}
namespace = self.controller.update(NAMESPACE1, **properties)
self.assertEqual(NAMESPACE1, namespace.namespace)
def test_update_namespace_invalid_property(self):
properties = {'protected': '123'}
self.assertRaises(TypeError, self.controller.update, NAMESPACE1,
**properties)
def test_delete_namespace(self):
self.controller.delete(NAMESPACE1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s' % NAMESPACE1,
{},
None)]
self.assertEqual(expect, self.api.calls)

318
tests/v2/test_metadefs_objects.py Normal file

@ -0,0 +1,318 @@
# Copyright 2012 OpenStack Foundation.
# 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 six
import testtools
import warlock
from glanceclient.v2 import metadefs
from tests import utils
NAMESPACE1 = 'Namespace1'
OBJECT1 = 'Object1'
OBJECT2 = 'Object2'
OBJECTNEW = 'ObjectNew'
PROPERTY1 = 'Property1'
PROPERTY2 = 'Property2'
PROPERTY3 = 'Property3'
PROPERTY4 = 'Property4'
def _get_object_fixture(ns_name, obj_name, **kwargs):
obj = {
"description": "DESCRIPTION",
"name": obj_name,
"self": "/v2/metadefs/namespaces/%s/objects/%s" %
(ns_name, obj_name),
"required": [],
"properties": {
PROPERTY1: {
"type": "integer",
"description": "DESCRIPTION",
"title": "Quota: CPU Shares"
},
PROPERTY2: {
"minimum": 1000,
"type": "integer",
"description": "DESCRIPTION",
"maximum": 1000000,
"title": "Quota: CPU Period"
},
},
"schema": "/v2/schemas/metadefs/object",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
obj.update(kwargs)
return obj
fixtures = {
"/v2/metadefs/namespaces/%s/objects" % NAMESPACE1: {
"GET": (
{},
{
"objects": [
_get_object_fixture(NAMESPACE1, OBJECT1),
_get_object_fixture(NAMESPACE1, OBJECT2)
],
"schema": "v2/schemas/metadefs/objects"
}
),
"POST": (
{},
_get_object_fixture(NAMESPACE1, OBJECTNEW)
),
"DELETE": (
{},
{}
)
},
"/v2/metadefs/namespaces/%s/objects/%s" % (NAMESPACE1, OBJECT1): {
"GET": (
{},
_get_object_fixture(NAMESPACE1, OBJECT1)
),
"PUT": (
{},
_get_object_fixture(NAMESPACE1, OBJECT1)
),
"DELETE": (
{},
{}
)
}
}
fake_object_schema = {
"additionalProperties": False,
"definitions": {
"property": {
"additionalProperties": {
"required": [
"title",
"type"
],
"type": "object",
"properties": {
"additionalItems": {
"type": "boolean"
},
"enum": {
"type": "array"
},
"description": {
"type": "string"
},
"title": {
"type": "string"
},
"default": {},
"minLength": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"required": {
"$ref": "#/definitions/stringArray"
},
"maximum": {
"type": "number"
},
"minItems": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"readonly": {
"type": "boolean"
},
"minimum": {
"type": "number"
},
"maxItems": {
"$ref": "#/definitions/positiveInteger"
},
"maxLength": {
"$ref": "#/definitions/positiveInteger"
},
"uniqueItems": {
"default": False,
"type": "boolean"
},
"pattern": {
"type": "string",
"format": "regex"
},
"items": {
"type": "object",
"properties": {
"enum": {
"type": "array"
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
},
"type": "object"
},
"positiveIntegerDefault0": {
"allOf": [
{
"$ref": "#/definitions/positiveInteger"
},
{
"default": 0
}
]
},
"stringArray": {
"uniqueItems": True,
"items": {
"type": "string"
},
"type": "array"
},
"positiveInteger": {
"minimum": 0,
"type": "integer"
}
},
"required": [
"name"
],
"name": "object",
"properties": {
"created_at": {
"type": "string",
"description": "Date and time of object creation (READ-ONLY)",
"format": "date-time"
},
"description": {
"type": "string"
},
"name": {
"type": "string"
},
"self": {
"type": "string"
},
"required": {
"$ref": "#/definitions/stringArray"
},
"properties": {
"$ref": "#/definitions/property"
},
"schema": {
"type": "string"
},
"updated_at": {
"type": "string",
"description": "Date and time of the last object modification "
"(READ-ONLY)",
"format": "date-time"
},
}
}
FakeObjectModel = warlock.model_factory(fake_object_schema)
class TestObjectController(testtools.TestCase):
def setUp(self):
super(TestObjectController, self).setUp()
self.api = utils.FakeAPI(fixtures)
self.controller = metadefs.ObjectController(self.api,
FakeObjectModel)
def test_list_object(self):
objects = list(self.controller.list(NAMESPACE1))
actual = [obj.name for obj in objects]
self.assertEqual([OBJECT1, OBJECT2], actual)
def test_get_object(self):
obj = self.controller.get(NAMESPACE1, OBJECT1)
self.assertEqual(OBJECT1, obj.name)
self.assertEqual([PROPERTY1, PROPERTY2],
list(six.iterkeys(obj.properties)))
def test_create_object(self):
properties = {
'name': OBJECTNEW,
'description': 'DESCRIPTION'
}
obj = self.controller.create(NAMESPACE1, **properties)
self.assertEqual(OBJECTNEW, obj.name)
def test_create_object_invalid_property(self):
properties = {
'namespace': NAMESPACE1
}
self.assertRaises(TypeError, self.controller.create, **properties)
def test_update_object(self):
properties = {
'description': 'UPDATED_DESCRIPTION'
}
obj = self.controller.update(NAMESPACE1, OBJECT1, **properties)
self.assertEqual(OBJECT1, obj.name)
def test_update_object_invalid_property(self):
properties = {
'required': 'INVALID'
}
self.assertRaises(TypeError, self.controller.update, NAMESPACE1,
OBJECT1, **properties)
def test_delete_object(self):
self.controller.delete(NAMESPACE1, OBJECT1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s/objects/%s' % (NAMESPACE1, OBJECT1),
{},
None)]
self.assertEqual(expect, self.api.calls)
def test_delete_all_objects(self):
self.controller.delete_all(NAMESPACE1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s/objects' % NAMESPACE1,
{},
None)]
self.assertEqual(expect, self.api.calls)

296
tests/v2/test_metadefs_properties.py Normal file

@ -0,0 +1,296 @@
# Copyright 2012 OpenStack Foundation.
# 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 testtools
import warlock
from glanceclient.v2 import metadefs
from tests import utils
NAMESPACE1 = 'Namespace1'
PROPERTY1 = 'Property1'
PROPERTY2 = 'Property2'
PROPERTYNEW = 'PropertyNew'
fixtures = {
"/v2/metadefs/namespaces/%s/properties" % NAMESPACE1: {
"GET": (
{},
{
"properties": {
PROPERTY1: {
"default": "1",
"type": "integer",
"description": "Number of cores.",
"title": "cores"
},
PROPERTY2: {
"items": {
"enum": [
"Intel",
"AMD"
],
"type": "string"
},
"type": "array",
"description": "Specifies the CPU manufacturer.",
"title": "Vendor"
},
}
}
),
"POST": (
{},
{
"items": {
"enum": [
"Intel",
"AMD"
],
"type": "string"
},
"type": "array",
"description": "UPDATED_DESCRIPTION",
"title": "Vendor",
"name": PROPERTYNEW
}
),
"DELETE": (
{},
{}
)
},
"/v2/metadefs/namespaces/%s/properties/%s" % (NAMESPACE1, PROPERTY1): {
"GET": (
{},
{
"items": {
"enum": [
"Intel",
"AMD"
],
"type": "string"
},
"type": "array",
"description": "Specifies the CPU manufacturer.",
"title": "Vendor"
}
),
"PUT": (
{},
{
"items": {
"enum": [
"Intel",
"AMD"
],
"type": "string"
},
"type": "array",
"description": "UPDATED_DESCRIPTION",
"title": "Vendor"
}
),
"DELETE": (
{},
{}
)
},
}
fake_property_schema = {
"additionalProperties": False,
"definitions": {
"positiveIntegerDefault0": {
"allOf": [
{
"$ref": "#/definitions/positiveInteger"
},
{
"default": 0
}
]
},
"stringArray": {
"minItems": 1,
"items": {
"type": "string"
},
"uniqueItems": True,
"type": "array"
},
"positiveInteger": {
"minimum": 0,
"type": "integer"
}
},
"required": [
"name",
"title",
"type"
],
"name": "property",
"properties": {
"description": {
"type": "string"
},
"minLength": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"enum": {
"type": "array"
},
"minimum": {
"type": "number"
},
"maxItems": {
"$ref": "#/definitions/positiveInteger"
},
"maxLength": {
"$ref": "#/definitions/positiveInteger"
},
"uniqueItems": {
"default": False,
"type": "boolean"
},
"additionalItems": {
"type": "boolean"
},
"name": {
"type": "string"
},
"title": {
"type": "string"
},
"default": {},
"pattern": {
"type": "string",
"format": "regex"
},
"required": {
"$ref": "#/definitions/stringArray"
},
"maximum": {
"type": "number"
},
"minItems": {
"$ref": "#/definitions/positiveIntegerDefault0"
},
"readonly": {
"type": "boolean"
},
"items": {
"type": "object",
"properties": {
"enum": {
"type": "array"
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
},
"type": {
"enum": [
"array",
"boolean",
"integer",
"number",
"object",
"string",
"null"
],
"type": "string"
}
}
}
FakePropertyModel = warlock.model_factory(fake_property_schema)
class TestPropertyController(testtools.TestCase):
def setUp(self):
super(TestPropertyController, self).setUp()
self.api = utils.FakeAPI(fixtures)
self.controller = metadefs.PropertyController(self.api,
FakePropertyModel)
def test_list_property(self):
properties = list(self.controller.list(NAMESPACE1))
actual = [prop.name for prop in properties]
self.assertEqual([PROPERTY1, PROPERTY2], actual)
def test_get_property(self):
prop = self.controller.get(NAMESPACE1, PROPERTY1)
self.assertEqual(PROPERTY1, prop.name)
def test_create_property(self):
properties = {
'name': PROPERTYNEW,
'title': 'TITLE',
'type': 'string'
}
obj = self.controller.create(NAMESPACE1, **properties)
self.assertEqual(PROPERTYNEW, obj.name)
def test_create_property_invalid_property(self):
properties = {
'namespace': NAMESPACE1
}
self.assertRaises(TypeError, self.controller.create, **properties)
def test_update_property(self):
properties = {
'description': 'UPDATED_DESCRIPTION'
}
prop = self.controller.update(NAMESPACE1, PROPERTY1, **properties)
self.assertEqual(PROPERTY1, prop.name)
def test_update_property_invalid_property(self):
properties = {
'type': 'INVALID'
}
self.assertRaises(TypeError, self.controller.update, NAMESPACE1,
PROPERTY1, **properties)
def test_delete_property(self):
self.controller.delete(NAMESPACE1, PROPERTY1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s/properties/%s' % (NAMESPACE1,
PROPERTY1),
{},
None)]
self.assertEqual(expect, self.api.calls)
def test_delete_all_properties(self):
self.controller.delete_all(NAMESPACE1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s/properties' % NAMESPACE1,
{},
None)]
self.assertEqual(expect, self.api.calls)

176
tests/v2/test_metadefs_resource_types.py Normal file

@ -0,0 +1,176 @@
# Copyright 2012 OpenStack Foundation.
# 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 testtools
import warlock
from glanceclient.v2 import metadefs
from tests import utils
NAMESPACE1 = 'Namespace1'
RESOURCE_TYPE1 = 'ResourceType1'
RESOURCE_TYPE2 = 'ResourceType2'
RESOURCE_TYPE3 = 'ResourceType3'
RESOURCE_TYPE4 = 'ResourceType4'
RESOURCE_TYPENEW = 'ResourceTypeNew'
fixtures = {
"/v2/metadefs/namespaces/%s/resource_types" % NAMESPACE1: {
"GET": (
{},
{
"resource_type_associations": [
{
"name": RESOURCE_TYPE3,
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
},
{
"name": RESOURCE_TYPE4,
"prefix": "PREFIX:",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
]
}
),
"POST": (
{},
{
"name": RESOURCE_TYPENEW,
"prefix": "PREFIX:",
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
),
},
"/v2/metadefs/namespaces/%s/resource_types/%s" % (NAMESPACE1,
RESOURCE_TYPE1):
{
"DELETE": (
{},
{}
),
},
"/v2/metadefs/resource_types": {
"GET": (
{},
{
"resource_types": [
{
"name": RESOURCE_TYPE1,
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
},
{
"name": RESOURCE_TYPE2,
"created_at": "2014-08-14T09:07:06Z",
"updated_at": "2014-08-14T09:07:06Z",
}
]
}
)
},
}
fake_resource_type_schema = {
"name": "resource_type",
"properties": {
"prefix": {
"type": "string",
"description": "Specifies the prefix to use for the given "
"resource type. Any properties in the namespace "
"should be prefixed with this prefix when being "
"applied to the specified resource type. Must "
"include prefix separator (e.g. a colon :).",
"maxLength": 80
},
"properties_target": {
"type": "string",
"description": "Some resource types allow more than one "
"key / value pair per instance. For example, "
"Cinder allows user and image metadata on volumes. "
"Only the image properties metadata is evaluated "
"by Nova (scheduling or drivers). This property "
"allows a namespace target to remove the "
"ambiguity.",
"maxLength": 80
},
"name": {
"type": "string",
"description": "Resource type names should be aligned with Heat "
"resource types whenever possible: http://docs."
"openstack.org/developer/heat/template_guide/"
"openstack.html",
"maxLength": 80
},
"created_at": {
"type": "string",
"description": "Date and time of resource type association"
" (READ-ONLY)",
"format": "date-time"
},
"updated_at": {
"type": "string",
"description": "Date and time of the last resource type "
"association modification (READ-ONLY)",
"format": "date-time"
},
}
}
FakeRTModel = warlock.model_factory(fake_resource_type_schema)
class TestResoureTypeController(testtools.TestCase):
def setUp(self):
super(TestResoureTypeController, self).setUp()
self.api = utils.FakeAPI(fixtures)
self.controller = metadefs.ResourceTypeController(self.api,
FakeRTModel)
def test_list_resource_types(self):
resource_types = list(self.controller.list())
names = [rt.name for rt in resource_types]
self.assertEqual([RESOURCE_TYPE1, RESOURCE_TYPE2], names)
def test_get_resource_types(self):
resource_types = list(self.controller.get(NAMESPACE1))
names = [rt.name for rt in resource_types]
self.assertEqual([RESOURCE_TYPE3, RESOURCE_TYPE4], names)
def test_associate_resource_types(self):
resource_types = self.controller.associate(NAMESPACE1,
name=RESOURCE_TYPENEW)
self.assertEqual(RESOURCE_TYPENEW, resource_types['name'])
def test_associate_resource_types_invalid_property(self):
longer = '1234' * 50
properties = {'name': RESOURCE_TYPENEW, 'prefix': longer}
self.assertRaises(TypeError, self.controller.associate, NAMESPACE1,
**properties)
def test_deassociate_resource_types(self):
self.controller.deassociate(NAMESPACE1, RESOURCE_TYPE1)
expect = [
('DELETE',
'/v2/metadefs/namespaces/%s/resource_types/%s' % (NAMESPACE1,
RESOURCE_TYPE1),
{},
None)]
self.assertEqual(expect, self.api.calls)

499
tests/v2/test_shell_v2.py

@ -385,3 +385,502 @@ class ShellV2Test(testtools.TestCase):
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 = {}
expect_namespace['namespace'] = 'MyNamespace'
expect_namespace['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 = {}
expect_namespace['namespace'] = 'MyNamespace'
expect_namespace['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 = {}
expect_namespace['namespace'] = 'MyNamespace'
expect_namespace['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 = {}
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 = {}
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 = {}
expect_rt['namespace'] = 'MyNamespace'
expect_rt['name'] = 'MyResourceType'
expect_rt['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)
mocked_list.assert_called_once()
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",
'schema': '{}'})
with mock.patch.object(self.gc.metadefs_property,
'create') as mocked_create:
expect_property = {}
expect_property['namespace'] = 'MyNamespace'
expect_property['name'] = 'MyProperty'
expect_property['title'] = 'Title'
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')
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",
'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 = {}
expect_property['namespace'] = 'MyNamespace'
expect_property['name'] = 'MyProperty'
expect_property['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 = {}
expect_property['namespace'] = 'MyNamespace'
expect_property['property'] = 'MyProperty'
expect_property['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 = {}
expect_object['namespace'] = 'MyNamespace'
expect_object['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 = {}
expect_object['namespace'] = 'MyNamespace'
expect_object['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 = {}
expect_object['namespace'] = 'MyNamespace'
expect_object['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}})