text: # Copyright (c) 2016 IBM Corp.
  #
  # 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 os
  import sys
  from unittest import mock
  
  from neutron_lib.agent import topics
  from neutron_lib import constants
  from neutron_lib.utils import helpers
  from oslo_config import cfg
  from oslo_service import service
  
  from neutron.agent.linux import ip_lib
  from neutron.common import config as common_config
  from neutron.plugins.ml2.drivers.agent import _agent_manager_base as amb
  from neutron.plugins.ml2.drivers.macvtap.agent import macvtap_neutron_agent
  from neutron.plugins.ml2.drivers.macvtap import macvtap_common
  from neutron.tests import base
  
  
  INTERFACE_MAPPINGS = {'physnet1': 'eth1'}
  NETWORK_ID = 'net-id123'
  NETWORK_SEGMENT_VLAN = amb.NetworkSegment('vlan', 'physnet1', 1)
  NETWORK_SEGMENT_FLAT = amb.NetworkSegment('flat', 'physnet1', None)
  
  
  class TestMacvtapRPCCallbacks(base.BaseTestCase):
      def setUp(self):
          super(TestMacvtapRPCCallbacks, self).setUp()
  
          agent = mock.Mock()
          agent.mgr = mock.Mock()
          agent.mgr.interface_mappings = INTERFACE_MAPPINGS
          self.rpc = macvtap_neutron_agent.MacvtapRPCCallBack(mock.Mock(), agent,
                                                              mock.Mock())
  
      def test_network_delete_vlan(self):
          self.rpc.network_map = {NETWORK_ID: NETWORK_SEGMENT_VLAN}
          with mock.patch.object(ip_lib.IpLinkCommand, 'delete') as mock_del,\
                  mock.patch.object(macvtap_common, 'get_vlan_device_name',
                                    return_value='vlan1'),\
                  mock.patch.object(ip_lib.IPDevice, 'exists',
                                    return_value=True):
              self.rpc.network_delete("anycontext", network_id=NETWORK_ID)
              self.assertTrue(mock_del.called)
  
      def test_network_delete_flat(self):
          self.rpc.network_map = {NETWORK_ID: NETWORK_SEGMENT_FLAT}
          with mock.patch.object(ip_lib.IpLinkCommand, 'delete') as mock_del:
              self.rpc.network_delete(
                  "anycontext", network_id=NETWORK_SEGMENT_FLAT.segmentation_id)
              self.assertFalse(mock_del.called)
  
      def test_port_update(self):
          port = {'id': 'port-id123', 'mac_address': 'mac1'}
          self.rpc.port_update(context=None, port=port)
          self.assertEqual(set(['mac1']), self.rpc.updated_devices)
  
  
  class TestMacvtapManager(base.BaseTestCase):
      def setUp(self):
          super(TestMacvtapManager, self).setUp()
          with mock.patch.object(ip_lib, 'device_exists', return_value=True):
              self.mgr = macvtap_neutron_agent.MacvtapManager(INTERFACE_MAPPINGS)
  
      def test_validate_interface_mappings_dev_exists(self):
          good_mapping = {'physnet1': 'eth1', 'physnet2': 'eth2'}
          self.mgr.interface_mappings = good_mapping
          with mock.patch.object(ip_lib, 'device_exists', return_value=True)\
                  as mock_de:
              self.mgr.validate_interface_mappings()
              mock_de.assert_any_call('eth1')
              mock_de.assert_any_call('eth2')
              self.assertEqual(2, mock_de.call_count)
  
      def test_validate_interface_mappings_dev_not_exists(self):
          bad_mapping = {'physnet1': 'foo'}
          self.mgr.interface_mappings = bad_mapping
          with mock.patch.object(ip_lib, 'device_exists', return_value=False)\
                  as mock_de, mock.patch.object(sys, 'exit') as mock_exit:
              self.mgr.validate_interface_mappings()
              mock_de.assert_called_with('foo')
              mock_exit.assert_called_once_with(1)
  
      def _test_ensure_port_admin_state(self, admin_state):
          dev = 'macvtap1'
          mac = 'mac1'
  
          self.mgr.mac_device_name_mappings = {mac: dev}
          with mock.patch.object(ip_lib, 'IPDevice') as mock_ip_dev:
              self.mgr.ensure_port_admin_state(mac, admin_state)
              self.assertEqual(admin_state, mock_ip_dev(dev).link.set_up.called)
              self.assertNotEqual(admin_state,
                                  mock_ip_dev(dev).link.set_down.called)
  
      def test_ensure_port_admin_state_up(self):
          self._test_ensure_port_admin_state(True)
  
      def test_ensure_port_admin_state_down(self):
          self._test_ensure_port_admin_state(False)
  
      def test_get_all_devices(self):
          listing = ['foo', 'macvtap0', 'macvtap1', 'bar']
          # set some mac mappings to make sure they are cleaned up
          self.mgr.mac_device_name_mappings = {'foo': 'bar'}
          with mock.patch.object(os, 'listdir',
                                 return_value=listing) as mock_ld,\
                  mock.patch.object(ip_lib, 'get_device_mac') as mock_gdn:
              mock_gdn.side_effect = ['mac0', 'mac1']
  
              result = self.mgr.get_all_devices()
              mock_ld.assert_called_once_with(macvtap_neutron_agent.MACVTAP_FS)
              self.assertEqual(set(['mac0', 'mac1']), result)
              self.assertEqual({'mac0': 'macvtap0', 'mac1': 'macvtap1'},
                               self.mgr.mac_device_name_mappings)
  
      def test_get_agent_configurations(self):
          expected = {'interface_mappings': INTERFACE_MAPPINGS}
          self.assertEqual(expected, self.mgr.get_agent_configurations())
  
      def test_get_agent_id_ok(self):
          mock_devices = [ip_lib.IPDevice('macvtap1')]
          with mock.patch.object(ip_lib.IPWrapper, 'get_devices',
                                 return_value=mock_devices),\
              mock.patch.object(ip_lib, 'get_device_mac',
                                return_value='foo:bar'):
              self.assertEqual('macvtapfoobar', self.mgr.get_agent_id())
  
      def test_get_agent_id_fail(self):
          mock_devices = []
          with mock.patch.object(ip_lib.IPWrapper, 'get_devices',
                                 return_value=mock_devices),\
                  mock.patch.object(sys, 'exit') as mock_exit:
              self.mgr.get_agent_id()
              mock_exit.assert_called_once_with(1)
  
      def test_get_agent_id_no_mac(self):
          mock_devices = [ip_lib.IPDevice('macvtap0'),
                          ip_lib.IPDevice('macvtap1')]
          with mock.patch.object(ip_lib.IPWrapper, 'get_devices',
                                 return_value=mock_devices),\
              mock.patch.object(ip_lib, 'get_device_mac',
                                side_effect=[None, 'foo:bar:1']) as mock_gdm:
              self.assertEqual('macvtapfoobar1', self.mgr.get_agent_id())
              mock_gdm.assert_has_calls([mock.call('macvtap0'),
                                         mock.call('macvtap1')])
  
      def test_get_extension_driver_type(self):
          self.assertEqual('macvtap', self.mgr.get_extension_driver_type())
  
      def test_get_rpc_callbacks(self):
          context = mock.Mock()
          agent = mock.Mock()
          sg_agent = mock.Mock()
          obj = self.mgr.get_rpc_callbacks(context, agent, sg_agent)
          self.assertIsInstance(obj, macvtap_neutron_agent.MacvtapRPCCallBack)
  
      def test_get_rpc_consumers(self):
          consumers = [[topics.PORT, topics.UPDATE],
                       [topics.NETWORK, topics.DELETE],
                       [topics.SECURITY_GROUP, topics.UPDATE]]
          self.assertEqual(consumers, self.mgr.get_rpc_consumers())
  
      def test_plug_interface(self):
          self.mgr.mac_device_name_mappings['mac1'] = 'macvtap0'
          with mock.patch.object(ip_lib.IpLinkCommand, 'set_allmulticast_on')\
                  as mock_sao:
              self.mgr.plug_interface('network_id', 'network_segment', 'mac1',
                                      'device_owner')
              self.assertTrue(mock_sao.called)
  
  
  class TestMacvtapMain(base.BaseTestCase):
      def test_parse_interface_mappings_good(self):
          cfg.CONF.set_override('physical_interface_mappings', 'good_mapping',
                                'macvtap')
          with mock.patch.object(helpers, 'parse_mappings',
                                 return_value=INTERFACE_MAPPINGS):
              mappings = macvtap_neutron_agent.parse_interface_mappings()
              self.assertEqual(INTERFACE_MAPPINGS, mappings)
  
      def test_parse_interface_mappings_bad(self):
          cfg.CONF.set_override('physical_interface_mappings', 'bad_mapping',
                                'macvtap')
          with mock.patch.object(helpers, 'parse_mappings',
                                 side_effect=ValueError('bad mapping')),\
                  mock.patch.object(sys, 'exit') as mock_exit:
              macvtap_neutron_agent.parse_interface_mappings()
              mock_exit.assert_called_with(1)
  
      def test_parse_interface_mappings_no_mapping(self):
          with mock.patch.object(sys, 'exit') as mock_exit:
              macvtap_neutron_agent.parse_interface_mappings()
              mock_exit.assert_called_with(1)
  
      def test_validate_firewall_driver_noop_long(self):
          cfg.CONF.set_override('firewall_driver',
                                'neutron.agent.firewall.NoopFirewallDriver',
                                'SECURITYGROUP')
          macvtap_neutron_agent.validate_firewall_driver()
  
      def test_validate_firewall_driver_noop(self):
          cfg.CONF.set_override('firewall_driver',
                                'noop',
                                'SECURITYGROUP')
          macvtap_neutron_agent.validate_firewall_driver()
  
      def test_validate_firewall_driver_other(self):
          cfg.CONF.set_override('firewall_driver',
                                'foo',
                                'SECURITYGROUP')
          with mock.patch.object(sys, 'exit')as mock_exit:
              macvtap_neutron_agent.validate_firewall_driver()
              mock_exit.assert_called_with(1)
  
      def test_main(self):
          cfg.CONF.set_override('quitting_rpc_timeout', 1, 'AGENT')
          cfg.CONF.set_override('polling_interval', 2, 'AGENT')
  
          mock_manager_return = mock.Mock(spec=amb.CommonAgentManagerBase)
          mock_launch_return = mock.Mock()
  
          with mock.patch.object(common_config, 'init'),\
              mock.patch.object(common_config, 'setup_logging'),\
              mock.patch.object(service, 'launch',
                                return_value=mock_launch_return) as mock_launch,\
              mock.patch.object(macvtap_neutron_agent,
                                'parse_interface_mappings',
                                return_value=INTERFACE_MAPPINGS) as mock_pim,\
              mock.patch.object(macvtap_neutron_agent,
                                'validate_firewall_driver') as mock_vfd,\
              mock.patch('neutron.plugins.ml2.drivers.agent._common_agent.'
                         'CommonAgentLoop') as mock_loop,\
              mock.patch('neutron.plugins.ml2.drivers.macvtap.agent.'
                         'macvtap_neutron_agent.MacvtapManager',
                         return_value=mock_manager_return) as mock_manager:
              macvtap_neutron_agent.main()
              self.assertTrue(mock_vfd.called)
              self.assertTrue(mock_pim.called)
              mock_manager.assert_called_with(INTERFACE_MAPPINGS)
              mock_loop.assert_called_with(mock_manager_return, 2, 1,
                                           constants.AGENT_TYPE_MACVTAP,
                                           constants.AGENT_PROCESS_MACVTAP)
              self.assertTrue(mock_launch.called)
              self.assertTrue(mock_launch_return.wait.called)
  