Browse Source

Initialize plugin for OSC

Initial Octavia client command to list load balancers in a given
project.

Sets up many of the base classes and base files needed for
the rest of the client when it comes up.

Change-Id: I5d426e1a3a364abbe77edea5e8aaad2c8c2213c1
changes/23/446223/15 0.1.0
Ankur Gupta 5 years ago
parent
commit
970a232dc0
  1. 3
      .testr.conf
  2. 2
      doc/source/data/lbaas.csv
  3. 1
      doc/source/index.rst
  4. 13
      doc/source/usage/osc/v2/load-balancer.rst
  5. 0
      octaviaclient/api/__init__.py
  6. 40
      octaviaclient/api/load_balancer_v2.py
  7. 0
      octaviaclient/osc/__init__.py
  8. 66
      octaviaclient/osc/plugin.py
  9. 0
      octaviaclient/osc/v2/__init__.py
  10. 41
      octaviaclient/osc/v2/load_balancer.py
  11. 23
      octaviaclient/tests/base.py
  12. 254
      octaviaclient/tests/fakes.py
  13. 0
      octaviaclient/tests/functional/__init__.py
  14. 123
      octaviaclient/tests/functional/base.py
  15. 0
      octaviaclient/tests/functional/osc/__init__.py
  16. 0
      octaviaclient/tests/functional/osc/v2/__init__.py
  17. 28
      octaviaclient/tests/test_octaviaclient.py
  18. 0
      octaviaclient/tests/unit/__init__.py
  19. 0
      octaviaclient/tests/unit/api/__init__.py
  20. 53
      octaviaclient/tests/unit/api/test_load_balancer.py
  21. 0
      octaviaclient/tests/unit/osc/__init__.py
  22. 0
      octaviaclient/tests/unit/osc/v2/__init__.py
  23. 79
      octaviaclient/tests/unit/osc/v2/fakes.py
  24. 81
      octaviaclient/tests/unit/osc/v2/test_load_balancer.py
  25. 75
      octaviaclient/tests/utils.py
  26. 4
      releasenotes/notes/initial-list-command-90c9fa39fc10540e.yaml
  27. 5
      setup.cfg
  28. 5
      test-requirements.txt
  29. 5
      tox.ini

3
.testr.conf

@ -2,6 +2,7 @@
test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-60} \
${PYTHON:-python} -m subunit.run discover -t ./ . $LISTOPT $IDOPTION
${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./octaviaclient/tests/unit} $LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE
test_list_option=--list

2
doc/source/data/lbaas.csv

@ -20,7 +20,7 @@ lbaas-listener-show,,LBaaS v2 Show information of a given listener.
lbaas-listener-update,,LBaaS v2 Update a given listener.
lbaas-loadbalancer-create,,LBaaS v2 Create a loadbalancer.
lbaas-loadbalancer-delete,,LBaaS v2 Delete a given loadbalancer.
lbaas-loadbalancer-list,,LBaaS v2 List loadbalancers that belong to a given tenant.
lbaas-loadbalancer-list,loadbalancer list,LBaaS v2 List loadbalancers that belong to a given tenant.
lbaas-loadbalancer-list-on-agent,,List the loadbalancers on a loadbalancer v2 agent.
lbaas-loadbalancer-show,,LBaaS v2 Show information of a given loadbalancer.
lbaas-loadbalancer-stats,,Retrieve stats for a given loadbalancer.

1
doc/source/index.rst

@ -17,7 +17,6 @@ Getting Started
readme
installation
usage/osc_cli_plugins
contributing
Usage

13
doc/source/usage/osc/v2/load-balancer.rst

@ -0,0 +1,13 @@
============
loadbalancer
============
loadbalancer list
-----------------
List load balancers
.. program:: loadbalancer list
.. code:: bash
openstack loadbalancer list

0
octaviaclient/api/__init__.py

40
octaviaclient/api/load_balancer_v2.py

@ -0,0 +1,40 @@
# 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.
#
"""Load Balancer v2 API Library"""
from osc_lib.api import api
class APIv2(api.BaseAPI):
"""Load Balancer v2 API"""
_endpoint_suffix = '/v2.0/lbaas'
def __init__(self, endpoint=None, **kwargs):
super(APIv2, self).__init__(endpoint=endpoint, **kwargs)
self.endpoint = self.endpoint.rstrip('/')
self._build_url()
def _build_url(self):
if not self.endpoint.endswith(self._endpoint_suffix):
self.endpoint = self.endpoint + self._endpoint_suffix
def load_balancer_list(
self,
**filter
):
url = '/loadbalancers'
load_balancer_list = self.list(url, **filter)['loadbalancers']
return load_balancer_list

