Browse Source

Add pool commands to client

This patch implements pool create, list,
show, set, and delete within the openstack client.

Change-Id: I08852b0e612ef8cd9e501c6de4d45f546374ccda
changes/92/461992/14
Jude Cross 4 years ago
parent
commit
755eaedb0b
  1. 8
      doc/source/data/lbaas.csv
  2. 165
      doc/source/usage/osc/v2/load-balancer.rst
  3. 68
      octaviaclient/api/load_balancer_v2.py
  4. 27
      octaviaclient/osc/v2/constants.py
  5. 19
      octaviaclient/osc/v2/listener.py
  6. 231
      octaviaclient/osc/v2/pool.py
  7. 67
      octaviaclient/osc/v2/utils.py
  8. 58
      octaviaclient/tests/unit/api/test_load_balancer.py
  9. 30
      octaviaclient/tests/unit/osc/v2/fakes.py
  10. 7
      octaviaclient/tests/unit/osc/v2/test_listener.py
  11. 198
      octaviaclient/tests/unit/osc/v2/test_pool.py
  12. 5
      setup.cfg

8
doc/source/data/lbaas.csv

@ -31,7 +31,7 @@ lbaas-member-delete,,LBaaS v2 Delete a given member.
lbaas-member-list,,LBaaS v2 List members that belong to a given pool.
lbaas-member-show,,LBaaS v2 Show information of a given member.
lbaas-member-update,,LBaaS v2 Update a given member.
lbaas-pool-create,,LBaaS v2 Create a pool.
lbaas-pool-delete,,LBaaS v2 Delete a given pool.
lbaas-pool-list,,LBaaS v2 List pools that belong to a given tenant.
lbaas-pool-show,,LBaaS v2 Show information of a given pool.
lbaas-pool-create,loadbalancer pool create,LBaaS v2 Create a pool.
lbaas-pool-delete,loadbalancer pool delete,LBaaS v2 Delete a given pool.
lbaas-pool-list,loadbalancer pool list,LBaaS v2 List pools that belong to a given tenant.
lbaas-pool-show,loadbalancer pool show,LBaaS v2 Show information of a given pool.

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

@ -348,3 +348,168 @@ Delete a listener
.. describe:: <listener>
Listener to delete (name or ID).
====
pool
====
loadbalancer pool list
----------------------
List pools
.. program:: loadbalancer pool list
.. code:: bash
openstack loadbalancer pool list
loadbalancer pool show
----------------------
Show the details of a single pool
.. program:: loadbalancer pool show
.. code:: bash
openstack loadbalancer pool show
<pool>
.. _loadbalancer_pool_list-pool:
.. describe:: <pool>
Name or UUID of the pool.
loadbalancer pool create
------------------------
Create a pool
.. program:: loadbalancer pool show
.. code:: bash
openstack loadbalancer pool create
[--name <name>]
[--description <description>]
--protocol {'TERMINATED_HTTPS','HTTP','HTTPS','TCP','PROXY'}
[--listener <listener>]
[--loadbalancer <load_balancer>]
[--session-persistence <session persistence>]
--lb-algorithm {'SOURCE_IP','ROUND_ROBIN','LEAST_CONNECTIONS'}
[--project <project>]
[--enable | --disable]
.. option:: --name <name>
Set pool name.
.. option:: --description <description>
Set pool description.
.. option:: --protocol {'TERMINATED_HTTPS','HTTP','HTTPS','TCP','PROXY'}
Set the pool protocol.
.. option:: --listener <listener>
Listener to add the pool to (name or ID).
.. option:: --loadbalancer <load_balancer>
Load balancer to add the pool to (name or ID).
.. option:: --session-persistence <session persistence>
Set the session persistence for the listener (key=value).
.. option:: --lb-algorithm {'SOURCE_IP','ROUND_ROBIN','LEAST_CONNECTIONS'}
Load balancing algorithm to use.
.. option:: --project <project>
Set the project owning this pool (name or ID).
.. option:: --enable
Enable pool (default).
.. option:: --disable
Disable pool.
loadbalancer pool set
---------------------
Update a pool
.. program:: loadbalancer pool set
.. code:: bash
openstack loadbalancer pool set
[--name <name>]
[--description <description>]
[--protocol {'TERMINATED_HTTPS','HTTP','HTTPS','TCP','PROXY'}]
[--loadbalancer <load_balancer>]
[--listener <listener>]
[--session-persistence <session_persistence>]
[--lb-algorithm {'SOURCE_IP','ROUND_ROBIN','LEAST_CONNECTIONS'}]
[--enable | --disable]
<pool>
.. option:: --name <name>
Set the name of the pool.
.. option:: --description <description>
Set the description of the pool.
.. option:: --protocol {'TERMINATED_HTTPS','HTTP','HTTPS','TCP','PROXY'}
Set protocol for the pool.
.. option:: --loadbalancer <load_balancer>
Load balncer to add the pool to (name or ID).
.. option:: --listener <listener>
Listener to add the pool to (name or ID).
.. option:: --session-persistence <session_persistence>
Set the session persistence for the listener (key=value).
.. option:: --lb-algorithm {'SOURCE_IP','ROUND_ROBIN','LEAST_CONNECTIONS'}
Set the load balancing algorithm to use.
.. option:: --enable
Enable pool.
.. option:: --disable
Disable pool.
.. _loadbalancer_pool_set-pool:
.. describe:: <pool>
Pool to update (name or ID).
loadbalancer pool delete
------------------------
Delete a pool
.. program:: loadbalancer pool delete
.. code:: bash
openstack loadbalancer pool delete
<pool>
.. _loadbalancer_pool_delete-pool:
.. describe:: <pool>
Pool to delete (name or ID).

68
octaviaclient/api/load_balancer_v2.py

@ -159,9 +159,75 @@ class APIv2(api.BaseAPI):
:param kwargs:
A dict of arguments to update a listener
:return:
A dict of the updated listener's settings
Response Code from the API
"""
url = const.BASE_SINGLE_LISTENER_URL.format(uuid=listener_id)
response = self.create(url, method='PUT', **kwargs)
return response
def pool_list(self, **kwargs):
"""List all pools
:param kwargs:
Parameters to filter on (not implemented)
:return:
List of pools
"""
url = const.BASE_POOL_URL
pool_list = self.list(url, **kwargs)
return pool_list
def pool_create(self, **kwargs):
"""Create a pool
:param kwargs:
Parameters to create a listener with (expects json=)
:return:
A dict of the created pool's settings
"""
url = const.BASE_POOL_URL
pool = self.create(url, **kwargs)
return pool
def pool_delete(self, pool_id):
"""Delete a pool
:param string pool_id:
ID of of listener to delete
:return:
Response Code from the API
"""
url = const.BASE_SINGLE_POOL_URL.format(uuid=pool_id)
deleted_pool = self.delete(url)
return deleted_pool
def pool_show(self, pool_id):
"""Show a pool's settings
:param string pool_id:
ID of the pool to show
:return:
Dict of the specified pool's settings
"""
pool = self.find(path=const.BASE_POOL_URL, value=pool_id)
return pool
def pool_set(self, pool_id, **kwargs):
"""Update a pool's settings
:param pool_id:
ID of the pool to update
:param kwargs:
A dict of arguments to update a pool
:return:
Response Code from the API
"""
url = const.BASE_SINGLE_POOL_URL.format(uuid=pool_id)
pool = self.create(url, method='PUT', **kwargs)
return pool

27
octaviaclient/osc/v2/constants.py

@ -66,3 +66,30 @@ LISTENER_COLUMNS = (
'protocol',
'protocol_port',
'admin_state_up')
POOL_ROWS = (
'admin_state_up',
'created_at',
'description',
'healthmonitor_id',
'id',
'lb_algorithm',
'listeners',
'loadbalancers',
'members',
'name',
'operating_status',
'project_id',
'protocol',
'provisioning_status',
'session_persistence',
'updated_at')
POOL_COLUMNS = (
'id',
'name',
'project_id',
'provisioning status',
'protocol',
'lb_algorithm',
'admin_state_up')

19
octaviaclient/osc/v2/listener.py

@ -15,7 +15,6 @@
from cliff import lister
import json
from osc_lib.command import command
from osc_lib import utils
@ -60,9 +59,9 @@ class CreateListener(command.ShowOne):
)
parser.add_argument(
'--default-pool',
metavar='<pool-id>',
help="The ID of the pool used by the listener if no L7 policies "
"match."
metavar='<pool>',
help="The name or ID of the pool used by the listener if no "
"L7 policies match."
)
parser.add_argument(
'--default-tls-container-ref',
@ -81,9 +80,7 @@ class CreateListener(command.ShowOne):
)
parser.add_argument(
'--insert-headers',
metavar='<header=value>',
nargs='*',
type=json.loads,
metavar='<header=value,...>',
help="A dictionary of optional headers to insert into the request "
"before it is sent to the backend member."
)
@ -118,7 +115,8 @@ class CreateListener(command.ShowOne):
json=body)
formatters = {'loadbalancers': v2_utils.format_list,
'pools': v2_utils.format_list,
'l7policies': v2_utils.format_list}
'l7policies': v2_utils.format_list,
'insert_headers': v2_utils.format_hash}
return (rows,
(utils.get_dict_properties(data['listener'],
@ -202,7 +200,7 @@ class ShowListener(command.ShowOne):
parser.add_argument(
'listener',
metavar='<listener>',
help='UUID of the listener'
help='Name or UUID of the listener'
)
return parser
@ -219,7 +217,8 @@ class ShowListener(command.ShowOne):
)
formatters = {'loadbalancers': v2_utils.format_list,
'pools': v2_utils.format_list,
'l7policies': v2_utils.format_list}
'l7policies': v2_utils.format_list,
'insert_headers': v2_utils.format_hash}
return (rows,
(utils.get_dict_properties(data, rows, formatters=formatters)))

231
octaviaclient/osc/v2/pool.py

@ -0,0 +1,231 @@
# 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.
#
"""Pool action implementation"""
from cliff import lister
from osc_lib.command import command
from osc_lib import utils
from octaviaclient.osc.v2 import constants as const
from octaviaclient.osc.v2 import utils as v2_utils
class CreatePool(command.ShowOne):
"""Create a pool"""
def get_parser(self, prog_name):
parser = super(CreatePool, self).get_parser(prog_name)
parser.add_argument(
'--name',
metavar='<name>',
help="Set pool name"
)
parser.add_argument(
'--description',
metavar='<description>',
help="Set pool description"
)
parser.add_argument(
'--protocol',
metavar='<protocol>',
required=True,
choices=['TERMINATED_HTTPS', 'HTTP', 'HTTPS', 'TCP', 'PROXY'],
help="Set the pool protocol"
)
parent_group = parser.add_mutually_exclusive_group(required=True)
parent_group.add_argument(
'--listener',
metavar='<listener>',
help="Listener to add to pool (name or ID)"
)
parent_group.add_argument(
'--loadbalancer',
metavar='<load_balancer>',
help="Load balncer to add the pool to (name or ID)"
)
parser.add_argument(
'--session-persistence',
metavar='<session persistence>',
help="Set the session persistence for the listener (key=value)"
)
parser.add_argument(
'--lb-algorithm',
metavar='<lb_algorithm>',
required=True,
choices=['SOURCE_IP', 'ROUND_ROBIN', 'LEAST_CONNECTIONS'],
help="Load balancing algorithm to use"
)
admin_group = parser.add_mutually_exclusive_group()
admin_group.add_argument(
'--enable',
action='store_true',
default=True,
help="Enable pool (default)"
)
admin_group.add_argument(
'--disable',
action='store_true',
default=None,
help="Disable pool"
)
return parser
def take_action(self, parsed_args):
rows = const.POOL_ROWS
attrs = v2_utils.get_pool_attrs(self.app.client_manager, parsed_args)
body = {"pool": attrs}
data = self.app.client_manager.load_balancer.pool_create(
json=body)
formatters = {'loadbalancers': v2_utils.format_list,
'members': v2_utils.format_list,
'listeners': v2_utils.format_list,
'session_persistence': v2_utils.format_hash}
return (rows, (utils.get_dict_properties(
data['pool'], rows, formatters=formatters)))
class DeletePool(command.Command):
"""Delete a pool"""
def get_parser(self, prog_name):
parser = super(DeletePool, self).get_parser(prog_name)
parser.add_argument(
'pool',
metavar="<pool>",
help="Pool to delete (name or ID)"
)
return parser
def take_action(self, parsed_args):
attrs = v2_utils.get_pool_attrs(self.app.client_manager, parsed_args)
pool_id = attrs.pop('pool_id')
self.app.client_manager.load_balancer.pool_delete(
pool_id=pool_id)
class ListPool(lister.Lister):
"""List pools"""
def get_parser(self, prog_name):
parser = super(ListPool, self).get_parser(prog_name)
return parser
def take_action(self, parsed_args):
columns = const.POOL_COLUMNS
data = self.app.client_manager.load_balancer.pool_list()
formatters = {'loadbalancers': v2_utils.format_list,
'members': v2_utils.format_list,
'listeners': v2_utils.format_list}
return (columns,
(utils.get_dict_properties(
s, columns, formatters=formatters) for s in data['pools']))
class ShowPool(command.ShowOne):
"""Show the details of a single pool"""
def get_parser(self, prog_name):
parser = super(ShowPool, self).get_parser(prog_name)
parser.add_argument(
'pool',
metavar='<pool>',
help='Name or UUID of the pool.'
)
return parser
def take_action(self, parsed_args):
rows = const.POOL_ROWS
attrs = v2_utils.get_pool_attrs(self.app.client_manager, parsed_args)
pool_id = attrs.pop('pool_id')
data = self.app.client_manager.load_balancer.pool_show(
pool_id=pool_id,
)
formatters = {'loadbalancers': v2_utils.format_list,
'members': v2_utils.format_list,
'listeners': v2_utils.format_list,
'session_persistence': v2_utils.format_hash}
return (rows, (utils.get_dict_properties(
data, rows, formatters=formatters)))
class SetPool(command.Command):
"""Update a pool"""
def get_parser(self, prog_name):
parser = super(SetPool, self).get_parser(prog_name)
parser.add_argument(
'pool',
metavar="<pool>",
help="Pool to update (name or ID)"
)
parser.add_argument(
'--name',
metavar='<name>',
help="New pool name"
)
parser.add_argument(
'--description',
metavar='<description>',
help="Set pool description"
)
parser.add_argument(
'--session-persistence',
metavar='<session_persistence>',
help="Set the session persistence for the listener (key=value)"
)
parser.add_argument(
'--lb-algorithm',
metavar='<lb_algorithm>',
choices=['SOURCE_IP', 'ROUND_ROBIN', 'LEAST_CONNECTIONS'],
help="Load balancing algorithm to use"
)
admin_group = parser.add_mutually_exclusive_group()
admin_group.add_argument(
'--enable',
action='store_true',
default=None,
help="Enable pool (default)"
)
admin_group.add_argument(
'--disable',
action='store_true',
default=None,
help="Disable pool"
)
return parser
def take_action(self, parsed_args):
attrs = v2_utils.get_pool_attrs(self.app.client_manager, parsed_args)
pool_id = attrs.pop('pool_id')
body = {'pool': attrs}
self.app.client_manager.load_balancer.pool_set(
pool_id, json=body)

67
octaviaclient/osc/v2/utils.py

@ -65,7 +65,6 @@ def get_loadbalancer_attrs(client_manager, parsed_args):
),
'connection_limit': ('connection_limit', str),
'protocol_port': ('protocol_port', int),
'default_pool': ('default_pool_id', str),
'project': (
'project_id',
'project',
@ -128,7 +127,11 @@ def get_listener_attrs(client_manager, parsed_args):
),
'connection_limit': ('connection_limit', str),
'protocol_port': ('protocol_port', int),
'default_pool': ('default_pool_id', str),
'default_pool': (
'default_pool_id',
'pools',
client_manager.load_balancer.pool_list
),
'project': (
'project_id',
'project',
@ -136,7 +139,7 @@ def get_listener_attrs(client_manager, parsed_args):
),
'enable': ('admin_state_up', lambda x: True),
'disable': ('admin_state_up', lambda x: False),
'insert_headers': ('insert_headers', _format_headers)
'insert_headers': ('insert_headers', _format_kv)
}
_attrs = vars(parsed_args)
@ -145,20 +148,64 @@ def get_listener_attrs(client_manager, parsed_args):
return attrs
def _format_headers(headers):
formatted_headers = {}
headers = headers.split(',')
for header in headers:
k, v = header.split('=')
formatted_headers[k] = v
def get_pool_attrs(client_manager, parsed_args):
attr_map = {
'name': ('name', str),
'description': ('description', str),
'protocol': ('protocol', str),
'pool': (
'pool_id',
'pools',
client_manager.load_balancer.pool_list
),
'loadbalancer': (
'loadbalancer_id',
'loadbalancers',
client_manager.load_balancer.load_balancer_list
),
'lb_algorithm': ('lb_algorithm', str),
'listener': (
'listener_id',
'listeners',
client_manager.load_balancer.listener_list
),
'project': (
'project_id',
'project',
client_manager.identity
),
'session_persistence': ('session_persistence', _format_kv),
'enable': ('admin_state_up', lambda x: True),
'disable': ('admin_state_up', lambda x: False)
}
_attrs = vars(parsed_args)
attrs = _map_attrs(_attrs, attr_map)
return formatted_headers
return attrs
def format_list(data):
return '\n'.join(i['id'] for i in data)
def format_hash(data):
if data:
return '\n'.join('{}={}'.format(k, v) for k, v in data.items())
else:
return None
def _format_kv(data):
formatted_kv = {}
values = data.split(',')
for value in values:
k, v = value.split('=')
formatted_kv[k] = v
return formatted_kv
def _map_attrs(attrs, attr_map):
mapped_attrs = {}
for k, v in attrs.items():

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

@ -26,6 +26,7 @@ FAKE_URL = 'http://example.com/v2.0/lbaas/'
FAKE_LB = uuidutils.generate_uuid()
FAKE_LI = uuidutils.generate_uuid()
FAKE_PO = uuidutils.generate_uuid()
LIST_LB_RESP = {
'loadbalancers':
@ -39,6 +40,11 @@ LIST_LI_RESP = {
{'name': 'lb2'}]
}
LIST_PO_RESP = {
'pools':
[{'name': 'po1'},
{'name': 'po2'}]
}
SINGLE_LB_RESP = {'loadbalancer': {'id': FAKE_LB, 'name': 'lb1'}}
SINGLE_LB_UPDATE = {"loadbalancer": {"admin_state_up": False}}
@ -46,6 +52,9 @@ SINGLE_LB_UPDATE = {"loadbalancer": {"admin_state_up": False}}
SINGLE_LI_RESP = {'listener': {'id': FAKE_LI, 'name': 'li1'}}
SINGLE_LI_UPDATE = {"listener": {"admin_state_up": False}}
SINGLE_PO_RESP = {'pool': {'id': FAKE_PO, 'name': 'li1'}}
SINGLE_PO_UPDATE = {"pool": {"admin_state_up": False}}
class TestLoadBalancerv2(utils.TestCase):
@ -155,3 +164,52 @@ class TestLoadBalancer(TestLoadBalancerv2):
)
ret = self.api.listener_delete(FAKE_LI)
self.assertEqual(200, ret.status_code)
def test_list_pool_no_options(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + 'pools',
json=LIST_PO_RESP,
status_code=200,
)
ret = self.api.pool_list()
self.assertEqual(LIST_PO_RESP, ret)
def test_show_pool(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + 'pools/' + FAKE_PO,
json=SINGLE_PO_RESP,
status_code=200
)
ret = self.api.pool_show(FAKE_PO)
self.assertEqual(SINGLE_PO_RESP['pool'], ret)
def test_create_pool(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + 'pools',
json=SINGLE_PO_RESP,
status_code=200
)
ret = self.api.pool_create(json=SINGLE_PO_RESP)
self.assertEqual(SINGLE_PO_RESP, ret)
def test_set_pool(self):
self.requests_mock.register_uri(
'PUT',
FAKE_URL + 'pools/' + FAKE_PO,
json=SINGLE_PO_UPDATE,
status_code=200
)
ret = self.api.pool_set(FAKE_PO, json=SINGLE_PO_UPDATE)
self.assertEqual(SINGLE_PO_UPDATE, ret)
def test_delete_pool(self):
self.requests_mock.register_uri(
'DELETE',
FAKE_URL + 'pools/' + FAKE_PO,
status_code=200
)
ret = self.api.pool_delete(FAKE_PO)
self.assertEqual(200, ret.status_code)

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

@ -107,3 +107,33 @@ class FakeListener(object):
loaded=True)
return li
class FakePool(object):
"""Fake one or more listeners."""
@staticmethod
def create_one_pool(attrs=None):
attrs = attrs or {}
po_info = {
'admin_state_up': True,
'description': 'fake desc',
'id': str(uuid.uuid4()),
'lb_algorithm': 'ROUND_ROBIN',
'listeners': [{'id': str(uuid.uuid4())}],
'loadbalancers': [{'id': str(uuid.uuid4())}],
'members': [{'id': str(uuid.uuid4())}],
'name': 'po-name-' + uuid.uuid4().hex,
'project_id': uuid.uuid4().hex,
'protocol': 'HTTP',
'provisioning_status': 'ACTIVE',
}
po_info.update(attrs)
po = fakes.FakeResource(
info=copy.deepcopy(po_info),
loaded=True)
return po

7
octaviaclient/tests/unit/osc/v2/test_listener.py

@ -63,7 +63,6 @@ class TestListener(li_fakes.TestLoadBalancerv2):
'admin_state_up': _li.admin_state_up
}]
}
li_info = copy.deepcopy(info)
def setUp(self):
@ -140,7 +139,7 @@ class TestListenerCreate(TestListener):
super(TestListenerCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.listener_create.return_value = {
'listener': self.li_info}
'listener': self.li_info['listeners'][0]}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -148,7 +147,7 @@ class TestListenerCreate(TestListener):
@mock.patch('octaviaclient.osc.v2.utils.get_listener_attrs')
def test_listener_create(self, mock_client):
mock_client.return_value = self.li_info
mock_client.return_value = self.li_info['listeners'][0]
arglist = ['mock_lb_id',
'--name', self._li.name,
'--protocol', 'HTTP',
@ -163,7 +162,7 @@ class TestListenerCreate(TestListener):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.listener_create.assert_called_with(
json={'listener': self.li_info})
json={'listener': self.li_info['listeners'][0]})
class TestListenerShow(TestListener):

198
octaviaclient/tests/unit/osc/v2/test_pool.py

@ -0,0 +1,198 @@
# 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 osc_lib import exceptions
from octaviaclient.osc.v2 import pool as pool
from octaviaclient.tests.unit.osc.v2 import fakes as po_fakes
AUTH_TOKEN = "foobar"
AUTH_URL = "http://192.0.2.2"
class TestPool(po_fakes.TestLoadBalancerv2):
_po = po_fakes.FakePool.create_one_pool()
columns = ('id',
'name',
'project_id',
'provisioning status',
'protocol',
'lb_algorithm',
'admin_state_up')
datalist = (
(
_po.id,
_po.name,
_po.project_id,
_po.provisioning_status,
_po.protocol,
_po.lb_algorithm,
True
),
)
info = {
'pools':
[{'id': _po.id,
'name': _po.name,
'project_id': _po.project_id,
'provisioning_status': _po.provisioning_status,
'members': _po.members,
'protocol': _po.protocol,
'lb_algorithm': _po.lb_algorithm,
'loadbalancers': _po.loadbalancers,
'listeners': _po.listeners,
'pool_id': _po.id,
'admin_state_up': True,
'session_persistance': {'k': 'v'}
}]
}
po_info = copy.deepcopy(info)
def setUp(self):
super(TestPool, self).setUp()
self.li_mock = self.app.client_manager.load_balancer.load_balancers
self.li_mock.reset_mock()
self.api_mock = mock.Mock()
self.api_mock.pool_list.return_value = self.po_info
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
class TestPoolList(TestPool):
def setUp(self):
super(TestPoolList, self).setUp()
self.cmd = pool.ListPool(self.app, None)
def test_pool_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.pool_list.assert_called_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
class TestPoolDelete(TestPool):
def setUp(self):
super(TestPoolDelete, self).setUp()
self.cmd = pool.DeletePool(self.app, None)
def test_pool_delete(self):
arglist = [self._po.id]
verifylist = [
('pool', self._po.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.pool_delete.assert_called_with(
pool_id=self._po.id)
def test_listener_delete_failure(self):
arglist = ['unknown_pool']
verifylist = [
('pool', 'unknown_pool')
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(exceptions.CommandError, self.cmd.take_action,
parsed_args)
self.assertNotCalled(self.api_mock.pool_delete)
class TestPoolCreate(TestPool):
def setUp(self):
super(TestPoolCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.pool_create.return_value = {
'pool': self.po_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = pool.CreatePool(self.app, None)
@mock.patch('octaviaclient.osc.v2.utils.get_pool_attrs')
def test_pool_create(self, mock_attrs):
mock_attrs.return_value = self.po_info
arglist = ['--loadbalancer', 'mock_lb_id',
'--name', self._po.name,
'--protocol', 'HTTP',
'--lb-algorithm', 'ROUND_ROBIN']
verifylist = [
('loadbalancer', 'mock_lb_id'),
('name', self._po.name),
('protocol', 'HTTP'),
('lb_algorithm', 'ROUND_ROBIN')
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.pool_create.assert_called_with(
json={'pool': self.po_info})
class TestPoolShow(TestPool):
def setUp(self):
super(TestPoolShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.pool_list.return_value = self.po_info
self.api_mock.pool_show.return_value = {
'pool': self.po_info['pools'][0]}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = pool.ShowPool(self.app, None)
def test_pool_show(self,):
arglist = [self._po.id]
verifylist = [
('pool', self._po.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.pool_show.assert_called_with(pool_id=self._po.id)
class TestPoolSet(TestPool):
def setUp(self):
super(TestPoolSet, self).setUp()
self.cmd = pool.SetPool(self.app, None)
def test_pool_set(self):
arglist = [self._po.id, '--name', 'new_name']
verifylist = [
('pool', self._po.id),
('name', 'new_name')
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.pool_set.assert_called_with(
self._po.id, json={'pool': {'name': 'new_name'}})

5
setup.cfg

@ -37,6 +37,11 @@ openstack.load_balancer.v2 =
loadbalancer_listener_show = octaviaclient.osc.v2.listener:ShowListener
loadbalancer_listener_delete = octaviaclient.osc.v2.listener:DeleteListener
loadbalancer_listener_set = octaviaclient.osc.v2.listener:SetListener
loadbalancer_pool_create = octaviaclient.osc.v2.pool:CreatePool
loadbalancer_pool_list = octaviaclient.osc.v2.pool:ListPool
loadbalancer_pool_show = octaviaclient.osc.v2.pool:ShowPool
loadbalancer_pool_delete = octaviaclient.osc.v2.pool:DeletePool
loadbalancer_pool_set = octaviaclient.osc.v2.pool:SetPool
[build_sphinx]
source-dir = doc/source

Loading…
Cancel
Save