neutron/tests/unit/notifiers/test_batch_notifier.py

   1 # Copyright (c) 2014 OpenStack Foundation.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17
  18 from neutron.notifiers import batch_notifier
  19 from neutron.tests import base
  20
  21
  22 class TestBatchNotifier(base.BaseTestCase):
  23     def setUp(self):
  24         super(TestBatchNotifier, self).setUp()
  25         self.notifier = batch_notifier.BatchNotifier(0.1, lambda x: x)
  26         self.spawn_n = mock.patch('eventlet.spawn_n').start()
  27
  28     def test_queue_event_no_event(self):
  29         self.notifier.queue_event(None)
  30         self.assertEqual(0, len(self.notifier.pending_events))
  31         self.assertEqual(0, self.spawn_n.call_count)
  32
  33     def test_queue_event_first_event(self):
  34         self.notifier.queue_event(mock.Mock())
  35         self.assertEqual(1, len(self.notifier.pending_events))
  36         self.assertEqual(1, self.spawn_n.call_count)
  37
  38     def test_queue_event_multiple_events(self):
  39         events = 6
  40         for i in range(0, events):
  41             self.notifier.queue_event(mock.Mock())
  42         self.assertEqual(events, len(self.notifier.pending_events))
  43         self.assertEqual(1, self.spawn_n.call_count)
  44
  45     def test_queue_event_call_send_events(self):
  46         with mock.patch.object(self.notifier,
  47                                'callback') as send_events:
  48             self.spawn_n.side_effect = lambda func: func()
  49             self.notifier.queue_event(mock.Mock())
  50             self.assertFalse(self.notifier._waiting_to_send)
  51             self.assertTrue(send_events.called)