openstack
/
python-octaviaclient
Code Issues Proposed changes

Improve unit testing scalability 

The use of class members in unittest's TestCase derivatives
is discouraged, and the abundance of class members in the
testing code base has crippled the ability to add tests.

This patch addresses the scalability limitations by removing all class
members and relocates them in the relevant setUp function.

Change-Id: Ic0380a44749ea9bb506f67d876c3ecb7d363e726
This commit is contained in:
Bar Elharar 2017-12-31 04:59:54 +02:00
parent f06396cb87
commit c5bb79c8ec
11 changed files with 181 additions and 493 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
octaviaclient/tests/fakes.py
View File

@ -13,6 +13,7 @@
# under the License.
#
import copy
import json
import mock
import sys
@ -217,7 +218,7 @@ class FakeResource(object):
return list(self._info)
def to_dict(self):
return self._info
return copy.deepcopy(self._info)
@property
def info(self):

1
octaviaclient/tests/unit/osc/v2/constants.py
View File

@ -62,6 +62,7 @@ LISTENER_ATTRS = {
"protocol_port": 80,
"provisioning_status": "ACTIVE",
"default_pool_id": None,
"loadbalancers": None,
"connection_limit": 10,
"admin_state_up": True,
"default_tls_container_ref": uuidutils.generate_uuid(dashed=True),

86
octaviaclient/tests/unit/osc/v2/test_amphora.py
View File

@ -18,88 +18,30 @@ import osc_lib.tests.utils as osc_test_utils
from octaviaclient.osc.v2 import amphora
from octaviaclient.osc.v2 import constants
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestAmphora(fakes.TestOctaviaClient):
_amp = fakes.createFakeResource('amphora')
columns = constants.AMPHORA_COLUMNS
rows = constants.AMPHORA_ROWS
data_show = (
(
_amp.id,
_amp.loadbalancer_id,
_amp.compute_id,
_amp.lb_network_ip,
_amp.vrrp_ip,
_amp.ha_ip,
_amp.vrrp_port_id,
_amp.ha_port_id,
_amp.cert_expiration,
_amp.cert_busy,
_amp.role,
_amp.status,
_amp.vrrp_interface,
_amp.vrrp_id,
_amp.vrrp_priority,
_amp.cached_zone,
),
)
info_show = {
'id': _amp.id,
'loadbalancer_id': _amp.loadbalancer_id,
'compute_id': _amp.compute_id,
'lb_network_ip': _amp.lb_network_ip,
'vrrp_ip': _amp.vrrp_ip,
'ha_ip': _amp.ha_ip,
'vrrp_port_id': _amp.vrrp_port_id,
'ha_port_id': _amp.ha_port_id,
'cert_expiration': _amp.cert_expiration,
'cert_busy': _amp.cert_busy,
'role': _amp.role,
'status': _amp.status,
'vrrp_interface': _amp.vrrp_interface,
'vrrp_id': _amp.vrrp_id,
'vrrp_priority': _amp.vrrp_priority,
'cached_zone': _amp.cached_zone,
}
data_list = (
(
_amp.id,
_amp.loadbalancer_id,
_amp.status,
_amp.role,
_amp.lb_network_ip,
_amp.ha_ip,
),
)
info_list = {
'amphorae':
[{
'id': _amp.id,
'loadbalancer_id': _amp.loadbalancer_id,
'status': _amp.status,
'role': _amp.role,
'lb_network_ip': _amp.lb_network_ip,
'ha_ip': _amp.ha_ip,
}],
}
amp_info = copy.deepcopy(info_list)
def setUp(self):
super(TestAmphora, self).setUp()
self._amp = fakes.createFakeResource('amphora')
self.amp_info = copy.deepcopy(attr_consts.AMPHORA_ATTRS)
self.columns = copy.deepcopy(constants.AMPHORA_COLUMNS)
self.rows = copy.deepcopy(constants.AMPHORA_ROWS)
info_list = {'amphorae': [
{k: v for k, v in attr_consts.AMPHORA_ATTRS.items() if (
k in self.columns)},
]}
self.api_mock = mock.Mock()
self.api_mock.amphora_list.return_value = self.amp_info
self.api_mock.amphora_list.return_value = info_list
self.api_mock.amphora_show.return_value = {
"amphora": self.amp_info['amphorae'][0],
"amphora": info_list['amphorae'][0],
}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -108,6 +50,8 @@ class TestAmphoraList(TestAmphora):
def setUp(self):
super(TestAmphoraList, self).setUp()
self.data_list = (tuple(
attr_consts.AMPHORA_ATTRS[k] for k in self.columns),)
self.cmd = amphora.ListAmphora(self.app, None)
def test_amphora_list_no_options(self):

63
octaviaclient/tests/unit/osc/v2/test_health_monitor.py
View File

@ -16,54 +16,27 @@ import mock
from osc_lib import exceptions
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import health_monitor
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestHealthMonitor(fakes.TestOctaviaClient):
_hm = fakes.createFakeResource('hm')
columns = ('id', 'name', 'project_id', 'type', 'admin_state_up')
datalist = (
(
_hm.id,
_hm.name,
_hm.project_id,
_hm.type,
_hm.admin_state_up
),
)
info = {
'healthmonitors':
[{
"project_id": _hm.project_id,
"name": _hm.name,
"admin_state_up": True,
"pools": _hm.pools,
"created_at": _hm.created_at,
"delay": _hm.delay,
"expected_codes": _hm.expected_codes,
"max_retries": _hm.max_retries,
"http_method": _hm.http_method,
"timeout": _hm.timeout,
"max_retries_down": _hm.max_retries_down,
"url_path": _hm.url_path,
"type": _hm.type,
"id": _hm.id
}]
}
hm_info = copy.deepcopy(info)
def setUp(self):
super(TestHealthMonitor, self).setUp()
self.li_mock = self.app.client_manager.load_balancer.load_balancers
self.li_mock.reset_mock()
self._hm = fakes.createFakeResource('hm')
self.hm_info = copy.deepcopy(attr_consts.HM_ATTRS)
self.columns = copy.deepcopy(constants.MONITOR_COLUMNS)
info_list = {'healthmonitors': [
{k: v for k, v in attr_consts.HM_ATTRS.items() if (
k in self.columns)},
]}
self.api_mock = mock.Mock()
self.api_mock.health_monitor_list.return_value = self.hm_info
self.api_mock.health_monitor_list.return_value = info_list
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -72,6 +45,7 @@ class TestHealthMonitorList(TestHealthMonitor):
def setUp(self):
super(TestHealthMonitorList, self).setUp()
self.datalist = (tuple(attr_consts.HM_ATTRS[k] for k in self.columns),)
self.cmd = health_monitor.ListHealthMonitor(self.app, None)
def test_health_monitor_list_no_options(self):
@ -120,7 +94,7 @@ class TestHealthMonitorCreate(TestHealthMonitor):
super(TestHealthMonitorCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.health_monitor_create.return_value = {
'healthmonitor': self.hm_info['healthmonitors'][0]}
'healthmonitor': self.hm_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -128,7 +102,7 @@ class TestHealthMonitorCreate(TestHealthMonitor):
@mock.patch('octaviaclient.osc.v2.utils.get_health_monitor_attrs')
def test_health_monitor_create(self, mock_client):
mock_client.return_value = self.hm_info['healthmonitors'][0]
mock_client.return_value = self.hm_info
arglist = ['mock_pool_id',
'--name', self._hm.name,
'--delay', str(self._hm.delay),
@ -149,17 +123,16 @@ class TestHealthMonitorCreate(TestHealthMonitor):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.health_monitor_create.assert_called_with(
json={'healthmonitor': self.hm_info['healthmonitors'][0]})
json={'healthmonitor': self.hm_info})
class TestHealthMonitorShow(TestHealthMonitor):
def setUp(self):
super(TestHealthMonitorShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.health_monitor_list.return_value = self.hm_info
self.api_mock.health_monitor_show.return_value = (
self.hm_info['healthmonitors'][0])
self.api_mock.health_monitor_show.return_value = {
'healthmonitor': self.hm_info,
}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock

55
octaviaclient/tests/unit/osc/v2/test_l7policy.py
View File

@ -16,58 +16,25 @@ import mock
from osc_lib import exceptions
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import l7policy
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestL7Policy(fakes.TestOctaviaClient):
_l7po = fakes.createFakeResource('l7policy')
columns = (
'id',
'name',
'project_id',
'provisioning_status',
'action',
'position',
'admin_state_up'
)
datalist = (
(
_l7po.id,
_l7po.name,
_l7po.project_id,
_l7po.provisioning_status,
_l7po.action,
_l7po.position,
_l7po.admin_state_up
),
)
info = {'l7policies': [{
'listener_id': _l7po.listener_id,
'description': _l7po.description,
'admin_state_up': _l7po.admin_state_up,
'rules': _l7po.rules,
'provisioning_status': _l7po.provisioning_status,
'redirect_pool_id': _l7po.redirect_pool_id,
'action': _l7po.action,
'position': _l7po.position,
'project_id': _l7po.project_id,
'id': _l7po.id,
'name': _l7po.name
}]}
l7po_info = copy.deepcopy(info)
def setUp(self):
super(TestL7Policy, self).setUp()
self.l7po_mock = self.app.client_manager.load_balancer.load_balancers
self.l7po_mock.reset_mock()
self._l7po = fakes.createFakeResource('l7policy')
self.l7po_info = copy.deepcopy(attr_consts.L7POLICY_ATTRS)
self.columns = copy.deepcopy(constants.L7POLICY_COLUMNS)
self.api_mock = mock.Mock()
self.api_mock.l7policy_list.return_value = self.l7po_info
self.api_mock.l7policy_list.return_value = copy.deepcopy(
{'l7policies': [attr_consts.L7POLICY_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -76,6 +43,8 @@ class TestL7PolicyList(TestL7Policy):
def setUp(self):
super(TestL7PolicyList, self).setUp()
self.datalist = (tuple(
attr_consts.L7POLICY_ATTRS[k] for k in self.columns),)
self.cmd = l7policy.ListL7Policy(self.app, None)
def test_l7policy_list_no_options(self):
@ -122,7 +91,6 @@ class TestL7PolicyCreate(TestL7Policy):
def setUp(self):
super(TestL7PolicyCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.l7policy_create.return_value = {
'l7policy': self.l7po_info}
lb_client = self.app.client_manager
@ -165,7 +133,6 @@ class TestL7PolicyShow(TestL7Policy):
def setUp(self):
super(TestL7PolicyShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.l7policy_list.return_value = [{'id': self._l7po.id}]
self.api_mock.l7policy_show.return_value = self.l7po_info
lb_client = self.app.client_manager

81
octaviaclient/tests/unit/osc/v2/test_l7rule.py
View File

@ -14,68 +14,28 @@
import copy
import mock
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import l7rule
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
from octaviaclient.tests.unit.osc.v2 import test_l7policy
class TestL7Rule(fakes.TestOctaviaClient):
_l7ru = fakes.createFakeResource('l7rule')
_l7po = fakes.createFakeResource('l7policy')
columns = (
'id',
'project_id',
'provisioning_status',
'compare_type',
'type',
'key',
'value',
'invert',
'admin_state_up')
datalist = (
(
_l7ru.id,
_l7ru.project_id,
_l7ru.provisioning_status,
_l7ru.compare_type,
_l7ru.type,
_l7ru.key,
_l7ru.value,
_l7ru.invert,
_l7ru.admin_state_up
),
)
info = {'rules': [{
"provisioning_status": _l7ru.provisioning_status,
"compare_type": _l7ru.compare_type,
"type": _l7ru.type,
"key": _l7ru.key,
"project_id": _l7ru.project_id,
"id": _l7ru.id,
"value": _l7ru.value,
'l7rule_id': _l7ru.id,
'l7policy_id': _l7po.id,
'admin_state_up': _l7ru.admin_state_up,
'invert': _l7ru.invert
}]}
po_info = test_l7policy.TestL7Policy.info
l7po_info = copy.deepcopy(po_info)
l7ru_info = copy.deepcopy(info)
def setUp(self):
super(TestL7Rule, self).setUp()
self.l7ru_mock = self.app.client_manager.load_balancer.load_balancers
self.l7ru_mock.reset_mock()
self._l7ru = fakes.createFakeResource('l7rule')
self.l7rule_info = copy.deepcopy(attr_consts.L7RULE_ATTRS)
self.columns = copy.deepcopy(constants.L7RULE_COLUMNS)
self._l7po = fakes.createFakeResource('l7policy')
self.api_mock = mock.Mock()
self.api_mock.l7rule_list.return_value = self.l7ru_info
self.api_mock.l7pool_list.return_value = self.l7po_info
self.api_mock.l7rule_list.return_value = copy.deepcopy(
{'rules': [attr_consts.L7RULE_ATTRS]})
self.api_mock.l7policy_list.return_value = copy.deepcopy(
{'l7policies': [attr_consts.L7POLICY_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -84,11 +44,13 @@ class TestL7RuleList(TestL7Rule):
def setUp(self):
super(TestL7RuleList, self).setUp()
self.datalist = (tuple(
attr_consts.L7RULE_ATTRS[k] for k in self.columns),)
self.cmd = l7rule.ListL7Rule(self.app, None)
@mock.patch('octaviaclient.osc.v2.utils.get_l7rule_attrs')
def test_l7rule_list_no_options(self, mock_attrs):
mock_attrs.return_value = self.l7ru_info['rules'][0]
mock_attrs.return_value = {'l7policy_id': self._l7po.id}
arglist = [self._l7po.id]
verifylist = [('l7policy', self._l7po.id)]
@ -108,7 +70,10 @@ class TestL7RuleDelete(TestL7Rule):
@mock.patch('octaviaclient.osc.v2.utils.get_l7rule_attrs')
def test_l7rule_delete(self, mock_attrs):
mock_attrs.return_value = self.l7ru_info['rules'][0]
mock_attrs.return_value = {
'l7policy_id': self._l7po.id,
'l7rule_id': self._l7ru.id,
}
arglist = [self._l7po.id, self._l7ru.id]
verifylist = [
('l7policy', self._l7po.id),
@ -127,9 +92,8 @@ class TestL7RuleCreate(TestL7Rule):
def setUp(self):
super(TestL7RuleCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.l7rule_create.return_value = {
'rule': self.l7ru_info}
'rule': self.l7rule_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -170,10 +134,7 @@ class TestL7RuleShow(TestL7Rule):
def setUp(self):
super(TestL7RuleShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.l7policy_list.return_value = self.l7po_info
self.api_mock.l7rule_list.return_value = self.l7ru_info
self.api_mock.l7rule_show.return_value = self.l7ru_info['rules'][0]
self.api_mock.l7rule_show.return_value = self.l7rule_info
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock

130
octaviaclient/tests/unit/osc/v2/test_listener.py
View File

@ -16,62 +16,24 @@ import mock
from osc_lib import exceptions
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import listener
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestListener(fakes.TestOctaviaClient):
_li = fakes.createFakeResource('listener')
columns = (
'id',
'default_pool_id',
'name',
'project_id',
'protocol',
'protocol_port',
'admin_state_up',
)
datalist = (
(
_li.id,
_li.default_pool_id,
_li.name,
_li.project_id,
_li.protocol,
_li.protocol_port,
_li.admin_state_up,
),
)
info = {
'listeners':
[{
'id': _li.id,
'name': _li.name,
'project_id': _li.project_id,
'loadbalancers': None,
'provisioning_status': _li.provisioning_status,
'default_pool_id': _li.default_pool_id,
'connection_limit': _li.connection_limit,
'protocol': _li.protocol,
'protocol_port': _li.protocol_port,
'admin_state_up': _li.admin_state_up,
'default_tls_container_ref': _li.default_tls_container_ref,
'sni_container_refs': _li.sni_container_refs
}]
}
li_info = copy.deepcopy(info)
def setUp(self):
super(TestListener, self).setUp()
self.li_mock = self.app.client_manager.load_balancer.load_balancers
self.li_mock.reset_mock()
self._listener = fakes.createFakeResource('listener')
self.listener_info = copy.deepcopy(attr_consts.LISTENER_ATTRS)
self.columns = copy.deepcopy(constants.LISTENER_COLUMNS)
self.api_mock = mock.Mock()
self.api_mock.listener_list.return_value = self.li_info
self.api_mock.listener_list.return_value = copy.deepcopy(
{'listeners': [attr_consts.LISTENER_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -80,6 +42,8 @@ class TestListenerList(TestListener):
def setUp(self):
super(TestListenerList, self).setUp()
self.datalist = (tuple(
attr_consts.LISTENER_ATTRS[k] for k in self.columns),)
self.cmd = listener.ListListener(self.app, None)
def test_listener_list_no_options(self):
@ -112,15 +76,15 @@ class TestListenerDelete(TestListener):
self.cmd = listener.DeleteListener(self.app, None)
def test_listener_delete(self):
arglist = [self._li.id]
arglist = [self._listener.id]
verifylist = [
('listener', self._li.id)
('listener', self._listener.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.listener_delete.assert_called_with(
listener_id=self._li.id)
listener_id=self._listener.id)
def test_listener_delete_failure(self):
arglist = ['unknown_lb']
@ -137,9 +101,8 @@ class TestListenerCreate(TestListener):
def setUp(self):
super(TestListenerCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.listener_create.return_value = {
'listener': self.li_info['listeners'][0]}
'listener': self.listener_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -147,14 +110,14 @@ 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['listeners'][0]
mock_client.return_value = self.listener_info
arglist = ['mock_lb_id',
'--name', self._li.name,
'--name', self._listener.name,
'--protocol', 'HTTP',
'--protocol-port', '80']
verifylist = [
('loadbalancer', 'mock_lb_id'),
('name', self._li.name),
('name', self._listener.name),
('protocol', 'HTTP'),
('protocol_port', '80')
]
@ -162,56 +125,56 @@ 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['listeners'][0]})
json={'listener': self.listener_info})
@mock.patch('octaviaclient.osc.v2.utils.get_listener_attrs')
def test_tls_listener_create(self, mock_client):
mock_client.return_value = self.li_info['listeners'][0]
mock_client.return_value = self.listener_info
arglist = ['mock_lb_id',
'--name', self._li.name,
'--name', self._listener.name,
'--protocol', 'TERMINATED_HTTPS'.lower(),
'--protocol-port', '443',
'--sni-container-refs',
self._li.sni_container_refs[0],
self._li.sni_container_refs[1],
self._listener.sni_container_refs[0],
self._listener.sni_container_refs[1],
'--default-tls-container-ref',
self._li.default_tls_container_ref]
self._listener.default_tls_container_ref]
verifylist = [
('loadbalancer', 'mock_lb_id'),
('name', self._li.name),
('name', self._listener.name),
('protocol', 'TERMINATED_HTTPS'),
('protocol_port', '443'),
('sni_container_refs', self._li.sni_container_refs),
('default_tls_container_ref', self._li.default_tls_container_ref)
('sni_container_refs', self._listener.sni_container_refs),
('default_tls_container_ref',
self._listener.default_tls_container_ref)
]
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['listeners'][0]})
json={'listener': self.listener_info})
class TestListenerShow(TestListener):
def setUp(self):
super(TestListenerShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.listener_list.return_value = self.li_info
self.api_mock.listener_show.return_value = self.li_info['listeners'][0]
self.api_mock.listener_show.return_value = self.listener_info
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = listener.ShowListener(self.app, None)
def test_listener_show(self):
arglist = [self._li.id]
arglist = [self._listener.id]
verifylist = [
('listener', self._li.id),
('listener', self._listener.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.listener_show.assert_called_with(listener_id=self._li.id)
self.api_mock.listener_show.assert_called_with(
listener_id=self._listener.id)
class TestListenerSet(TestListener):
@ -221,28 +184,29 @@ class TestListenerSet(TestListener):
self.cmd = listener.SetListener(self.app, None)
def test_listener_set(self):
arglist = [self._li.id, '--name', 'new_name',
arglist = [self._listener.id, '--name', 'new_name',
'--sni-container-refs',
self._li.sni_container_refs[0],
self._li.sni_container_refs[1],
self._listener.sni_container_refs[0],
self._listener.sni_container_refs[1],
'--default-tls-container-ref',
self._li.default_tls_container_ref]
self._listener.default_tls_container_ref]
verifylist = [
('listener', self._li.id),
('listener', self._listener.id),
('name', 'new_name'),
('sni_container_refs', self._li.sni_container_refs),
('default_tls_container_ref', self._li.default_tls_container_ref)
('sni_container_refs', self._listener.sni_container_refs),
('default_tls_container_ref',
self._listener.default_tls_container_ref)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.listener_set.assert_called_with(
self._li.id, json={
self._listener.id, json={
'listener': {
'name': 'new_name',
'sni_container_refs': self._li.sni_container_refs,
'sni_container_refs': self._listener.sni_container_refs,
'default_tls_container_ref':
self._li.default_tls_container_ref
self._listener.default_tls_container_ref
}})
@ -258,12 +222,12 @@ class TestListenerStatsShow(TestListener):
self.cmd = listener.ShowListenerStats(self.app, None)
def test_listener_stats_show(self):
arglist = [self._li.id]
arglist = [self._listener.id]
verifylist = [
('listener', self._li.id),
('listener', self._listener.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.listener_stats_show.assert_called_with(
listener_id=self._li.id)
listener_id=self._listener.id)

75
octaviaclient/tests/unit/osc/v2/test_load_balancer.py
View File

@ -18,54 +18,25 @@ import mock
from osc_lib import exceptions
from oslo_utils import uuidutils
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import load_balancer
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestLoadBalancer(fakes.TestOctaviaClient):
_lb = fakes.createFakeResource('loadbalancer')
columns = (
'id',
'name',
'project_id',
'vip_address',
'provisioning_status',
'provider'
)
datalist = (
(
_lb.id,
_lb.name,
_lb.project_id,
_lb.vip_address,
_lb.provisioning_status,
_lb.provider
),
)
info = {
'loadbalancers':
[{'id': _lb.id,
'name': _lb.name,
'project_id': _lb.project_id,
'vip_address': _lb.vip_address,
'vip_network_id': _lb.vip_network_id,
'provisioning_status': _lb.provisioning_status,
'provider': _lb.provider
}]
}
lb_info = copy.deepcopy(info)
def setUp(self):
super(TestLoadBalancer, self).setUp()
self.lb_mock = self.app.client_manager.load_balancer.load_balancers
self.lb_mock.reset_mock()
self._lb = fakes.createFakeResource('loadbalancer')
self.lb_info = copy.deepcopy(attr_consts.LOADBALANCER_ATTRS)
self.columns = copy.deepcopy(constants.LOAD_BALANCER_COLUMNS)
self.api_mock = mock.Mock()
self.api_mock.load_balancer_list.return_value = self.lb_info
self.api_mock.load_balancer_list.return_value = copy.deepcopy(
{'loadbalancers': [attr_consts.LOADBALANCER_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
lb_client.neutronclient = mock.MagicMock()
@ -75,6 +46,8 @@ class TestLoadBalancerList(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerList, self).setUp()
self.datalist = (tuple(
attr_consts.LOADBALANCER_ATTRS[k] for k in self.columns),)
self.cmd = load_balancer.ListLoadBalancer(self.app, None)
def test_load_balancer_list_no_options(self):
@ -132,9 +105,9 @@ class TestLoadBalancerCreate(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerCreate, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.load_balancer_create.return_value = {
'loadbalancer': self.lb_info['loadbalancers'][0]
'loadbalancer': self.lb_info
}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -143,7 +116,7 @@ class TestLoadBalancerCreate(TestLoadBalancer):
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create(self, mock_client):
mock_client.return_value = self.lb_info['loadbalancers'][0]
mock_client.return_value = self.lb_info
arglist = ['--name', self._lb.name,
'--vip-network-id', self._lb.vip_network_id,
'--project', self._lb.project_id]
@ -156,12 +129,12 @@ class TestLoadBalancerCreate(TestLoadBalancer):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_create.assert_called_with(
json={'loadbalancer': self.lb_info['loadbalancers'][0]})
json={'loadbalancer': self.lb_info})
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create_with_qos_policy(self, mock_client):
qos_policy_id = 'qos_id'
lb_info = copy.deepcopy(self.lb_info['loadbalancers'][0])
lb_info = copy.deepcopy(self.lb_info)
lb_info.update({'vip_qos_policy_id': qos_policy_id})
mock_client.return_value = lb_info
@ -185,10 +158,9 @@ class TestLoadBalancerCreate(TestLoadBalancer):
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create_missing_args(self, mock_client):
# Clone load balancer to avoid race conditions
lb = fakes.createFakeResource('loadbalancer')
attrs_list = lb.to_dict()
attrs_list = self.lb_info
# init missing keys
args = ("vip_subnet_id", "vip_network_id", "vip_port_id")
for a in args:
# init missing keys
@ -198,7 +170,7 @@ class TestLoadBalancerCreate(TestLoadBalancer):
for comb in itertools.combinations(args, n):
# subtract comb's keys from attrs_list
filtered_attrs = {k: v for k, v in attrs_list.items() if (
k not in comb)}
k not in comb)}
mock_client.return_value = filtered_attrs
if not any(k in filtered_attrs for k in args) or all(
k in filtered_attrs for k in ("vip_network_id",
@ -219,10 +191,8 @@ class TestLoadBalancerShow(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.load_balancer_list.return_value = self.lb_info
self.api_mock.load_balancer_show.return_value = (
self.lb_info['loadbalancers'][0])
self.api_mock.load_balancer_show.return_value = {
'loadbalancer': self.lb_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -242,9 +212,6 @@ class TestLoadBalancerSet(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerSet, self).setUp()
self.api_mock.load_balancer_list.return_value = self.lb_info
self.api_mock.load_balancer_show.return_value = {
'loadbalancer': self.lb_info['loadbalancers'][0]}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.SetLoadBalancer(self.app, None)

61
octaviaclient/tests/unit/osc/v2/test_member.py
View File

@ -17,59 +17,28 @@ import mock
import osc_lib.tests.utils as osc_test_utils
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import member
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestMember(fakes.TestOctaviaClient):
_mem = fakes.createFakeResource('member')
columns = (
'id',
'name',
'project_id',
'provisioning_status',
'address',
'protocol_port',
'operating_status',
'weight'
)
datalist = (
(
_mem.id,
_mem.name,
_mem.project_id,
_mem.provisioning_status,
_mem.address,
_mem.protocol_port,
_mem.operating_status,
_mem.weight
),
)
info = {'members': [{
'id': _mem.id,
'name': _mem.name,
'project_id': _mem.project_id,
'provisioning_status': _mem.provisioning_status,
'address': _mem.address,
'protocol_port': _mem.protocol_port,
'operating_status': _mem.operating_status,
'weight': _mem.weight,
'pool_id': _mem.pool_id}]
}
mem_info = copy.deepcopy(info)
def setUp(self):
super(TestMember, self).setUp()
self.mem_mock = self.app.client_manager.load_balancer.load_balancers
self.mem_mock.reset_mock()
self._mem = fakes.createFakeResource('member')
self.mem_info = copy.deepcopy(attr_consts.MEMBER_ATTRS)
self.columns = copy.deepcopy(constants.MEMBER_COLUMNS)
info_list = {'members': [
{k: v for k, v in attr_consts.MEMBER_ATTRS.items() if (
k in self.columns)}
]}
self.api_mock = mock.Mock()
self.api_mock.member_list.return_value = self.mem_info
self.api_mock.member_list.return_value = info_list
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
lb_client.neutronclient = mock.MagicMock()
@ -79,6 +48,8 @@ class TestListMember(TestMember):
def setUp(self):
super(TestListMember, self).setUp()
self.datalist = (tuple(
attr_consts.MEMBER_ATTRS[k] for k in self.columns),)
self.cmd = member.ListMember(self.app, None)
def test_member_list_no_options(self):
@ -189,9 +160,7 @@ class TestMemberShow(TestMember):
def setUp(self):
super(TestMemberShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.member_list.return_value = self.mem_info
self.api_mock.member_show.return_value = self.mem_info['members'][0]
self.api_mock.member_show.return_value = self.mem_info
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock

65
octaviaclient/tests/unit/osc/v2/test_pool.py
View File

@ -16,59 +16,24 @@ import mock
from osc_lib import exceptions
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import pool as pool
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestPool(fakes.TestOctaviaClient):
_po = fakes.createFakeResource('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._po = fakes.createFakeResource('pool')
self.pool_info = copy.deepcopy(attr_consts.POOL_ATTRS)
self.columns = copy.deepcopy(constants.POOL_COLUMNS)
self.api_mock = mock.Mock()
self.api_mock.pool_list.return_value = self.po_info
self.api_mock.pool_list.return_value = copy.deepcopy(
{'pools': [attr_consts.POOL_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -77,6 +42,9 @@ class TestPoolList(TestPool):
def setUp(self):
super(TestPoolList, self).setUp()
self.datalist = (tuple(
attr_consts.POOL_ATTRS[k] for k in self.columns
),)
self.cmd = pool.ListPool(self.app, None)
def test_pool_list_no_options(self):
@ -123,9 +91,8 @@ 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}
'pool': self.pool_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -133,7 +100,7 @@ class TestPoolCreate(TestPool):
@mock.patch('octaviaclient.osc.v2.utils.get_pool_attrs')
def test_pool_create(self, mock_attrs):
mock_attrs.return_value = self.po_info
mock_attrs.return_value = self.pool_info
arglist = ['--loadbalancer', 'mock_lb_id',
'--name', self._po.name,
'--protocol', 'HTTP',
@ -149,16 +116,14 @@ class TestPoolCreate(TestPool):
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})
json={'pool': self.pool_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 = self.po_info['pools'][0]
self.api_mock.pool_show.return_value = self.pool_info
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock

54
octaviaclient/tests/unit/osc/v2/test_quota.py
View File

@ -16,48 +16,25 @@ import mock
from osc_lib import exceptions
from octaviaclient.osc.v2 import constants
from octaviaclient.osc.v2 import quota
from octaviaclient.tests.unit.osc.v2 import constants as attr_consts
from octaviaclient.tests.unit.osc.v2 import fakes
class TestQuota(fakes.TestOctaviaClient):
_qt = fakes.createFakeResource('quota')
columns = ('project_id', 'load_balancer', 'listener', 'pool',
'health_monitor', 'member')
datalist = (
(
_qt.project_id,
_qt.load_balancer,
_qt.listener,
_qt.pool,
_qt.health_monitor,
_qt.member
),
)
info = {
'quotas':
[{
"project_id": _qt.project_id,
"load_balancer": _qt.load_balancer,
"listener": _qt.listener,
"pool": _qt.pool,
"health_monitor": _qt.health_monitor,
"member": _qt.member
}]
}
qt_info = copy.deepcopy(info)
def setUp(self):
super(TestQuota, self).setUp()
self.qt_mock = self.app.client_manager.load_balancer.load_balancers
self.qt_mock.reset_mock()
self._qt = fakes.createFakeResource('quota')
self.quota_info = copy.deepcopy(attr_consts.QUOTA_ATTRS)
self.columns = copy.deepcopy(constants.QUOTA_COLUMNS)
self.api_mock = mock.Mock()
self.api_mock.quota_list.return_value = self.qt_info
self.api_mock.quota_list.return_value = copy.deepcopy(
{'quotas': [attr_consts.QUOTA_ATTRS]})
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -66,6 +43,8 @@ class TestQuotaList(TestQuota):
def setUp(self):
super(TestQuotaList, self).setUp()
self.datalist = (tuple(
attr_consts.QUOTA_ATTRS[k] for k in self.columns),)
self.cmd = quota.ListQuota(self.app, None)
def test_quota_list_no_options(self):
@ -84,10 +63,8 @@ class TestQuotaShow(TestQuota):
def setUp(self):
super(TestQuotaShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.quota_list.return_value = self.qt_info
self.api_mock.quota_show.return_value = {
'quota': self.qt_info['quotas'][0]}
'quota': self.quota_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
@ -95,7 +72,7 @@ class TestQuotaShow(TestQuota):
@mock.patch('octaviaclient.osc.v2.utils.get_quota_attrs')
def test_quota_show(self, mock_attrs):
mock_attrs.return_value = self.qt_info['quotas'][0]
mock_attrs.return_value = self.quota_info
arglist = [self._qt.project_id]
verifylist = [
('project', self._qt.project_id),
@ -120,7 +97,6 @@ class TestQuotaDefaultsShow(TestQuota):
def setUp(self):
super(TestQuotaDefaultsShow, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.quota_defaults_show.return_value = {
'quota': self.qt_defaults}
@ -145,9 +121,9 @@ class TestQuotaSet(TestQuota):
def setUp(self):
super(TestQuotaSet, self).setUp()
self.api_mock = mock.Mock()
self.api_mock.quota_set.return_value = {
'quota': self.qt_info['quotas'][0]}
'quota': self.quota_info}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock