def convert(ctx: Context, url: str) -> str:
        """This converter checks whether the given URL can be reached with a status code of 200."""
        try:
            async with ctx.bot.http_session.get(url) as resp:
                if resp.status != 200:
                    raise BadArgument(
                        f"HTTP GET on `{url}` returned status `{resp.status}`, expected 200"
                    )
        except CertificateError:
            if url.startswith('https'):
                raise BadArgument(
                    f"Got a `CertificateError` for URL `{url}`. Does it support HTTPS?"
                )
            raise BadArgument(f"Got a `CertificateError` for URL `{url}`.")
        except ValueError:
            raise BadArgument(f"`{url}` doesn't look like a valid hostname to me.")
        except ClientConnectorError:
            raise BadArgument(f"Cannot connect to host with URL `{url}`.")
        return url 

def get_cat_image_url(timeout: float) -> str:
    api_url = 'http://thecatapi.com/api/images/get'
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                async with session.get(
                    api_url, params={'format': 'xml', 'type': 'jpg,png'}
                ) as res:
                    if res.status != 200:
                        raise APIServerError
                    xml_result = await res.read()
                    tree = etree.fromstring(xml_result)
                    url = tree.find('data/images/image/url').text
            except aiohttp.client_exceptions.ServerDisconnectedError:
                await asyncio.sleep(0.1)
                continue
            try:
                async with async_timeout.timeout(timeout=timeout):
                    async with session.get(url) as res:
                        async with res:
                            if res.status == 200:
                                return url
            except (aiohttp.ClientConnectorError, asyncio.TimeoutError):
                continue 

def get_dog_image_url(timeout: float) -> str:
    api_url = 'https://dog.ceo/api/breeds/image/random'
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                async with session.get(api_url) as res:
                    if res.status != 200:
                        raise APIServerError
                    data = await res.json(loads=json.loads)
                    url = data['message']
            except aiohttp.client_exceptions.ServerDisconnectedError:
                await asyncio.sleep(0.1)
                continue
            try:
                async with async_timeout.timeout(timeout=timeout):
                    async with session.get(url) as res:
                        async with res:
                            if res.status == 200:
                                return url
            except (aiohttp.ClientConnectorError, asyncio.TimeoutError):
                continue 

def get_stratz_match(match_id):
	url = f"https://api.stratz.com/api/v1/match/{match_id}"
	cached_data = httpgetter.cache.get(url, "json")
	
	if cached_data:
		if is_stratz_parsed(cached_data):
			return cached_data
		else:
			await httpgetter.cache.remove(url)

	try:
		return await httpgetter.get(url, cache=True, errors={
			500: "Looks like something wrong with the STRATZ api",
			204: "STRATZ hasn't recieved this match yet. Try again a bit later"
		})
	except aiohttp.ClientConnectorError:
		print("ClientConnectorError on stratz api result")
		raise StratzMatchNotParsedError(match_id) 

def analyze_nodes(self, address, port):
        found_nodes = []
        async with self.semaphore:
            full_host = f'http://{address}:{port}'
            self.logger.info(f'[+] Scanning host at {full_host}')
            try:
                async with aiohttp.ClientSession(loop=asyncio.get_event_loop(), timeout=self.timeout) as client:
                    ros_master_client = ServerProxy(full_host, client=client)
                    code, msg, val = await ros_master_client.getSystemState('')
                    if code == 1:
                        nodes = list(map(lambda x: x[0], map(lambda x: x[1], reduce(lambda x, y: x + y, val))))
                        for node in nodes:
                            if node in self.rosin_nodes:
                                found_nodes.append(node)
                if len(found_nodes) > 0:
                    ros_host = ROSHost(address, port)
                    ros_host.nodes = found_nodes
                    self.hosts.append(ros_host)
            except ClientConnectorError:
                self.logger.debug(f'[-] Unable to connect to host {address}')
            except Exception as e:
                ex, msg, tb = sys.exc_info()
                traceback.print_tb(tb)
                self.logger.debug(f'[-] Connection error on host {address}') 

