packages.valory.skills.abstract_round_abci.tests.test_behaviours_utils

Test the behaviours_utils.py module of the skill.

mock_yield_and_return

def mock_yield_and_return(
        return_value: Any) -> Callable[[], Generator[None, None, Any]]

Wrapper for a Dummy generator that returns a bool.

yield_and_return_bool_wrapper

def yield_and_return_bool_wrapper(
        flag_value: bool
) -> Callable[[], Generator[None, None, Optional[bool]]]

Wrapper for a Dummy generator that returns a bool.

yield_and_return_int_wrapper

def yield_and_return_int_wrapper(
    value: Optional[int]
) -> Callable[[], Generator[None, None, Optional[int]]]

Wrapper for a Dummy generator that returns an int.

AsyncBehaviourTest Objects

class AsyncBehaviourTest(AsyncBehaviour, ABC)

Concrete AsyncBehaviour class for testing purposes.

async_act_wrapper

def async_act_wrapper() -> Generator

Do async act wrapper. Forwards to 'async_act'.

async_act

def async_act() -> Generator

Do 'async_act'.

test_async_behaviour_ticks

def test_async_behaviour_ticks() -> None

Test "AsyncBehaviour", only ticks.

test_async_behaviour_wait_for_message

def test_async_behaviour_wait_for_message() -> None

Test 'wait_for_message'.

test_async_behaviour_wait_for_message_raises_timeout_exception

def test_async_behaviour_wait_for_message_raises_timeout_exception() -> None

Test 'wait_for_message' when it raises TimeoutException.

test_async_behaviour_wait_for_condition

def test_async_behaviour_wait_for_condition() -> None

Test 'wait_for_condition' method.

test_async_behaviour_wait_for_condition_with_timeout

def test_async_behaviour_wait_for_condition_with_timeout() -> None

Test 'wait_for_condition' method with timeout expired.

test_async_behaviour_sleep

def test_async_behaviour_sleep() -> None

Test 'sleep' method.

test_async_behaviour_without_yield

def test_async_behaviour_without_yield() -> None

Test AsyncBehaviour, async_act without yield/yield from.

test_async_behaviour_raise_stopiteration

def test_async_behaviour_raise_stopiteration() -> None

Test AsyncBehaviour, async_act raising 'StopIteration'.

test_async_behaviour_stop

def test_async_behaviour_stop() -> None

Test AsyncBehaviour.stop method.

RoundA Objects

class RoundA(AbstractRound)

Concrete ABCI round.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Handle end block.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check payload.

process_payload

def process_payload(payload: BaseTxPayload) -> None

Process payload.

BehaviourATest Objects

class BehaviourATest(BaseBehaviour)

Concrete BaseBehaviour class.

async_act

def async_act() -> Generator

Do the 'async_act'.

dummy_generator_wrapper

def dummy_generator_wrapper(
        return_value: Any = None) -> Callable[[Any], Generator]

A wrapper around a dummy generator that yields nothing and returns the given return value.

TestBaseBehaviour Objects

class TestBaseBehaviour()

Tests for the 'BaseBehaviour' class.

setup_method

def setup_method() -> None

Set up the tests.

dummy_put_message

def dummy_put_message(*args: Any, **kwargs: Any) -> None

A dummy implementation of Outbox.put_message

test_behaviour_id

def test_behaviour_id() -> None

Test behaviour_id on instance.

test_send_to_ipfs

@pytest.mark.parametrize(
    "ipfs_response, expected_log",
    [
        (
            MagicMock(ipfs_hash="test",
                      performative=IpfsMessage.Performative.IPFS_HASH),
            "Successfully stored dummy_filename to IPFS with hash: test",
        ),
        (
            MagicMock(ipfs_hash="test",
                      performative=IpfsMessage.Performative.ERROR),
            f"Expected performative {IpfsMessage.Performative.IPFS_HASH} but got {IpfsMessage.Performative.ERROR}.",
        ),
    ],
)
def test_send_to_ipfs(caplog: LogCaptureFixture, ipfs_response: IpfsMessage,
                      expected_log: str) -> None

