# Copyright 2018 Rackspace, US Inc. # # 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 from unittest import mock from octavia_lib.api.drivers import data_models as driver_dm from octavia_lib.api.drivers import exceptions as lib_exceptions from octavia.api.drivers import utils from octavia.common import constants from octavia.common import data_models from octavia.common import exceptions from octavia.tests.common import sample_data_models from octavia.tests.unit import base class TestUtils(base.TestCase): def setUp(self): super(TestUtils, self).setUp() self.sample_data = sample_data_models.SampleDriverDataModels() def test_call_provider(self): mock_driver_method = mock.MagicMock() # Test happy path utils.call_provider("provider_name", mock_driver_method, "arg1", foo="arg2") mock_driver_method.assert_called_with("arg1", foo="arg2") # Test driver raising DriverError mock_driver_method.side_effect = lib_exceptions.DriverError self.assertRaises(exceptions.ProviderDriverError, utils.call_provider, "provider_name", mock_driver_method) # Test driver raising different types of NotImplementedError mock_driver_method.side_effect = NotImplementedError self.assertRaises(exceptions.ProviderNotImplementedError, utils.call_provider, "provider_name", mock_driver_method) mock_driver_method.side_effect = lib_exceptions.NotImplementedError self.assertRaises(exceptions.ProviderNotImplementedError, utils.call_provider, "provider_name", mock_driver_method) # Test driver raising UnsupportedOptionError mock_driver_method.side_effect = ( lib_exceptions.UnsupportedOptionError) self.assertRaises(exceptions.ProviderUnsupportedOptionError, utils.call_provider, "provider_name", mock_driver_method) # Test driver raising ProviderDriverError mock_driver_method.side_effect = Exception self.assertRaises(exceptions.ProviderDriverError, utils.call_provider, "provider_name", mock_driver_method) def test_base_to_provider_dict(self): test_dict = {'provisioning_status': constants.ACTIVE, 'operating_status': constants.ONLINE, 'provider': 'octavia', 'created_at': 'now', 'updated_at': 'then', 'enabled': True, 'project_id': 1} result_dict = utils._base_to_provider_dict(test_dict, include_project_id=True) self.assertEqual({'admin_state_up': True, 'project_id': 1}, result_dict) result_dict = utils._base_to_provider_dict(test_dict, include_project_id=False) self.assertEqual({'admin_state_up': True}, result_dict) @mock.patch('octavia.db.repositories.FlavorRepository.' 'get_flavor_metadata_dict') @mock.patch('octavia.db.api.get_session') @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_lb_dict_to_provider_dict(self, mock_load_cert, mock_secret, mock_get_session, mock_get_flavor): cert1 = data_models.TLSContainer(certificate='cert 1') cert2 = data_models.TLSContainer(certificate='cert 2') cert3 = data_models.TLSContainer(certificate='cert 3') mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE', 'ca cert', 'X509 CRL FILE', 'ca cert', 'X509 CRL FILE', 'X509 POOL CA CERT FILE', 'X509 CRL FILE'] listener_certs = {'tls_cert': cert1, 'sni_certs': [cert2, cert3]} pool_cert = data_models.TLSContainer(certificate='pool cert') pool_certs = {'tls_cert': pool_cert, 'sni_certs': []} mock_load_cert.side_effect = [pool_certs, listener_certs, listener_certs, listener_certs, listener_certs] mock_get_flavor.return_value = {'shaved_ice': 'cherry'} test_lb_dict = {'name': 'lb1', 'project_id': self.sample_data.project_id, 'vip_subnet_id': self.sample_data.subnet_id, 'vip_port_id': self.sample_data.port_id, 'vip_address': self.sample_data.ip_address, 'vip_network_id': self.sample_data.network_id, 'vip_qos_policy_id': self.sample_data.qos_policy_id, 'id': self.sample_data.lb_id, 'listeners': [], 'pools': [], 'description': '', 'admin_state_up': True, 'provisioning_status': constants.PENDING_CREATE, 'operating_status': constants.OFFLINE, 'flavor_id': 'flavor_id', 'provider': 'noop_driver'} ref_listeners = copy.deepcopy(self.sample_data.provider_listeners) ref_prov_lb_dict = { 'vip_address': self.sample_data.ip_address, 'admin_state_up': True, 'loadbalancer_id': self.sample_data.lb_id, 'vip_subnet_id': self.sample_data.subnet_id, 'listeners': ref_listeners, 'description': '', 'project_id': self.sample_data.project_id, 'vip_port_id': self.sample_data.port_id, 'vip_qos_policy_id': self.sample_data.qos_policy_id, 'vip_network_id': self.sample_data.network_id, 'pools': self.sample_data.provider_pools, 'flavor': {'shaved_ice': 'cherry'}, 'name': 'lb1'} vip = data_models.Vip(ip_address=self.sample_data.ip_address, network_id=self.sample_data.network_id, port_id=self.sample_data.port_id, subnet_id=self.sample_data.subnet_id, qos_policy_id=self.sample_data.qos_policy_id) provider_lb_dict = utils.lb_dict_to_provider_dict( test_lb_dict, vip=vip, db_pools=self.sample_data.test_db_pools, db_listeners=self.sample_data.test_db_listeners) self.assertEqual(ref_prov_lb_dict, provider_lb_dict) @mock.patch('octavia.db.repositories.FlavorRepository.' 'get_flavor_metadata_dict') @mock.patch('octavia.db.api.get_session') def test_db_loadbalancer_to_provider_loadbalancer(self, mock_get_session, mock_get_flavor): mock_get_flavor.return_value = {'shaved_ice': 'cherry'} vip = data_models.Vip(ip_address=self.sample_data.ip_address, network_id=self.sample_data.network_id, port_id=self.sample_data.port_id, subnet_id=self.sample_data.subnet_id) test_db_lb = data_models.LoadBalancer(id=1, flavor_id='2', vip=vip) provider_lb = utils.db_loadbalancer_to_provider_loadbalancer( test_db_lb) ref_provider_lb = driver_dm.LoadBalancer( loadbalancer_id=1, flavor={'shaved_ice': 'cherry'}, vip_address=self.sample_data.ip_address, vip_network_id=self.sample_data.network_id, vip_port_id=self.sample_data.port_id, vip_subnet_id=self.sample_data.subnet_id) self.assertEqual(ref_provider_lb.to_dict(render_unsets=True), provider_lb.to_dict(render_unsets=True)) def test_db_listener_to_provider_listener(self): test_db_list = data_models.Listener(id=1) provider_list = utils.db_listener_to_provider_listener(test_db_list) ref_provider_list = driver_dm.Listener(listener_id=1, insert_headers={}) self.assertEqual(ref_provider_list.to_dict(render_unsets=True), provider_list.to_dict(render_unsets=True)) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_db_listeners_to_provider_listeners(self, mock_load_cert, mock_secret): mock_secret.side_effect = ['ca cert', 'X509 CRL FILE', 'ca cert', 'X509 CRL FILE', 'ca cert', 'X509 CRL FILE'] cert1 = data_models.TLSContainer(certificate='cert 1') cert2 = data_models.TLSContainer(certificate='cert 2') cert3 = data_models.TLSContainer(certificate='cert 3') mock_load_cert.return_value = {'tls_cert': cert1, 'sni_certs': [cert2, cert3]} provider_listeners = utils.db_listeners_to_provider_listeners( self.sample_data.test_db_listeners) ref_listeners = copy.deepcopy(self.sample_data.provider_listeners) self.assertEqual(ref_listeners, provider_listeners) @mock.patch('oslo_context.context.RequestContext', return_value=None) def test_get_secret_data_errors(self, mock_context): mock_cert_mngr = mock.MagicMock() mock_cert_mngr.get_secret.side_effect = [Exception, Exception] # Test for_delete == False path self.assertRaises(exceptions.CertificateRetrievalException, utils._get_secret_data, mock_cert_mngr, 'fake_project_id', 1) # Test for_delete == True path self.assertIsNone( utils._get_secret_data(mock_cert_mngr, 'fake_project_id', 2, for_delete=True)) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_listener_dict_to_provider_dict(self, mock_load_cert, mock_secret): mock_secret.side_effect = ['ca cert', 'X509 CRL FILE', 'X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] cert1 = data_models.TLSContainer(certificate='cert 1') cert2 = data_models.TLSContainer(certificate='cert 2') cert3 = data_models.TLSContainer(certificate='cert 3') listener_certs = {'tls_cert': cert1, 'sni_certs': [cert2, cert3]} pool_cert = data_models.TLSContainer(certificate='pool cert') pool_certs = {'tls_cert': pool_cert, 'sni_certs': []} mock_load_cert.side_effect = [listener_certs, pool_certs] # The reason to do this, as before the logic arrives the test func, # there are two data sources, one is from db_dict, the other is from # the api layer model_dict, actually, they are different and contain # different fields. That's why the test_listener1_dict from sample data # just contain the client_ca_tls_certificate_id for client certificate, # not any other related fields. So we need to delete them. expect_prov = copy.deepcopy(self.sample_data.provider_listener1_dict) expect_pool_prov = copy.deepcopy(self.sample_data.provider_pool1_dict) expect_prov['default_pool'] = expect_pool_prov provider_listener = utils.listener_dict_to_provider_dict( self.sample_data.test_listener1_dict) self.assertEqual(expect_prov, provider_listener) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_listener_dict_to_provider_dict_load_cert_error( self, mock_load_cert, mock_secret): mock_secret.side_effect = ['ca cert', 'X509 CRL FILE', 'X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] mock_load_cert.side_effect = [exceptions.OctaviaException, Exception] # Test load_cert exception for_delete == False path self.assertRaises(exceptions.OctaviaException, utils.listener_dict_to_provider_dict, self.sample_data.test_listener1_dict) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_listener_dict_to_provider_dict_load_cert_error_for_delete( self, mock_load_cert, mock_secret): mock_secret.side_effect = ['ca cert', 'X509 CRL FILE', 'X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] mock_load_cert.side_effect = [Exception] # Test load_cert exception for_delete == True path expect_prov = copy.deepcopy(self.sample_data.provider_listener1_dict) expect_pool_prov = copy.deepcopy(self.sample_data.provider_pool1_dict) del expect_pool_prov['tls_container_data'] expect_prov['default_pool'] = expect_pool_prov del expect_prov['default_tls_container_data'] del expect_prov['sni_container_data'] provider_listener = utils.listener_dict_to_provider_dict( self.sample_data.test_listener1_dict, for_delete=True) self.assertEqual(expect_prov, provider_listener) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_listener_dict_to_provider_dict_SNI(self, mock_load_cert, mock_secret): mock_secret.return_value = 'ca cert' cert1 = data_models.TLSContainer(certificate='cert 1') cert2 = data_models.TLSContainer(certificate='cert 2') cert3 = data_models.TLSContainer(certificate='cert 3') mock_load_cert.return_value = {'tls_cert': cert1, 'sni_certs': [cert2, cert3]} # Test with bad SNI content test_listener = copy.deepcopy(self.sample_data.test_listener1_dict) test_listener['sni_containers'] = [()] self.assertRaises(exceptions.ValidationException, utils.listener_dict_to_provider_dict, test_listener) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_db_pool_to_provider_pool(self, mock_load_cert, mock_secret): pool_cert = data_models.TLSContainer(certificate='pool cert') mock_load_cert.return_value = {'tls_cert': pool_cert, 'sni_certs': None, 'client_ca_cert': None} mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] provider_pool = utils.db_pool_to_provider_pool( self.sample_data.db_pool1) self.assertEqual(self.sample_data.provider_pool1, provider_pool) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_db_pool_to_provider_pool_partial(self, mock_load_cert, mock_secret): pool_cert = data_models.TLSContainer(certificate='pool cert') mock_load_cert.return_value = {'tls_cert': pool_cert, 'sni_certs': None, 'client_ca_cert': None} mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] test_db_pool = self.sample_data.db_pool1 test_db_pool.members = [self.sample_data.db_member1] provider_pool = utils.db_pool_to_provider_pool(test_db_pool) self.assertEqual(self.sample_data.provider_pool1, provider_pool) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_db_pools_to_provider_pools(self, mock_load_cert, mock_secret): pool_cert = data_models.TLSContainer(certificate='pool cert') mock_load_cert.return_value = {'tls_cert': pool_cert, 'sni_certs': None, 'client_ca_cert': None} mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] provider_pools = utils.db_pools_to_provider_pools( self.sample_data.test_db_pools) self.assertEqual(self.sample_data.provider_pools, provider_pools) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_pool_dict_to_provider_dict(self, mock_load_cert, mock_secret): pool_cert = data_models.TLSContainer(certificate='pool cert') mock_load_cert.return_value = {'tls_cert': pool_cert, 'sni_certs': None, 'client_ca_cert': None} mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] expect_prov = copy.deepcopy(self.sample_data.provider_pool1_dict) expect_prov.pop('crl_container_ref') provider_pool_dict = utils.pool_dict_to_provider_dict( self.sample_data.test_pool1_dict) provider_pool_dict.pop('crl_container_ref') self.assertEqual(expect_prov, provider_pool_dict) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_pool_dict_to_provider_dict_load_cert_error( self, mock_load_cert, mock_secret): mock_load_cert.side_effect = [exceptions.OctaviaException, Exception] # Test load_cert exception for_delete == False path self.assertRaises(exceptions.OctaviaException, utils.pool_dict_to_provider_dict, self.sample_data.test_pool1_dict) @mock.patch('octavia.api.drivers.utils._get_secret_data') @mock.patch('octavia.common.tls_utils.cert_parser.load_certificates_data') def test_pool_dict_to_provider_dict_load_cert_error_for_delete( self, mock_load_cert, mock_secret): mock_load_cert.side_effect = [Exception] # Test load_cert exception for_delete == True path mock_secret.side_effect = ['X509 POOL CA CERT FILE', 'X509 POOL CRL FILE'] expect_prov = copy.deepcopy(self.sample_data.provider_pool1_dict) expect_prov.pop('crl_container_ref') del expect_prov['tls_container_data'] provider_pool_dict = utils.pool_dict_to_provider_dict( self.sample_data.test_pool1_dict, for_delete=True) provider_pool_dict.pop('crl_container_ref') self.assertEqual(expect_prov, provider_pool_dict) def test_db_HM_to_provider_HM(self): provider_hm = utils.db_HM_to_provider_HM(self.sample_data.db_hm1) self.assertEqual(self.sample_data.provider_hm1, provider_hm) def test_hm_dict_to_provider_dict(self): provider_hm_dict = utils.hm_dict_to_provider_dict( self.sample_data.test_hm1_dict) self.assertEqual(self.sample_data.provider_hm1_dict, provider_hm_dict) def test_HM_to_provider_HM_with_http_version_and_domain_name(self): provider_hm = utils.db_HM_to_provider_HM(self.sample_data.db_hm2) self.assertEqual(self.sample_data.provider_hm2, provider_hm) provider_hm_dict = utils.hm_dict_to_provider_dict( self.sample_data.test_hm2_dict) self.assertEqual(self.sample_data.provider_hm2_dict, provider_hm_dict) def test_hm_dict_to_provider_dict_partial(self): provider_hm_dict = utils.hm_dict_to_provider_dict({'id': 1}) self.assertEqual({'healthmonitor_id': 1}, provider_hm_dict) def test_db_members_to_provider_members(self): provider_members = utils.db_members_to_provider_members( self.sample_data.db_pool1_members) self.assertEqual(self.sample_data.provider_pool1_members, provider_members) def test_member_dict_to_provider_dict(self): provider_member_dict = utils.member_dict_to_provider_dict( self.sample_data.test_member1_dict) self.assertEqual(self.sample_data.provider_member1_dict, provider_member_dict) def test_db_l7policies_to_provider_l7policies(self): provider_rules = utils.db_l7policies_to_provider_l7policies( self.sample_data.db_l7policies) self.assertEqual(self.sample_data.provider_l7policies, provider_rules) def test_l7policy_dict_to_provider_dict(self): provider_l7policy_dict = utils.l7policy_dict_to_provider_dict( self.sample_data.test_l7policy1_dict) self.assertEqual(self.sample_data.provider_l7policy1_dict, provider_l7policy_dict) def test_db_l7rules_to_provider_l7rules(self): provider_rules = utils.db_l7rules_to_provider_l7rules( self.sample_data.db_l7Rules) self.assertEqual(self.sample_data.provider_rules, provider_rules) def test_l7rule_dict_to_provider_dict(self): provider_rules_dict = utils.l7rule_dict_to_provider_dict( self.sample_data.test_l7rule1_dict) self.assertEqual(self.sample_data.provider_l7rule1_dict, provider_rules_dict) def test_vip_dict_to_provider_dict(self): new_vip_dict = utils.vip_dict_to_provider_dict( self.sample_data.test_vip_dict) self.assertEqual(self.sample_data.provider_vip_dict, new_vip_dict) def test_vip_dict_to_provider_dict_partial(self): new_vip_dict = utils.vip_dict_to_provider_dict( {'ip_address': '192.0.2.44'}) self.assertEqual({'vip_address': '192.0.2.44'}, new_vip_dict) def test_provider_vip_dict_to_vip_obj(self): new_provider_vip = utils.provider_vip_dict_to_vip_obj( self.sample_data.provider_vip_dict) self.assertEqual(self.sample_data.db_vip, new_provider_vip)