Skip to content

packages.valory.connections.abci.tests.test_abci

Tests for valory/abci connection.

hypothesis_cleanup

@pytest.fixture(scope="session", autouse=True)
def hypothesis_cleanup() -> Generator

Fixture to remove hypothesis directory after tests.

AsyncBytesIO Objects

class AsyncBytesIO()

Utility class to emulate asyncio.StreamReader.

__init__

def __init__(data: bytes) -> None

Initialize the buffer.

read

async def read(n: int) -> bytes

Read n bytes.

ABCIAppTest Objects

class ABCIAppTest()

A dummy ABCI application for testing purposes.

AbciDialogues Objects

class AbciDialogues(BaseAbciDialogues)

The dialogues class keeps track of all ABCI dialogues.

__init__

def __init__(address: str) -> None

Initialize dialogues.

__init__

def __init__(address: str)

Initialize.

handle

def handle(request: AbciMessage) -> AbciMessage

Process a request.

echo

def echo(request: AbciMessage) -> AbciMessage

Process echo request

info

def info(request: AbciMessage) -> AbciMessage

Process an info request.

flush

def flush(request: AbciMessage) -> AbciMessage

Process a flush request.

init_chain

def init_chain(request: AbciMessage) -> AbciMessage

Process an init_chain request.

query

def query(request: AbciMessage) -> AbciMessage

Process a query request.

check_tx

def check_tx(request: AbciMessage) -> AbciMessage

Process a check_tx request.

deliver_tx

def deliver_tx(request: AbciMessage) -> AbciMessage

Process a deliver_tx request.

begin_block

def begin_block(request: AbciMessage) -> AbciMessage

Process a begin_block request.

end_block

def end_block(request: AbciMessage) -> AbciMessage

Process an end_block request.

commit

def commit(request: AbciMessage) -> AbciMessage

Process a commit request.

set_option

def set_option(request: AbciMessage) -> AbciMessage

Process a commit request.

no_match

def no_match(request: AbciMessage) -> NoReturn

No match.

BaseABCITest Objects

class BaseABCITest()

Base class for ABCI test (mixin).

make_app

def make_app() -> ABCIAppTest

Make an ABCI app.

start_tendermint_docker_image

def start_tendermint_docker_image(use_grpc: bool) -> Container

Start TendermintDockerImage

BaseTestABCITendermintIntegration Objects

@pytest.mark.integration
class BaseTestABCITendermintIntegration(BaseThreadedAsyncLoop, ABC)

Integration test between ABCI connection and Tendermint node.

To use this class:

  • inherit from this class, and write test methods;
  • overwrite the method "make_app". The app must implement a 'handle(message)' method.

tendermint_port

noqa: F811

setup_and_teardown

@pytest.fixture(autouse=True, params=[False, True])
def setup_and_teardown(request: SubRequest) -> Generator

Set up the test.

make_app

@abstractmethod
def make_app() -> Any

Make an ABCI app.

health_check

def health_check() -> bool

Do a health-check.

tendermint_url

def tendermint_url() -> str

Get the current Tendermint URL.

process_incoming_messages

async def process_incoming_messages() -> None

Receive requests and send responses from/to the Tendermint node

TestNoop Objects

@skip_docker_tests
class TestNoop(BaseABCITest, BaseTestABCITendermintIntegration)

Test integration between ABCI connection and Tendermint, without txs.

test_run

def test_run() -> None

Assert the node produces at least one new block within the deadline.

Polling for monotonic latest_block_height growth distinguishes "Tendermint is slow to start" from "Tendermint has stalled."

TestQuery Objects

@skip_docker_tests
class TestQuery(BaseABCITest, BaseTestABCITendermintIntegration)

Test integration ABCI-Tendermint with a query.

test_run

def test_run() -> None

Run the test.

Send a query request to the Tendermint node, which will trigger a query request to the ABCI connection.

TestTransaction Objects

@skip_docker_tests
class TestTransaction(BaseABCITest, BaseTestABCITendermintIntegration)

Test integration ABCI-Tendermint by sending a transaction.

test_run

def test_run() -> None

Run the test.

Send a query request to the Tendermint node, which will trigger a query request to the ABCI connection.

test_connection_standalone_tendermint_setup

@pytest.mark.asyncio
async def test_connection_standalone_tendermint_setup() -> None

Test the setup of the connection configured with Tendermint.

test_ensure_connected_raises_connection_error

def test_ensure_connected_raises_connection_error() -> None

Test '_ensure_connected' raises ConnectionError if the channel is not connected.

test_encode_varint_method

def test_encode_varint_method() -> None

Test encode_varint (uint64 Protobuf) method

test_encode_decode_varint

@settings(database=database.InMemoryExampleDatabase())
@given(integers(min_value=0, max_value=(1 << 64) - 1))
@pytest.mark.asyncio
async def test_encode_decode_varint(value: int) -> None

Test that encoding and decoding works.

test_encoding_raises

@pytest.mark.parametrize("value", [-1, 1 << 64])
@pytest.mark.asyncio
async def test_encoding_raises(value: int) -> None

Test encoding raises

test_decode_varint_raises_exception_when_failing

@pytest.mark.asyncio
async def test_decode_varint_raises_exception_when_failing() -> None

Test that decode_varint raises exception when the decoding fails.

test_decode_varint_raises_eof_error

@pytest.mark.asyncio
async def test_decode_varint_raises_eof_error() -> None

Test that decode_varint raises exception when the EOF of the stream is reached.

test_dep_util

def test_dep_util() -> None

Test dependency utils.

test_varint_message_reader

@pytest.mark.asyncio
async def test_varint_message_reader() -> None

Test VarintMessageReader

TestTendermintNodeStop Objects

class TestTendermintNodeStop()

Tests for TendermintNode stop behavior.

test_stop_tm_process_closes_stdout

def test_stop_tm_process_closes_stdout() -> None

Test that _stop_tm_process closes stdout before nulling _process.

test_stop_monitoring_thread_uses_timeout

def test_stop_monitoring_thread_uses_timeout() -> None

Test that _stop_monitoring_thread calls join with a timeout.

test_stop_completes_with_blocking_readline

def test_stop_completes_with_blocking_readline() -> None

Test that stop() completes even when readline() would block.

Simulates a monitoring thread blocked on readline() and verifies that closing stdout unblocks it, allowing stop() to complete within a reasonable time.