Load Balancing as a Service (LBaaS) for OpenStack
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

171 lines
7.2 KiB

# Copyright 2014, Author: Min Wang,German Eichberger
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from unittest import mock
from oslo_utils import uuidutils
from octavia.amphorae.drivers.noop_driver import driver
from octavia.common import constants
from octavia.common import data_models
from octavia.network import data_models as network_models
from octavia.tests.unit import base
FAKE_UUID_1 = uuidutils.generate_uuid()
class TestLoggingUpdate(base.TestCase):
def setUp(self):
super(TestLoggingUpdate, self).setUp()
self.mixin = driver.LoggingUpdate()
def test_update_stats(self):
self.mixin.update_stats('test update stats')
self.assertEqual('test update stats', self.mixin.stats)
def test_update_health(self):
self.mixin.update_health('test update health')
self.assertEqual('test update health', self.mixin.health)
class TestNoopAmphoraLoadBalancerDriver(base.TestCase):
FAKE_UUID_1 = uuidutils.generate_uuid()
def setUp(self):
super(TestNoopAmphoraLoadBalancerDriver, self).setUp()
self.driver = driver.NoopAmphoraLoadBalancerDriver()
self.listener = data_models.Listener()
self.listener.id = uuidutils.generate_uuid()
self.listener.protocol_port = 80
self.vip = data_models.Vip()
self.vip.ip_address = "192.51.100.1"
self.amphora = data_models.Amphora()
self.amphora.id = self.FAKE_UUID_1
self.load_balancer = data_models.LoadBalancer(
id=FAKE_UUID_1, amphorae=[self.amphora], vip=self.vip,
listeners=[self.listener])
self.listener.load_balancer = self.load_balancer
self.network = network_models.Network(id=self.FAKE_UUID_1)
self.port = network_models.Port(id=uuidutils.generate_uuid())
self.amphorae_net_configs = {
self.amphora.id:
network_models.AmphoraNetworkConfig(
amphora=self.amphora,
vip_subnet=network_models.Subnet(id=self.FAKE_UUID_1))
}
self.pem_file = 'test_pem_file'
self.agent_config = 'test agent config'
self.timeout_dict = {constants.REQ_CONN_TIMEOUT: 1,
constants.REQ_READ_TIMEOUT: 2,
constants.CONN_MAX_RETRIES: 3,
constants.CONN_RETRY_INTERVAL: 4}
def test_update_amphora_listeners(self):
self.driver.update_amphora_listeners(self.load_balancer, self.amphora,
self.timeout_dict)
self.assertEqual((self.listener, self.amphora.id, self.timeout_dict,
'update_amp'),
self.driver.driver.amphoraconfig[(
self.listener.id,
self.amphora.id)])
def test_update(self):
self.driver.update(self.load_balancer)
self.assertEqual(([self.listener], self.vip, 'active'),
self.driver.driver.amphoraconfig[(
(self.listener.protocol_port,),
self.vip.ip_address)])
def test_start(self):
mock_amphora = mock.MagicMock()
mock_amphora.id = '321'
self.driver.start(self.load_balancer, amphora=mock_amphora)
self.assertEqual((self.load_balancer, mock_amphora, 'start'),
self.driver.driver.amphoraconfig[(
self.load_balancer.id, '321')])
def test_reload(self):
mock_amphora = mock.MagicMock()
mock_amphora.id = '321'
self.driver.reload(self.load_balancer, amphora=mock_amphora)
self.assertEqual((self.load_balancer, mock_amphora, 'reload'),
self.driver.driver.amphoraconfig[(
self.load_balancer.id, '321')])
def test_delete(self):
self.driver.delete(self.listener)
self.assertEqual((self.listener, self.vip, 'delete'),
self.driver.driver.amphoraconfig[(
self.listener.protocol_port,
self.vip.ip_address)])
def test_get_info(self):
self.driver.get_info(self.amphora)
self.assertEqual((self.amphora.id, 'get_info'),
self.driver.driver.amphoraconfig[
self.amphora.id])
def test_get_diagnostics(self):
self.driver.get_diagnostics(self.amphora)
self.assertEqual((self.amphora.id, 'get_diagnostics'),
self.driver.driver.amphoraconfig[
self.amphora.id])
def test_finalize_amphora(self):
self.driver.finalize_amphora(self.amphora)
self.assertEqual((self.amphora.id, 'finalize amphora'),
self.driver.driver.amphoraconfig[
self.amphora.id])
def test_post_network_plug(self):
self.driver.post_network_plug(self.amphora, self.port)
self.assertEqual((self.amphora.id, self.port.id, 'post_network_plug'),
self.driver.driver.amphoraconfig[(
self.amphora.id, self.port.id)])
def test_post_vip_plug(self):
self.driver.post_vip_plug(self.amphora, self.load_balancer,
self.amphorae_net_configs)
expected_method_and_args = (self.load_balancer.id,
self.amphorae_net_configs,
'post_vip_plug')
actual_method_and_args = self.driver.driver.amphoraconfig[(
self.load_balancer.id, id(self.amphorae_net_configs)
)]
self.assertEqual(expected_method_and_args, actual_method_and_args)
def test_upload_cert_amp(self):
self.driver.upload_cert_amp(self.amphora, self.pem_file)
self.assertEqual(
(self.amphora.id, self.pem_file, 'update_amp_cert_file'),
self.driver.driver.amphoraconfig[(
self.amphora.id, self.pem_file)])
def test_update_agent_config(self):
self.driver.update_amphora_agent_config(self.amphora,
self.agent_config)
self.assertEqual(
(self.amphora.id, self.agent_config,
'update_amphora_agent_config'),
self.driver.driver.amphoraconfig[(
self.amphora.id, self.agent_config)])
def test_get_interface_from_ip(self):
result = self.driver.get_interface_from_ip(self.amphora,
'198.51.100.99')
self.assertEqual('noop0', result)
result = self.driver.get_interface_from_ip(self.amphora,
'198.51.100.9')
self.assertIsNone(result)