def test_single_proxy(self, proxy):
        """
        text one proxy, if valid, put them to usable_proxies.
        """
        try:
            async with aiohttp.ClientSession() as session:
                try:
                    if isinstance(proxy, bytes):
                        proxy = proxy.decode('utf-8')
                    real_proxy = 'http://' + proxy
                    print('Testing', proxy)
                    async with session.get(self.test_api, proxy=real_proxy, timeout=get_proxy_timeout) as response:
                        if response.status == 200:
                            self._conn.put(proxy)
                            print('Valid proxy', proxy)
                except (ProxyConnectionError, TimeoutError, ValueError):
                    print('Invalid proxy', proxy)
        except (ServerDisconnectedError, ClientResponseError,ClientConnectorError) as s:
            print(s)
            pass 

def post_one(item, headers, session):
    payload = {
        'actionType': 'APPEND',
        'entities': item
    }

    payload = dumps(payload)

    url = orion + '/v2/op/update'
    try:
        async with session.post(url, headers=headers, data=payload) as response:
            status = response.status
    except ClientConnectorError:
        return 'connection problem'
    except ToE:
        return 'timeout problem'

    if status not in http_ok:
        return 'response code ' + str(status)

    return True 

def post_one(item, headers, session):
    payload = {
        'actionType': 'APPEND',
        'entities': item
    }

    payload = dumps(payload)

    url = orion + '/v2/op/update'
    try:
        async with session.post(url, headers=headers, data=payload) as response:
            status = response.status
    except ClientConnectorError:
        return 'connection problem'
    except ToE:
        return 'timeout problem'

    if status not in http_ok:
        return 'response code ' + str(status)

    return True 

def post_one(item, headers, session):
    payload = {
        'actionType': 'APPEND',
        'entities': item
    }

    payload = dumps(payload)

    url = orion + '/v2/op/update'
    try:
        async with session.post(url, headers=headers, data=payload) as response:
            status = response.status
    except ClientConnectorError:
        return 'connection problem'
    except ToE:
        return 'timeout problem'

    if status not in http_ok:
        return 'response code ' + str(status)

    return True 

def post_one(item, headers, session):
    payload = {
        'actionType': 'APPEND',
        'entities': item
    }

    payload = dumps(payload)

    url = orion + '/v2/op/update'
    try:
        async with session.post(url, headers=headers, data=payload) as response:
            status = response.status
    except ClientConnectorError:
        return 'connection problem'
    except ToE:
        return 'timeout problem'

    if status not in http_ok:
        return 'response code ' + str(status)

    return True 

def post_one(el, headers, session):
    payload = {
        'actionType': 'APPEND',
        'entities': el
    }

    payload = dumps(payload)

    url = orion + '/v2/op/update'
    try:
        async with session.post(url, headers=headers, data=payload) as response:
            status = response.status
    except ClientConnectorError:
        return 'connection problem'
    except ToE:
        return 'timeout problem'

    if status not in http_ok:
        return 'response code ' + str(status)

    return True 

def test_exceptions(self):
        import aiohttp

        exceptions = aiohttp_.AiohttpClient.exceptions

        with pytest.raises(exceptions.BaseClientException):
            raise aiohttp.ClientError()

        with pytest.raises(exceptions.BaseClientException):
            # Test polymorphism
            raise aiohttp.InvalidURL("invalid")

        with pytest.raises(exceptions.ConnectionError):
            raise aiohttp.ClientConnectionError()

        with pytest.raises(exceptions.ConnectionTimeout):
            raise aiohttp.ClientConnectorError.__new__(
                aiohttp.ClientConnectorError
            )

        with pytest.raises(exceptions.ServerTimeout):
            raise aiohttp.ServerTimeoutError()

        with pytest.raises(exceptions.SSLError):
            raise aiohttp.ClientSSLError.__new__(aiohttp.ClientSSLError)

        with pytest.raises(exceptions.InvalidURL):
            raise aiohttp.InvalidURL("invalid") 

