test_asynchronous.py 文件源码

def test_run(self, m_count):
        event = mock.sentinel.event
        group = mock.sentinel.group
        m_queue = mock.Mock()
        m_queue.empty.return_value = True
        m_queue.get.return_value = event
        m_handler = mock.Mock()
        m_count.return_value = [1]
        async_handler = h_async.Async(m_handler, mock.Mock(), mock.Mock(),
                                      queue_depth=1)

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

        m_handler.assert_called_once_with(event)