6832082aa5
As suggested by Andrey, we should have a consistent format for the arguments in the octavia services. It makes sense to use just the id of the resource rather than the entire dict. This patch fixes the load balancer resource Change-Id: If642649c078e1182807582c287141509579b2d21 Signed-off-by: Lucas H. Xu <hanxu@redhat.com>
459 lines
20 KiB
Python
459 lines
20 KiB
Python
# Copyright 2018: Red Hat Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import fixtures
|
|
import mock
|
|
|
|
from rally.common import cfg
|
|
from rally import exceptions
|
|
from rally_openstack.services.loadbalancer import octavia
|
|
from tests.unit import test
|
|
|
|
BASE_PATH = "rally_openstack.services.loadbalancer"
|
|
CONF = cfg.CONF
|
|
|
|
|
|
class LoadBalancerServiceTestCase(test.TestCase):
|
|
def setUp(self):
|
|
super(LoadBalancerServiceTestCase, self).setUp()
|
|
self.clients = mock.MagicMock()
|
|
self.name_generator = mock.MagicMock()
|
|
self.service = octavia.Octavia(self.clients,
|
|
name_generator=self.name_generator)
|
|
self.mock_wait_for_status = fixtures.MockPatch(
|
|
"rally.task.utils.wait_for_status")
|
|
self.useFixture(self.mock_wait_for_status)
|
|
|
|
def _get_context(self):
|
|
context = test.get_test_context()
|
|
context.update({
|
|
"user": {
|
|
"id": "fake_user",
|
|
"tenant_id": "fake_tenant",
|
|
"credential": mock.MagicMock()
|
|
},
|
|
"tenant": {"id": "fake_tenant",
|
|
"networks": [{"id": "fake_net",
|
|
"subnets": ["fake_subnet"]}]}})
|
|
return context
|
|
|
|
def atomic_actions(self):
|
|
return self.service._atomic_actions
|
|
|
|
def test_load_balancer_list(self):
|
|
self.service.load_balancer_list(),
|
|
self.service._clients.octavia().load_balancer_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_list")
|
|
|
|
def test_load_balancer_show(self):
|
|
lb = {"id": "loadbalancer-id"}
|
|
self.service.load_balancer_show(lb["id"])
|
|
self.service._clients.octavia().load_balancer_show \
|
|
.assert_called_once_with(lb["id"])
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_show")
|
|
|
|
def test_load_balancer_show_fail_404(self):
|
|
fake_lb = {"id": "fake_lb"}
|
|
ex = Exception()
|
|
ex.code = 404
|
|
self.service._clients.octavia().load_balancer_show.side_effect = ex
|
|
self.assertRaises(
|
|
exceptions.GetResourceNotFound,
|
|
self.service.load_balancer_show, fake_lb["id"])
|
|
|
|
def test_load_balancer_show_resource_fail(self):
|
|
fake_lb = {"id": "fake_lb"}
|
|
ex = Exception()
|
|
self.service._clients.octavia().load_balancer_show.side_effect = ex
|
|
self.assertRaises(
|
|
exceptions.GetResourceFailure,
|
|
self.service.load_balancer_show, fake_lb["id"])
|
|
|
|
def test_load_balancer_create(self):
|
|
self.service.generate_random_name = mock.MagicMock(
|
|
return_value="lb")
|
|
self.service.load_balancer_create("subnet_id")
|
|
self.service._clients.octavia().load_balancer_create \
|
|
.assert_called_once_with(json={
|
|
"loadbalancer": {"name": "lb",
|
|
"admin_state_up": True,
|
|
"vip_qos_policy_id": None,
|
|
"listeners": None,
|
|
"provider": None,
|
|
"vip_subnet_id": "subnet_id",
|
|
"description": None}})
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_create")
|
|
|
|
def test_load_balancer_delete(self):
|
|
self.service.load_balancer_delete("lb-id")
|
|
self.service._clients.octavia().load_balancer_delete \
|
|
.assert_called_once_with("lb-id", cascade=False)
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_delete")
|
|
|
|
def test_load_balancer_set(self):
|
|
self.service.generate_random_name = mock.MagicMock(
|
|
return_value="new_lb")
|
|
lb_update_args = {"name": "new_lb_name"}
|
|
self.service.load_balancer_set(
|
|
"lb-id", lb_update_args=lb_update_args)
|
|
self.service._clients.octavia().load_balancer_set \
|
|
.assert_called_once_with(
|
|
"lb-id", json={"loadbalancer": {"name": "new_lb_name"}})
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_set")
|
|
|
|
def test_load_balancer_stats_show(self):
|
|
lb = {"id": "new_lb"}
|
|
self.assertEqual(
|
|
self.service.load_balancer_stats_show(lb, kwargs={}),
|
|
self.service._clients.octavia()
|
|
.load_balancer_stats_show.return_value)
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_stats_show")
|
|
|
|
def test_load_balancer_failover(self):
|
|
lb = {"id": "new_lb"}
|
|
self.service.load_balancer_failover(lb["id"])
|
|
self.service._clients.octavia().load_balancer_failover \
|
|
.assert_called_once_with(lb["id"])
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.load_balancer_failover")
|
|
|
|
def test_listener_list(self):
|
|
self.service.listener_list()
|
|
self.service._clients.octavia().listener_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_list")
|
|
|
|
def test_listener_show(self):
|
|
self.service.listener_show(listener_id="listener_id")
|
|
self.service._clients.octavia().listener_show \
|
|
.assert_called_once_with("listener_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_show")
|
|
|
|
def test_listener_create(self):
|
|
self.service.listener_create()
|
|
self.service._clients.octavia().listener_create \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_create")
|
|
|
|
def test_listener_delete(self):
|
|
self.service.listener_delete(listener_id="listener_id")
|
|
self.service._clients.octavia().listener_delete \
|
|
.assert_called_once_with("listener_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_delete")
|
|
|
|
def test_listener_set(self):
|
|
self.service.listener_set(listener_id="listener_id")
|
|
self.service._clients.octavia().listener_set \
|
|
.assert_called_once_with("listener_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_set")
|
|
|
|
def test_listener_stats_show(self):
|
|
self.service.listener_stats_show(listener_id="listener_id")
|
|
self.service._clients.octavia().listener_stats_show \
|
|
.assert_called_once_with("listener_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.listener_stats_show")
|
|
|
|
def test_pool_list(self):
|
|
self.service.pool_list()
|
|
self.service._clients.octavia().pool_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.pool_list")
|
|
|
|
def test_update_pool_resource(self):
|
|
fake_pool = {"id": "pool-id"}
|
|
self.service.update_pool_resource(fake_pool)
|
|
self.service._clients.octavia().pool_show \
|
|
.assert_called_once_with("pool-id")
|
|
|
|
def test_update_pool_resource_fail_404(self):
|
|
fake_pool = {"id": "pool-id"}
|
|
ex = Exception()
|
|
ex.status_code = 404
|
|
self.service._clients.octavia().pool_show.side_effect = ex
|
|
self.assertRaises(
|
|
exceptions.GetResourceNotFound,
|
|
self.service.update_pool_resource, fake_pool)
|
|
|
|
def test_update_pool_resource_fail(self):
|
|
fake_pool = {"id": "pool-id"}
|
|
ex = Exception()
|
|
self.service._clients.octavia().pool_show.side_effect = ex
|
|
self.assertRaises(
|
|
exceptions.GetResourceFailure,
|
|
self.service.update_pool_resource, fake_pool)
|
|
|
|
def test_pool_create(self):
|
|
self.service.generate_random_name = mock.MagicMock(
|
|
return_value="pool")
|
|
self.service.pool_create(
|
|
lb_id="loadbalancer-id",
|
|
protocol="HTTP",
|
|
lb_algorithm="ROUND_ROBIN")
|
|
self.service._clients.octavia().pool_create \
|
|
.assert_called_once_with(
|
|
json={"pool": {
|
|
"lb_algorithm": "ROUND_ROBIN",
|
|
"protocol": "HTTP",
|
|
"listener_id": None,
|
|
"description": None,
|
|
"admin_state_up": True,
|
|
"session_persistence": None,
|
|
"loadbalancer_id": "loadbalancer-id",
|
|
"name": "pool"}})
|
|
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.pool_create")
|
|
|
|
def test_pool_delete(self):
|
|
self.service.pool_delete(pool_id="fake_pool")
|
|
self.service._clients.octavia().pool_delete \
|
|
.assert_called_once_with("fake_pool")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.pool_delete")
|
|
|
|
def test_pool_show(self):
|
|
self.service.pool_show(pool_id="fake_pool")
|
|
self.service._clients.octavia().pool_show \
|
|
.assert_called_once_with("fake_pool")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.pool_show")
|
|
|
|
def test_pool_set(self):
|
|
pool_update_args = {"name": "new-pool-name"}
|
|
self.service.pool_set(
|
|
pool_id="fake_pool",
|
|
pool_update_args=pool_update_args)
|
|
self.service._clients.octavia().pool_set \
|
|
.assert_called_once_with(
|
|
"fake_pool",
|
|
json={"pool": {"name": "new-pool-name"}})
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.pool_set")
|
|
|
|
def test_member_list(self):
|
|
self.service.member_list(pool_id="fake_pool")
|
|
self.service._clients.octavia().member_list \
|
|
.assert_called_once_with("fake_pool")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.member_list")
|
|
|
|
def test_member_show(self):
|
|
self.service.member_show(pool_id="fake_pool", member_id="fake_member")
|
|
self.service._clients.octavia().member_show \
|
|
.assert_called_once_with("fake_pool", "fake_member")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.member_show")
|
|
|
|
def test_member_create(self):
|
|
self.service.member_create(pool_id="fake_pool")
|
|
self.service._clients.octavia().member_create \
|
|
.assert_called_once_with("fake_pool")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.member_create")
|
|
|
|
def test_member_delete(self):
|
|
self.service.member_delete(
|
|
pool_id="fake_pool", member_id="fake_member")
|
|
self.service._clients.octavia().member_delete \
|
|
.assert_called_once_with("fake_pool", "fake_member")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.member_delete")
|
|
|
|
def test_member_set(self):
|
|
self.service.member_set(pool_id="fake_pool", member_id="fake_member")
|
|
self.service._clients.octavia().member_set \
|
|
.assert_called_once_with("fake_pool", "fake_member")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.member_set")
|
|
|
|
def test_l7policy_list(self):
|
|
self.service.l7policy_list()
|
|
self.service._clients.octavia().l7policy_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7policy_list")
|
|
|
|
def test_l7policy_create(self):
|
|
self.service.l7policy_create()
|
|
self.service._clients.octavia().l7policy_create \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7policy_create")
|
|
|
|
def test_l7policy_delete(self):
|
|
self.service.l7policy_delete(l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7policy_delete \
|
|
.assert_called_once_with("fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7policy_delete")
|
|
|
|
def test_l7policy_show(self):
|
|
self.service.l7policy_show(l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7policy_show \
|
|
.assert_called_once_with("fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7policy_show")
|
|
|
|
def test_l7policy_set(self):
|
|
self.service.l7policy_set(l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7policy_set \
|
|
.assert_called_once_with("fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7policy_set")
|
|
|
|
def test_l7rule_list(self):
|
|
self.service.l7rule_list(l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7rule_list \
|
|
.assert_called_once_with("fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7rule_list")
|
|
|
|
def test_l7rule_create(self):
|
|
self.service.l7rule_create(l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7rule_create \
|
|
.assert_called_once_with("fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7rule_create")
|
|
|
|
def test_l7rule_delete(self):
|
|
self.service.l7rule_delete(
|
|
l7rule_id="fake_id", l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7rule_delete \
|
|
.assert_called_once_with("fake_id", "fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7rule_delete")
|
|
|
|
def test_l7rule_show(self):
|
|
self.service.l7rule_show(
|
|
l7rule_id="fake_id", l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7rule_show \
|
|
.assert_called_once_with("fake_id", "fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7rule_show")
|
|
|
|
def test_l7rule_set(self):
|
|
self.service.l7rule_set(l7rule_id="fake_id", l7policy_id="fake_policy")
|
|
self.service._clients.octavia().l7rule_set \
|
|
.assert_called_once_with("fake_id", "fake_policy")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.l7rule_set")
|
|
|
|
def test_health_monitor_list(self):
|
|
self.service.health_monitor_list()
|
|
self.service._clients.octavia().health_monitor_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.health_monitor_list")
|
|
|
|
def test_health_monitor_create(self):
|
|
self.service.health_monitor_create()
|
|
self.service._clients.octavia().health_monitor_create \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.health_monitor_create")
|
|
|
|
def test_health_monitor_delete(self):
|
|
self.service.health_monitor_delete(health_monitor_id="fake_monitor_id")
|
|
self.service._clients.octavia().health_monitor_delete \
|
|
.assert_called_once_with("fake_monitor_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.health_monitor_delete")
|
|
|
|
def test_health_monitor_show(self):
|
|
self.service.health_monitor_show(health_monitor_id="fake_monitor_id")
|
|
self.service._clients.octavia().health_monitor_show \
|
|
.assert_called_once_with("fake_monitor_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.health_monitor_show")
|
|
|
|
def test_health_monitor_set(self):
|
|
self.service.health_monitor_set(health_monitor_id="fake_monitor_id")
|
|
self.service._clients.octavia().health_monitor_set \
|
|
.assert_called_once_with("fake_monitor_id")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.health_monitor_set")
|
|
|
|
def test_quota_list(self):
|
|
self.service.quota_list(params="fake_params")
|
|
self.service._clients.octavia().quota_list \
|
|
.assert_called_once_with("fake_params")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.quota_list")
|
|
|
|
def test_quota_show(self):
|
|
self.service.quota_show(project_id="fake_project")
|
|
self.service._clients.octavia().quota_show \
|
|
.assert_called_once_with("fake_project")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.quota_show")
|
|
|
|
def test_quota_reset(self):
|
|
self.service.quota_reset(project_id="fake_project")
|
|
self.service._clients.octavia().quota_reset \
|
|
.assert_called_once_with("fake_project")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.quota_reset")
|
|
|
|
def test_quota_set(self):
|
|
self.service.quota_set(project_id="fake_project",
|
|
params="fake_params")
|
|
self.service._clients.octavia().quota_set \
|
|
.assert_called_once_with("fake_project", "fake_params")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.quota_set")
|
|
|
|
def test_quota_defaults_show(self):
|
|
self.service.quota_defaults_show()
|
|
self.service._clients.octavia().quota_defaults_show \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.quota_defaults_show")
|
|
|
|
def test_amphora_show(self):
|
|
self.service.amphora_show(amphora_id="fake_amphora")
|
|
self.service._clients.octavia().amphora_show \
|
|
.assert_called_once_with("fake_amphora")
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.amphora_show")
|
|
|
|
def test_amphora_list(self):
|
|
self.service.amphora_list()
|
|
self.service._clients.octavia().amphora_list \
|
|
.assert_called_once_with()
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.amphora_list")
|
|
|
|
@mock.patch("%s.Ocvita.wait_for_loadbalancer_prov_status" % BASE_PATH)
|
|
def wait_for_loadbalancer_prov_status(self, mock_wait_for_status):
|
|
fake_lb = {}
|
|
self.service.wait_for_loadbalancer_prov_status(lb=fake_lb)
|
|
self.assertTrue(mock_wait_for_status.called)
|
|
self._test_atomic_action_timer(self.atomic_actions(),
|
|
"octavia.wait_for_loadbalancers")
|