python-octaviaclient/octaviaclient/tests/unit/osc/v2/test_load_balancer.py
Michael Johnson 98f7c574a6 Add load balancer flavor support to the OSC client
This patch adds the flavor fields for load balancers.

It also adds .stestr to the .gitigonre list.

Change-Id: I9153895eb5e7ebba50a2c31b5375ecf173d8bfa2
2019-02-20 08:39:34 +00:00

479 lines
17 KiB
Python

# 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 itertools
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):
def setUp(self):
super(TestLoadBalancer, self).setUp()
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 = 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()
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):
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))
def test_load_balancer_list_with_name(self):
arglist = ['--name', 'rainbarrel']
verifylist = [('name', 'rainbarrel')]
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(name='rainbarrel')
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_network(self, mock_client):
mock_client.return_value = {
'vip_network_id': self._lb.vip_network_id,
}
arglist = [
'--vip-network-id', self._lb.vip_network_id,
]
verify_list = [
('vip_network_id', self._lb.vip_network_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_subnet(self, mock_client):
mock_client.return_value = {
'vip_subnet_id': self._lb.vip_subnet_id,
}
arglist = [
'--vip-subnet-id', self._lb.vip_subnet_id,
]
verify_list = [
('vip_subnet_id', self._lb.vip_subnet_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_qos_policy(self, mock_client):
mock_client.return_value = {
'vip_qos_policy_id': self._lb.vip_qos_policy_id,
}
arglist = [
'--vip-qos-policy-id', self._lb.vip_qos_policy_id,
]
verify_list = [
('vip_qos_policy_id', self._lb.vip_qos_policy_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_provisioning_status(self, mock_client):
mock_client.return_value = {
'provisioning_status': self._lb.provisioning_status,
}
arglist = [
'--provisioning-status', 'active',
]
verify_list = [
('provisioning_status', 'ACTIVE'),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_operating_status(self, mock_client):
mock_client.return_value = {
'operating_status': self._lb.operating_status,
}
arglist = [
'--operating-status', 'ONLiNE',
]
verify_list = [
('operating_status', 'ONLINE'),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_provider(self, mock_client):
mock_client.return_value = {
'provider': self._lb.provider,
}
arglist = [
'--provider', 'octavia',
]
verify_list = [
('provider', 'octavia'),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_list_with_flavor(self, mock_client):
mock_client.return_value = {
'flavor_id': self._lb.flavor_id,
}
arglist = [
'--flavor', self._lb.flavor_id,
]
verify_list = [
('flavor', self._lb.flavor_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verify_list)
columns, data = self.cmd.take_action(parsed_args)
self.assertEqual(self.columns, columns)
self.assertEqual(self.datalist, tuple(data))
class TestLoadBalancerDelete(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerDelete, self).setUp()
self.cmd = load_balancer.DeleteLoadBalancer(self.app, None)
def test_load_balancer_delete(self):
arglist = [self._lb.id]
verifylist = [
('loadbalancer', self._lb.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_delete.assert_called_with(
lb_id=self._lb.id)
def test_load_balancer_delete_failure(self):
arglist = ['unknown_lb']
verifylist = [
('loadbalancer', 'unknown_lb')
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(exceptions.CommandError, self.cmd.take_action,
parsed_args)
self.assertNotCalled(self.api_mock.load_balancer_delete)
class TestLoadBalancerCreate(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerCreate, self).setUp()
self.api_mock.load_balancer_create.return_value = {
'loadbalancer': self.lb_info
}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.CreateLoadBalancer(self.app, None)
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create(self, mock_client):
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,
'--flavor', self._lb.flavor_id]
verifylist = [
('name', self._lb.name),
('vip_network_id', self._lb.vip_network_id),
('project', self._lb.project_id),
('flavor', self._lb.flavor_id),
]
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})
@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)
lb_info.update({'vip_qos_policy_id': qos_policy_id})
mock_client.return_value = lb_info
arglist = [
'--name', self._lb.name,
'--vip-network-id', self._lb.vip_network_id,
'--project', self._lb.project_id,
'--vip-qos-policy-id', qos_policy_id,
]
verifylist = [
('name', self._lb.name),
('vip_network_id', self._lb.vip_network_id),
('project', self._lb.project_id),
('vip_qos_policy_id', qos_policy_id),
]
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': lb_info})
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create_with_provider(self, mock_client):
provider = 'foobar'
lb_info = copy.deepcopy(self.lb_info)
lb_info.update({'provider': provider})
mock_client.return_value = lb_info
arglist = [
'--name', self._lb.name,
'--vip-network-id', self._lb.vip_network_id,
'--project', self._lb.project_id,
'--provider', provider,
]
verifylist = [
('name', self._lb.name),
('vip_network_id', self._lb.vip_network_id),
('project', self._lb.project_id),
('provider', provider),
]
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': lb_info})
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_create_missing_args(self, mock_client):
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
attrs_list[a] = ''
# verify all valid combinations of args
for n in range(len(args)+1):
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)}
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",
"vip_port_id")
):
self.assertRaises(
exceptions.CommandError,
self.cmd.take_action,
filtered_attrs)
else:
try:
self.cmd.take_action(filtered_attrs)
except exceptions.CommandError as e:
self.fail("%s raised unexpectedly" % e)
class TestLoadBalancerShow(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerShow, self).setUp()
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
self.cmd = load_balancer.ShowLoadBalancer(self.app, None)
def test_load_balancer_show(self):
arglist = [self._lb.id]
verifylist = [
('loadbalancer', self._lb.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_show.assert_called_with(lb_id=self._lb.id)
class TestLoadBalancerSet(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerSet, self).setUp()
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.SetLoadBalancer(self.app, None)
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_set(self, mock_attrs):
qos_policy_id = uuidutils.generate_uuid()
mock_attrs.return_value = {
'loadbalancer_id': self._lb.id,
'name': 'new_name',
'vip_qos_policy_id': qos_policy_id,
}
arglist = [self._lb.id, '--name', 'new_name',
'--vip-qos-policy-id', qos_policy_id]
verifylist = [
('loadbalancer', self._lb.id),
('name', 'new_name'),
('vip_qos_policy_id', qos_policy_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_set.assert_called_with(
self._lb.id, json={
'loadbalancer': {
'name': 'new_name',
'vip_qos_policy_id': qos_policy_id,
}
})
@mock.patch('octaviaclient.osc.v2.utils.get_loadbalancer_attrs')
def test_load_balancer_remove_qos_policy(self, mock_attrs):
mock_attrs.return_value = {
'loadbalancer_id': self._lb.id,
'vip_qos_policy_id': None,
}
arglist = [self._lb.id, '--vip-qos-policy-id', 'None']
verifylist = [
('loadbalancer', self._lb.id),
('vip_qos_policy_id', 'None'),
]
try:
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
except Exception as e:
self.fail("%s raised unexpectedly" % e)
class TestLoadBalancerStats(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerStats, self).setUp()
lb_stats_info = {'stats': {'bytes_in': '0'}}
self.api_mock.load_balancer_stats_show.return_value = {
'stats': lb_stats_info['stats']}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.ShowLoadBalancerStats(self.app, None)
def test_load_balancer_stats_show(self):
arglist = [self._lb.id]
verifylist = [
('loadbalancer', self._lb.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_stats_show.assert_called_with(
lb_id=self._lb.id)
class TestLoadBalancerStatus(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerStatus, self).setUp()
expected_res = {'statuses': {'operating_status': 'ONLINE',
'provisioning_status': 'ACTIVE'}}
self.api_mock.load_balancer_status_show.return_value = {
'statuses': expected_res['statuses']
}
lb_client = self.app.client_manager
lb_client.load_balancer = self.api_mock
self.cmd = load_balancer.ShowLoadBalancerStatus(self.app, None)
def test_load_balancer_status_show(self):
# lbaas-loadbalancer-status test_id.
arglist = [self._lb.id]
verifylist = [
('loadbalancer', self._lb.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_status_show.assert_called_with(
lb_id=self._lb.id)
class TestLoadBalancerFailover(TestLoadBalancer):
def setUp(self):
super(TestLoadBalancerFailover, self).setUp()
self.cmd = load_balancer.FailoverLoadBalancer(self.app, None)
def test_load_balancer_failover(self):
arglist = [self._lb.id]
verifylist = [
('loadbalancer', self._lb.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
self.api_mock.load_balancer_failover.assert_called_with(
lb_id=self._lb.id)