test_paraproxio.py 文件源码

def _go(self, file_path, check_resp: Callable[[ClientResponse], None] = None):
        """ Make a test request to the web server through the proxy."""
        expected_content = TEST_WEB_SERVER_FILES.get(file_path)
        expected_content_length = len(expected_content)
        connector = aiohttp.ProxyConnector(proxy=PROXY_ADDRESS, loop=self.loop)
        async with aiohttp.client.ClientSession(connector=connector, loop=self.loop) as session:
            url = create_host_url(file_path)
            async with session.get(url) as resp:  # type: ClientResponse
                self.assertEqual(resp.status, 200)
                if resp.headers.get(hdrs.TRANSFER_ENCODING) != 'chunked':
                    self.assertEqual(resp.headers.get(hdrs.CONTENT_LENGTH), str(expected_content_length))
                content = await resp.read()
                self.assertEqual(content, expected_content)
                if check_resp:
                    check_resp(resp)
                await asyncio.sleep(1, loop=self.loop)  # Wait a little bit before closing the session.