packages.valory.skills.transaction_settlement_abci.tests.test_behaviours

Tests for valory/registration_abci skill's behaviours.

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.

test_skill_public_id

def test_skill_public_id() -> None

Test skill module public ID

TransactionSettlementFSMBehaviourBaseCase Objects

class TransactionSettlementFSMBehaviourBaseCase(FSMBehaviourBaseCase)

Base case for testing TransactionSettlement FSMBehaviour.

ffw_signature

def ffw_signature(db_items: Optional[Dict] = None) -> None

Fast-forward to the SignatureBehaviour.

TestTransactionSettlementBaseBehaviour Objects

class TestTransactionSettlementBaseBehaviour(
        TransactionSettlementFSMBehaviourBaseCase)

Test TransactionSettlementBaseBehaviour.

test__get_tx_data

@pytest.mark.parametrize(
    "message, tx_digest, rpc_status, expected_data, replacement",
    (
        (
            MagicMock(performative=ContractApiMessage.Performative.ERROR,
                      message="GS026"),
            None,
            RPCResponseStatus.SUCCESS,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.VERIFIED,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(performative=ContractApiMessage.Performative.ERROR,
                      message="test"),
            None,
            RPCResponseStatus.SUCCESS,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.ERROR,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_MESSAGE),
            None,
            RPCResponseStatus.SUCCESS,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION),
            None,
            RPCResponseStatus.INCORRECT_NONCE,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.ERROR,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION),
            None,
            RPCResponseStatus.INSUFFICIENT_FUNDS,
            {
                "blacklisted_keepers": {"agent_1" + "-" * 35},
                "keeper_retries": 1,
                "keepers": deque(("agent_3" + "-" * 35, )),
                "status": VerificationStatus.INSUFFICIENT_FUNDS,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION),
            None,
            RPCResponseStatus.UNCLASSIFIED_ERROR,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION),
            None,
            RPCResponseStatus.UNDERPRICED,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION),
            "test_digest_0",
            RPCResponseStatus.ALREADY_KNOWN,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "test_digest_0",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION,
                raw_transaction=MagicMock(
                    body={
                        "nonce": 0,
                        "maxPriorityFeePerGas": 10,
                        "maxFeePerGas": 20,
                        "gas": 0,
                    }),
            ),
            "test_digest_1",
            RPCResponseStatus.SUCCESS,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "test_digest_1",
            },
            False,
        ),
        (
            MagicMock(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION,
                raw_transaction=MagicMock(
                    body={
                        "nonce": 0,
                        "maxPriorityFeePerGas": 10,
                        "maxFeePerGas": 20,
                        "gas": 0,
                    }),
            ),
            "test_digest_2",
            RPCResponseStatus.SUCCESS,
            {
                "blacklisted_keepers": set(),
                "keeper_retries": 2,
                "keepers": deque(("agent_1" + "-" * 35, "agent_3" + "-" * 35)),
                "status": VerificationStatus.PENDING,
                "tx_digest": "test_digest_2",
            },
            True,
        ),
    ),
)
def test__get_tx_data(message: ContractApiMessage, tx_digest: Optional[str],
                      rpc_status: RPCResponseStatus, expected_data: TxDataType,
                      replacement: bool, monkeypatch: MonkeyPatch) -> None

Test _get_tx_data.

test_get_gas_price_params

@pytest.mark.parametrize(
    argnames=["tx_body", "expected_params"],
    argvalues=[
        [
            {
                "maxPriorityFeePerGas": "dummy",
                "maxFeePerGas": "dummy"
            },
            ["maxPriorityFeePerGas", "maxFeePerGas"],
        ],
        [{
            "gasPrice": "dummy"
        }, ["gasPrice"]],
        [
            {
                "maxPriorityFeePerGas": "dummy"
            },
            [],
        ],
        [
            {
                "maxFeePerGas": "dummy"
            },
            [],
        ],
        [
            {},
            [],
        ],
        [
            {
                "maxPriorityFeePerGas": "dummy",
                "maxFeePerGas": "dummy",
                "gasPrice": "dummy",
            },
            ["maxPriorityFeePerGas", "maxFeePerGas"],
        ],
    ],
)
def test_get_gas_price_params(tx_body: dict,
                              expected_params: List[str]) -> None

Test the get_gas_price_params method

test_parse_revert_reason_successful

def test_parse_revert_reason_successful() -> None

Test _parse_revert_reason method.

test_parse_revert_reason_unsuccessful

