def _query_qna_service(
        self, turn_context: TurnContext, options: QnAMakerOptions
    ) -> QueryResults:
        url = f"{ self._endpoint.host }/knowledgebases/{ self._endpoint.knowledge_base_id }/generateAnswer"

        question = GenerateAnswerRequestBody(
            question=turn_context.activity.text,
            top=options.top,
            score_threshold=options.score_threshold,
            strict_filters=options.strict_filters,
            context=options.context,
            qna_id=options.qna_id,
            is_test=options.is_test,
            ranker_type=options.ranker_type,
        )

        http_request_helper = HttpRequestUtils(self._http_client)

        response: ClientResponse = await http_request_helper.execute_http_request(
            url, question, self._endpoint, options.timeout
        )

        result: QueryResults = await self._format_qna_result(response, options)

        return result 

def fetch(self, url, **kwargs) -> aiohttp.ClientResponse:
        headers = {"User-Agent": get_user_agent()}
        async with aiohttp.ClientSession(
            conn_timeout=self.config["response_timeout"],
            read_timeout=self.config["response_timeout"],
        ) as session:
            try:
                async with session.get(
                    urljoin(self.host, url), headers=headers, **kwargs
                ) as response:
                    await response.text()
                    return response
            except aiohttp.ClientConnectionError:
                raise exceptions.SlaveDoesNotExist(
                    f"Unable to connect to the slave at {self.host}"
                ) 

def _read_content(self, response: aiohttp.ClientResponse):
        content_length = response.headers.get('Content-Length')
        if content_length:
            content_length = int(content_length)
            if content_length > self.max_content_length:
                msg = 'content length {} larger than limit {}'.format(
                    content_length, self.max_content_length)
                raise ContentTooLargeError(msg)
        content_length = 0
        content = bytearray()
        async for chunk in response.content.iter_chunked(8 * 1024):
            content_length += len(chunk)
            if content_length > self.max_content_length:
                msg = 'content length larger than limit {}'.format(
                    self.max_content_length)
                raise ContentTooLargeError(msg)
            content.extend(chunk)
        return content 

def _read(
            self, url, etag=None, last_modified=None, referer=None,
            headers=None, ignore_content=False
    ) -> aiohttp.ClientResponse:
        headers = self._prepare_headers(
            etag=etag,
            last_modified=last_modified,
            referer=referer,
            headers=headers,
        )
        await self._async_init()
        if not self.allow_private_address:
            await self.check_private_address(url)
        async with self.session.get(url, headers=headers) as response:
            content = None
            if not is_ok_status(response.status) or not ignore_content:
                content = await self._read_content(response)
            if not is_ok_status(response.status):
                return response.headers, content, url, response.status
            self.check_content_type(response)
        return response.headers, content, str(response.url), response.status 

def setUp(self):
        self.main_page_path = generate_unique_path()
        os.makedirs(self.main_page_path)
        self.dorks = dict(response={'dorks': "test_dorks"})
        self.loop = asyncio.new_event_loop()
        aiohttp.ClientSession.get = AsyncMock(
            return_value=aiohttp.ClientResponse(
                url=yarl.URL("http://www.example.com"),
                method="GET",
                writer=None,
                continue100=1,
                timer=None,
                request_info=None,
                traces=None,
                loop=self.loop,
                session=None))
        no_dorks = True
        tanner = "tanner.mushmush.org"
        self.handler = HtmlHandler(no_dorks, tanner)
        self.data = None 

def test_event_result(self):
        aiohttp.ClientResponse.json = AsyncMock(
            return_value=dict(
                detection={
                    'type': 1},
                sess_uuid="test_uuid"))

        async def test():
            self.result = await self.handler.submit_data(self.data)

        self.loop.run_until_complete(test())
        self.assertEqual(
            self.result,
            dict(
                detection={
                    'type': 1},
                sess_uuid="test_uuid")) 