def test_proxy_failure_async(self):
        client: WebClient = WebClient(
            token=self.bot_token,
            proxy=self.proxy,
            run_async=True
        )
        with self.assertRaises(ClientConnectorError):
            await client.auth_test() 

def test_fail_proxy_request(aa_fail_proxy_config, s3_client):
    # based on test_can_make_request

    with pytest.raises(aiohttp.ClientConnectorError):
        await s3_client.list_buckets() 

def test_resolve_host_fail(loop, remote_resolve):
    tr, proto = mock.Mock(name='transport'), mock.Mock(name='protocol')

    with mock.patch('aiosocks.connector.create_connection',
                    make_mocked_coro((tr, proto))):
        req = ProxyClientRequest(
            'GET', URL('http://python.org'), loop=loop,
            proxy=URL('socks5://proxy.example'))
        connector = ProxyConnector(loop=loop, remote_resolve=remote_resolve)
        connector._resolve_host = make_mocked_coro(raise_exception=OSError())

        with pytest.raises(aiohttp.ClientConnectorError):
            await connector.connect(req, [], ClientTimeout()) 

def running(self):
        """Start websocket connection."""
        try:
            async with self.session.ws_connect(
                self.url, ssl=self.ssl_context, heartbeat=15
            ) as ws:
                self.state = STATE_RUNNING

                async for msg in ws:

                    if self.state == STATE_STOPPED:
                        break

                    if msg.type == aiohttp.WSMsgType.TEXT:
                        self._data = json.loads(msg.data)
                        self.session_handler_callback(SIGNAL_DATA)
                        LOGGER.debug(msg.data)

                    elif msg.type == aiohttp.WSMsgType.CLOSED:
                        LOGGER.warning("AIOHTTP websocket connection closed")
                        break

                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        LOGGER.error("AIOHTTP websocket error")
                        break

        except aiohttp.ClientConnectorError:
            if self.state != STATE_STOPPED:
                LOGGER.error("Client connection error")
                self.state = STATE_DISCONNECTED

        except Exception as err:
            if self.state != STATE_STOPPED:
                LOGGER.error("Unexpected error %s", err)
                self.state = STATE_DISCONNECTED

        else:
            if self.state != STATE_STOPPED:
                self.state = STATE_DISCONNECTED 

def running(self):
        """Start websocket connection."""
        url = f"http://{self.host}:{self.port}"

        try:
            async with self.session.ws_connect(url, heartbeat=15) as ws:
                self.state = STATE_RUNNING

                async for msg in ws:

                    if self.state == STATE_STOPPED:
                        break

                    elif msg.type == aiohttp.WSMsgType.TEXT:
                        self._data = json.loads(msg.data)
                        self.session_handler_callback("data")
                        LOGGER.debug(msg.data)

                    elif msg.type == aiohttp.WSMsgType.CLOSED:
                        LOGGER.warning("pydeCONZ websocket connection closed")
                        break

                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        LOGGER.error("pydeCONZ websocket error")
                        break

        except aiohttp.ClientConnectorError:
            LOGGER.error("Client connection error")
            if self.state != STATE_STOPPED:
                self.retry()

        except Exception as err:
            LOGGER.error("Unexpected error %s", err)
            if self.state != STATE_STOPPED:
                self.retry()

        else:
            if self.state != STATE_STOPPED:
                self.retry() 

def listen(slack_client, url):
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(3)) as session:
        print(f"connecting to {url}")
        try:
            ws = await session.ws_connect(url)
        except (aiohttp.ClientConnectorError, asyncio.TimeoutError):
            print(f"failed to connect to {url}")
            return
        print(f"connected to {url}")

        async for msg in ws:
            r = json.loads(msg.data)
            try:
                queries = r["api"]["search"]["interrupted_queries"]
            except KeyError:
                continue

            for q in queries:
                # clean = re.sub(r"\s+", " ", q)
                clean = sqlparse.format(q, reindent=True, keyword_case='upper')
                print(f'{url}: {clean}')
                response = await slack_client.chat_postMessage(
                    username=url,
                    icon_emoji=":hourglass_flowing_sand:",
                    channel='#clubhouse-de-obscure',
                    text="*Query timed out:* " + clean
                )
                if not response["ok"]:
                    print("SLACK ERROR:\n", response)
                print() 

