rpc_test.py 文件源码

def test_timeout_plus_one(self):
        delay = self.client.timeout+1

        # NOTE: This function is only called sometimes, probably depending on
        # which task finishes first, advance() or client.connect().
        async def delay_response(request):
            await asyncio.sleep(delay, loop=self.loop)
            return web.json_response(rsrc.SESSION_GET_RESPONSE)
        self.daemon.response = delay_response

        with self.assertRaises(ConnectionError) as cm:
            await asyncio.gather(self.advance(delay),
                                 self.client.connect(),
                                 loop=self.loop)
        self.assertIn('timeout', str(cm.exception).lower())
        self.assertIn(str(self.client.timeout), str(cm.exception))
        self.assertIn(str(self.url), str(cm.exception))

        self.assert_cb_connected_called(calls=0)
        self.assert_cb_disconnected_called(calls=0)
        self.assert_cb_error_called(calls=1,
                                    args=[(self.url,)],
                                    kwargs=[{'error': r'{}.*{}'.format(self.client.timeout,
                                                                       self.url)}])