Merge "OSC 1/4 Add CT create and UT framework"

This commit is contained in:
Jenkins 2017-07-27 14:18:46 +00:00 committed by Gerrit Code Review
commit be24874fa5
9 changed files with 958 additions and 5 deletions

View File

@ -12,16 +12,248 @@
# License for the specific language governing permissions and limitations
# under the License.
from magnumclient.common import utils as magnum_utils
from magnumclient.i18n import _
from osc_lib.command import command
from osc_lib import utils
from osc_lib import utils as osc_utils
from oslo_log import log as logging
CLUSTER_TEMPLATE_ATTRIBUTES = [
'insecure_registry',
'labels',
'updated_at',
'floating_ip_enabled',
'fixed_subnet',
'master_flavor_id',
'uuid',
'no_proxy',
'https_proxy',
'tls_disabled',
'keypair_id',
'public',
'http_proxy',
'docker_volume_size',
'server_type',
'external_network_id',
'cluster_distro',
'image_id',
'volume_driver',
'registry_enabled',
'docker_storage_driver',
'apiserver_port',
'name',
'created_at',
'network_driver',
'fixed_network',
'coe',
'flavor_id',
'master_lb_enabled',
'dns_nameserver'
]
def _show_cluster_template(cluster_template):
del cluster_template._info['links']
for field in cluster_template._info:
if cluster_template._info[field] is None:
setattr(cluster_template, field, '-')
columns = CLUSTER_TEMPLATE_ATTRIBUTES
return columns, osc_utils.get_item_properties(cluster_template, columns)
class CreateClusterTemplate(command.ShowOne):
"""Create a Cluster Template."""
_description = _("Create a Cluster Template.")
def get_parser(self, prog_name):
parser = super(CreateClusterTemplate, self).get_parser(prog_name)
parser.add_argument(
'--name',
metavar='<name>',
help=_('Name of the cluster template to create.'))
parser.add_argument(
'--coe',
required=True,
metavar='<coe>',
help=_('Specify the Container Orchestration Engine to use.'))
parser.add_argument(
'--image',
required=True,
metavar='<image>',
help=_('The name or UUID of the base image to customize for the '
'Cluster.'))
parser.add_argument(
'--external-network',
dest='external_network',
required=True,
metavar='<external-network>',
help=_('The external Neutron network name or UUID to connect to '
'this Cluster Template.'))
parser.add_argument(
'--keypair',
metavar='<keypair>',
help=_('The name or UUID of the SSH keypair to load into the '
'Cluster nodes.'))
parser.add_argument(
'--fixed-network',
dest='fixed_network',
metavar='<fixed-network>',
help=_('The private Neutron network name to connect to this '
'Cluster model.'))
parser.add_argument(
'--fixed-subnet',
dest='fixed_subnet',
metavar='<fixed-subnet>',
help=_('The private Neutron subnet name to connect to Cluster.'))
parser.add_argument(
'--network-driver',
dest='network_driver',
metavar='<network-driver>',
help=_('The network driver name for instantiating container '
'networks.'))
parser.add_argument(
'--volume-driver',
dest='volume_driver',
metavar='<volume-driver>',
help=_('The volume driver name for instantiating container '
'volume.'))
parser.add_argument(
'--dns-nameserver',
dest='dns_nameserver',
metavar='<dns-nameserver>',
default='8.8.8.8',
help=_('The DNS nameserver to use for this cluster template.'))
parser.add_argument(
'--flavor',
metavar='<flavor>',
default='m1.medium',
help=_('The nova flavor name or UUID to use when launching the '
'Cluster.'))
parser.add_argument(
'--master-flavor',
dest='master_flavor',
metavar='<master-flavor>',
help=_('The nova flavor name or UUID to use when launching the '
'master node of the Cluster.'))
parser.add_argument(
'--docker-volume-size',
dest='docker_volume_size',
metavar='<docker-volume-size>',
type=int,
help=_('Specify the number of size in GB for the docker volume '
'to use.'))
parser.add_argument(
'--docker-storage-driver',
dest='docker_storage_driver',
metavar='<docker-storage-driver>',
default='devicemapper',
help=_('Select a docker storage driver. Supported: devicemapper, '
'overlay. Default: devicemapper'))
parser.add_argument(
'--http-proxy',
dest='http_proxy',
metavar='<http-proxy>',
help=_('The http_proxy address to use for nodes in Cluster.'))
parser.add_argument(
'--https-proxy',
dest='https_proxy',
metavar='<https-proxy>',
help=_('The https_proxy address to use for nodes in Cluster.'))
parser.add_argument(
'--no-proxy',
dest='no_proxy',
metavar='<no-proxy>',
help=_('The no_proxy address to use for nodes in Cluster.'))
parser.add_argument(
'--labels',
metavar='<KEY1=VALUE1,KEY2=VALUE2;KEY3=VALUE3...>',
action='append',
default=[],
help=_('Arbitrary labels in the form of key=value pairs to '
'associate with a cluster template. May be used multiple '
'times.'))
parser.add_argument(
'--tls-disabled',
dest='tls_disabled',
action='store_true',
default=False,
help=_('Disable TLS in the Cluster.'))
parser.add_argument(
'--public',
action='store_true',
default=False,
help=_('Make cluster template public.'))
parser.add_argument(
'--registry-enabled',
dest='registry_enabled',
action='store_true',
default=False,
help=_('Enable docker registry in the Cluster'))
parser.add_argument(
'--server-type',
dest='server_type',
metavar='<server-type>',
default='vm',
help=_('Specify the server type to be used for example vm. '
'For this release default server type will be vm.'))
parser.add_argument(
'--master-lb-enabled',
dest='master_lb_enabled',
action='store_true',
default=False,
help=_('Indicates whether created Clusters should have a load '
'balancer for master nodes or not.'))
parser.add_argument(
'--floating-ip-enabled',
dest='floating_ip_enabled',
action='store_true',
default=True,
help=_('Indicates whether created Clusters should have a '
'floating ip or not.'))
return parser
def take_action(self, parsed_args):
self.log.debug("take_action(%s)", parsed_args)
mag_client = self.app.client_manager.container_infra
args = {
'name': parsed_args.name,
'image_id': parsed_args.image,
'keypair_id': parsed_args.keypair,
'external_network_id': parsed_args.external_network,
'coe': parsed_args.coe,
'fixed_network': parsed_args.fixed_network,
'fixed_subnet': parsed_args.fixed_subnet,
'network_driver': parsed_args.network_driver,
'volume_driver': parsed_args.volume_driver,
'dns_nameserver': parsed_args.dns_nameserver,
'flavor_id': parsed_args.flavor,
'master_flavor_id': parsed_args.master_flavor,
'docker_volume_size': parsed_args.docker_volume_size,
'docker_storage_driver': parsed_args.docker_storage_driver,
'http_proxy': parsed_args.http_proxy,
'https_proxy': parsed_args.https_proxy,
'no_proxy': parsed_args.no_proxy,
'labels': magnum_utils.handle_labels(parsed_args.labels),
'tls_disabled': parsed_args.tls_disabled,
'public': parsed_args.public,
'registry_enabled': parsed_args.registry_enabled,
'server_type': parsed_args.server_type,
'master_lb_enabled': parsed_args.master_lb_enabled,
'floating_ip_enabled': parsed_args.floating_ip_enabled,
}
ct = mag_client.cluster_templates.create(**args)
print("Request to create cluster template %s accepted"
% parsed_args.name)
return _show_cluster_template(ct)
class ListTemplateCluster(command.Lister):
"""List Cluster Templates."""
_description = _("List Cluster Templates.")
log = logging.getLogger(__name__ + ".ListTemplateCluster")
@ -58,8 +290,10 @@ class ListTemplateCluster(command.Lister):
mag_client = self.app.client_manager.container_infra
columns = ['uuid', 'name']
cts = mag_client.cluster_templates.list()
cts = mag_client.cluster_templates.list(limit=parsed_args.limit,
sort_key=parsed_args.sort_key,
sort_dir=parsed_args.sort_dir)
return (
columns,
(utils.get_item_properties(ct, columns) for ct in cts)
(osc_utils.get_item_properties(ct, columns) for ct in cts)
)