def _wrap_response(
        self,
        response: aiohttp.ClientResponse,
        url: str,
        **kwargs: Union[int, Optional[str]],
    ) -> Dict[str, Any]:
        """Parses the response as json, then runs check_response and
        add_jikan_metadata
        """
        json_response: Dict[str, Any] = {}
        try:
            json_response = await response.json()
            if not isinstance(json_response, dict):
                json_response = {"data": json_response}
        except (json.decoder.JSONDecodeError, simplejson.JSONDecodeError):
            json_response = {"error": await response.text()}
        if response.status >= 400:
            raise APIException(response.status, json_response, **kwargs)
        return utils.add_jikan_metadata(response, json_response, url) 

def get(
        self, path: str, required_serial: Optional[int], **kw: Any
    ) -> AsyncGenerator[aiohttp.ClientResponse, None]:
        logger.debug(f"Getting {path} (serial {required_serial})")
        if not path.startswith(("https://", "http://")):
            path = self.url + path

        async with self.session.get(path, **kw) as r:
            got_serial = (
                int(r.headers[PYPI_SERIAL_HEADER])
                if PYPI_SERIAL_HEADER in r.headers
                else None
            )
            await self.check_for_stale_cache(path, required_serial, got_serial)
            yield r

    # TODO: Add storage backend support / refactor - #554 

def validate(self, response: aiohttp.ClientResponse, checker_result: CheckerResult) -> bool:
        if response.status != 200:
            return False

        json_result = await response.json()
        if 'ip' in json_result:
            checker_result.ipv4 = json_result['ip']
        if 'city' in json_result:
            checker_result.city = json_result['city']
        if 'region' in json_result:
            checker_result.region = json_result['region']
        if 'country' in json_result:
            checker_result.country_code = json_result['country']
        if 'loc' in json_result:
            checker_result.location_coordinates = tuple(float(x) for x in json_result['loc'].split(','))
        if 'org' in json_result:
            checker_result.organization_name = json_result['org']

        return True 

def test_iter_lines_generator():
    """Test that lines are split correctly."""
    async def mock_iter_content(n):
        for chunk in [b'1\r\n2\r\n', b'3\r', b'\n4', b'\r\n5']:
            yield chunk

    response = ClientResponse(
        'get', ST_URL,
        request_info=Mock(),
        writer=Mock(),
        continue100=None,
        timer=TimerNoop(),
        traces=[],
        loop=Mock(),
        session=Mock(),
    )
    response._headers = {'Content-Type': 'application/json;charset=utf-8'}
    with patch.object(response, 'content', Mock(iter_chunked=mock_iter_content)):
        result = [
            line async for line in _iter_lines_generator(
                response=response, decode_unicode=True
            )
        ]
        assert result == ['1', '2', '3', '4', '5'] 

def get(self, path, *path_args, params=None):
    """Sends a GET request to a given API endpoint.

    This is a low-level function that returns a raw HTTP response, no error
    checking nor response parsing is performed. See :func:`get_json`,
    :func:`get_data` and :func:`get_object` for higher-level functions.

    :param path: Path to API endpoint, can contain format placeholders {}.
    :param path_args: A variable number of arguments that are put into any
      placeholders used in path.
    :param params: Parameters sent in the request.
    :type path: str
    :type params: dict
    :returns: An instance of :class:`ClientResponse`.
    """
    return _make_sync(self.get_async(path, *path_args, params=params)) 

def get_error_async(self, response):
    """Given a :class:`ClientResponse` returns a :class:`APIError`

    This function checks if the response from the VirusTotal backend was an
    error and returns the appropiate :class:`APIError` or None if no error
    occurred.

    :param response: A :class:`ClientResponse` instance.
    :returns: An instance of :class:`APIError` or None.
    """
    if response.status == 200:
      return None
    if response.status >= 400 and response.status <= 499:
      if response.content_type == 'application/json':
        json_response = await response.json_async()
        error = json_response.get('error')
        if error:
          return APIError.from_dict(error)
      return APIError('ClientError', await response.text_async())
    return APIError('ServerError', await response.text_async()) 