def version_check(ev):
    """
    :param ev: The event object referenced in the event.
    :type ev: sigma.core.mechanics.event.SigmaEvent
    """
    version_url = 'https://gitlab.com/lu-ci/sigma/apex-sigma/raw/master/info/version.yml'
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(version_url) as version_data:
                data = await version_data.read()
                data = yaml.safe_load(data)
    except (aiohttp.ClientConnectorError, YAMLError):
        data = None
    if data:
        official_stamp = data['build_date']
        version = ev.bot.info.get_version()
        current_stamp = version.timestamp
        if official_stamp > current_stamp:
            current = f'{version.major}.{version.minor}.{version.patch} {version.codename}'
            latest = f'{data["version"]["major"]}.{data["version"]["minor"]}.{data["version"]["patch"]}'
            latest += f' {data["codename"]}'
            ev.log.warning('---------------------------------')
            ev.log.warning('Your Sigma version is outdated.')
            ev.log.warning(f'CURRENT: {current}')
            ev.log.warning(f'LATEST:  {latest}')
            ev.log.warning('Updating is strongly suggested.')
            ev.log.warning('---------------------------------')
    else:
        ev.log.warning('Could not retrieve latest version information.') 

def test_connection_errors(self):
        trace_configs = [
            opentelemetry.ext.aiohttp_client.create_trace_config()
        ]

        for url, expected_status in (
            ("http://this-is-unknown.local/", StatusCanonicalCode.UNKNOWN),
            ("http://127.0.0.1:1/", StatusCanonicalCode.UNAVAILABLE),
        ):
            with self.subTest(expected_status=expected_status):

                async def do_request(url):
                    async with aiohttp.ClientSession(
                        trace_configs=trace_configs
                    ) as session:
                        async with session.get(url):
                            pass

                loop = asyncio.get_event_loop()
                with self.assertRaises(aiohttp.ClientConnectorError):
                    loop.run_until_complete(do_request(url))

            self.assert_spans(
                [
                    (
                        "GET",
                        (expected_status, None),
                        {
                            "component": "http",
                            "http.method": "GET",
                            "http.url": url,
                        },
                    )
                ]
            )
            self.memory_exporter.clear() 

def test_get_dog_image_url(response_mock):
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random', body='', status=500,
    )
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random',
        exception=aiohttp.client_exceptions.ServerDisconnectedError(),
    )
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random',
        body=json.dumps(
            {'status': 'success', 'message': 'http://dog.com/404.jpg'}
        ),
        headers={'Content-Type': 'application/json'},
    )
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random',
        body=json.dumps(
            {
                'status': 'success',
                'message': 'http://cannotresolve.com/200.jpg',
            }
        ),
        headers={'Content-Type': 'application/json'},
    )
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random',
        body=json.dumps(
            {'status': 'success', 'message': 'http://timeout.com/200.jpg'}
        ),
        headers={'Content-Type': 'application/json'},
    )
    response_mock.get(
        'https://dog.ceo/api/breeds/image/random',
        body=json.dumps(
            {'status': 'success', 'message': 'http://dog.com/200.jpg'}
        ),
        headers={'Content-Type': 'application/json'},
    )
    response_mock.get(
        'http://cannotresolve.com/200.jpg',
        exception=aiohttp.ClientConnectorError(None, OSError()),
    )
    response_mock.get(
        'http://timeout.com/200.jpg', exception=asyncio.TimeoutError(),
    )
    response_mock.get('http://dog.com/404.jpg', status=404)
    response_mock.get('http://dog.com/200.jpg', status=200)

    with pytest.raises(APIServerError):
        await get_dog_image_url(0.001)

    url = await get_dog_image_url(0.001)
    assert url == 'http://dog.com/200.jpg' 