Test send_to_ipfs

test_ipfs_store_fails

def test_ipfs_store_fails(caplog: LogCaptureFixture) -> None

Test for failure during building store_file_req.

test_do_ipfs_request

def test_do_ipfs_request() -> None

Test _do_ipfs_request

test_get_from_ipfs

@pytest.mark.parametrize(
    "ipfs_response, expected_log",
    [
        (
            MagicMock(
                files={"dummy_file_name": "test"},
                performative=IpfsMessage.Performative.FILES,
            ),
            "Retrieved 1 objects from ipfs.",
        ),
        (
            MagicMock(ipfs_hash="test",
                      performative=IpfsMessage.Performative.ERROR),
            f"Expected performative {IpfsMessage.Performative.FILES} but got {IpfsMessage.Performative.ERROR}.",
        ),
    ],
)
def test_get_from_ipfs(caplog: LogCaptureFixture, ipfs_response: IpfsMessage,
                       expected_log: str) -> None

Test get_from_ipfs

test_ipfs_get_fails

def test_ipfs_get_fails(caplog: LogCaptureFixture) -> None

Test for failure during building get_files req.

test_params_property

def test_params_property() -> None

Test the 'params' property.

test_synchronized_data_property

def test_synchronized_data_property() -> None

Test the 'synchronized_data' property.

test_check_in_round

def test_check_in_round() -> None

Test 'BaseBehaviour' initialization.

test_check_in_last_round

def test_check_in_last_round() -> None

Test 'BaseBehaviour' initialization.

test_check_round_height_has_changed

def test_check_round_height_has_changed() -> None

Test 'check_round_height_has_changed'.

test_wait_until_round_end_negative_last_round_or_matching_round

def test_wait_until_round_end_negative_last_round_or_matching_round() -> None

Test 'wait_until_round_end' method, negative case (not in matching nor last round).

test_wait_until_round_end_positive

@mock.patch.object(BaseBehaviour, "wait_for_condition")
@mock.patch.object(BaseBehaviour, "check_not_in_round", return_value=False)
@mock.patch.object(BaseBehaviour,
                   "check_not_in_last_round",
                   return_value=False)
def test_wait_until_round_end_positive(*_: Any) -> None

Test 'wait_until_round_end' method, positive case.

test_wait_from_last_timestamp

def test_wait_from_last_timestamp() -> None

Test 'wait_from_last_timestamp'.

test_wait_from_last_timestamp_negative

def test_wait_from_last_timestamp_negative() -> None

Test 'wait_from_last_timestamp'.

test_set_done

def test_set_done() -> None

Test 'set_done' method.

test_send_a2a_transaction_positive

@mock.patch.object(BaseBehaviour, "_send_transaction")
def test_send_a2a_transaction_positive(*_: Any) -> None

Test 'send_a2a_transaction' method, positive case.

test_check_sync_logs_version_mismatch_for_schema_drift

def test_check_sync_logs_version_mismatch_for_schema_drift(
        caplog: LogCaptureFixture) -> None

Schema drift triggers the distinct 'version mismatch' warning.

A KeyError/TypeError/ValueError from parsing /status indicates the response shape has drifted (e.g. Tendermint renamed a field). The widened catch must log the distinct 'version mismatch' warning, not the legacy 'not accepting transactions yet' debug message that fits the JSONDecodeError case.

test_check_sync_keeps_legacy_message_for_json_decode_error

def test_check_sync_keeps_legacy_message_for_json_decode_error(
        caplog: LogCaptureFixture) -> None

Non-JSON status response keeps the legacy debug log.

JSONDecodeError (Tendermint not started yet, returns non-JSON) keeps the legacy 'not accepting transactions yet' debug log, preserving the distinction that motivated splitting the catch.

test_async_act_wrapper_agent_sync_mode

def test_async_act_wrapper_agent_sync_mode() -> None

Test 'async_act_wrapper' in sync mode.

test_async_act_wrapper_agent_sync_mode_where_height_dont_match