View File

View File

View File

@ -0,0 +1,254 @@
# Copyright 2013 Nebula Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import json
import mock
import sys
from keystoneauth1 import fixture
import requests
import six
AUTH_TOKEN = "foobar"
AUTH_URL = "http://0.0.0.0"
USERNAME = "itchy"
PASSWORD = "scratchy"
PROJECT_NAME = "poochie"
REGION_NAME = "richie"
INTERFACE = "catchy"
VERSION = "3"
TEST_RESPONSE_DICT = fixture.V2Token(token_id=AUTH_TOKEN,
user_name=USERNAME)
_s = TEST_RESPONSE_DICT.add_service('identity', name='keystone')
_s.add_endpoint(AUTH_URL + ':5000/v2.0')
_s = TEST_RESPONSE_DICT.add_service('network', name='neutron')
_s.add_endpoint(AUTH_URL + ':9696')
_s = TEST_RESPONSE_DICT.add_service('compute', name='nova')
_s.add_endpoint(AUTH_URL + ':8774/v2.1')
_s = TEST_RESPONSE_DICT.add_service('image', name='glance')
_s.add_endpoint(AUTH_URL + ':9292')
_s = TEST_RESPONSE_DICT.add_service('object', name='swift')
_s.add_endpoint(AUTH_URL + ':8080/v1')
TEST_RESPONSE_DICT_V3 = fixture.V3Token(user_name=USERNAME)
TEST_RESPONSE_DICT_V3.set_project_scope()
TEST_VERSIONS = fixture.DiscoveryList(href=AUTH_URL)
def to_unicode_dict(catalog_dict):
"""Converts dict to unicode dict
"""
if isinstance(catalog_dict, dict):
return {to_unicode_dict(key): to_unicode_dict(value)
for key, value in catalog_dict.items()}
elif isinstance(catalog_dict, list):
return [to_unicode_dict(element) for element in catalog_dict]
elif isinstance(catalog_dict, str):
return catalog_dict + u""
else:
return catalog_dict
class FakeStdout(object):
def __init__(self):
self.content = []
def write(self, text):
self.content.append(text)
def make_string(self):
result = ''
for line in self.content:
result = result + line
return result
class FakeLog(object):
def __init__(self):
self.messages = {}
def debug(self, msg):
self.messages['debug'] = msg
def info(self, msg):
self.messages['info'] = msg
def warning(self, msg):
self.messages['warning'] = msg
def error(self, msg):
self.messages['error'] = msg
def critical(self, msg):
self.messages['critical'] = msg
class FakeApp(object):
def __init__(self, _stdout, _log):
self.stdout = _stdout
self.client_manager = None
self.stdin = sys.stdin
self.stdout = _stdout or sys.stdout
self.stderr = sys.stderr
self.log = _log
class FakeOptions(object):
def __init__(self, **kwargs):
self.os_beta_command = False
class FakeClient(object):
def __init__(self, **kwargs):
self.endpoint = kwargs['endpoint']
self.token = kwargs['token']
class FakeClientManager(object):
_api_version = {
'image': '2',
}
def __init__(self):
self.compute = None
self.identity = None
self.image = None
self.object_store = None
self.volume = None
self.network = None
self.session = None
self.auth_ref = None
self.auth_plugin_name = None
self.network_endpoint_enabled = True
def get_configuration(self):
return {
'auth': {
'username': USERNAME,
'password': PASSWORD,
'token': AUTH_TOKEN,
},
'region': REGION_NAME,
'identity_api_version': VERSION,
}
def is_network_endpoint_enabled(self):
return self.network_endpoint_enabled
class FakeModule(object):
def __init__(self, name, version):
self.name = name
self.__version__ = version
# Workaround for openstacksdk case
self.version = mock.Mock()
self.version.__version__ = version
class FakeResource(object):
def __init__(self, manager=None, info=None, loaded=False, methods=None):
"""Set attributes and methods for a resource.
:param manager:
The resource manager
:param Dictionary info:
A dictionary with all attributes
:param bool loaded:
True if the resource is loaded in memory
:param Dictionary methods:
A dictionary with all methods
"""
info = info or {}
methods = methods or {}
self.__name__ = type(self).__name__
self.manager = manager
self._info = info
self._add_details(info)
self._add_methods(methods)
self._loaded = loaded
def _add_details(self, info):
for (k, v) in six.iteritems(info):
setattr(self, k, v)
def _add_methods(self, methods):
"""Fake methods with MagicMock objects.
For each <@key, @value> pairs in methods, add an callable MagicMock
object named @key as an attribute, and set the mock's return_value to
@value. When users access the attribute with (), @value will be
returned, which looks like a function call.
"""
for (name, ret) in six.iteritems(methods):
method = mock.Mock(return_value=ret)
setattr(self, name, method)
def __repr__(self):
reprkeys = sorted(k for k in self.__dict__.keys() if k[0] != '_' and
k != 'manager')
info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
return "<%s %s>" % (self.__class__.__name__, info)
def keys(self):
return self._info.keys()
def to_dict(self):
return self._info
@property
def info(self):
return self._info
def __getitem__(self, item):
return self._info.get(item)
def get(self, item, default=None):
return self._info.get(item, default)
class FakeResponse(requests.Response):
def __init__(self, headers=None, status_code=200,
data=None, encoding=None):
super(FakeResponse, self).__init__()
headers = headers or {}
self.status_code = status_code
self.headers.update(headers)
self._content = json.dumps(data)
if not isinstance(self._content, six.binary_type):
self._content = self._content.encode()
class FakeModel(dict):
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)