def post(self, path, *path_args, data=None):
    """Sends a POST request to a given API endpoint.

    This is a low-level function that returns a raw HTTP response, no error
    checking nor response parsing is performed. See :func:`post_object` for
    a higher-level function.

    :param path: Path to API endpoint, can contain format placeholders {}.
    :param path_args: A variable number of arguments that are put into any
      placeholders used in path.
    :param data: Data sent in the request body.
    :type path: str
    :type data: A string or bytes
    :returns: An instance of :class:`ClientResponse`.
    """
    return _make_sync(self.post_async(path, *path_args, data=data)) 

def request(self, method: str, url: str,
                      **kwargs: Any
                      ) -> Tuple[aiohttp.ClientResponse, Union[str, dict]]:
        try:
            params = kwargs['params']
            if isinstance(params, dict):
                kwargs['params'] = {k: (str(v).lower() if isinstance(v, bool)
                                        else v) for k, v in params.items()}
            else:
                kwargs['params'] = [(k, (str(v).lower() if isinstance(v, bool)
                                         else v)) for k, v in params]
        except KeyError:
            pass

        pre_time = time.time()
        async with self.__session.request(method, url, **kwargs) as r:
            log.debug('{0} {1} has returned {2.status} in {3:.2f}s'.format(
                method,
                url,
                r,
                time.time() - pre_time
            ))

            data = await self.json_or_text(r)
            return r, data 

def get_snapshot(client: aiohttp.ClientSession, trading_pair: str, limit: int = 1000) -> Dict[str, Any]:
        original_trading_pair: str = trading_pair
        params: Dict[str, str] = {"count": str(limit), "pair": trading_pair} if limit != 0 else {"pair": trading_pair}
        async with client.get(SNAPSHOT_REST_URL, params=params) as response:
            response: aiohttp.ClientResponse = response
            if response.status != 200:
                raise IOError(f"Error fetching Kraken market snapshot for {original_trading_pair}. "
                              f"HTTP status is {response.status}.")
            response_json = await response.json()
            if len(response_json["error"]) > 0:
                raise IOError(f"Error fetching Kraken market snapshot for {original_trading_pair}. "
                              f"Error is {response_json['error']}.")
            data: Dict[str, Any] = next(iter(response_json["result"].values()))
            data = {"trading_pair": trading_pair, **data}
            data["latest_update"] = max([*map(lambda x: x[2], data["bids"] + data["asks"])], default=0.)

            # Need to add the symbol into the snapshot message for the Kafka message queue.
            # Because otherwise, there'd be no way for the receiver to know which market the
            # snapshot belongs to.

            return data 

def get_active_exchange_markets(cls,
                                          api_endpoint: str = "https://rest.bamboorelay.com/",
                                          api_prefix: str = "main/0x") -> pd.DataFrame:
        """
        Returned data frame should have trading_pair as index and include usd volume, baseAsset and quoteAsset
        """
        client: aiohttp.ClientSession = cls.http_client()
        async with client.get(f"{api_endpoint}{api_prefix}/markets?perPage=1000&include=ticker,stats") as response:
            response: aiohttp.ClientResponse = response
            if response.status != 200:
                raise IOError(f"Error fetching active Bamboo Relay markets. HTTP status is {response.status}.")
            data = await response.json()
            data: List[Dict[str, any]] = [
                {**item, **{"baseAsset": item["id"].split("-")[0], "quoteAsset": item["id"].split("-")[1]}}
                for item in data
            ]
            all_markets: pd.DataFrame = pd.DataFrame.from_records(data=data, index="id")

            quote_volume: List[float] = []
            for row in all_markets.itertuples():
                base_volume: float = float(row.stats["volume24Hour"])
                quote_volume.append(base_volume)

            all_markets.loc[:, "volume"] = quote_volume
            return all_markets.sort_values("USDVolume", ascending=False) 

