Python源码示例:aiohttp.test_utils()

示例1
def _http_request(
        trace_config,
        url: str,
        method: str = "GET",
        status_code: int = HTTPStatus.OK,
        request_handler: typing.Callable = None,
        **kwargs
    ) -> typing.Tuple[str, int]:
        """Helper to start an aiohttp test server and send an actual HTTP request to it."""

        async def do_request():
            async def default_handler(request):
                assert "traceparent" in request.headers
                return aiohttp.web.Response(status=int(status_code))

            handler = request_handler or default_handler

            app = aiohttp.web.Application()
            parsed_url = urllib.parse.urlparse(url)
            app.add_routes([aiohttp.web.get(parsed_url.path, handler)])
            app.add_routes([aiohttp.web.post(parsed_url.path, handler)])
            app.add_routes([aiohttp.web.patch(parsed_url.path, handler)])

            with contextlib.suppress(aiohttp.ClientError):
                async with aiohttp.test_utils.TestServer(app) as server:
                    netloc = (server.host, server.port)
                    async with aiohttp.test_utils.TestClient(
                        server, trace_configs=[trace_config]
                    ) as client:
                        await client.start_server()
                        await client.request(
                            method, url, trace_request_ctx={}, **kwargs
                        )
            return netloc

        loop = asyncio.get_event_loop()
        return loop.run_until_complete(do_request())