@mock.patch.object(BaseBehaviour, "_get_status", _get_status_wrong_patch)
def test_async_act_wrapper_agent_sync_mode_where_height_dont_match() -> None

Test 'async_act_wrapper' in sync mode.

test_async_act_wrapper_exception

@pytest.mark.parametrize("exception_cls", [StopIteration])
def test_async_act_wrapper_exception(exception_cls: Exception) -> None

Test 'async_act_wrapper'.

test_get_request_nonce_from_dialogue

def test_get_request_nonce_from_dialogue() -> None

Test '_get_request_nonce_from_dialogue' helper method.

test_send_transaction_stop_condition

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=False)
def test_send_transaction_stop_condition(*_: Any) -> None

Test '_send_transaction' method's stop_condition as provided by send_a2a_transaction.

test_send_transaction_positive_false_condition

def test_send_transaction_positive_false_condition() -> None

Test '_send_transaction', positive case (false condition)

test_send_transaction_positive

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
def test_send_transaction_positive(*_: Any) -> None

Test '_send_transaction', positive case.

test_send_transaction_invalid_transaction

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
@mock.patch.object(
    BaseBehaviour,
    "_wait_until_transaction_delivered",
    new=_wait_until_transaction_delivered_patch,
)
def test_send_transaction_invalid_transaction(*_: Any) -> None

Test '_send_transaction', positive case.

test_send_transaction_valid_transaction

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(BaseBehaviour,
                   "_is_invalid_transaction",
                   return_value=False)
@mock.patch.object(BaseBehaviour, "_tx_not_found", return_value=True)
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
@mock.patch.object(
    BaseBehaviour,
    "_wait_until_transaction_delivered",
    new=_wait_until_transaction_delivered_patch,
)
def test_send_transaction_valid_transaction(*_: Any) -> None

Test '_send_transaction', positive case.

test_tx_not_found

def test_tx_not_found(*_: Any) -> None

Test _tx_not_found

test_is_invalid_transaction

@pytest.mark.parametrize(
    "body, expected",
    [
        (
            '{"tx_result": {"info": "LateArrivingTransaction: request \'RedeemPayload(...round_count=368...\'."}}',
            True,
        ),
        (
            '{"tx_result": {"info": "TransactionNotValidError: ..."}}',
            True,
        ),
        (
            '{"tx_result": {"info": ""}}',
            False,
        ),
    ],
)
def test_is_invalid_transaction(body: str, expected: bool) -> None

Test _is_invalid_transaction recognizes various transaction error types.

test_send_transaction_signing_error

@mock.patch.object(BaseBehaviour, "_send_signing_request")
def test_send_transaction_signing_error(*_: Any) -> None

Test '_send_transaction', signing error.

test_send_transaction_timeout_exception_submit_tx

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
def test_send_transaction_timeout_exception_submit_tx(*_: Any) -> None

Test '_send_transaction', timeout exception.

test_send_transaction_timeout_exception_wait_until_transaction_delivered

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
def test_send_transaction_timeout_exception_wait_until_transaction_delivered(
        *_: Any) -> None

Test '_send_transaction', timeout exception.

test_send_transaction_transaction_not_delivered

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
def test_send_transaction_transaction_not_delivered(*_: Any) -> None

Test '_send_transaction', timeout exception.

test_send_transaction_wrong_ok_code

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
def test_send_transaction_wrong_ok_code(*_: Any) -> None

Test '_send_transaction', positive case.

test_send_transaction_wait_delivery_timeout_exception

@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(
    BaseBehaviour,
    "_check_http_return_code_200",
    return_value=True,
)
@mock.patch("json.loads",
            return_value={"result": {
                "hash": "",
                "code": OK_CODE
            }})
def test_send_transaction_wait_delivery_timeout_exception(*_: Any) -> None

Test '_send_transaction', timeout exception on tx delivery.

test_send_transaction_error_status_code