def get_active_exchange_markets(cls) -> pd.DataFrame:
        """
        Returned data frame should have trading pair as index and include usd volume, baseAsset and quoteAsset
        """
        client: aiohttp.ClientSession = cls.http_client()
        async with client.get(f"{MARKETS_URL}?include=ticker,stats") as response:
            response: aiohttp.ClientResponse = response
            if response.status != 200:
                raise IOError(f"Error fetching active Radar Relay markets. HTTP status is {response.status}.")
            data = await response.json()
            data: List[Dict[str, any]] = [
                {**item, **{"baseAsset": item["id"].split("-")[0], "quoteAsset": item["id"].split("-")[1]}}
                for item in data
            ]
            all_markets: pd.DataFrame = pd.DataFrame.from_records(data=data, index="id")

            quote_volume: List[float] = []
            for row in all_markets.itertuples():
                base_volume: float = float(row.stats["volume24Hour"])
                quote_volume.append(base_volume)

            all_markets.loc[:, "volume"] = quote_volume
            return all_markets.sort_values("USDVolume", ascending=False) 

def _format_qna_result(
        self, result, options: QnAMakerOptions
    ) -> QueryResults:
        json_res = result
        if isinstance(result, ClientResponse):
            json_res = await result.json()

        answers_within_threshold = [
            {**answer, "score": answer["score"] / 100}
            for answer in json_res["answers"]
            if answer["score"] / 100 > options.score_threshold
        ]
        sorted_answers = sorted(
            answers_within_threshold, key=lambda ans: ans["score"], reverse=True
        )

        answers_as_query_results = [
            QueryResult().deserialize(answer) for answer in sorted_answers
        ]

        active_learning_enabled = (
            json_res["activeLearningEnabled"]
            if "activeLearningEnabled" in json_res
            else True
        )

        query_answer_response = QueryResults(
            answers_as_query_results, active_learning_enabled
        )

        return query_answer_response 

def test_get_response(self):
        """This test will test out:
        - BaseAttrDict.response
        """
        tag = '2P0LYQ'
        chests = await self.cr.get_player_chests(tag)
        self.assertTrue(isinstance(chests.response, aiohttp.ClientResponse)) 

def threaded_callback(callback):
    coroutine_callback = asyncio.coroutine(callback)

    @asyncio.coroutine
    def new_callback(response):
        if isinstance(response, aiohttp.ClientResponse):
            yield from response.text()
            response = ThreadedResponse(response)
        response = yield from coroutine_callback(response)
        if isinstance(response, ThreadedResponse):
            return response.unwrap()
        else:
            return response

    return new_callback 

def __init__(
        self,
        response: aiohttp.ClientResponse,
        response_json: Optional[dict] = None,
        response_text: str = ""
    ):
        self.status = response.status
        self.response_json = response_json or {}
        self.response_text = response_text
        self.response = response 

def maybe_raise_for_status(self, response: aiohttp.ClientResponse, should_raise: bool) -> None:
        """Raise ResponseCodeError for non-OK response if an exception should be raised."""
        if should_raise and response.status >= 400:
            try:
                response_json = await response.json()
                raise ResponseCodeError(response=response, response_json=response_json)
            except aiohttp.ContentTypeError:
                response_text = await response.text()
                raise ResponseCodeError(response=response, response_text=response_text) 

def store_async(url: str, path: str) -> aiohttp.ClientResponse:
    print('Async storing {url} in {path}'.format(url=url, path=path))
    try:
        async with aiohttp.ClientSession() as aios:
            response = await aios.get(url)
            with open(path, 'wb') as fout:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    fout.write(chunk)
            return response
    # type: ignore # urllib isn't fully stubbed
    except (urllib.error.HTTPError, aiohttp.ClientError) as e:
        raise FetchException(e) 

def get_url(url: str, headers: dict = None) -> Tuple[ClientResponse, dict]:
    headers = headers or get_headers()
    async with ClientSession(headers=headers) as session:
        async with timeout(10):
            async with session.get(url) as resp:
                return resp, await resp.json() 

