import time
from http import HTTPStatus
from itertools import count
from typing import Sequence

import gevent
import grequests
import pytest
import structlog
from eth_utils import to_canonical_address
from flask import url_for

from raiden.api.python import RaidenAPI
from raiden.api.rest import APIServer, RestAPI
from raiden.constants import RoutingMode
from raiden.message_handler import MessageHandler
from raiden.network.transport import MatrixTransport
from raiden.raiden_event_handler import RaidenEventHandler
from raiden.raiden_service import RaidenService
from raiden.settings import RestApiConfig
from raiden.tests.integration.api.utils import wait_for_listening_port
from raiden.tests.integration.fixtures.raiden_network import RestartNode
from raiden.tests.utils.detect_failure import raise_on_failure
from raiden.tests.utils.protocol import HoldRaidenEventHandler
from raiden.tests.utils.transfer import (
    assert_synced_channel_state,
    wait_assert,
    watch_for_unlock_failures,
)
from raiden.transfer import views
from raiden.ui.startup import RaidenBundle
from raiden.utils.formatting import to_checksum_address
from raiden.utils.typing import (
    Address,
    BlockNumber,
    Host,
    Iterator,
    List,
    Port,
    TokenAddress,
    TokenAmount,
    TokenNetworkAddress,
    Tuple,
)

log = structlog.get_logger(__name__)


def iwait_and_get(items: Sequence[gevent.Greenlet]) -> None:
    """Iteratively wait and get on passed greenlets.

    This ensures exceptions in the greenlets are re-raised as soon as possible.
    """
    for item in gevent.iwait(items):
        item.get()


def _url_for(apiserver: APIServer, endpoint: str, **kwargs) -> str:
    # url_for() expects binary address so we have to convert here
    for key, val in kwargs.items():
        if isinstance(val, str) and val.startswith("0x"):
            kwargs[key] = to_canonical_address(val)

    with apiserver.flask_app.app_context():
        return url_for(f"v1_resources.{endpoint}", **kwargs)


def start_apiserver(raiden_app: RaidenService, rest_api_port_number: Port) -> APIServer:
    raiden_api = RaidenAPI(raiden_app)
    rest_api = RestAPI(raiden_api)
    api_server = APIServer(
        rest_api, config=RestApiConfig(host=Host("localhost"), port=rest_api_port_number)
    )

    # required for url_for
    api_server.flask_app.config["SERVER_NAME"] = f"localhost:{rest_api_port_number}"

    api_server.start()

    wait_for_listening_port(rest_api_port_number)

    return api_server


def start_apiserver_for_network(
    raiden_network: List[RaidenService], port_generator: Iterator[Port]
) -> List[APIServer]:
    return [start_apiserver(app, next(port_generator)) for app in raiden_network]


def restart_app(app: RaidenService, restart_node: RestartNode) -> RaidenService:
    new_transport = MatrixTransport(
        config=app.config.transport, environment=app.config.environment_type
    )
    raiden_event_handler = RaidenEventHandler()
    hold_handler = HoldRaidenEventHandler(raiden_event_handler)

    app = RaidenService(
        config=app.config,
        rpc_client=app.rpc_client,
        proxy_manager=app.proxy_manager,
        query_start_block=BlockNumber(0),
        raiden_bundle=RaidenBundle(
            app.default_registry,
            app.default_secret_registry,
        ),
        services_bundle=app.default_services_bundle,
        transport=new_transport,
        raiden_event_handler=hold_handler,
        message_handler=MessageHandler(),
        routing_mode=RoutingMode.PRIVATE,
    )

    restart_node(app)

    return app


def restart_network(
    raiden_network: List[RaidenService], restart_node: RestartNode
) -> List[RaidenService]:
    for app in raiden_network:
        app.stop()

    wait_network = (gevent.spawn(restart_app, app, restart_node) for app in raiden_network)

    gevent.joinall(set(wait_network), raise_error=True)

    new_network = [greenlet.get() for greenlet in wait_network]

    return new_network


def restart_network_and_apiservers(
    raiden_network: List[RaidenService],
    restart_node: RestartNode,
    api_servers: List[APIServer],
    port_generator: Iterator[Port],
) -> Tuple[List[RaidenService], List[APIServer]]:
    """Stop an app and start it back"""
    for rest_api in api_servers:
        rest_api.stop()

    new_network = restart_network(raiden_network, restart_node)
    new_servers = start_apiserver_for_network(new_network, port_generator)

    return (new_network, new_servers)