0
octaviaclient/osc/__init__.py

66
octaviaclient/osc/plugin.py

@ -0,0 +1,66 @@
# 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.
"""OpenStackClient plugin for Load Balancer service."""
import logging
from octaviaclient.api import load_balancer_v2
from osc_lib import utils
LOG = logging.getLogger(__name__)
DEFAULT_LOADBALANCER_API_VERSION = '2.0'
API_VERSION_OPTION = 'os_loadbalancer_api_version'
API_NAME = 'load_balancer'
LOAD_BALANCER_API_TYPE = 'loadbalancer'
LOAD_BALANCER_API_VERSIONS = {
'2.0': 'octaviaclient.api.load_balancer_v2.APIv2',
}
def make_client(instance):
"""Returns a load balancer service client"""
endpoint = instance.get_endpoint_for_service_type(
'load-balancer',
region_name=instance.region_name,
interface=instance.interface,
)
client = load_balancer_v2.APIv2(
session=instance.session,
service_type='load-balancer',
endpoint=endpoint,
)
return client
def build_option_parser(parser):
"""Hook to add global options
Called from openstackclient.shell.OpenStackShell.__init__()
after the builtin parser has been initialized. This is
where a plugin can add global options such as an API version.
:param argparse.ArgumentParser parser: The parser object that
has been initialized by OpenStackShell.
"""
parser.add_argument(
'--os-loadbalancer-api-version',
metavar='<loadbalancer-api-version>',
default=utils.env(
'OS_LOADBALANCER_API_VERSION',
default=DEFAULT_LOADBALANCER_API_VERSION),
help='OSC Plugin API version, default=' +
DEFAULT_LOADBALANCER_API_VERSION +
' (Env: OS_LOADBALANCER_API_VERSION)')
return parser

0
octaviaclient/osc/v2/__init__.py

41
octaviaclient/osc/v2/load_balancer.py

@ -0,0 +1,41 @@
# 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.
#
"""Load Balancer action implementation"""
from cliff import lister
from osc_lib import utils
class ListLoadBalancer(lister.Lister):
"""List load balancers"""
def parsed_args(self, prog_name):
parser = super(ListLoadBalancer, self).get_parser(prog_name)
return parser
def take_action(self, parsed_args):
columns = (
'ID',
'Name',
'Project ID',
'VIP Address',
'Provisioning Status',)
data = self.app.client_manager.load_balancer.load_balancer_list()
return (columns,
(utils.get_dict_properties(
s, columns,
formatters={},
) for s in data))

23
octaviaclient/tests/base.py

@ -1,23 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# 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.
from oslotest import base
class TestCase(base.BaseTestCase):
"""Test case base class for all unit tests."""

254
octaviaclient/tests/fakes.py

@ -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)

0
octaviaclient/tests/functional/__init__.py

123
octaviaclient/tests/functional/base.py