def request(
        self,
        url: str,
        method: str = "GET",
        payload: dict = None,
        return_response: bool = False,
        headers: dict = None,
    ) -> Union[ClientResponse, dict, str]:
        """
        Makes a HTTP request.

        Parameters
        ----------
        url : str
            The destination URL of the request.
        method : str
            The HTTP method (POST, GET, PUT, DELETE, FETCH, etc.).
        payload : Dict[str, Any]
            The json payload to be sent along the request.
        return_response : bool
            Whether the `ClientResponse` object should be returned.
        headers : Dict[str, str]
            Additional headers to `headers`.

        Returns
        -------
        ClientResponse or Dict[str, Any] or List[Any] or str
            `ClientResponse` if `return_response` is `True`.
            `dict` if the returned data is a json object.
            `list` if the returned data is a json list.
            `str` if the returned data is not a valid json data,
            the raw response.
        """
        async with self.session.request(method, url, headers=headers, json=payload) as resp:
            if return_response:
                return resp
            try:
                return await resp.json()
            except (JSONDecodeError, ClientResponseError):
                return await resp.text() 

def _request(
        self, url: str, method: str = "GET", cached: bool = False, **kwargs
    ) -> aiohttp.ClientResponse:
        headers = {"User-Agent": get_user_agent()}

        if cached and self.config.get("use_mesos_cache", False):
            # TODO: fall back to original host if this fails?
            host = self.cache_host
        else:
            host = self.host

        try:
            async with aiohttp.ClientSession(
                conn_timeout=self.config["response_timeout"],
                read_timeout=self.config["response_timeout"],
            ) as session:
                async with session.request(
                    method=method, url=urljoin(host, url), headers=headers, **kwargs
                ) as resp:
                    # if nobody awaits resp.text() or resp.json() before we exit the session context manager, then the
                    # http connection gets closed before we read the response; then later calls to resp.text/json will
                    # fail.
                    await resp.text()
                    return resp

        except aiohttp.client_exceptions.ClientConnectionError:
            raise exceptions.MasterNotAvailableException(MISSING_MASTER.format(host))
        except aiohttp.client_exceptions.TooManyRedirects:
            raise exceptions.MasterTemporarilyNotAvailableException(
                (
                    "Unable to connect to master at %s, likely due to "
                    "an ongoing leader election"
                )
                % host
            ) 

def aiohttp_raise_for_status(response: aiohttp.ClientResponse):
    # workaround aiohttp bug, can remove after fixed in aiohttp
    # issue: https://github.com/aio-libs/aiohttp/issues/3906
    if response.status >= 400:
        response.release()
        raise aiohttp.ClientResponseError(
            response.request_info,
            response.history,
            status=response.status,
            message=response.reason,
            headers=response.headers,
        ) 

def _read_text(self, response: aiohttp.ClientResponse):
        content = await self._read_content(response)
        return content.decode('utf-8', errors='ignore') 

def _read_by_proxy(
        self, url, etag=None, last_modified=None, referer=None,
        headers=None, ignore_content=False
    ):
        if not self.has_rss_proxy:
            raise ValueError("rss_proxy_url not provided")
        headers = self._prepare_headers(
            etag=etag,
            last_modified=last_modified,
            referer=referer,
            headers=headers,
        )
        data = dict(
            url=url,
            token=self.rss_proxy_token,
            headers=headers,
        )
        await self._async_init()
        async with self.session.post(self.rss_proxy_url, json=data) as response:
            response: aiohttp.ClientResponse
            if not is_ok_status(response.status):
                body = await self._read_text(response)
                message = f'status={response.status} body={body!r}'
                raise RSSProxyError(message)
            proxy_status = response.headers.get('x-rss-proxy-status', None)
            if proxy_status and proxy_status.upper() == 'ERROR':
                body = await self._read_text(response)
                message = f'status={response.status} body={body!r}'
                raise RSSProxyError(message)
            proxy_status = int(proxy_status) if proxy_status else HTTPStatus.OK.value
            content = None
            if not is_ok_status(proxy_status) or not ignore_content:
                content = await self._read_content(response)
            if not is_ok_status(proxy_status):
                return response.headers, content, url, proxy_status
            self.check_content_type(response)
        return response.headers, content, url, proxy_status 

def test_get_dorks(self):
        aiohttp.ClientResponse.json = AsyncMock(
            return_value=dict(response={'dorks': "test_dorks"}))

        async def test():
            self.data = await self.handler.get_dorks()

        self.loop.run_until_complete(test())
        aiohttp.ClientSession.get.assert_called_with(
            'http://tanner.mushmush.org:8090/dorks', timeout=10.0)