neutron/tests/unit/agent/linux/test_polling.py

   1 # Copyright 2013 Red Hat, Inc.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import mock
  16
  17 from neutron.agent.common import base_polling
  18 from neutron.agent.linux import polling
  19 from neutron.tests import base
  20
  21
  22 class TestGetPollingManager(base.BaseTestCase):
  23
  24     def test_return_always_poll_by_default(self):
  25         with polling.get_polling_manager() as pm:
  26             self.assertEqual(pm.__class__, base_polling.AlwaysPoll)
  27
  28     def test_manage_polling_minimizer(self):
  29         mock_target = 'neutron.agent.linux.polling.InterfacePollingMinimizer'
  30         with mock.patch('%s.start' % mock_target) as mock_start:
  31             with mock.patch('%s.stop' % mock_target) as mock_stop:
  32                 with polling.get_polling_manager(minimize_polling=True) as pm:
  33                     self.assertEqual(pm.__class__,
  34                                      polling.InterfacePollingMinimizer)
  35                 mock_stop.assert_has_calls([mock.call()])
  36             mock_start.assert_has_calls([mock.call()])
  37
  38
  39 class TestInterfacePollingMinimizer(base.BaseTestCase):
  40
  41     def setUp(self):
  42         super(TestInterfacePollingMinimizer, self).setUp()
  43         self.pm = polling.InterfacePollingMinimizer()
  44
  45     def test_start_calls_monitor_start(self):
  46         with mock.patch.object(self.pm._monitor, 'start') as mock_start:
  47             self.pm.start()
  48         mock_start.assert_called_with()
  49
  50     def test_stop_calls_monitor_stop(self):
  51         with mock.patch.object(self.pm._monitor, 'stop') as mock_stop:
  52             self.pm.stop()
  53         mock_stop.assert_called_with()
  54
  55     def mock_has_updates(self, return_value):
  56         target = ('neutron.agent.linux.ovsdb_monitor.SimpleInterfaceMonitor'
  57                   '.has_updates')
  58         return mock.patch(
  59             target,
  60             new_callable=mock.PropertyMock(return_value=return_value),
  61         )
  62
  63     def test__is_polling_required_returns_when_updates_are_present(self):
  64         with self.mock_has_updates(True):
  65             self.assertTrue(self.pm._is_polling_required())