@ -0,0 +1,123 @@
# 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 os
import re
import shlex
import subprocess
import testtools
from tempest.lib.cli import output_parser
from tempest.lib import exceptions
COMMON_DIR = os.path.dirname(os.path.abspath(__file__))
FUNCTIONAL_DIR = os.path.normpath(os.path.join(COMMON_DIR, '..'))
ROOT_DIR = os.path.normpath(os.path.join(FUNCTIONAL_DIR, '..'))
EXAMPLE_DIR = os.path.join(ROOT_DIR, 'examples')
def execute(cmd, fail_ok=False, merge_stderr=False):
"""Executes specified command for the given action."""
cmdlist = shlex.split(cmd)
result = ''
result_err = ''
stdout = subprocess.PIPE
stderr = subprocess.STDOUT if merge_stderr else subprocess.PIPE
proc = subprocess.Popen(cmdlist, stdout=stdout, stderr=stderr)
result, result_err = proc.communicate()
result = result.decode('utf-8')
if not fail_ok and proc.returncode != 0:
raise exceptions.CommandFailed(proc.returncode, cmd, result,
result_err)
return result
class TestCase(testtools.TestCase):
delimiter_line = re.compile('^\+\-[\+\-]+\-\+$')
@classmethod
def openstack(cls, cmd, fail_ok=False):
"""Executes openstackclient command for the given action."""
return execute('openstack ' + cmd, fail_ok=fail_ok)
@classmethod
def get_openstack_configuration_value(cls, configuration):
opts = cls.get_opts([configuration])
return cls.openstack('configuration show ' + opts)
@classmethod
def get_openstack_extention_names(cls):
opts = cls.get_opts(['Name'])
return cls.openstack('extension list ' + opts)
@classmethod
def get_opts(cls, fields, output_format='value'):
return ' -f {0} {1}'.format(output_format,
' '.join(['-c ' + it for it in fields]))
@classmethod
def assertOutput(cls, expected, actual):
if expected != actual:
raise Exception(expected + ' != ' + actual)
@classmethod
def assertInOutput(cls, expected, actual):
if expected not in actual:
raise Exception(expected + ' not in ' + actual)
@classmethod
def assertsOutputNotNone(cls, observed):
if observed is None:
raise Exception('No output observed')
def assert_table_structure(self, items, field_names):
"""Verify that all items have keys listed in field_names."""
for item in items:
for field in field_names:
self.assertIn(field, item)
def assert_show_fields(self, show_output, field_names):
"""Verify that all items have keys listed in field_names."""
# field_names = ['name', 'description']
# show_output = [{'name': 'fc2b98d8faed4126b9e371eda045ade2'},
# {'description': 'description-821397086'}]
# this next line creates a flattened list of all 'keys' (like 'name',
# and 'description' out of the output
all_headers = [item for sublist in show_output for item in sublist]
for field_name in field_names:
self.assertIn(field_name, all_headers)
def parse_show_as_object(self, raw_output):
"""Return a dict with values parsed from cli output."""
items = self.parse_show(raw_output)
o = {}
for item in items:
o.update(item)
return o
def parse_show(self, raw_output):
"""Return list of dicts with item values parsed from cli output."""
items = []
table_ = output_parser.table(raw_output)
for row in table_['values']:
item = {}
item[row[0]] = row[1]
items.append(item)
return items
def parse_listing(self, raw_output):
"""Return list of dicts with basic item parsed from cli output."""
return output_parser.listing(raw_output)

0
octaviaclient/tests/functional/osc/__init__.py

0
octaviaclient/tests/functional/osc/v2/__init__.py

28
octaviaclient/tests/test_octaviaclient.py

@ -1,28 +0,0 @@
# -*- coding: utf-8 -*-
# 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.
"""
test_octaviaclient
----------------------------------
Tests for `octaviaclient` module.
"""
from octaviaclient.tests import base
class TestOctaviaclient(base.TestCase):
def test_something(self):
pass

0
octaviaclient/tests/unit/__init__.py

0
octaviaclient/tests/unit/api/__init__.py

53
octaviaclient/tests/unit/api/test_load_balancer.py

@ -0,0 +1,53 @@
# 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.
#
"""Load Balancer v2 API Library Tests"""
from keystoneauth1 import session
from requests_mock.contrib import fixture
from octaviaclient.api import load_balancer_v2 as lb
from osc_lib.tests import utils
FAKE_ACCOUNT = 'q12we34r'
FAKE_AUTH = '11223344556677889900'
FAKE_URL = 'http://example.com/v2.0/lbaas/'
FAKE_LB = 'rainbarrel'
LIST_LB_RESP = [
{'name': 'lb1'},
{'name': 'lb2'},
]
class TestLoadBalancerv2(utils.TestCase):
def setUp(self):
super(TestLoadBalancerv2, self).setUp()
sess = session.Session()
self.api = lb.APIv2(session=sess, endpoint=FAKE_URL)
self.requests_mock = self.useFixture(fixture.Fixture())
class TestLoadBalancer(TestLoadBalancerv2):
def test_list_load_balancer_no_options(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + 'loadbalancers',
json={'loadbalancers': LIST_LB_RESP},
status_code=200,
)
ret = self.api.load_balancer_list()
self.assertEqual(LIST_LB_RESP, ret)

0
octaviaclient/tests/unit/osc/__init__.py

0
octaviaclient/tests/unit/osc/v2/__init__.py

79
octaviaclient/tests/unit/osc/v2/fakes.py