def address_from_apiserver(apiserver: APIServer) -> Address:
    return apiserver.rest_api.raiden_api.address


def transfer_and_assert(
    server_from: APIServer,
    server_to: APIServer,
    token_address: TokenAddress,
    identifier: int,
    amount: TokenAmount,
) -> None:
    url = _url_for(
        server_from,
        "token_target_paymentresource",
        token_address=to_checksum_address(token_address),
        target_address=to_checksum_address(address_from_apiserver(server_to)),
    )
    json = {"amount": amount, "identifier": identifier}

    log.debug("PAYMENT REQUEST", url=url, json=json)

    request = grequests.post(url, json=json)

    start = time.monotonic()
    response = request.send().response
    duration = time.monotonic() - start

    log.debug("PAYMENT RESPONSE", url=url, json=json, response=response, duration=duration)

    assert getattr(request, "exception", None) is None
    assert response is not None
    assert response.status_code == HTTPStatus.OK, f"Payment failed, reason: {response.content}"
    assert response.headers["Content-Type"] == "application/json"


def sequential_transfers(
    server_from: APIServer,
    server_to: APIServer,
    number_of_transfers: int,
    token_address: TokenAddress,
    identifier_generator: Iterator[int],
) -> None:
    for _ in range(number_of_transfers):
        transfer_and_assert(
            server_from=server_from,
            server_to=server_to,
            token_address=token_address,
            identifier=next(identifier_generator),
            amount=TokenAmount(1),
        )


def stress_send_serial_transfers(
    rest_apis: List[APIServer],
    token_address: TokenAddress,
    identifier_generator: Iterator[int],
    deposit: TokenAmount,
) -> None:
    """Send `deposit` transfers of value `1` one at a time, without changing
    the initial capacity.
    """
    pairs = list(zip(rest_apis, rest_apis[1:] + [rest_apis[0]]))

    # deplete the channels in one direction
    for server_from, server_to in pairs:
        sequential_transfers(
            server_from=server_from,
            server_to=server_to,
            number_of_transfers=deposit,
            token_address=token_address,
            identifier_generator=identifier_generator,
        )

    # deplete the channels in the backwards direction
    for server_to, server_from in pairs:
        sequential_transfers(
            server_from=server_from,
            server_to=server_to,
            number_of_transfers=deposit * 2,
            token_address=token_address,
            identifier_generator=identifier_generator,
        )

    # reset the balances balances by sending the "extra" deposit forward
    for server_from, server_to in pairs:
        sequential_transfers(
            server_from=server_from,
            server_to=server_to,
            number_of_transfers=deposit,
            token_address=token_address,
            identifier_generator=identifier_generator,
        )


def stress_send_parallel_transfers(
    rest_apis: List[APIServer],
    token_address: TokenAddress,
    identifier_generator: Iterator[int],
    deposit: TokenAmount,
) -> None:
    """Send `deposit` transfers in parallel, without changing the initial capacity."""
    pairs = list(zip(rest_apis, rest_apis[1:] + [rest_apis[0]]))

    # deplete the channels in one direction
    iwait_and_get(
        [
            gevent.spawn(
                sequential_transfers,
                server_from=server_from,
                server_to=server_to,
                number_of_transfers=deposit,
                token_address=token_address,
                identifier_generator=identifier_generator,
            )
            for server_from, server_to in pairs
        ]
    )

    # deplete the channels in the backwards direction
    iwait_and_get(
        [
            gevent.spawn(
                sequential_transfers,
                server_from=server_from,
                server_to=server_to,
                number_of_transfers=deposit * 2,
                token_address=token_address,
                identifier_generator=identifier_generator,
            )
            for server_to, server_from in pairs
        ]
    )

    # reset the balances balances by sending the "extra" deposit forward
    iwait_and_get(
        [
            gevent.spawn(
                sequential_transfers,
                server_from=server_from,
                server_to=server_to,
                number_of_transfers=deposit,
                token_address=token_address,
                identifier_generator=identifier_generator,
            )
            for server_from, server_to in pairs
        ]
    )


