text: # Copyright (c) 2016 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 mock
  
  import oslo_messaging
  from oslo_serialization import jsonutils
  from oslo_utils import uuidutils
  import testtools
  
  from neutron.api.rpc.handlers import resources_rpc
  from neutron.common import utils
  from neutron.objects import trunk as trunk_obj
  from neutron.services.trunk import constants
  from neutron.services.trunk.drivers.openvswitch.agent import exceptions
  from neutron.services.trunk.drivers.openvswitch.agent import ovsdb_handler
  from neutron.services.trunk.drivers.openvswitch.agent import trunk_manager
  from neutron.tests import base
  
  
  class TestLockOnBridgeName(base.BaseTestCase):
      def setUp(self):
          super(TestLockOnBridgeName, self).setUp()
          self.lock_mock = mock.patch('oslo_concurrency.lockutils.lock').start()
          self.method_called = False
  
      def test_positional_argument(self):
          @ovsdb_handler.lock_on_bridge_name(required_parameter='bridge_name')
          def testing_method(param, bridge_name, another):
              self.method_called = True
  
          testing_method('foo', 'bridge', 'bar')
          self.lock_mock.assert_called_once_with('bridge')
          self.assertTrue(self.method_called)
  
      def test_keyword_argument(self):
          @ovsdb_handler.lock_on_bridge_name(required_parameter='bridge_name')
          def testing_method(param, bridge_name, another):
              self.method_called = True
  
          testing_method('foo', another='baz', bridge_name='bridge')
          self.lock_mock.assert_called_once_with('bridge')
          self.assertTrue(self.method_called)
  
      def test_missing_argument(self):
          with testtools.ExpectedException(RuntimeError):
              @ovsdb_handler.lock_on_bridge_name(
                  required_parameter='bridge_name')
              def testing_method(one, two):
                  pass
  
  
  class TestIsTrunkServicePort(base.BaseTestCase):
      def test_with_bridge_name(self):
          observed = ovsdb_handler.is_trunk_service_port('tbr-foo')
          self.assertTrue(observed)
  
      def test_with_subport_patch_port_int_side(self):
          observed = ovsdb_handler.is_trunk_service_port('spi-foo')
          self.assertTrue(observed)
  
      def test_with_subport_patch_port_trunk_side(self):
          observed = ovsdb_handler.is_trunk_service_port('spt-foo')
          self.assertTrue(observed)
  
      def test_with_trunk_patch_port_int_side(self):
          observed = ovsdb_handler.is_trunk_service_port('tpi-foo')
          self.assertTrue(observed)
  
      def test_with_trunk_patch_port_trunk_side(self):
          observed = ovsdb_handler.is_trunk_service_port('tpt-foo')
          self.assertTrue(observed)
  
      def test_with_random_string(self):
          observed = ovsdb_handler.is_trunk_service_port('foo')
          self.assertFalse(observed)
  
  
  class TestBridgeHasInstancePort(base.BaseTestCase):
      def setUp(self):
          super(TestBridgeHasInstancePort, self).setUp()
          self.bridge = mock.Mock()
          self.present_interfaces = []
          self.bridge.get_iface_name_list.return_value = self.present_interfaces
  
      def test_only_service_ports_on_bridge(self):
          """Test when only with patch ports and bridge name are on trunk bridge.
          """
          self.present_interfaces.extend(
              ['tbr-foo', 'spt-foo', 'tpt-foo'])
          self.assertFalse(ovsdb_handler.bridge_has_instance_port(self.bridge))
  
      def test_device_on_bridge(self):
          """Condition is True decause of foo device is present on bridge."""
          self.present_interfaces.extend(
              ['tbr-foo', 'spt-foo', 'tpt-foo', 'foo'])
          self.assertTrue(ovsdb_handler.bridge_has_instance_port(self.bridge))
  
      def test_ovsdb_error(self):
          self.bridge.get_iface_name_list.side_effect = RuntimeError
          self.assertFalse(ovsdb_handler.bridge_has_instance_port(self.bridge))
  
  
  class TestOVSDBHandler(base.BaseTestCase):
      """Test that RPC or OVSDB failures do not cause crash."""
      def setUp(self):
          super(TestOVSDBHandler, self).setUp()
          self.ovsdb_handler = ovsdb_handler.OVSDBHandler(mock.sentinel.manager)
          mock.patch.object(self.ovsdb_handler, 'trunk_rpc').start()
          mock.patch.object(self.ovsdb_handler, 'trunk_manager').start()
          self.trunk_manager = self.ovsdb_handler.trunk_manager
          self.trunk_id = uuidutils.generate_uuid()
          self.fake_subports = [
                  trunk_obj.SubPort(
                      id=uuidutils.generate_uuid(),
                      port_id=uuidutils.generate_uuid(),
                      segmentation_id=1)]
          self.fake_port = {
              'name': 'foo',
              'external_ids': {
                  'trunk_id': 'trunk_id',
                  'subport_ids': jsonutils.dumps(
                      [s.id for s in self.fake_subports]),
              }
          }
          self.subport_bindings = {
              'trunk_id': [
                  {'id': subport.port_id,
                   'mac_address': 'mac'} for subport in self.fake_subports]}
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      @mock.patch.object(utils, 'wait_until_true',
                         side_effect=utils.WaitTimeout)
      def test_handle_trunk_add_interface_wont_appear(self, wut, br):
          self.ovsdb_handler.handle_trunk_add('foo')
          self.assertTrue(self.trunk_manager.dispose_trunk.called)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_handle_trunk_add_rpc_failure(self, br):
          with mock.patch.object(self.ovsdb_handler, '_wire_trunk',
                  side_effect=oslo_messaging.MessagingException):
              with mock.patch.object(ovsdb_handler, 'bridge_has_instance_port',
                      return_value=True):
                  self.ovsdb_handler.handle_trunk_add('foo')
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_handle_trunk_add_ovsdb_failure(self, br):
          with mock.patch.object(self.ovsdb_handler, '_wire_trunk',
                  side_effect=RuntimeError):
              with mock.patch.object(ovsdb_handler, 'bridge_has_instance_port',
                      return_value=True):
                  self.ovsdb_handler.handle_trunk_add('foo')
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_handle_trunk_add_parent_port_not_found(self, br):
          with mock.patch.object(self.ovsdb_handler, '_get_parent_port',
                  side_effect=exceptions.ParentPortNotFound):
              # do not wait the default timeout
              self.ovsdb_handler.timeout = 1
              self.ovsdb_handler.handle_trunk_add('foo')
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_handle_trunk_add_missing_bridge(self, br):
          br.return_value.bridge_exists.return_value = False
          with mock.patch.object(
                  ovsdb_handler, 'bridge_has_instance_port') as has_port_mock:
              self.ovsdb_handler.handle_trunk_add('foo')
          self.assertFalse(has_port_mock.called)
  
      def test_handle_trunk_remove_trunk_manager_failure(self):
          with mock.patch.object(self.ovsdb_handler, '_get_trunk_metadata',
                  side_effect=trunk_manager.TrunkManagerError(error='error')):
              with mock.patch.object(ovsdb_handler, 'bridge_has_instance_port',
                      return_value=True):
                  self.ovsdb_handler.handle_trunk_remove('foo', self.fake_port)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_handle_trunk_remove_rpc_failure(self, br):
          self.ovsdb_handler.trunk_rpc.update_trunk_status = (
              oslo_messaging.MessagingException)
          self.ovsdb_handler.handle_trunk_remove('foo', self.fake_port)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_wire_subports_for_trunk_trunk_manager_failure(self, br):
          with mock.patch.object(
                  self.ovsdb_handler, '_update_trunk_metadata') as f:
              trunk_rpc = self.ovsdb_handler.trunk_rpc
              trunk_rpc.update_subport_bindings.return_value = (
                  self.subport_bindings)
              self.trunk_manager.add_sub_port.side_effect = (
                  trunk_manager.TrunkManagerError(error='error'))
  
              status = self.ovsdb_handler.wire_subports_for_trunk(
                  None, 'trunk_id', self.fake_subports)
              self.assertTrue(f.call_count)
              self.assertEqual(constants.DEGRADED_STATUS, status)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_wire_subports_for_trunk_ovsdb_failure(self, br):
          self.ovsdb_handler.trunk_rpc.update_subport_bindings.return_value = (
              self.subport_bindings)
          with mock.patch.object(self.ovsdb_handler, '_set_trunk_metadata',
                  side_effect=RuntimeError):
              status = self.ovsdb_handler.wire_subports_for_trunk(
                  None, 'trunk_id', self.fake_subports)
          self.assertEqual(constants.DEGRADED_STATUS, status)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_unwire_subports_for_trunk_port_not_found(self, br):
          self.ovsdb_handler.trunk_rpc.update_subport_bindings.return_value = (
              self.subport_bindings)
          with mock.patch.object(self.ovsdb_handler, '_update_trunk_metadata',
                  side_effect=exceptions.ParentPortNotFound(bridge='foo_br')):
              status = self.ovsdb_handler.unwire_subports_for_trunk(
                  'trunk_id', ['subport_id'])
          self.assertEqual(constants.ACTIVE_STATUS, status)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test_unwire_subports_for_trunk_trunk_manager_failure(self, br):
          with mock.patch.object(
              self.ovsdb_handler, '_update_trunk_metadata') as f:
              self.trunk_manager.remove_sub_port.side_effect = (
                  trunk_manager.TrunkManagerError(error='error'))
              status = self.ovsdb_handler.unwire_subports_for_trunk(
                  'foo_trunk_id', ['subport_id'])
              self.assertTrue(f.call_count)
              self.assertEqual(constants.DEGRADED_STATUS, status)
  
      def test__wire_trunk_get_trunk_details_failure(self):
          self.trunk_manager.get_port_uuid_from_external_ids.side_effect = (
              trunk_manager.TrunkManagerError(error='error'))
          self.ovsdb_handler._wire_trunk(mock.Mock(), self.fake_port)
  
      def test__wire_trunk_trunk_not_associated(self):
          self.ovsdb_handler.trunk_rpc.get_trunk_details.side_effect = (
              resources_rpc.ResourceNotFound(
                  resource_id='id', resource_type='type'))
          self.ovsdb_handler._wire_trunk(mock.Mock(), self.fake_port)
          trunk_rpc = self.ovsdb_handler.trunk_rpc
          self.assertFalse(trunk_rpc.update_trunk_status.called)
  
      def test__wire_trunk_create_trunk_failure(self):
          self.trunk_manager.create_trunk.side_effect = (
              trunk_manager.TrunkManagerError(error='error'))
          self.ovsdb_handler._wire_trunk(mock.Mock(), self.fake_port)
          trunk_rpc = self.ovsdb_handler.trunk_rpc
          trunk_rpc.update_trunk_status.assert_called_once_with(
              mock.ANY, mock.ANY, constants.ERROR_STATUS)
  
      def test__wire_trunk_rewire_trunk_failure(self):
          with mock.patch.object(self.ovsdb_handler,
                                 'unwire_subports_for_trunk') as f,\
                  mock.patch.object(self.ovsdb_handler,
                                    'get_connected_subports_for_trunk') as g:
              g.return_value = ['stale_port']
              f.return_value = constants.DEGRADED_STATUS
              self.ovsdb_handler._wire_trunk(
                  mock.Mock(), self.fake_port, rewire=True)
              trunk_rpc = self.ovsdb_handler.trunk_rpc
              trunk_rpc.update_trunk_status.assert_called_once_with(
                  mock.ANY, mock.ANY, constants.DEGRADED_STATUS)
  
      def test__wire_trunk_report_trunk_called_on_wiring(self):
          with mock.patch.object(self.trunk_manager, 'create_trunk'),\
                  mock.patch.object(self.ovsdb_handler,
                                    'wire_subports_for_trunk'):
              self.ovsdb_handler._wire_trunk(mock.Mock(), self.fake_port)
              trunk_rpc = self.ovsdb_handler.trunk_rpc
              self.assertTrue(trunk_rpc.update_trunk_status.called)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test__set_trunk_metadata_with_None_params(self, br):
          mock_br = br.return_value
          with mock.patch.object(
                  self.ovsdb_handler, "_get_parent_port",
                  return_value={'name': 'foo', 'external_ids': {}}):
              self.ovsdb_handler._set_trunk_metadata(None, None, 'foo', [])
              mock_br.set_db_attribute.assert_called_once_with(
                  'Interface', 'foo', 'external_ids',
                  {'bridge_name': mock.ANY, 'trunk_id': 'foo',
                   'subport_ids': '[]'})
  
      def test__get_current_status_active(self):
          self.assertEqual(constants.ACTIVE_STATUS,
                           self.ovsdb_handler._get_current_status([], []))
  
      def test__get_current_status_degraded(self):
          self.assertEqual(constants.DEGRADED_STATUS,
                           self.ovsdb_handler._get_current_status(
                               [mock.ANY], []))
  
      def _test__update_trunk_metadata_wire_flag(self, mock_br, wire,
                                                 external_ids,
                                                 subport_ids,
                                                 expected_subport_ids):
          with mock.patch.object(
                  self.ovsdb_handler, "_get_parent_port",
                  return_value={'name': 'foo', 'external_ids': external_ids}):
              self.ovsdb_handler._update_trunk_metadata(
                  None, None, 'foo', subport_ids, wire=wire)
              external_ids = mock_br.set_db_attribute.call_args[0][3]
              self.assertEqual(1, mock_br.set_db_attribute.call_count)
              self.assertEqual(
                  sorted(expected_subport_ids),
                  sorted(external_ids['subport_ids']))
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test__update_trunk_metadata_wire(self, br):
          mock_br = br.return_value
          external_ids = {
              'subport_ids': '["foo_subport_1"]'
          }
          subport_ids = ['foo_subport_2']
          expected_subport_ids = '["foo_subport_1", "foo_subport_2"]'
          self._test__update_trunk_metadata_wire_flag(
              mock_br, True, external_ids, subport_ids, expected_subport_ids)
  
      @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
      def test__update_trunk_metadata_unwire(self, br):
          mock_br = br.return_value
          external_ids = {
              'subport_ids': '["foo_subport_1", "foo_subport_2"]'
          }
          subport_ids = ['foo_subport_2']
          expected_subport_ids = '["foo_subport_1"]'
          self._test__update_trunk_metadata_wire_flag(
              mock_br, False, external_ids, subport_ids, expected_subport_ids)
  