def test_websocket_non_regression_bug_105(event_loop, server):

    # This test will check a fix to a race condition which happens if the user is trying
    # to connect using the same client twice at the same time
    # See bug #105

    url = f"ws://{server.hostname}:{server.port}/graphql"
    print(f"url = {url}")

    sample_transport = WebsocketsTransport(url=url)

    client = Client(transport=sample_transport)

    # Create a coroutine which start the connection with the transport but does nothing
    async def client_connect(client):
        async with client:
            await asyncio.sleep(2 * MS)

    # Create two tasks which will try to connect using the same client (not allowed)
    connect_task1 = asyncio.ensure_future(client_connect(client))
    connect_task2 = asyncio.ensure_future(client_connect(client))

    with pytest.raises(TransportAlreadyConnected):
        await asyncio.gather(connect_task1, connect_task2) 

def test_websocket_sending_invalid_payload():

    # Get Websockets transport
    sample_transport = WebsocketsTransport(
        url="wss://countries.trevorblades.com/graphql", ssl=True
    )

    # Instanciate client
    async with Client(transport=sample_transport):

        invalid_payload = '{"id": "1", "type": "start", "payload": "BLAHBLAH"}'

        print(f">>> {invalid_payload}")
        await sample_transport.websocket.send(invalid_payload)

        await asyncio.sleep(2) 

def server_countdown_close_connection_in_middle(ws, path):
    await WebSocketServer.send_connection_ack(ws)

    result = await ws.recv()
    json_result = json.loads(result)
    assert json_result["type"] == "start"
    payload = json_result["payload"]
    query = payload["query"]
    query_id = json_result["id"]

    count_found = search("count: {:d}", query)
    count = count_found[0]
    stopping_before = count // 2
    print(f"Countdown started from: {count}, stopping server before {stopping_before}")
    for number in range(count, stopping_before, -1):
        await ws.send(countdown_server_answer.format(query_id=query_id, number=number))
        await asyncio.sleep(2 * MS)

    print("Closing server while subscription is still running now")
    await ws.close()
    await ws.wait_closed()
    print("Server is now closed") 

def test_websocket_subscription_slow_consumer(
    event_loop, client_and_server, subscription_str
):

    session, server = client_and_server

    count = 10
    subscription = gql(subscription_str.format(count=count))

    async for result in session.subscribe(subscription):
        await asyncio.sleep(10 * MS)

        number = result["number"]
        print(f"Number received: {number}")

        assert number == count

        count -= 1

    assert count == -1 

def server_starwars(ws, path):
    await WebSocketServer.send_connection_ack(ws)

    try:
        await ws.recv()

        reviews = [starwars_expected_one, starwars_expected_two]

        for review in reviews:

            data = (
                '{"type":"data","id":"1","payload":{"data":{"reviewAdded": '
                + json.dumps(review)
                + "}}}"
            )
            await ws.send(data)
            await asyncio.sleep(2 * MS)

        await WebSocketServer.send_complete(ws, 1)
        await WebSocketServer.wait_connection_terminate(ws)

    except websockets.exceptions.ConnectionClosedOK:
        pass

    print("Server is now closed") 

def manipulate_condition(condition):
    print('starting manipulate_condition')

    # pause to let consumers start
    await asyncio.sleep(0.1)

    for i in range(1, 3):
        with await condition:
            print('notifying {} consumers'.format(i))
            condition.notify(n=i)
        await asyncio.sleep(0.1)

    with await condition:
        print('notifying remaining consumers')
        condition.notify_all()

    print('ending manipulate_condition') 

def patcher(self):
        await self.bot.wait_until_ready()
        try:
            await asyncio.sleep(6)  # be safe lolz
            while True:
                if not hasattr(self.bot.send_message, 'old'):
                    print(
                        '[WARNING:] -- Overwriting bot.send_message with '
                        'send_lolz. If bot.send_message is not reloaded,')
                    print(
                        '[WARNING:] -- in the event of a crash of the lolz '
                        'cog, you may not be able revert to bot.send_message '
                        'without a restart/reloading lolz')
                    self.bot.send_message = self.send_lolz(self.bot.send_message)
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            pass 

