test_coros.py 文件源码

def mocked_websocket(*args, **kwargs):
        class MockedWebsocket(mock.Mock):
            def __init__(self, **kwargs):
                super().__init__(**kwargs)
                self.count = 0

            @asyncio.coroutine
            def recv(self, *args, **kwargs):
                self.count += 1
                if self.count > 3:
                    raise ConnectionClosed(1000, "Peace out homie!")
                return "Hello World!"

        return MockedWebsocket()