test_sockets.py 文件源码

def test_run(self, chat):
        client = Mock()

        chat.pubsub = Mock()
        chat.pubsub.channels = ['quorum']
        chat.pubsub.listen.return_value = [{
            'type': 'message',
            'channel': 'quorum',
            'data': 'Calloo! Callay!',
        }]
        chat.greenlet = Mock()

        chat.subscribe(client, 'quorum')
        chat.run()

        gevent.wait()  # wait for event loop
        client.send.assert_called_once_with('quorum:Calloo! Callay!')