test_transfer.py 文件源码

def test_healthcheck_with_bad_peer(raiden_network, nat_keepalive_retries, nat_keepalive_timeout):
    """ If the Ping messages are not answered, the node must be set to
    unreachable.
    """
    app0, app1 = raiden_network  # pylint: disable=unbalanced-tuple-unpacking

    address0 = app0.raiden.address
    address1 = app1.raiden.address

    assert app0.raiden.protocol.nodeaddresses_networkstatuses[address1] == NODE_NETWORK_REACHABLE
    assert app1.raiden.protocol.nodeaddresses_networkstatuses[address0] == NODE_NETWORK_REACHABLE

    # Drop all Ping and Ack messages
    app0.raiden.protocol.transport.droprate = 1
    app1.raiden.protocol.transport.droprate = 1

    gevent.sleep(
        (nat_keepalive_retries + 2) * nat_keepalive_timeout
    )

    assert app0.raiden.protocol.nodeaddresses_networkstatuses[address1] == NODE_NETWORK_UNREACHABLE
    assert app1.raiden.protocol.nodeaddresses_networkstatuses[address0] == NODE_NETWORK_UNREACHABLE