def stress_send_and_receive_parallel_transfers(
    rest_apis: List[APIServer],
    token_address: TokenAddress,
    identifier_generator: Iterator[int],
    deposit: TokenAmount,
) -> None:
    """Send transfers of value one in parallel"""
    pairs = list(zip(rest_apis, rest_apis[1:] + [rest_apis[0]]))

    forward_transfers = [
        gevent.spawn(
            sequential_transfers,
            server_from=server_from,
            server_to=server_to,
            number_of_transfers=deposit,
            token_address=token_address,
            identifier_generator=identifier_generator,
        )
        for server_from, server_to in pairs
    ]

    backwards_transfers = [
        gevent.spawn(
            sequential_transfers,
            server_from=server_from,
            server_to=server_to,
            number_of_transfers=deposit,
            token_address=token_address,
            identifier_generator=identifier_generator,
        )
        for server_to, server_from in pairs
    ]

    iwait_and_get(forward_transfers + backwards_transfers)


def assert_channels(
    raiden_network: List[RaidenService],
    token_network_address: TokenNetworkAddress,
    deposit: TokenAmount,
) -> None:
    pairs = list(zip(raiden_network, raiden_network[1:] + [raiden_network[0]]))

    for first, second in pairs:
        wait_assert(
            assert_synced_channel_state,
            token_network_address,
            first,
            deposit,
            [],
            second,
            deposit,
            [],
        )


@pytest.mark.skip(reason="flaky, see https://github.com/raiden-network/raiden/issues/4803")
@raise_on_failure
@pytest.mark.parametrize("number_of_nodes", [3])
@pytest.mark.parametrize("number_of_tokens", [1])
@pytest.mark.parametrize("channels_per_node", [2])
@pytest.mark.parametrize("deposit", [2])
@pytest.mark.parametrize("reveal_timeout", [15])
@pytest.mark.parametrize("settle_timeout", [120])
def test_stress(
    raiden_network: List[RaidenService],
    restart_node: RestartNode,
    deposit: TokenAmount,
    token_addresses: List[TokenAddress],
    port_generator: Iterator[Port],
) -> None:
    token_address = token_addresses[0]
    rest_apis = start_apiserver_for_network(raiden_network, port_generator)
    identifier_generator = count(start=1)

    token_network_address = views.get_token_network_address_by_token_address(
        views.state_from_raiden(raiden_network[0]),
        raiden_network[0].default_registry.address,
        token_address,
    )
    assert token_network_address

    for _ in range(2):
        assert_channels(raiden_network, token_network_address, deposit)

        with watch_for_unlock_failures(*raiden_network):
            stress_send_serial_transfers(rest_apis, token_address, identifier_generator, deposit)

        raiden_network, rest_apis = restart_network_and_apiservers(
            raiden_network, restart_node, rest_apis, port_generator
        )

        assert_channels(raiden_network, token_network_address, deposit)

        with watch_for_unlock_failures(*raiden_network):
            stress_send_parallel_transfers(rest_apis, token_address, identifier_generator, deposit)

        raiden_network, rest_apis = restart_network_and_apiservers(
            raiden_network, restart_node, rest_apis, port_generator
        )

        assert_channels(raiden_network, token_network_address, deposit)

        with watch_for_unlock_failures(*raiden_network):
            stress_send_and_receive_parallel_transfers(
                rest_apis, token_address, identifier_generator, deposit
            )

        raiden_network, rest_apis = restart_network_and_apiservers(
            raiden_network, restart_node, rest_apis, port_generator
        )

    restart_network(raiden_network, restart_node)

