openstack/aodh
Code Issues Proposed changes

Remove eventlet.sleep(0) in collector tests

Browse Source 
This change removes eventlet.sleep(0) in collector tests

Change-Id: I72df104720c4c0b6cf41adfcafea0c769e872953
Closes-bug: #1312826
This commit is contained in:
Mehdi Abaakouk 2014-04-25 18:37:20 +02:00
parent ad72ca88ab
commit 4f958acbff
1 changed files with 18 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
ceilometer/tests/test_collector.py
View File

@ -15,9 +15,9 @@
# 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 contextlib
import socket
import eventlet
import mock
import msgpack
import oslo.messaging
@ -159,32 +159,36 @@ class TestCollector(tests_base.BaseTestCase):
self._verify_udp_socket(udp_socket)
@staticmethod
def _dummy_thread_group_add_thread(method):
method()
@mock.patch.object(oslo.messaging.MessageHandlingServer, 'start')
@mock.patch.object(collector.CollectorService, 'start_udp')
def test_only_udp(self, udp_start, rpc_start):
"""Check that only UDP is started if rpc_backend is empty."""
self.CONF.set_override('rpc_backend', '')
udp_socket = self._make_fake_socket(self.counter)
with mock.patch('socket.socket', return_value=udp_socket):
with contextlib.nested(
mock.patch.object(
self.srv.tg, 'add_thread',
side_effect=self._dummy_thread_group_add_thread),
mock.patch('socket.socket', return_value=udp_socket)):
self.srv.start()
# UDP run into its own thread, so we need to sleep to get
# the thread start
eventlet.sleep(0)
self.assertEqual(0, rpc_start.call_count)
self.assertEqual(1, udp_start.call_count)
self.assertEqual(0, rpc_start.call_count)
self.assertEqual(1, udp_start.call_count)
@mock.patch.object(oslo.messaging.MessageHandlingServer, 'start')
@mock.patch.object(collector.CollectorService, 'start_udp')
def test_only_rpc(self, udp_start, rpc_start):
"""Check that only RPC is started if udp_address is empty."""
self.CONF.set_override('udp_address', '', group='collector')
self.srv.start()
# UDP run into its own thread, so we need to sleep to get
# the thread start
eventlet.sleep(0)
self.assertEqual(1, rpc_start.call_count)
self.assertEqual(0, udp_start.call_count)
with mock.patch.object(
self.srv.tg, 'add_thread',
side_effect=self._dummy_thread_group_add_thread):
self.srv.start()
self.assertEqual(1, rpc_start.call_count)
self.assertEqual(0, udp_start.call_count)
def test_udp_receive_valid_encoding(self):
self.data_sent = []