@pytest.mark.parametrize("resetting", (True, False))
@pytest.mark.parametrize(
    "non_200_count",
    (
        0,
        NON_200_RETURN_CODE_DURING_RESET_THRESHOLD,
        NON_200_RETURN_CODE_DURING_RESET_THRESHOLD + 1,
    ),
)
@mock.patch.object(BaseBehaviour, "_send_signing_request")
@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch("json.loads")
def test_send_transaction_error_status_code(_: Any, __: Any, ___: Any,
                                            ____: Any, resetting: bool,
                                            non_200_count: int) -> None

Test '_send_transaction', error status code.

test_send_signing_request

@mock.patch.object(BaseBehaviour, "_get_request_nonce_from_dialogue")
@mock.patch.object(behaviour_utils, "RawMessage")
@mock.patch.object(behaviour_utils, "Terms")
def test_send_signing_request(*_: Any) -> None

Test '_send_signing_request'.

test_fuzz_send_signing_request

@given(st.binary())
def test_fuzz_send_signing_request(input_bytes: bytes) -> None

Fuzz '_send_signing_request'.

Mock context manager decorators don't work here.

Arguments:

• input_bytes: fuzz input

test_send_transaction_signing_request

@mock.patch.object(BaseBehaviour, "_get_request_nonce_from_dialogue")
@mock.patch.object(behaviour_utils, "RawMessage")
@mock.patch.object(behaviour_utils, "Terms")
def test_send_transaction_signing_request(*_: Any) -> None

Test '_send_signing_request'.

test_send_transaction_request

@pytest.mark.parametrize(
    "chain_id, expected_kwargs",
    (
        (
            None,
            dict(
                counterparty=LEDGER_API_ADDRESS,
                performative=LedgerApiMessage.Performative.
                SEND_SIGNED_TRANSACTION,
                signed_transaction=SignedTransaction(
                    ledger_id="ethereum", body={"test_tx": "test_tx"}),
            ),
        ),
        (
            "ethereum",
            dict(
                counterparty=LEDGER_API_ADDRESS,
                performative=LedgerApiMessage.Performative.
                SEND_SIGNED_TRANSACTION,
                signed_transaction=SignedTransaction(
                    ledger_id="ethereum", body={"test_tx": "test_tx"}),
                kwargs=LedgerApiMessage.Kwargs({"chain_id": "ethereum"}),
            ),
        ),
    ),
)
def test_send_transaction_request(chain_id: Optional[str],
                                  expected_kwargs: Any) -> None

Test '_send_transaction_request'.

test_send_transaction_request_flashbots_args_are_inert

def test_send_transaction_request_flashbots_args_are_inert() -> None

Flashbots-era kwargs are accepted for downstream API compatibility but do not change the request.

The skill-level use_flashbots / target_block_numbers / raise_on_failed_simulation parameters were intentionally preserved after the upstream flashbots plugin was removed (HISTORY.md v0.21.17). Locks in that passing them produces the same plain SEND_SIGNED_TRANSACTION shape as omitting them.

test_send_transaction_receipt_request

def test_send_transaction_receipt_request() -> None

Test '_send_transaction_receipt_request'.

test_build_http_request_message

def test_build_http_request_message(*_: Any) -> None

Test '_build_http_request_message'.

test_wait_until_transaction_delivered

@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
@mock.patch.object(BaseBehaviour, "sleep")
@mock.patch("json.loads")
def test_wait_until_transaction_delivered(*_: Any) -> None

Test '_wait_until_transaction_delivered' method.

test_wait_until_transaction_delivered_failed

@mock.patch.object(Transaction, "encode", return_value=MagicMock())
@mock.patch.object(
    BaseBehaviour,
    "_build_http_request_message",
    return_value=(MagicMock(), MagicMock()),
)
@mock.patch.object(BaseBehaviour,
                   "_check_http_return_code_200",
                   return_value=True)
@mock.patch.object(BaseBehaviour, "sleep")
@mock.patch("json.loads")
def test_wait_until_transaction_delivered_failed(*_: Any) -> None

Test '_wait_until_transaction_delivered' method.

test_wait_until_transaction_delivered_raises_timeout