View File

@ -0,0 +1,75 @@
# Copyright 2012-2013 OpenStack Foundation
# Copyright 2013 Nebula Inc.
#
# 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 fixtures
import os
import testtools
from openstackclient.tests.unit import fakes
class ParserException(Exception):
pass
class TestCase(testtools.TestCase):
def setUp(self):
testtools.TestCase.setUp(self)
if (os.environ.get("OS_STDOUT_CAPTURE") == "True" or
os.environ.get("OS_STDOUT_CAPTURE") == "1"):
stdout = self.useFixture(fixtures.StringStream("stdout")).stream
self.useFixture(fixtures.MonkeyPatch("sys.stdout", stdout))
if (os.environ.get("OS_STDERR_CAPTURE") == "True" or
os.environ.get("OS_STDERR_CAPTURE") == "1"):
stderr = self.useFixture(fixtures.StringStream("stderr")).stream
self.useFixture(fixtures.MonkeyPatch("sys.stderr", stderr))
def assertNotCalled(self, m, msg=None):
"""Assert a function was not called"""
if m.called:
if not msg:
msg = 'method %s should not have been called' % m
self.fail(msg)
class TestCommand(TestCase):
"""Test cliff command classes"""
def setUp(self):
super(TestCommand, self).setUp()
# Build up a fake app
self.fake_stdout = fakes.FakeStdout()
self.fake_log = fakes.FakeLog()
self.app = fakes.FakeApp(self.fake_stdout, self.fake_log)
self.app.client_manager = fakes.FakeClientManager()
self.app.options = fakes.FakeOptions()
def check_parser(self, cmd, args, verify_args):
cmd_parser = cmd.get_parser('check_parser')
try:
parsed_args = cmd_parser.parse_args(args)
except SystemExit:
raise ParserException("Argument parse failed")
for av in verify_args:
attr, value = av
if attr:
self.assertIn(attr, parsed_args)
self.assertEqual(value, getattr(parsed_args, attr))
return parsed_args