@pytest.mark.parametrize(
    "message",
    (
        MagicMock(
            performative=ContractApiMessage.Performative.ERROR,
            message="Non existing code should be invalid GS086.",
        ),
        MagicMock(
            performative=ContractApiMessage.Performative.ERROR,
            message="Code not matching the regex should be invalid GS0265.",
        ),
        MagicMock(
            performative=ContractApiMessage.Performative.ERROR,
            message="No code in the message should be invalid.",
        ),
        MagicMock(
            performative=ContractApiMessage.Performative.ERROR,
            message="",  # empty message should be invalid
        ),
        MagicMock(
            performative=ContractApiMessage.Performative.ERROR,
            message=None,  # `None` message should be invalid
        ),
    ),
)
def test_parse_revert_reason_unsuccessful(message: ContractApiMessage) -> None

Test _parse_revert_reason method.

TestRandomnessInOperation Objects

class TestRandomnessInOperation(BaseRandomnessBehaviourTest)

Test randomness in operation.

TestSelectKeeperTransactionSubmissionBehaviourA Objects

class TestSelectKeeperTransactionSubmissionBehaviourA(
        BaseSelectKeeperBehaviourTest)

Test SelectKeeperBehaviour.

TestSelectKeeperTransactionSubmissionBehaviourB Objects

class TestSelectKeeperTransactionSubmissionBehaviourB(
        TestSelectKeeperTransactionSubmissionBehaviourA)

Test SelectKeeperBehaviour.

test_select_keeper

@mock.patch.object(
    TransactionSettlementSynchronizedSata,
    "keepers",
    new_callable=mock.PropertyMock,
)
@mock.patch.object(
    TransactionSettlementSynchronizedSata,
    "keeper_retries",
    new_callable=mock.PropertyMock,
)
@mock.patch.object(
    TransactionSettlementSynchronizedSata,
    "final_verification_status",
    new_callable=mock.PropertyMock,
)
@pytest.mark.parametrize(
    "keepers, keeper_retries, blacklisted_keepers, final_verification_status",
    (
        (
            deque(f"keeper_{i}" for i in range(4)),
            1,
            set(),
            VerificationStatus.NOT_VERIFIED,
        ),
        (deque(("test_keeper", )), 2, set(), VerificationStatus.PENDING),
        (deque(("test_keeper", )), 2, set(), VerificationStatus.NOT_VERIFIED),
        (deque(("test_keeper", )), 2, {"a1"}, VerificationStatus.NOT_VERIFIED),
        (
            deque(("test_keeper", )),
            2,
            {"test_keeper"},
            VerificationStatus.NOT_VERIFIED,
        ),
        (
            deque(("test_keeper", )),
            2,
            {"a_1", "a_2", "test_keeper"},
            VerificationStatus.NOT_VERIFIED,
        ),
        (deque(("test_keeper", )), 1, set(), VerificationStatus.NOT_VERIFIED),
        (deque(("test_keeper", )), 3, set(), VerificationStatus.NOT_VERIFIED),
    ),
)
def test_select_keeper(final_verification_status_mock: mock.PropertyMock,
                       keeper_retries_mock: mock.PropertyMock,
                       keepers_mock: mock.PropertyMock, keepers: Deque[str],
                       keeper_retries: int, blacklisted_keepers: Set[str],
                       final_verification_status: VerificationStatus) -> None

Test select keeper agent.

test_select_keeper_tx_pending

@mock.patch.object(
    TransactionSettlementSynchronizedSata,
    "final_verification_status",
    new_callable=mock.PropertyMock,
    return_value=VerificationStatus.PENDING,
)
@pytest.mark.skip
def test_select_keeper_tx_pending(_: mock.PropertyMock,
                                  caplog: LogCaptureFixture) -> None

Test select keeper while tx is pending

TestSignatureBehaviour Objects

class TestSignatureBehaviour(TransactionSettlementFSMBehaviourBaseCase)

Test SignatureBehaviour.

test_signature_behaviour

def test_signature_behaviour() -> None

Test signature behaviour.

TestFinalizeBehaviour Objects

class TestFinalizeBehaviour(TransactionSettlementFSMBehaviourBaseCase)

Test FinalizeBehaviour.

test_non_sender_act

def test_non_sender_act() -> None

Test finalize behaviour.

test_sender_act

