test_asynchronous.py 文件源码

def test_run_empty(self, m_count):
        events = [mock.sentinel.event1, mock.sentinel.event2]
        group = mock.sentinel.group
        m_queue = mock.Mock()
        m_queue.empty.return_value = True
        m_queue.get.side_effect = events + [six_queue.Empty()]
        m_handler = mock.Mock()
        m_count.return_value = list(range(5))
        async_handler = h_async.Async(m_handler, mock.Mock(), mock.Mock())

        with mock.patch('time.sleep'):
            async_handler._run(group, m_queue)

        m_handler.assert_has_calls([mock.call(event) for event in events])
        self.assertEqual(len(events), m_handler.call_count)