View File

@ -0,0 +1,163 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import argparse
import copy
import datetime
import uuid
from magnumclient.tests.osc.unit import osc_fakes
from magnumclient.tests.osc.unit import osc_utils
class FakeBaseModel(object):
def __repr__(self):
return "<" + self.__class__.model_name + "%s>" % self._info
class FakeBaseModelManager(object):
def list(self, limit=None, marker=None, sort_key=None,
sort_dir=None, detail=False):
pass
def get(self, id):
pass
def create(self, **kwargs):
pass
def delete(self, id):
pass
def update(self, id, patch):
pass
class MagnumFakeContainerInfra(object):
def __init__(self):
self.cluster_templates = FakeBaseModelManager()
class MagnumFakeClientManager(osc_fakes.FakeClientManager):
def __init__(self):
super(MagnumFakeClientManager, self).__init__()
self.container_infra = MagnumFakeContainerInfra()
class MagnumParseException(Exception):
"""The base exception class for all exceptions this library raises."""
def __init__(self, message=None, details=None):
self.message = message or "Argument parse exception"
self.details = details or None
def __str__(self):
return self.message
class TestMagnumClientOSCV1(osc_utils.TestCase):
def setUp(self):
super(TestMagnumClientOSCV1, self).setUp()
self.fake_stdout = osc_fakes.FakeStdout()
self.fake_log = osc_fakes.FakeLog()
self.app = osc_fakes.FakeApp(self.fake_stdout, self.fake_log)
self.namespace = argparse.Namespace()
self.app.client_manager = MagnumFakeClientManager()
def check_parser(self, cmd, args, verify_args):
cmd_parser = cmd.get_parser('check_parser')
try:
parsed_args = cmd_parser.parse_args(args)
except SystemExit:
raise MagnumParseException()
for av in verify_args:
attr, value = av
if attr:
self.assertIn(attr, parsed_args)
self.assertEqual(value, getattr(parsed_args, attr))
return parsed_args
class FakeClusterTemplate(object):
"""Fake one or more ClusterTemplate."""
@staticmethod
def create_one_cluster_template(attrs=None):
"""Create a fake ClusterTemplate.
:param Dictionary attrs:
A dictionary with all attributes
:return:
A FakeResource object, with flavor_id, image_id, and so on
"""
attrs = attrs or {}
# set default attributes.
ct_info = {
'links': [],
'insecure_registry': None,
'labels': {},
'updated_at': None,
'floating_ip_enabled': True,
'fixed_subnet': None,
'master_flavor_id': None,
'uuid': uuid.uuid4().hex,
'no_proxy': None,
'https_proxy': None,
'tls_disabled': False,
'keypair_id': None,
'public': False,
'http_proxy': None,
'docker_volume_size': None,
'server_type': 'vm',
'external_network_id': 'public',
'cluster_distro': 'fedora-atomic',
'image_id': 'fedora-atomic-latest',
'volume_driver': None,
'registry_enabled': False,
'docker_storage_driver': 'devicemapper',
'apiserver_port': None,
'name': 'fake-ct-' + uuid.uuid4().hex,
'created_at': datetime.datetime.now(),
'network_driver': 'flannel',
'fixed_network': None,
'coe': 'kubernetes',
'flavor_id': 'm1.medium',
'master_lb_enabled': False,
'dns_nameserver': '8.8.8.8'
}
# Overwrite default attributes.
ct_info.update(attrs)
ct = osc_fakes.FakeResource(info=copy.deepcopy(ct_info), loaded=True)
return ct
@staticmethod
def create_cluster_templates(attrs=None, count=2):
"""Create multiple fake cluster templates.
:param Dictionary attrs:
A dictionary with all attributes
:param int count:
The number of cluster templates to fake
:return:
A list of FakeResource objects faking the cluster templates
"""
cts = []
for i in range(0, count):
cts.append(FakeClusterTemplate.create_one_cluster_template(attrs))
return cts

