text: # Copyright (c) 2017 RedHat, 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 mock
  import munch
  from openstack import exceptions as os_exc
  
  from kuryr_kubernetes.controller.drivers import lb_public_ip\
      as d_lb_public_ip
  from kuryr_kubernetes.controller.drivers import public_ip
  from kuryr_kubernetes.objects import lbaas as obj_lbaas
  from kuryr_kubernetes.tests import base as test_base
  from kuryr_kubernetes.tests.unit import kuryr_fixtures as k_fix
  from oslo_config import cfg
  
  
  class TestFloatingIpServicePubIPDriverDriver(test_base.TestCase):
  
      def test_acquire_service_pub_ip_info_clusterip(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          project_id = mock.sentinel.project_id
          cur_service_pub_ip_info = None
          service = {'spec': {'type': 'ClusterIP'}}
  
          result = cls.acquire_service_pub_ip_info(m_driver, service, project_id,
                                                   cur_service_pub_ip_info)
          self.assertIsNone(result)
  
      def test_acquire_service_pub_ip_info_usr_specified_ip(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.4',
                             'port_id': None,
                             'id': 'a2a62ea7-e3bf-40df-8c09-aa0c29876a6b'})
          os_net.ips.return_value = (ip for ip in [fip])
          project_id = mock.sentinel.project_id
          spec_type = 'LoadBalancer'
          spec_lb_ip = '1.2.3.4'
  
          expected_resp = (obj_lbaas
                           .LBaaSPubIp(ip_id=fip.id,
                                       ip_addr=fip.floating_ip_address,
                                       alloc_method='user'))
  
          result = cls.acquire_service_pub_ip_info(m_driver, spec_type,
                                                   spec_lb_ip,  project_id)
          self.assertEqual(result, expected_resp)
  
      def test_acquire_service_pub_ip_info_user_specified_non_exist_fip(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'port_id': None})
          os_net.ips.return_value = (ip for ip in [fip])
  
          project_id = mock.sentinel.project_id
  
          spec_type = 'LoadBalancer'
          spec_lb_ip = '1.2.3.4'
  
          result = cls.acquire_service_pub_ip_info(m_driver, spec_type,
                                                   spec_lb_ip,  project_id)
          self.assertIsNone(result)
  
      def test_acquire_service_pub_ip_info_user_specified_occupied_fip(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.4',
                             'port_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'})
          os_net.ips.return_value = (ip for ip in [fip])
  
          project_id = mock.sentinel.project_id
          spec_type = 'LoadBalancer'
          spec_lb_ip = '1.2.3.4'
  
          result = cls.acquire_service_pub_ip_info(m_driver, spec_type,
                                                   spec_lb_ip,  project_id)
          self.assertIsNone(result)
  
      @mock.patch('kuryr_kubernetes.config.CONF')
      def test_acquire_service_pub_ip_info_pool_net_not_defined(self, m_cfg):
          driver = d_lb_public_ip.FloatingIpServicePubIPDriver()
          public_net = ''
          m_cfg.neutron_defaults.external_svc_net = public_net
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.ips.return_value = (ip for ip in [])
          project_id = mock.sentinel.project_id
          spec_type = 'LoadBalancer'
          spec_lb_ip = None
  
          self.assertRaises(cfg.RequiredOptError,
                            driver.acquire_service_pub_ip_info,
                            spec_type, spec_lb_ip, project_id)
  
      @mock.patch('kuryr_kubernetes.config.CONF')
      def test_acquire_service_pub_ip_info_pool_subnet_is_none(self, m_cfg):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          public_net = mock.sentinel.public_subnet
          m_cfg.neutron_defaults.external_svc_net = public_net
          m_cfg.neutron_defaults.external_svc_subnet = None
  
          os_net.get_subnet.return_value = munch.Munch(
              {'network_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'})
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          os_net.create_ip.return_value = fip
  
          project_id = mock.sentinel.project_id
          spec_type = 'LoadBalancer'
          spec_lb_ip = None
  
          expected_resp = obj_lbaas.LBaaSPubIp(ip_id=fip.id,
                                               ip_addr=fip.floating_ip_address,
                                               alloc_method='pool')
  
          result = cls.acquire_service_pub_ip_info(m_driver, spec_type,
                                                   spec_lb_ip,  project_id)
          self.assertEqual(result, expected_resp)
  
      @mock.patch('kuryr_kubernetes.config.CONF')
      def test_acquire_service_pub_ip_info_alloc_from_pool(self, m_cfg):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          m_cfg.neutron_defaults.external_svc_subnet = (mock.sentinel
                                                        .external_svc_subnet)
  
          os_net.get_subnet.return_value = munch.Munch(
              {'network_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'})
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          os_net.create_ip.return_value = fip
  
          project_id = mock.sentinel.project_id
          spec_type = 'LoadBalancer'
          spec_lb_ip = None
  
          expected_resp = obj_lbaas.LBaaSPubIp(ip_id=fip.id,
                                               ip_addr=fip.floating_ip_address,
                                               alloc_method='pool')
  
          result = cls.acquire_service_pub_ip_info(m_driver, spec_type,
                                                   spec_lb_ip,  project_id)
          self.assertEqual(result, expected_resp)
  
      def test_release_pub_ip_empty_lb_ip_info(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          service_pub_ip_info = None
  
          rc = cls.release_pub_ip(m_driver, service_pub_ip_info)
          self.assertIs(rc, True)
  
      def test_release_pub_ip_alloc_method_non_pool(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
  
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='kk'))
  
          rc = cls.release_pub_ip(m_driver, service_pub_ip_info)
          self.assertIs(rc, True)
  
      def test_release_pub_ip_alloc_method_user(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
  
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='user'))
          rc = cls.release_pub_ip(m_driver, service_pub_ip_info)
          self.assertIs(rc, True)
  
      def test_release_pub_ip_alloc_method_pool_neutron_exception(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.delete_ip.side_effect = os_exc.SDKException
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
  
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
          rc = cls.release_pub_ip(m_driver, service_pub_ip_info)
          self.assertIs(rc, False)
  
      def test_release_pub_ip_alloc_method_pool_neutron_succeeded(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          self.useFixture(k_fix.MockNetworkClient()).client
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
  
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
          rc = cls.release_pub_ip(m_driver, service_pub_ip_info)
          self.assertIs(rc, True)
  
      def test_associate_pub_ip_empty_params(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.update_floatingip.return_value = None
  
          service_pub_ip_info = None
          vip_port_id = None
  
          result = cls.associate_pub_ip(m_driver, service_pub_ip_info,
                                        vip_port_id)
          self.assertIsNone(result)
  
      def test_associate_lb_fip_id_not_exist(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.update_floatingip.return_value = None
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=0,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
  
          vip_port_id = 'ec29d641-fec4-4f67-928a-124a76b3a777'
  
          result = cls.associate_pub_ip(m_driver, service_pub_ip_info,
                                        vip_port_id)
          self.assertIsNone(result)
  
      def test_associate_lb_fip_id_not_exist_neutron_exception(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.update_ip.side_effect = os_exc.SDKException
  
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
          vip_port_id = 'ec29d641-fec4-4f67-928a-124a76b3a777'
  
          self.assertRaises(os_exc.SDKException, cls.associate_pub_ip,
                            m_driver, service_pub_ip_info, vip_port_id)
  
      def test_disassociate_pub_ip_empty_param(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          self.useFixture(k_fix.MockNetworkClient()).client
          service_pub_ip_info = None
  
          result = cls.disassociate_pub_ip(m_driver, service_pub_ip_info)
  
          self.assertIsNone(result)
  
      def test_disassociate_pub_ip_fip_id_not_exist(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.update_floatingip.return_value = None
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=0,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
  
          result = cls.disassociate_pub_ip(m_driver, service_pub_ip_info)
  
          self.assertIsNone(result)
  
      def test_disassociate_pub_ip_neutron_exception(self):
          cls = d_lb_public_ip.FloatingIpServicePubIPDriver
          m_driver = mock.Mock(spec=cls)
          m_driver._drv_pub_ip = public_ip.FipPubIpDriver()
          os_net = self.useFixture(k_fix.MockNetworkClient()).client
          os_net.update_ip.side_effect = os_exc.SDKException
          fip = munch.Munch({'floating_ip_address': '1.2.3.5',
                             'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'})
          service_pub_ip_info = (obj_lbaas
                                 .LBaaSPubIp(ip_id=fip.id,
                                             ip_addr=fip.floating_ip_address,
                                             alloc_method='pool'))
  
          self.assertRaises(os_exc.SDKException, cls.disassociate_pub_ip,
                            m_driver, service_pub_ip_info)
  