api.py 文件源码

def _coro_make_request(*,
                             endpoint: APITestEndPoint,
                             proxy: ProxyConfig = None,
                             download_content: bool = False,
                             accept_selfsigned_certs: bool = True):
    """
    A coroutine that makes the requests

    :param endpoint: APITestEndPoint object instance
    :type endpoint: APITestEndPoint

    :param proxy: ProxyConfig object instance
    :type proxy: ProxyConfig

    :return: the
    :rtype:
    """
    assert isinstance(endpoint, APITestEndPoint)
    assert isinstance(proxy, ProxyConfig)

    # Proxy needs authentication?
    proxy_auth = None
    if proxy.user is not None:
        proxy_auth = aiohttp.BasicAuth(proxy.user, proxy.password)

    # Extract info
    url = endpoint.request.url
    body = transform_apitest_body_to_queryable(endpoint.request.body)
    method = endpoint.request.method
    headers = {header.key: header.value for header in endpoint.request.headers}

    # Check SSL?
    conn = None
    if accept_selfsigned_certs:
        conn = aiohttp.TCPConnector(verify_ssl=False)

    # Do the Requests
    async with aiohttp.ClientSession(connector=conn) as session:
        async with session.request(method=method,
                                   url=url,
                                   headers=headers,
                                   data=body,
                                   proxy=proxy.url,
                                   proxy_auth=proxy_auth) as resp:

            if resp.status == 200:
                log.console("    - Response OK for URL: '{}'".format(url))
            else:
                log.console("    - Non-200 response for URL: '{}'".format(url))
                log.console("      \_ HTTP status code: '{}'".format(resp.status))

            # Download content?
            content = None
            if download_content:
                content = await resp.text()
                log.console("      \_{}".format(content))

            return RequestResponse(status=resp.status,
                                   content=content,
                                   headers=resp.headers)