View File

@ -0,0 +1,226 @@
# Copyright 2016 Easystack. 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 copy
import mock
from magnumclient.osc.v1 import cluster_templates as osc_ct
from magnumclient.tests.osc.unit.v1 import fakes as magnum_fakes
class TestClusterTemplate(magnum_fakes.TestMagnumClientOSCV1):
default_create_args = {
'coe': 'kubernetes',
'dns_nameserver': '8.8.8.8',
'docker_storage_driver': 'devicemapper',
'docker_volume_size': None,
'external_network_id': 'public',
'fixed_network': None,
'fixed_subnet': None,
'flavor_id': 'm1.medium',
'floating_ip_enabled': True,
'http_proxy': None,
'https_proxy': None,
'image_id': 'fedora-atomic-latest',
'keypair_id': None,
'labels': {},
'master_flavor_id': None,
'master_lb_enabled': False,
'name': 'fake-ct-1',
'network_driver': None,
'no_proxy': None,
'public': False,
'registry_enabled': False,
'server_type': 'vm',
'tls_disabled': False,
'volume_driver': None
}
def setUp(self):
super(TestClusterTemplate, self).setUp()
self.cluster_templates_mock = (
self.app.client_manager.container_infra.cluster_templates)
class TestClusterTemplateCreate(TestClusterTemplate):
def setUp(self):
super(TestClusterTemplateCreate, self).setUp()
attr = dict()
attr['name'] = 'fake-ct-1'
self.new_ct = (
magnum_fakes.FakeClusterTemplate.create_one_cluster_template(attr))
self.cluster_templates_mock.create = mock.Mock()
self.cluster_templates_mock.create.return_value = self.new_ct
self.cluster_templates_mock.get = mock.Mock()
self.cluster_templates_mock.get.return_value = copy.deepcopy(
self.new_ct)
self.cluster_templates_mock.update = mock.Mock()
self.cluster_templates_mock.update.return_value = self.new_ct
# Get the command object to test
self.cmd = osc_ct.CreateClusterTemplate(self.app, None)
self.data = tuple(map(lambda x: getattr(self.new_ct, x),
osc_ct.CLUSTER_TEMPLATE_ATTRIBUTES))
def test_cluster_template_create_required_args_pass(self):
"""Verifies required arguments."""
arglist = [
'--coe', self.new_ct.coe,
'--external-network', self.new_ct.external_network_id,
'--image', self.new_ct.image_id,
'--name', self.new_ct.name
]
verifylist = [
('coe', self.new_ct.coe),
('external_network', self.new_ct.external_network_id),
('image', self.new_ct.image_id),
('name', self.new_ct.name)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.cluster_templates_mock.create.assert_called_with(
**self.default_create_args)
def test_cluster_template_create_missing_required_arg(self):
"""Verifies missing required arguments."""
arglist = [
'--external-network', self.new_ct.external_network_id,
'--image', self.new_ct.image_id,
'--name', self.new_ct.name
]
verifylist = [
('external_network', self.new_ct.external_network_id),
('image', self.new_ct.image_id),
('name', self.new_ct.name)
]
self.assertRaises(magnum_fakes.MagnumParseException,
self.check_parser, self.cmd, arglist, verifylist)
# Verify all required args are checked and not just coe
arglist.append('--coe')
arglist.append(self.new_ct.coe)
verifylist.append(('coe', self.new_ct.coe))
arglist.remove('--image')
arglist.remove(self.new_ct.image_id)
verifylist.remove(('image', self.new_ct.image_id))
self.assertRaises(magnum_fakes.MagnumParseException,
self.check_parser, self.cmd, arglist, verifylist)
arglist.remove('--external-network')
arglist.remove(self.new_ct.external_network_id)
verifylist.remove(
('external_network', self.new_ct.external_network_id))
self.assertRaises(magnum_fakes.MagnumParseException,
self.check_parser, self.cmd, arglist, verifylist)
class TestClusterTemplateList(TestClusterTemplate):
attr = dict()
attr['name'] = 'fake-ct-1'
_cluster_template = (
magnum_fakes.FakeClusterTemplate.create_one_cluster_template(attr))
attr['name'] = 'fake-ct-2'
_cluster_template2 = (
magnum_fakes.FakeClusterTemplate.create_one_cluster_template(attr))
columns = [
'uuid',
'name'
]
datalist = (
(_cluster_template.uuid, _cluster_template.name),
(_cluster_template2.uuid, _cluster_template2.name)
)
def setUp(self):
super(TestClusterTemplateList, self).setUp()
self.cluster_templates_mock.list = mock.Mock()
self.cluster_templates_mock.list.return_value = [
self._cluster_template, self._cluster_template2
]
# Get the command object to test
self.cmd = osc_ct.ListTemplateCluster(self.app, None)
def test_cluster_template_list_no_options(self):
arglist = []
verifylist = [
('limit', None),
('sort_key', None),
('sort_dir', None),
('fields', None),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.cluster_templates_mock.list.assert_called_with(
limit=None,
sort_dir=None,
sort_key=None,
)
self.assertEqual(self.columns, columns)
index = 0
for d in data:
self.assertEqual(self.datalist[index], d)
index += 1
def test_cluster_template_list_options(self):
arglist = [
'--limit', '1',
'--sort-key', 'key',
'--sort-dir', 'asc',
'--fields', 'fields'
]
verifylist = [
('limit', 1),
('sort_key', 'key'),
('sort_dir', 'asc'),
('fields', 'fields'),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.cluster_templates_mock.list.assert_called_with(
limit=1,
sort_dir='asc',
sort_key='key',
)
def test_cluster_template_list_bad_sort_dir_fail(self):
arglist = [
'--sort-dir', 'foo'
]
verifylist = [
('limit', None),
('sort_key', None),
('sort_dir', 'foo'),
('fields', None),
]
self.assertRaises(magnum_fakes.MagnumParseException,
self.check_parser, self.cmd, arglist, verifylist)

View File

@ -30,7 +30,8 @@ openstack.cli.extension =
container_infra = magnumclient.osc.plugin
openstack.container_infra.v1 =
cluster_template_list = magnumclient.osc.v1.cluster_templates:ListTemplateCluster
coe_cluster_template_create = magnumclient.osc.v1.cluster_templates:CreateClusterTemplate
coe_cluster_template_list = magnumclient.osc.v1.cluster_templates:ListTemplateCluster
[build_sphinx]
source-dir = doc/source