def test_wait_until_transaction_delivered_raises_timeout(*_: Any) -> None

Test '_wait_until_transaction_delivered' method.

Uses a negative timeout to guarantee the deadline is already expired, avoiding timer-resolution issues on Windows (see 1477).

test_get_default_terms

@mock.patch.object(behaviour_utils, "Terms")
def test_get_default_terms(*_: Any) -> None

Test '_get_default_terms'.

test_send_raw_transaction

@mock.patch.object(BaseBehaviour, "_send_transaction_signing_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_receipt_request")
@mock.patch.object(behaviour_utils, "Terms")
@pytest.mark.parametrize(
    "ledger_message, expected_hash, expected_response_status",
    (
        (
            LedgerApiMessage(
                cast(
                    LedgerApiMessage.Performative,
                    LedgerApiMessage.Performative.TRANSACTION_DIGEST,
                ),
                ("", ""),
                transaction_digest=TransactionDigest("ledger_id", body="test"),
            ),
            "test",
            RPCResponseStatus.SUCCESS,
        ),
        (
            LedgerApiMessage(
                cast(
                    LedgerApiMessage.Performative,
                    LedgerApiMessage.Performative.TRANSACTION_DIGESTS,
                ),
                ("", ""),
                # Only the first hash will be considered
                # because we do not support sending multiple messages and receiving multiple tx hashes yet
                transaction_digests=TransactionDigests(
                    "ledger_id",
                    transaction_digests=["test", "will_not_be_considered"],
                ),
            ),
            "test",
            RPCResponseStatus.SUCCESS,
        ),
    ),
)
def test_send_raw_transaction(
        _send_transaction_signing_request: Any, _send_transaction_request: Any,
        _send_transaction_receipt_request: Any, _terms: Any,
        ledger_message: LedgerApiMessage, expected_hash: str,
        expected_response_status: RPCResponseStatus) -> None

Test 'send_raw_transaction'.

test_send_raw_transaction_with_wrong_signing_performative

@mock.patch.object(BaseBehaviour, "_send_transaction_signing_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_receipt_request")
@mock.patch.object(behaviour_utils, "Terms")
def test_send_raw_transaction_with_wrong_signing_performative(*_: Any) -> None

Test 'send_raw_transaction'.

test_send_raw_transaction_errors

@pytest.mark.parametrize(
    "message, expected_rpc_status",
    (
        ("Simulation failed for bundle", RPCResponseStatus.SIMULATION_FAILED),
        ("replacement transaction underpriced", RPCResponseStatus.UNDERPRICED),
        ("nonce too low", RPCResponseStatus.INCORRECT_NONCE),
        ("insufficient funds", RPCResponseStatus.INSUFFICIENT_FUNDS),
        ("already known", RPCResponseStatus.ALREADY_KNOWN),
        ("test", RPCResponseStatus.UNCLASSIFIED_ERROR),
    ),
)
@mock.patch.object(BaseBehaviour, "_send_transaction_signing_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_receipt_request")
@mock.patch.object(behaviour_utils, "Terms")
def test_send_raw_transaction_errors(
        _: Any, __: Any, ___: Any, ____: Any, message: str,
        expected_rpc_status: RPCResponseStatus) -> None

Test 'send_raw_transaction'.

test_send_raw_transaction_hashes_mismatch

@mock.patch.object(BaseBehaviour, "_send_transaction_signing_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_request")
@mock.patch.object(BaseBehaviour, "_send_transaction_receipt_request")
@mock.patch.object(behaviour_utils, "Terms")
def test_send_raw_transaction_hashes_mismatch(*_: Any) -> None

Test 'send_raw_transaction' when signature and tx responses' hashes mismatch.

test_get_transaction_receipt

def test_get_transaction_receipt(caplog: LogCaptureFixture) -> None

Test get_transaction_receipt.

test_get_transaction_receipt_error

def test_get_transaction_receipt_error(caplog: LogCaptureFixture) -> None

Test get_transaction_receipt with error performative.