def hal(self, ctx, *, question="none"):
        """Speak with HAL"""
        author = ctx.message.author
        msg = ""
        found = []
        for k,v in self.responses.items():
            if k in question.lower():
                found.append(v)
        if found:
            msg = randchoice(randchoice(found))
        if not msg:
            msg = randchoice(self.responses["none"])
        await asyncio.sleep(1)
        await self.bot.say(msg.format(author=author))
        if "sing it for me" in question.lower() and "Audio" in self.bot.cogs and author.voice_channel:
            audio = self.bot.get_cog("Audio")
            if audio.music_player.is_done():
                link = "https://www.youtube.com/watch?v=hchUl3QlJZE"
                # probably dont need. just too lazy to check.
                ctx.message.content = "{}play {}".format(ctx.prefix, link)
                if await audio.check_voice(ctx.message.author, ctx.message):
                    audio.queue.append(link) 

def check_output_is_expected(directory, capsys):
    """Create, move, and delete a file."""
    # Create file
    original_filename = os.path.join(directory, 'file.txt')
    pathlib.Path(original_filename).touch()
    await asyncio.sleep(0.1)  # force release to stdout
    captured = capsys.readouterr()
    assert captured.out == 'File created!\n'
    # Move file
    new_filename = os.path.join(directory, 'new_filename.txt')
    os.rename(original_filename, new_filename)
    await asyncio.sleep(0.1)  # force release to stdout
    captured = capsys.readouterr()
    assert captured.out == 'File moved!\n'
    # Delete file
    os.remove(new_filename)
    await asyncio.sleep(0.1)  # force release to stdout
    captured = capsys.readouterr()
    assert captured.out == 'File deleted!\n' 

def test_should_automatically_send_a_typing_indicator(self):
        async def aux(context):
            await asyncio.sleep(0.600)
            await context.send_activity(f"echo:{context.activity.text}")

        def assert_is_typing(activity, description):  # pylint: disable=unused-argument
            assert activity.type == ActivityTypes.typing

        adapter = TestAdapter(aux)
        adapter.use(ShowTypingMiddleware())

        step1 = await adapter.send("foo")
        step2 = await step1.assert_reply(assert_is_typing)
        step3 = await step2.assert_reply("echo:foo")
        step4 = await step3.send("bar")
        step5 = await step4.assert_reply(assert_is_typing)
        await step5.assert_reply("echo:bar") 

def test_get_clan_battles(self):
        """This test will test out:
        - Normal clan battles fetching
        - All battles fetching
        - Clan war battles only fetching
        """

        tag = '29UQQ282'
        battles = await self.cr.get_clan_battles(tag)
        self.assertTrue(isinstance(battles, list))
        await asyncio.sleep(2)
        battles = await self.cr.get_clan_battles(tag, type='all')
        self.assertTrue(isinstance(battles, list))
        await asyncio.sleep(2)
        battles = await self.cr.get_clan_battles(tag, type='war')
        self.assertTrue(isinstance(battles, list)) 

def test_websocket_sending_invalid_data(event_loop, client_and_server, query_str):

    session, server = client_and_server

    invalid_data = "QSDF"
    print(f">>> {invalid_data}")
    await session.transport.websocket.send(invalid_data)

    await asyncio.sleep(2 * MS) 

def test_websocket_server_sending_invalid_query_errors(event_loop, server):
    url = f"ws://{server.hostname}:{server.port}/graphql"
    print(f"url = {url}")

    sample_transport = WebsocketsTransport(url=url)

    # Invalid server message is ignored
    async with Client(transport=sample_transport):
        await asyncio.sleep(2 * MS) 

def test_websocket_sending_invalid_data():

    # Get Websockets transport
    sample_transport = WebsocketsTransport(
        url="wss://countries.trevorblades.com/graphql", ssl=True
    )

    # Instanciate client
    async with Client(transport=sample_transport) as session:

        query = gql(
            """
            query getContinents {
              continents {
                code
              }
            }
        """
        )

        # Execute query
        result = await session.execute(query)

        print(f"result = {result!r}")

        invalid_data = "QSDF"
        print(f">>> {invalid_data}")
        await sample_transport.websocket.send(invalid_data)

        await asyncio.sleep(2) 

def test_websocket_subscription_task_cancel(
    event_loop, client_and_server, subscription_str
):

    session, server = client_and_server

    count = 10
    subscription = gql(subscription_str.format(count=count))

    async def task_coro():
        nonlocal count
        async for result in session.subscribe(subscription):

            number = result["number"]
            print(f"Number received: {number}")

            assert number == count

            count -= 1

    task = asyncio.ensure_future(task_coro())

    async def cancel_task_coro():
        nonlocal task

        await asyncio.sleep(11 * MS)

        task.cancel()

    cancel_task = asyncio.ensure_future(cancel_task_coro())

    await asyncio.gather(task, cancel_task)

    assert count > 0 