@ -0,0 +1,79 @@
# 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
import uuid
from octaviaclient.tests import fakes
from osc_lib.tests import utils
LOADBALANCER = {
'id': 'lbid',
'name': 'lb1',
'project_id': 'dummyproject',
'vip_address': '192.0.2.2',
'provisioning_status': 'ONLINE',
}
class FakeLoadBalancerv2Client(object):
def __init__(self, **kwargs):
self.load_balancers = mock.Mock()
self.load_balancers.resource_class = fakes.FakeResource(None, {})
self.auth_token = kwargs['token']
self.management_url = kwargs['endpoint']
class TestLoadBalancerv2(utils.TestCommand):
def setUp(self):
super(TestLoadBalancerv2, self).setUp()
self.app.client_manager.load_balancer = FakeLoadBalancerv2Client(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,
)
class FakeLoadBalancer(object):
"""Fake one or more load balancers."""
@staticmethod
def create_one_load_balancer(attrs=None):
"""Create one load balancer.
:param Dictionary attrs:
A dictionary with all load balancer attributes
:return:
A FakeResource object
"""
attrs = attrs or {}
# Set default attribute
lb_info = {
'id': str(uuid.uuid4()),
'name': 'lb-name-' + uuid.uuid4().hex,
'project_id': uuid.uuid4().hex,
'vip_address': '192.0.2.2',
'provisioning_status': 'ONLINE',
}
lb_info.update(attrs)
lb = fakes.FakeResource(
info=copy.deepcopy(lb_info),
loaded=True)
return lb

81
octaviaclient/tests/unit/osc/v2/test_load_balancer.py

@ -0,0 +1,81 @@
# 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 octaviaclient.osc.v2 import load_balancer as load_balancer
from octaviaclient.tests.unit.osc.v2 import fakes as lb_fakes
AUTH_TOKEN = "foobar"
AUTH_URL = "http://192.0.2.2"
class TestLoadBalancer(lb_fakes.TestLoadBalancerv2):
def setUp(self):
super(TestLoadBalancer, self).setUp()
self.lb_mock = self.app.client_manager.load_balancer.load_balancers
self.lb_mock.reset_mock()
class TestLoadBalancerList(TestLoadBalancer):
_lb = lb_fakes.FakeLoadBalancer.create_one_load_balancer()
columns = (
'ID',
'Name',
'Project ID',
'VIP Address',
'Provisioning Status',
)
datalist = (
(
_lb.id,
_lb.name,
_lb.project_id,
_lb.vip_address,
_lb.provisioning_status,
),
)
info = {
'id': _lb.id,
'name': _lb.name,
'project_id': _lb.project_id,
'vip_address': _lb.vip_address,
'provisioning_status': _lb.provisioning_status,
}
lb_info = copy.deepcopy(info)
def setUp(self):
super(TestLoadBalancerList, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.load_balancer_list.return_value = [self.lb_info]
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.ListLoadBalancer(self.app, None)
def test_load_balancer_list_no_options(self):
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_list.assert_called_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))

75
octaviaclient/tests/utils.py

@ -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 octaviaclient.tests 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

4
releasenotes/notes/initial-list-command-90c9fa39fc10540e.yaml

@ -0,0 +1,4 @@
---
features:
- |
Add initial load balancer command ``loadbalancer list``.

5
setup.cfg

@ -24,7 +24,10 @@ packages =
[entry_points]
openstack.cli.extension =
load-balancer = octaviaclient.osc.plugin
load_balancer = octaviaclient.osc.plugin
openstack.load_balancer.v2 =
loadbalancer_list = octaviaclient.osc.v2.load_balancer:ListLoadBalancer
[build_sphinx]
source-dir = doc/source

5
test-requirements.txt

@ -3,9 +3,12 @@
# process, which may cause wedges in the gate later.
hacking>=0.12.0,!=0.13.0,<0.14 # Apache-2.0
requests!=2.12.2,!=2.13.0,>=2.10.0 # Apache-2.0
requests-mock>=1.1 # Apache-2.0
coverage>=4.0 # Apache-2.0
mock>=2.0 # BSD
python-subunit>=0.0.18 # Apache-2.0/BSD
python-openstackclient>=3.3.0 # Apache-2.0
sphinx>=1.5.1 # BSD
oslosphinx>=4.7.0 # Apache-2.0
oslotest>=1.10.0 # Apache-2.0

5
tox.ini

@ -29,7 +29,10 @@ commands =
sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html
[testenv:debug]
commands = oslo_debug_helper {posargs}
passenv = OS_*
commands =
pip install -q -U ipdb
oslo_debug_helper -t octaviaclient/tests {posargs}
[flake8]
# E123, E125 skipped as they are invalid PEP-8.

Loading…
Cancel
Save