test_get_contract_api_response

@pytest.mark.parametrize("contract_address", [None, "contract_address"])
def test_get_contract_api_response(contract_address: Optional[str]) -> None

Test 'get_contract_api_response'.

test_get_status

@mock.patch.object(BaseBehaviour,
                   "_build_http_request_message",
                   return_value=(None, None))
def test_get_status(_: mock.Mock) -> None

Test '_get_status'.

test_get_netinfo

def test_get_netinfo() -> None

Test _get_netinfo method

test_num_active_peers

@pytest.mark.parametrize(
    ("num_peers", "expected_num_peers", "netinfo_status_code"),
    [
        ("0", 1, 200),
        ("0", None, 500),
        ("0", None, None),
        (None, None, 200),
    ],
)
def test_num_active_peers(num_peers: Optional[str],
                          expected_num_peers: Optional[int],
                          netinfo_status_code: Optional[int]) -> None

Test num_active_peers.

test_default_callback_request_stopped

def test_default_callback_request_stopped() -> None

Test 'default_callback_request' when stopped.

test_default_callback_late_arriving_message

def test_default_callback_late_arriving_message(*_: Any) -> None

Test 'default_callback_request' when a message arrives late.

test_default_callback_request_waiting_message

def test_default_callback_request_waiting_message(*_: Any) -> None

Test 'default_callback_request' when waiting message.

test_default_callback_request_else

def test_default_callback_request_else(*_: Any) -> None

Test 'default_callback_request' else branch.

test_stop

def test_stop() -> None

Test the stop method.

test_acn_request_from_pending

@pytest.mark.parametrize(
    "performative",
    (
        TendermintMessage.Performative.GET_GENESIS_INFO,
        TendermintMessage.Performative.GET_RECOVERY_PARAMS,
    ),
)
@pytest.mark.parametrize(
    "address_to_acn_deliverable, n_pending",
    (
        ({}, 0),
        ({
            i: None
            for i in range(3)
        }, 3),
        ({
            0: "test",
            1: None,
            2: None
        }, 2),
        ({
            i: "test"
            for i in range(3)
        }, 0),
    ),
)
def test_acn_request_from_pending(performative: TendermintMessage.Performative,
                                  address_to_acn_deliverable: Dict[str, Any],
                                  n_pending: int) -> None

Test the _acn_request_from_pending method.

test_perform_acn_request

@pytest.mark.parametrize(
    "performative",
    (
        TendermintMessage.Performative.GET_GENESIS_INFO,
        TendermintMessage.Performative.GET_RECOVERY_PARAMS,
    ),
)
@pytest.mark.parametrize(
    "address_to_acn_deliverable_per_attempt, expected_result",
    (
        (
            tuple({"address": None} for _ in range(10)),
            None,
        ),  # an example in which no agent responds
        (
            (
                {
                    f"address{i}": None
                    for i in range(3)
                },
                {
                    "address1": None,
                    "address2": "test",
                    "address3": None
                },
            ) + tuple({
                "address1": None,
                "address2": "test",
                "address3": "malicious"
            } for _ in range(8)),
            None,
        ),  # an example in which no majority is reached
        (
            tuple({f"address{i}": None
                   for i in range(3)} for _ in range(3)) + ({
                       "address1": "test",
                       "address2": "test",
                       "address3": None
                   }, ),
            "test",
        ),  # an example in which majority is reached during the 4th ACN attempt
    ),
)
def test_perform_acn_request(performative: TendermintMessage.Performative,
                             address_to_acn_deliverable_per_attempt: Tuple[
                                 Dict[str, Any], ...],
                             expected_result: Any) -> None

Test the _perform_acn_request method.

test_request_recovery_params

@pytest.mark.parametrize("expected_result", (True, False))
def test_request_recovery_params(expected_result: bool) -> None

Test request_recovery_params.

test_start_reset

def test_start_reset() -> None

Test the _start_reset method.

test_end_reset

def test_end_reset() -> None

Test the _end_reset method.

test_is_timeout_expired