[((46, 6, 46, 36), 'structlog.get_logger', 'structlog.get_logger', ({(46, 27, 46, 35): '__name__'}, {}), '(__name__)', False, 'import structlog\n'), ((355, 1, 355, 91), 'pytest.mark.skip', 'pytest.mark.skip', (), '', False, 'import pytest\n'), ((357, 1, 357, 48), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(357, 25, 357, 42): '"""number_of_nodes"""', (357, 44, 357, 47): '[3]'}, {}), "('number_of_nodes', [3])", False, 'import pytest\n'), ((358, 1, 358, 49), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(358, 25, 358, 43): '"""number_of_tokens"""', (358, 45, 358, 48): '[1]'}, {}), "('number_of_tokens', [1])", False, 'import pytest\n'), ((359, 1, 359, 50), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(359, 25, 359, 44): '"""channels_per_node"""', (359, 46, 359, 49): '[2]'}, {}), "('channels_per_node', [2])", False, 'import pytest\n'), ((360, 1, 360, 40), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(360, 25, 360, 34): '"""deposit"""', (360, 36, 360, 39): '[2]'}, {}), "('deposit', [2])", False, 'import pytest\n'), ((361, 1, 361, 48), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(361, 25, 361, 41): '"""reveal_timeout"""', (361, 43, 361, 47): '[15]'}, {}), "('reveal_timeout', [15])", False, 'import pytest\n'), ((362, 1, 362, 49), 'pytest.mark.parametrize', 'pytest.mark.parametrize', ({(362, 25, 362, 41): '"""settle_timeout"""', (362, 43, 362, 48): '[120]'}, {}), "('settle_timeout', [120])", False, 'import pytest\n'), ((54, 16, 54, 35), 'gevent.iwait', 'gevent.iwait', ({(54, 29, 54, 34): 'items'}, {}), '(items)', False, 'import gevent\n'), ((69, 17, 69, 38), 'raiden.api.python.RaidenAPI', 'RaidenAPI', ({(69, 27, 69, 37): 'raiden_app'}, {}), '(raiden_app)', False, 'from raiden.api.python import RaidenAPI\n'), ((70, 15, 70, 34), 'raiden.api.rest.RestAPI', 'RestAPI', ({(70, 23, 70, 33): 'raiden_api'}, {}), '(raiden_api)', False, 'from raiden.api.rest import APIServer, RestAPI\n'), ((80, 4, 80, 49), 'raiden.tests.integration.api.utils.wait_for_listening_port', 'wait_for_listening_port', ({(80, 28, 80, 48): 'rest_api_port_number'}, {}), '(rest_api_port_number)', False, 'from raiden.tests.integration.api.utils import wait_for_listening_port\n'), ((92, 20, 94, 5), 'raiden.network.transport.MatrixTransport', 'MatrixTransport', (), '', False, 'from raiden.network.transport import MatrixTransport\n'), ((95, 27, 95, 47), 'raiden.raiden_event_handler.RaidenEventHandler', 'RaidenEventHandler', ({}, {}), '()', False, 'from raiden.raiden_event_handler import RaidenEventHandler\n'), ((96, 19, 96, 63), 'raiden.tests.utils.protocol.HoldRaidenEventHandler', 'HoldRaidenEventHandler', ({(96, 42, 96, 62): 'raiden_event_handler'}, {}), '(raiden_event_handler)', False, 'from raiden.tests.utils.protocol import HoldRaidenEventHandler\n'), ((171, 14, 171, 44), 'grequests.post', 'grequests.post', (), '', False, 'import grequests\n'), ((173, 12, 173, 28), 'time.monotonic', 'time.monotonic', ({}, {}), '()', False, 'import time\n'), ((372, 27, 372, 41), 'itertools.count', 'count', (), '', False, 'from itertools import count\n'), ((65, 15, 65, 60), 'flask.url_for', 'url_for', ({(65, 23, 65, 49): 'f"""v1_resources.{endpoint}"""'}, {}), "(f'v1_resources.{endpoint}', **kwargs)", False, 'from flask import url_for\n'), ((125, 20, 125, 64), 'gevent.spawn', 'gevent.spawn', ({(125, 33, 125, 44): 'restart_app', (125, 46, 125, 49): 'app', (125, 51, 125, 63): 'restart_node'}, {}), '(restart_app, app, restart_node)', False, 'import gevent\n'), ((175, 15, 175, 31), 'time.monotonic', 'time.monotonic', ({}, {}), '()', False, 'import time\n'), ((309, 8, 316, 9), 'gevent.spawn', 'gevent.spawn', (), '', False, 'import gevent\n'), ((321, 8, 328, 9), 'gevent.spawn', 'gevent.spawn', (), '', False, 'import gevent\n'), ((343, 8, 352, 9), 'raiden.tests.utils.transfer.wait_assert', 'wait_assert', ({(344, 12, 344, 39): 'assert_synced_channel_state', (345, 12, 345, 33): 'token_network_address', (346, 12, 346, 17): 'first', (347, 12, 347, 19): 'deposit', (348, 12, 348, 14): '[]', (349, 12, 349, 18): 'second', (350, 12, 350, 19): 'deposit', (351, 12, 351, 14): '[]'}, {}), '(assert_synced_channel_state, token_network_address, first,\n    deposit, [], second, deposit, [])', False, 'from raiden.tests.utils.transfer import assert_synced_channel_state, wait_assert, watch_for_unlock_failures\n'), ((375, 8, 375, 50), 'raiden.transfer.views.state_from_raiden', 'views.state_from_raiden', ({(375, 32, 375, 49): 'raiden_network[0]'}, {}), '(raiden_network[0])', False, 'from raiden.transfer import views\n'), ((62, 26, 62, 51), 'eth_utils.to_canonical_address', 'to_canonical_address', ({(62, 47, 62, 50): 'val'}, {}), '(val)', False, 'from eth_utils import to_canonical_address\n'), ((102, 26, 102, 40), 'raiden.utils.typing.BlockNumber', 'BlockNumber', ({(102, 38, 102, 39): '0'}, {}), '(0)', False, 'from raiden.utils.typing import Address, BlockNumber, Host, Iterator, List, Port, TokenAddress, TokenAmount, TokenNetworkAddress, Tuple\n'), ((103, 22, 106, 9), 'raiden.ui.startup.RaidenBundle', 'RaidenBundle', ({(104, 12, 104, 32): 'app.default_registry', (105, 12, 105, 39): 'app.default_secret_registry'}, {}), '(app.default_registry, app.default_secret_registry)', False, 'from raiden.ui.startup import RaidenBundle\n'), ((110, 24, 110, 40), 'raiden.message_handler.MessageHandler', 'MessageHandler', ({}, {}), '()', False, 'from raiden.message_handler import MessageHandler\n'), ((164, 22, 164, 56), 'raiden.utils.formatting.to_checksum_address', 'to_checksum_address', ({(164, 42, 164, 55): 'token_address'}, {}), '(token_address)', False, 'from raiden.utils.formatting import to_checksum_address\n'), ((256, 12, 263, 13), 'gevent.spawn', 'gevent.spawn', (), '', False, 'import gevent\n'), ((271, 12, 278, 13), 'gevent.spawn', 'gevent.spawn', (), '', False, 'import gevent\n'), ((286, 12, 293, 13), 'gevent.spawn', 'gevent.spawn', (), '', False, 'import gevent\n'), ((384, 13, 384, 55), 'raiden.tests.utils.transfer.watch_for_unlock_failures', 'watch_for_unlock_failures', ({(384, 39, 384, 54): '*raiden_network'}, {}), '(*raiden_network)', False, 'from raiden.tests.utils.transfer import assert_synced_channel_state, wait_assert, watch_for_unlock_failures\n'), ((393, 13, 393, 55), 'raiden.tests.utils.transfer.watch_for_unlock_failures', 'watch_for_unlock_failures', ({(393, 39, 393, 54): '*raiden_network'}, {}), '(*raiden_network)', False, 'from raiden.tests.utils.transfer import assert_synced_channel_state, wait_assert, watch_for_unlock_failures\n'), ((402, 13, 402, 55), 'raiden.tests.utils.transfer.watch_for_unlock_failures', 'watch_for_unlock_failures', ({(402, 39, 402, 54): '*raiden_network'}, {}), '(*raiden_network)', False, 'from raiden.tests.utils.transfer import assert_synced_channel_state, wait_assert, watch_for_unlock_failures\n'), ((198, 19, 198, 33), 'raiden.utils.typing.TokenAmount', 'TokenAmount', ({(198, 31, 198, 32): '(1)'}, {}), '(1)', False, 'from raiden.utils.typing import Address, BlockNumber, Host, Iterator, List, Port, TokenAddress, TokenAmount, TokenNetworkAddress, Tuple\n'), ((72, 44, 72, 61), 'raiden.utils.typing.Host', 'Host', ({(72, 49, 72, 60): '"""localhost"""'}, {}), "('localhost')", False, 'from raiden.utils.typing import Address, BlockNumber, Host, Iterator, List, Port, TokenAddress, TokenAmount, TokenNetworkAddress, Tuple\n')]