def test_websocket_subscription_close_transport(
    event_loop, client_and_server, subscription_str
):

    session, server = client_and_server

    count = 10
    subscription = gql(subscription_str.format(count=count))

    async def task_coro():
        nonlocal count
        async for result in session.subscribe(subscription):

            number = result["number"]
            print(f"Number received: {number}")

            assert number == count

            count -= 1

    task = asyncio.ensure_future(task_coro())

    async def close_transport_task_coro():
        nonlocal task

        await asyncio.sleep(11 * MS)

        await session.transport.close()

    close_transport_task = asyncio.ensure_future(close_transport_task_coro())

    await asyncio.gather(task, close_transport_task)

    assert count > 0 

def server_closing_while_we_are_doing_something_else(ws, path):
    await WebSocketServer.send_connection_ack(ws)
    result = await ws.recv()
    print(f"Server received: {result}")
    await ws.send(query1_server_answer.format(query_id=1))
    await WebSocketServer.send_complete(ws, 1)
    await asyncio.sleep(1 * MS)

    # Closing server after first query
    await ws.close() 

def getHeroAsync(episode):
    await asyncio.sleep(0.001)
    return getHero(episode) 

def subscribe_reviews(_root, _info, episode):
    for review in reviews[episode]:
        yield review
        await asyncio.sleep(0.1) 

def myFuture(future):
  await asyncio.sleep(1)
  future.set_result("My Future Has Completed") 

def slow_operation():
    yield from asyncio.sleep(1)
    return 'Future is done!' 

def main():
  await asyncio.sleep(1) 

def hello_world():
    yield from asyncio.sleep(1)
    print('Hello World')
    asyncio.async(hello_world()) 

def newsProducer(myQueue):
  while True:
    yield from myQueue.put(random.randint(1,5))
    yield from asyncio.sleep(1) 

def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y 

def myTask(n):
   time.sleep(1)
   print("Processing {}".format(n)) 

def myGenerator():
   for i in range(5):
       asyncio.ensure_future(myTask(i))
   print("Completed Tasks")
   yield from asyncio.sleep(2) 

def test_run_timed_queries_invalid_result_count_stop_task(
        self, query_tracker, config_data, make_query_loop
    ):
        """Timed queries returning invalid result counts are stopped."""
        config_data["queries"]["q"]["sql"] = "SELECT 100.0 AS a, 200.0 AS b"
        config_data["queries"]["q"]["interval"] = 1.0
        query_loop = make_query_loop()
        await query_loop.start()
        timed_call = query_loop._timed_calls["q"]
        await asyncio.sleep(1.1)
        await query_tracker.wait_failures()
        # the query has been stopped and removed
        assert not timed_call.running
        assert query_loop._timed_calls == {} 

def test_run_timed_queries_not_removed_if_not_failing_on_all_dbs(
        self, tmpdir, query_tracker, config_data, make_query_loop
    ):
        """Timed queries are removed when they fail on all databases."""
        db1 = tmpdir / "db1.sqlite"
        db2 = tmpdir / "db2.sqlite"
        config_data["databases"] = {
            "db1": {"dsn": f"sqlite:///{db1}"},
            "db2": {"dsn": f"sqlite:///{db2}"},
        }
        config_data["queries"]["q"].update(
            {"databases": ["db1", "db2"], "sql": "SELECT * FROM test", "interval": 1.0}
        )
        async with DataBase("db", f"sqlite:///{db1}") as db:
            await db.execute_sql("CREATE TABLE test (m INTEGER)")
            await db.execute_sql("INSERT INTO test VALUES (10)")
        # the query on the second database returns more columns
        async with DataBase("db", f"sqlite:///{db2}") as db:
            await db.execute_sql("CREATE TABLE test (m INTEGER, other INTERGER)")
            await db.execute_sql("INSERT INTO test VALUES (10, 20)")
        query_loop = make_query_loop()
        await query_loop.start()
        await asyncio.sleep(0.1)
        await query_tracker.wait_failures()
        assert len(query_tracker.queries) == 2
        assert len(query_tracker.results) == 1
        assert len(query_tracker.failures) == 1
        await asyncio.sleep(1.1)
        # succeeding query is run again, failing one is not
        assert len(query_tracker.results) == 2
        assert len(query_tracker.failures) == 1 

def _clear_code(self, uid):
        await asyncio.sleep(120)  # Codes last 2 minutes, or whenever they are used
        try:
            del self._uid_to_code[uid]
        except KeyError:
            pass  # Maybe the code has already been used