@pytest.mark.parametrize(
    "check_started, is_healthy, timeout, expiration_expected",
    (
        (None, True, 0, False),
        (None, False, 0, False),
        (datetime(1, 1, 1), True, 0, False),
        (datetime.now(), False, 3000, False),
        (datetime(1, 1, 1), False, 0, True),
    ),
)
def test_is_timeout_expired(check_started: Optional[datetime],
                            is_healthy: bool, timeout: float,
                            expiration_expected: bool) -> None

Test the _is_timeout_expired method.

test_get_reset_params

@pytest.mark.parametrize("default", (True, False))
@given(
    st.datetimes(
        min_value=MIN_DATETIME_WINDOWS,
        max_value=MAX_DATETIME_WINDOWS,
    ),
    st.integers(),
    st.integers(),
    st.integers(),
)
def test_get_reset_params(default: bool, timestamp: datetime, height: int,
                          interval: int, period: int) -> None

Test _get_reset_params method.

test_get_reset_params_before_first_transition

def test_get_reset_params_before_first_transition(
        caplog: LogCaptureFixture) -> None

Test _get_reset_params falls back to default params before the first round transition.

test_reset_tendermint_with_wait_timeout_expired

@mock.patch.object(BaseBehaviour, "_start_reset")
@mock.patch.object(BaseBehaviour, "_is_timeout_expired")
def test_reset_tendermint_with_wait_timeout_expired(*_: mock.Mock) -> None

Test tendermint reset.

test_reset_tendermint_with_wait

@mock.patch.object(BaseBehaviour, "_start_reset")
@mock.patch.object(BaseBehaviour,
                   "_build_http_request_message",
                   return_value=(None, None))
@pytest.mark.parametrize(
    "reset_response, status_response, local_height, on_startup, n_iter, expecting_success",
    (
        (
            {
                "message": "Tendermint reset was successful.",
                "status": True
            },
            {
                "result": {
                    "sync_info": {
                        "latest_block_height": 1
                    }
                }
            },
            1,
            False,
            3,
            True,
        ),
        (
            {
                "message": "Tendermint reset was successful.",
                "status": True
            },
            {
                "result": {
                    "sync_info": {
                        "latest_block_height": 1
                    }
                }
            },
            1,
            True,
            2,
            True,
        ),
        (
            {
                "message": "Tendermint reset was successful.",
                "status": True,
                "is_replay": True,
            },
            {
                "result": {
                    "sync_info": {
                        "latest_block_height": 1
                    }
                }
            },
            1,
            False,
            3,
            True,
        ),
        (
            {
                "message": "Tendermint reset was successful.",
                "status": True
            },
            {
                "result": {
                    "sync_info": {
                        "latest_block_height": 1
                    }
                }
            },
            3,
            False,
            3,
            False,
        ),
        (
            {
                "message": "Error resetting tendermint.",
                "status": False
            },
            {},
            0,
            False,
            2,
            False,
        ),
        ("wrong_response", {}, 0, False, 2, False),
        (
            {
                "message": "Reset Successful.",
                "status": True
            },
            "not_accepting_txs_yet",
            0,
            False,
            3,
            False,
        ),
    ),
)
def test_reset_tendermint_with_wait(build_http_request_message_mock: mock.Mock,
                                    _start_reset: mock.Mock,
                                    reset_response: Union[Dict[str,
                                                               Union[bool,
                                                                     str]],
                                                          str],
                                    status_response: Union[Dict[str,
                                                                Union[int,
                                                                      str]],
                                                           str],
                                    local_height: int, on_startup: bool,
                                    n_iter: int,
                                    expecting_success: bool) -> None

Test tendermint reset.

test_fuzz_submit_tx

@given(st.binary())
def test_fuzz_submit_tx(input_bytes: bytes) -> None

Fuzz '_submit_tx'.

Mock context manager decorators don't work here.

Arguments:

• input_bytes: fuzz input

test_degenerate_behaviour_async_act

def test_degenerate_behaviour_async_act() -> None

Test DegenerateBehaviour.async_act.

