testing.py 文件源码

def _local_request(self, method, uri, cookies=None, *args, **kwargs):
        import aiohttp
        if uri.startswith(('http:', 'https:', 'ftp:', 'ftps://' '//')):
            url = uri
        else:
            url = 'http://{host}:{port}{uri}'.format(
                host=HOST, port=PORT, uri=uri)

        log.info(url)
        conn = aiohttp.TCPConnector(verify_ssl=False)
        async with aiohttp.ClientSession(
                cookies=cookies, connector=conn) as session:
            async with getattr(
                    session, method.lower())(url, *args, **kwargs) as response:
                try:
                    response.text = await response.text()
                except UnicodeDecodeError as e:
                    response.text = None
                response.body = await response.read()
                return response