import asyncio
from typing import List

import pytest

from gql import Client, gql
from gql.transport.exceptions import (
    TransportConnectionFailed,
    TransportProtocolError,
    TransportQueryError,
)

from .conftest import WebSocketServerHelper

# Marking all tests in this file with the websockets marker
pytestmark = pytest.mark.websockets

invalid_query_str = """
    query getContinents {
      continents {
        code
        bloh
      }
    }
"""

invalid_query1_server_answer = (
    '{{"type":"next","id":"{query_id}",'
    '"payload":{{"errors":['
    '{{"message":"Cannot query field \\"bloh\\" on type \\"Continent\\".",'
    '"locations":[{{"line":4,"column":5}}],'
    '"extensions":{{"code":"INTERNAL_SERVER_ERROR"}}}}]}}}}'
)

invalid_query1_server = [invalid_query1_server_answer]


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [invalid_query1_server], indirect=True)
@pytest.mark.parametrize("query_str", [invalid_query_str])
async def test_graphqlws_invalid_query(client_and_graphqlws_server, query_str):

    session, server = client_and_graphqlws_server

    query = gql(query_str)

    with pytest.raises(TransportQueryError) as exc_info:
        await session.execute(query)

    exception = exc_info.value

    assert isinstance(exception.errors, List)

    error = exception.errors[0]

    assert error["extensions"]["code"] == "INTERNAL_SERVER_ERROR"


invalid_subscription_str = """
    subscription getContinents {
      continents {
        code
        bloh
      }
    }
"""


async def server_invalid_subscription(ws):
    await WebSocketServerHelper.send_connection_ack(ws)
    await ws.recv()
    await ws.send(invalid_query1_server_answer.format(query_id=1))
    await WebSocketServerHelper.send_complete(ws, 1)
    await ws.wait_closed()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "graphqlws_server", [server_invalid_subscription], indirect=True
)
@pytest.mark.parametrize("query_str", [invalid_subscription_str])
async def test_graphqlws_invalid_subscription(client_and_graphqlws_server, query_str):

    session, server = client_and_graphqlws_server

    query = gql(query_str)

    with pytest.raises(TransportQueryError) as exc_info:
        async for result in session.subscribe(query):
            pass

    exception = exc_info.value

    assert isinstance(exception.errors, List)

    error = exception.errors[0]

    assert error["extensions"]["code"] == "INTERNAL_SERVER_ERROR"


async def server_no_ack(ws):
    await ws.wait_closed()


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [server_no_ack], indirect=True)
@pytest.mark.parametrize("query_str", [invalid_query_str])
async def test_graphqlws_server_does_not_send_ack(graphqlws_server, query_str):
    from gql.transport.websockets import WebsocketsTransport

    url = f"ws://{graphqlws_server.hostname}:{graphqlws_server.port}/graphql"

    transport = WebsocketsTransport(url=url, ack_timeout=0.1)

    with pytest.raises(asyncio.TimeoutError):
        async with Client(transport=transport):
            pass


invalid_query_server_answer = (
    '{"id":"1","type":"error","payload":[{"message":"Cannot query field '
    '\\"helo\\" on type \\"Query\\". Did you mean \\"hello\\"?",'
    '"locations":[{"line":2,"column":3}]}]}'
)


async def server_invalid_query(ws):
    await WebSocketServerHelper.send_connection_ack(ws)
    result = await ws.recv()
    print(f"Server received: {result}")
    await ws.send(invalid_query_server_answer)
    await WebSocketServerHelper.wait_connection_terminate(ws)
    await ws.wait_closed()


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [server_invalid_query], indirect=True)
async def test_graphqlws_sending_invalid_query(client_and_graphqlws_server):

    session, server = client_and_graphqlws_server

    query = gql("{helo}")

    with pytest.raises(TransportQueryError) as exc_info:
        await session.execute(query)

    exception = exc_info.value

    assert isinstance(exception.errors, List)

    error = exception.errors[0]

    assert (
        error["message"]
        == 'Cannot query field "helo" on type "Query". Did you mean "hello"?'
    )


not_json_answer = ["BLAHBLAH"]
missing_type_answer = ["{}"]
missing_id_answer_1 = ['{"type": "next"}']
missing_id_answer_2 = ['{"type": "error"}']
missing_id_answer_3 = ['{"type": "complete"}']
data_without_payload = ['{"type": "next", "id":"1"}']
error_without_payload = ['{"type": "error", "id":"1"}']
error_with_payload_not_a_list = ['{"type": "error", "id":"1", "payload": "NOT A LIST"}']
payload_is_not_a_dict = ['{"type": "next", "id":"1", "payload": "BLAH"}']
empty_payload = ['{"type": "next", "id":"1", "payload": {}}']
sending_bytes = [b"\x01\x02\x03"]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "graphqlws_server",
    [
        not_json_answer,
        missing_type_answer,
        missing_id_answer_1,
        missing_id_answer_2,
        missing_id_answer_3,
        data_without_payload,
        error_without_payload,
        payload_is_not_a_dict,
        error_with_payload_not_a_list,
        empty_payload,
        sending_bytes,
    ],
    indirect=True,
)
async def test_graphqlws_transport_protocol_errors(client_and_graphqlws_server):

    session, server = client_and_graphqlws_server

    query = gql("query { hello }")

    with pytest.raises(TransportProtocolError):
        await session.execute(query)


async def server_without_ack(ws):
    # Sending something else than an ack
    await WebSocketServerHelper.send_complete(ws, 1)
    await ws.wait_closed()


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [server_without_ack], indirect=True)
async def test_graphqlws_server_does_not_ack(graphqlws_server):
    from gql.transport.websockets import WebsocketsTransport

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

    transport = WebsocketsTransport(url=url)

    with pytest.raises(TransportProtocolError):
        async with Client(transport=transport):
            pass


async def server_closing_directly(ws):
    await ws.close()


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [server_closing_directly], indirect=True)
async def test_graphqlws_server_closing_directly(graphqlws_server):
    from gql.transport.websockets import WebsocketsTransport

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

    transport = WebsocketsTransport(url=url)

    with pytest.raises(TransportConnectionFailed):
        async with Client(transport=transport):
            pass


async def server_closing_after_ack(ws):
    await WebSocketServerHelper.send_connection_ack(ws)
    await ws.close()


@pytest.mark.asyncio
@pytest.mark.parametrize("graphqlws_server", [server_closing_after_ack], indirect=True)
async def test_graphqlws_server_closing_after_ack(client_and_graphqlws_server):

    session, server = client_and_graphqlws_server

    query = gql("query { hello }")

    print("\n Trying to execute first query.\n")

    with pytest.raises(TransportConnectionFailed) as exc1:
        await session.execute(query)

    exc1_cause = exc1.value.__cause__
    exc1_cause_str = f"{type(exc1_cause).__name__}:{exc1_cause!s}"

    print(f"\n First query Exception cause: {exc1_cause_str}\n")

    assert (
        exc1_cause_str == "ConnectionClosedOK:received 1000 (OK); then sent 1000 (OK)"
    )

    await session.transport.wait_closed()

    print("\n Trying to execute second query.\n")

    with pytest.raises(TransportConnectionFailed) as exc2:
        await session.execute(query)

    exc2_cause = exc2.value.__cause__
    exc2_cause_str = f"{type(exc2_cause).__name__}:{exc2_cause!s}"

    print(f" Second query Exception cause: {exc2_cause_str}\n")

    assert (
        exc2_cause_str == "ConnectionClosedOK:received 1000 (OK); then sent 1000 (OK)"
    )