test_make_degenerate_behaviour

def test_make_degenerate_behaviour() -> None

Test 'make_degenerate_behaviour'.

TestTmManager Objects

class TestTmManager()

Class to test the TmManager behaviour.

setup_method

def setup_method() -> None

Set up the tests.

test_async_act

def test_async_act() -> None

Test the async_act method of the TmManager.

test_handle_unhealthy_tm

@given(latest_block_height=st.integers(min_value=0))
@pytest.mark.parametrize(
    "acn_communication_success",
    (
        True,
        False,
    ),
)
@pytest.mark.parametrize(
    "gentle_reset_attempted",
    (
        True,
        False,
    ),
)
@pytest.mark.parametrize(
    ("tm_reset_success", "num_active_peers"),
    [
        (True, 4),
        (False, 4),
        (True, 2),
        (False, None),
    ],
)
def test_handle_unhealthy_tm(latest_block_height: int,
                             acn_communication_success: bool,
                             gentle_reset_attempted: bool,
                             tm_reset_success: bool,
                             num_active_peers: Optional[int]) -> None

Test _handle_unhealthy_tm.

test_handle_unhealthy_tm_logging

@pytest.mark.parametrize(
    "n_repetitions",
    (
        1,
        2,
        1000,
    ),
)
def test_handle_unhealthy_tm_logging(n_repetitions: int) -> None

Verify if unintended logging repetition occurs during the execution of _handle_unhealthy_tm.

test_get_reset_params

@pytest.mark.parametrize(
    "expected_reset_params",
    (
        {
            "genesis_time": "genesis-time",
            "initial_height": "1"
        },
        None,
    ),
)
def test_get_reset_params(
        expected_reset_params: Optional[Dict[str, str]]) -> None

Test that reset params returns the correct params.

test_sleep_after_hard_reset

def test_sleep_after_hard_reset() -> None

Check that hard_reset_sleep returns the expected amount of time.

test_try_fix

@pytest.mark.parametrize(
    ("state", "notified", "message", "num_iter"),
    [
        (AsyncBehaviour.AsyncState.READY, False, None, 1),
        (AsyncBehaviour.AsyncState.WAITING_MESSAGE, True, Message(), 2),
        (AsyncBehaviour.AsyncState.WAITING_MESSAGE, True, Message(), 1),
    ],
)
def test_try_fix(state: AsyncBehaviour.AsyncState, notified: bool,
                 message: Optional[Message], num_iter: int) -> None

Tests try_fix.

test_get_callback_request

@pytest.mark.parametrize(
    "state",
    [
        AsyncBehaviour.AsyncState.WAITING_MESSAGE,
        AsyncBehaviour.AsyncState.READY,
    ],
)
def test_get_callback_request(state: AsyncBehaviour.AsyncState) -> None

Tests get_callback_request.

test_is_acting

def test_is_acting() -> None

Test is_acting.

test_meta_base_behaviour_when_instance_not_subclass_of_base_behaviour

def test_meta_base_behaviour_when_instance_not_subclass_of_base_behaviour(
) -> None

Test instantiation of meta class when instance not a subclass of BaseBehaviour.

test_base_behaviour_instantiation_without_attributes_raises_error

def test_base_behaviour_instantiation_without_attributes_raises_error(
) -> None

Test that definition of concrete subclass of BaseBehaviour without attributes raises error.

TestIPFSBehaviour Objects

class TestIPFSBehaviour()

Test IPFSBehaviour tests.

setup_method

def setup_method() -> None

Sets up the tests.

test_build_ipfs_message

def test_build_ipfs_message() -> None

Tests _build_ipfs_message.

test_build_ipfs_store_file_req

def test_build_ipfs_store_file_req() -> None

Tests _build_ipfs_store_file_req.

test_build_ipfs_get_file_req

def test_build_ipfs_get_file_req() -> None

Tests _build_ipfs_get_file_req.

test_deserialize_ipfs_objects

def test_deserialize_ipfs_objects() -> None

Tests _deserialize_ipfs_objects