def read(self, url, *args, use_proxy=False, **kwargs) -> FeedResponse:
        headers = content = None
        try:
            if use_proxy:
                headers, content, url, status = await self._read_by_proxy(url, *args, **kwargs)
            else:
                headers, content, url, status = await self._read(url, *args, **kwargs)
        except (socket.gaierror, aiodns.error.DNSError):
            status = FeedResponseStatus.DNS_ERROR.value
        except (socket.timeout, TimeoutError, aiohttp.ServerTimeoutError,
                asyncio.TimeoutError, concurrent.futures.TimeoutError):
            status = FeedResponseStatus.CONNECTION_TIMEOUT.value
        except (ssl.SSLError, ssl.CertificateError,
                aiohttp.ServerFingerprintMismatch,
                aiohttp.ClientSSLError,
                aiohttp.ClientConnectorSSLError,
                aiohttp.ClientConnectorCertificateError):
            status = FeedResponseStatus.SSL_ERROR.value
        except (aiohttp.ClientProxyConnectionError,
                aiohttp.ClientHttpProxyError):
            status = FeedResponseStatus.PROXY_ERROR.value
        except (ConnectionError,
                aiohttp.ServerDisconnectedError,
                aiohttp.ServerConnectionError,
                aiohttp.ClientConnectionError,
                aiohttp.ClientConnectorError):
            status = FeedResponseStatus.CONNECTION_RESET.value
        except (aiohttp.WSServerHandshakeError, aiohttp.ClientOSError):
            status = FeedResponseStatus.CONNECTION_ERROR.value
        except aiohttp.ClientPayloadError:
            status = FeedResponseStatus.CHUNKED_ENCODING_ERROR.value
        except UnicodeDecodeError:
            status = FeedResponseStatus.CONTENT_DECODING_ERROR.value
        except FeedReaderError as ex:
            status = ex.status
            LOG.warning(type(ex).__name__ + " url=%s %s", url, ex)
        except (aiohttp.ClientResponseError, aiohttp.ContentTypeError) as ex:
            status = ex.status
        except (aiohttp.ClientError, aiohttp.InvalidURL):
            status = FeedResponseStatus.UNKNOWN_ERROR.value
        builder = FeedResponseBuilder(use_proxy=use_proxy)
        builder.url(url)
        builder.status(status)
        builder.content(content)
        builder.headers(headers)
        return builder.build() 

def collect_one(station, session):

    try:
        async with session.get(stations[station]['url']) as response:
            result = await response.text()
            status = response.status
    except ClientConnectorError:
        logger.error('Collecting data from IPMA station %s failed due to the connection problem', station)
        return False
    except ToE:
        logger.error('Collecting link from IPMA station %s failed due to the timeout problem', station)
        return False

    if status not in http_ok:
        logger.error('Collecting data from IPMA station %s failed due to the return code %s', station, status)
        return False

    content = loads(result)

    result = dict()
    result['id'] = station
    result['retrieved'] = datetime.now().replace(microsecond=0)
    result['forecasts'] = dict()

    today = datetime.now(tz).strftime("%Y-%m-%d") + 'T00:00:00'
    tomorrow = (datetime.now(tz) + timedelta(days=1)).strftime("%Y-%m-%d") + 'T00:00:00'

    for forecast in content:
        if forecast['idPeriodo'] != 24:
            continue

        date = forecast['dataPrev']

        if date not in [today, tomorrow]:
            continue

        result['forecasts'][date] = dict()

        result['forecasts'][date]['feelsLikeTemperature'] = check_entity(forecast, 'utci')
        result['forecasts'][date]['issued'] = datetime.strptime(forecast['dataUpdate'], '%Y-%m-%dT%H:%M:%S')
        result['forecasts'][date]['period'] = forecast['idPeriodo']
        result['forecasts'][date]['precipitationProbability'] = check_entity(forecast, 'probabilidadePrecipita')
        result['forecasts'][date]['relativeHumidity'] = check_entity(forecast, 'hR')
        result['forecasts'][date]['temperature'] = check_entity(forecast, 'tMed')
        result['forecasts'][date]['tMax'] = check_entity(forecast, 'tMax')
        result['forecasts'][date]['tMin'] = check_entity(forecast, 'tMin')
        result['forecasts'][date]['weatherType'] = check_entity(forecast, 'idTipoTempo')
        result['forecasts'][date]['windDirection'] = check_entity(forecast, 'ddVento')
        result['forecasts'][date]['windSpeed'] = check_entity(forecast, 'ffVento')

    return result 