@pytest.mark.parametrize(
    "resubmitting, response_kwargs",
    (
        ((
            True,
            dict(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION,
                callable="get_deploy_transaction",
                raw_transaction=RawTransaction(
                    ledger_id="ethereum",
                    body={
                        "tx_hash": "0x3b",
                        "nonce": 0,
                        "maxFeePerGas": int(10e10),
                        "maxPriorityFeePerGas": int(10e10),
                    },
                ),
            ),
        )),
        (
            False,
            dict(
                performative=ContractApiMessage.Performative.RAW_TRANSACTION,
                callable="get_deploy_transaction",
                raw_transaction=RawTransaction(
                    ledger_id="ethereum",
                    body={
                        "tx_hash": "0x3b",
                        "nonce": 0,
                        "maxFeePerGas": int(10e10),
                        "maxPriorityFeePerGas": int(10e10),
                    },
                ),
            ),
        ),
        (
            False,
            dict(
                performative=ContractApiMessage.Performative.ERROR,
                callable="get_deploy_transaction",
                code=500,
                message="GS026",
                data=b"",
            ),
        ),
        (
            False,
            dict(
                performative=ContractApiMessage.Performative.ERROR,
                callable="get_deploy_transaction",
                code=500,
                message="other error",
                data=b"",
            ),
        ),
    ),
)
@mock.patch.object(SkillContext,
                   "agent_address",
                   new_callable=mock.PropertyMock)
def test_sender_act(
    agent_address_mock: mock.PropertyMock, resubmitting: bool,
    response_kwargs: Dict[
        str,
        Union[
            int,
            str,
            bytes,
            Dict[str, Union[int, str]],
            ContractApiMessage.Performative,
            RawTransaction,
        ],
    ]
) -> None

Test finalize behaviour.

test_sender_act_tx_data_contains_tx_digest

def test_sender_act_tx_data_contains_tx_digest() -> None

Test finalize behaviour.

test_handle_late_messages

def test_handle_late_messages() -> None

Test handle_late_messages.

TestValidateTransactionBehaviour Objects

class TestValidateTransactionBehaviour(
        TransactionSettlementFSMBehaviourBaseCase)

Test ValidateTransactionBehaviour.

test_validate_transaction_safe_behaviour

def test_validate_transaction_safe_behaviour() -> None

Test ValidateTransactionBehaviour.

test_validate_transaction_safe_behaviour_no_tx_sent

def test_validate_transaction_safe_behaviour_no_tx_sent() -> None

Test ValidateTransactionBehaviour when tx cannot be sent.

TestCheckTransactionHistoryBehaviour Objects

class TestCheckTransactionHistoryBehaviour(
        TransactionSettlementFSMBehaviourBaseCase)

Test CheckTransactionHistoryBehaviour.

test_check_tx_history_behaviour

@pytest.mark.parametrize(
    "verified, status, hashes_history, revert_reason",
    (
        (False, -1, "0x" + "t" * 64, "test"),
        (False, 0, "", "test"),
        (False, 0, "0x" + "t" * 64, "test"),
        (False, 0, "0x" + "t" * 64, "GS026"),
        (True, 1, "0x" + "t" * 64, "test"),
    ),
)
def test_check_tx_history_behaviour(verified: bool, status: int,
                                    hashes_history: str,
                                    revert_reason: str) -> None

Test CheckTransactionHistoryBehaviour.

test_check_tx_history_behaviour_negative

@pytest.mark.parametrize(
    "verified, status, hashes_history, revert_reason",
    ((False, 0, "0x" + "t" * 64, "test"), ),
)
def test_check_tx_history_behaviour_negative(verified: bool, status: int,
                                             hashes_history: str,
                                             revert_reason: str) -> None

Test CheckTransactionHistoryBehaviour.

TestCheckLateTxHashesBehaviour Objects

class TestCheckLateTxHashesBehaviour(TransactionSettlementFSMBehaviourBaseCase
                                     )

Test CheckLateTxHashesBehaviour.

test_check_tx_history_behaviour

def test_check_tx_history_behaviour() -> None

Test CheckTransactionHistoryBehaviour.

TestSynchronizeLateMessagesBehaviour Objects

class TestSynchronizeLateMessagesBehaviour(
        TransactionSettlementFSMBehaviourBaseCase)

Test SynchronizeLateMessagesBehaviour

test_async_act

@pytest.mark.parametrize("late_messages", ([], [MagicMock, MagicMock]))
def test_async_act(late_messages: List[MagicMock]) -> None

Test async_act

test_short_circuit_clears_nondefault_late_messages

def test_short_circuit_clears_nondefault_late_messages() -> None

Short-circuit must reset a pre-existing non-default late_messages.

With a non-empty late_messages going in plus a dummy _get_tx_data that returns an empty tx_digest, the local _tx_hashes accumulator stays empty after the iterator drains. That triggers the short-circuit, and the assertion below catches any regression that drops the late_messages = [] reset.

TestResetBehaviour Objects

class TestResetBehaviour(TransactionSettlementFSMBehaviourBaseCase)

Test the reset behaviour.

test_reset_behaviour

def test_reset_behaviour() -> None

Test reset behaviour.