packages.valory.connections.abci.tests.test_mock_server_channel

Unit tests for MockServerChannel.

MOCK_RPC_PORT

avoid conflicts with real TM

_SkillSideDialogues Objects

class _SkillSideDialogues(BaseAbciDialogues)

Simulates the skill-side ABCI dialogues (SERVER role).

__init__

def __init__(address: str) -> None

Initialise.

free_port

@pytest.fixture()
def free_port() -> int

Get a free port to avoid conflicts between parallel tests.

channel

@pytest.fixture()
def channel(free_port: int) -> MockServerChannel

Create a MockServerChannel instance.

connected_channel

@pytest_asyncio.fixture()
async def connected_channel(
        channel: MockServerChannel) -> AsyncGenerator[MockServerChannel, None]

Create and connect a MockServerChannel.

skill_dialogues

@pytest.fixture()
def skill_dialogues(channel: MockServerChannel) -> _SkillSideDialogues

Create a skill-side dialogues instance for proper response creation.

TestABCIMessageSequencing Objects

class TestABCIMessageSequencing()

Test the ABCI block lifecycle message sequence.

test_first_message_is_request_info

@pytest.mark.asyncio
async def test_first_message_is_request_info(
        connected_channel: MockServerChannel) -> None

The first message produced should be REQUEST_INFO.

test_init_chain_after_info_response

@pytest.mark.asyncio
async def test_init_chain_after_info_response(
        connected_channel: MockServerChannel,
        skill_dialogues: _SkillSideDialogues) -> None

After responding to INFO, the next message should be REQUEST_INIT_CHAIN.

test_full_block_lifecycle

@pytest.mark.asyncio
async def test_full_block_lifecycle(
        connected_channel: MockServerChannel,
        skill_dialogues: _SkillSideDialogues) -> None

Test the full sequence: info -> init_chain -> begin_block -> end_block -> commit.

test_deliver_tx_in_block

@pytest.mark.asyncio
async def test_deliver_tx_in_block(
        connected_channel: MockServerChannel,
        skill_dialogues: _SkillSideDialogues) -> None

Test that a submitted tx appears as REQUEST_DELIVER_TX in the next block.

TestRPCHandlers Objects

class TestRPCHandlers()

Test the mock Tendermint RPC HTTP endpoints.

test_broadcast_tx_sync

@pytest.mark.asyncio
async def test_broadcast_tx_sync(connected_channel: MockServerChannel) -> None

Test /broadcast_tx_sync returns hash and code 0.

test_tx_query_not_found

@pytest.mark.asyncio
async def test_tx_query_not_found(
        connected_channel: MockServerChannel) -> None

Test /tx?hash=... returns 500 when tx not yet delivered.

test_tx_query_found

@pytest.mark.asyncio
async def test_tx_query_found(connected_channel: MockServerChannel) -> None

Test /tx?hash=... returns 200 with tx_result after delivery.

test_status_returns_height

@pytest.mark.asyncio
async def test_status_returns_height(
        connected_channel: MockServerChannel) -> None

Test /status returns current block height.

test_net_info

@pytest.mark.asyncio
async def test_net_info(connected_channel: MockServerChannel) -> None

Test /net_info returns valid response.

test_hard_reset

@pytest.mark.asyncio
async def test_hard_reset(connected_channel: MockServerChannel) -> None

Test /hard_reset returns 200 (no-op).

test_gentle_reset

@pytest.mark.asyncio
async def test_gentle_reset(connected_channel: MockServerChannel) -> None

Test /gentle_reset returns 200 (no-op).

TestLifecycle Objects

class TestLifecycle()

Test connect and disconnect behaviour.

test_connect_sets_state

@pytest.mark.asyncio
async def test_connect_sets_state(channel: MockServerChannel) -> None

Test that connect initialises internal state.

test_disconnect_cleans_up

@pytest.mark.asyncio
async def test_disconnect_cleans_up(channel: MockServerChannel) -> None

Test that disconnect tears down all resources.

test_double_connect_is_noop

@pytest.mark.asyncio
async def test_double_connect_is_noop(channel: MockServerChannel) -> None

Test that connecting twice doesn't raise.

test_double_disconnect_is_noop

@pytest.mark.asyncio
async def test_double_disconnect_is_noop(channel: MockServerChannel) -> None

Test that disconnecting twice doesn't raise.

test_height_increments

@pytest.mark.asyncio
async def test_height_increments(connected_channel: MockServerChannel,
                                 skill_dialogues: _SkillSideDialogues) -> None

Test that block height increments after a full block cycle.

test_connect_rollback_on_port_conflict

@pytest.mark.asyncio
async def test_connect_rollback_on_port_conflict(free_port: int) -> None

Test that connect rolls back state if the port is already bound.

test_get_message_raises_on_producer_error

@pytest.mark.asyncio
async def test_get_message_raises_on_producer_error(
        channel: MockServerChannel) -> None

Test that get_message raises ConnectionError after producer dies.

test_broadcast_tx_invalid_hex

@pytest.mark.asyncio
async def test_broadcast_tx_invalid_hex(
        connected_channel: MockServerChannel) -> None

Test /broadcast_tx_sync with invalid hex returns 400.

test_unknown_route_returns_404

@pytest.mark.asyncio
async def test_unknown_route_returns_404(
        connected_channel: MockServerChannel) -> None

Test that an unknown path returns 404.