def collect_one(station, session, key):

    try:
        async with session.get(stations[station]['url'], headers={'api_key': key}, ssl=False) as response:
            result = await response.read()
            status = response.status
    except ClientConnectorError:
        logger.error('Collecting link from AEMET station %s failed due to the connection problem', station)
        return False
    except ToE:
        logger.error('Collecting link from AEMET station %s failed due to the timeout problem', station)
        return False

    if status not in http_ok:
        logger.error('Collecting link from AEMET station %s failed due to the return code %s', station, str(status))
        return False

    logger.debug('Remaining requests %s', response.headers.get('Remaining-request-count'))
    result = loads(result.decode('UTF-8'))

    try:
        content = get(result['datos'])
    except exceptions.ConnectionError:
        logger.error('Collecting data from AEMET station %s failed due to the connection problem', station)
        return False

    if content.status_code not in http_ok:
        logger.error('Collecting data from AEMET station %s failed due to the return code %s', station,
                     str(response.status))
        return False

    content = loads(content.text)

    result = dict()
    result['station'] = station
    issued = datetime.strptime(content[0]['elaborado'], "%Y-%m-%d")
    result['issued'] = issued.replace(tzinfo=tz).isoformat().replace('+00:00', 'Z')
    result['retrieved'] = datetime.now(tz).replace(tzinfo=tz).replace(microsecond=0).isoformat().replace('+00:00', 'Z')

    content = sorted(content[0]['prediccion']['dia'], key=lambda k: (k['fecha']), reverse=False)

    result['today'] = content[0]
    result['tomorrow'] = content[1]

    return result 

def retrieve_certificates(loop, url=None, ctl_offset=0, output_directory='/tmp/', concurrency_count=DOWNLOAD_CONCURRENCY):
    async with aiohttp.ClientSession(loop=loop, conn_timeout=10) as session:
        ctl_logs = await certlib.retrieve_all_ctls(session)

        if url:
            url = url.strip("'")

        for log in ctl_logs:
            if url and url not in log['url']:
                continue
            work_deque = deque()
            download_results_queue = asyncio.Queue(maxsize=MAX_QUEUE_SIZE)

            logging.info("Downloading certificates for {}".format(log['description']))
            try:
                log_info = await certlib.retrieve_log_info(log, session)
            except (aiohttp.ClientConnectorError, aiohttp.ServerTimeoutError, aiohttp.ClientOSError, aiohttp.ClientResponseError) as e:
                logging.error("Failed to connect to CTL! -> {} - skipping.".format(e))
                continue

            try:
                await certlib.populate_work(work_deque, log_info, start=ctl_offset)
            except Exception as e:
                logging.error("Log needs no update - {}".format(e))
                continue

            download_tasks = asyncio.gather(*[
                download_worker(session, log_info, work_deque, download_results_queue)
                for _ in range(concurrency_count)
            ])

            processing_task    = asyncio.ensure_future(processing_coro(download_results_queue, output_dir=output_directory))
            queue_monitor_task = asyncio.ensure_future(queue_monitor(log_info, work_deque, download_results_queue))

            asyncio.ensure_future(download_tasks)

            await download_tasks

            await download_results_queue.put(None) # Downloads are done, processing can stop

            await processing_task

            queue_monitor_task.cancel()

            logging.info("Completed {}, stored at {}!".format(
                log_info['description'],
                '/tmp/{}.csv'.format(log_info['url'].replace('/', '_'))
            ))

            logging.info("Finished downloading and processing {}".format(log_info['url']))