packages.valory.skills.transaction_settlement_abci.tests.test_rounds
Tests for valory/registration_abci skill's rounds.
get_participants
def get_participants() -> FrozenSet[str]
Participants
get_participant_to_randomness
def get_participant_to_randomness(
participants: FrozenSet[str],
round_id: int) -> Dict[str, RandomnessPayload]
participant_to_randomness
get_most_voted_randomness
def get_most_voted_randomness() -> str
most_voted_randomness
get_participant_to_selection
def get_participant_to_selection(
participants: FrozenSet[str],
keepers: str) -> Dict[str, SelectKeeperPayload]
participant_to_selection
get_participant_to_period_count
def get_participant_to_period_count(
participants: FrozenSet[str],
period_count: int) -> Dict[str, ResetPayload]
participant_to_selection
get_safe_contract_address
def get_safe_contract_address() -> str
safe_contract_address
get_participant_to_votes
def get_participant_to_votes(
participants: FrozenSet[str],
vote: Optional[bool] = True) -> Dict[str, ValidatePayload]
participant_to_votes
get_participant_to_votes_serialized
def get_participant_to_votes_serialized(
participants: FrozenSet[str],
vote: Optional[bool] = True) -> Dict[str, Dict[str, Any]]
participant_to_votes
get_most_voted_tx_hash
def get_most_voted_tx_hash() -> str
most_voted_tx_hash
get_participant_to_signature
def get_participant_to_signature(
participants: FrozenSet[str]) -> Dict[str, SignaturePayload]
participant_to_signature
get_final_tx_hash
def get_final_tx_hash() -> str
final_tx_hash
get_participant_to_check
def get_participant_to_check(
participants: FrozenSet[str], status: str,
tx_hash: str) -> Dict[str, CheckTransactionHistoryPayload]
Get participants to check
get_participant_to_late_arriving_tx_hashes
def get_participant_to_late_arriving_tx_hashes(
participants: FrozenSet[str]
) -> Dict[str, SynchronizeLateMessagesPayload]
participant_to_selection
get_late_arriving_tx_hashes_deserialized
def get_late_arriving_tx_hashes_deserialized() -> Dict[str, List[str]]
Get dummy late-arriving tx hashes.
get_late_arriving_tx_hashes_serialized
def get_late_arriving_tx_hashes_serialized() -> Dict[str, str]
Get dummy late-arriving tx hashes.
get_keepers
def get_keepers(keepers: Deque[str], retries: int = 1) -> str
Get dummy keepers.
BaseValidateRoundTest Objects
class BaseValidateRoundTest(BaseVotingRoundTest)
Test BaseValidateRound.
test_positive_votes
def test_positive_votes() -> None
Test ValidateRound.
test_negative_votes
def test_negative_votes() -> None
Test ValidateRound.
test_none_votes
def test_none_votes() -> None
Test ValidateRound.
BaseSelectKeeperRoundTest Objects
class BaseSelectKeeperRoundTest(BaseCollectSameUntilThresholdRoundTest)
Test SelectKeeperTransactionSubmissionARound
test_run
def test_run(most_voted_payload: str = "keeper",
keepers: str = "",
exit_event: Optional[Any] = None) -> None
Run tests.
TestSelectKeeperTransactionSubmissionARound Objects
class TestSelectKeeperTransactionSubmissionARound(BaseSelectKeeperRoundTest)
Test SelectKeeperTransactionSubmissionARound
test_run
@pytest.mark.parametrize(
"most_voted_payload, keepers, exit_event",
(
(
"incorrectly_serialized",
"",
TransactionSettlementEvent.INCORRECT_SERIALIZATION,
),
(
int(1).to_bytes(32, "big").hex() + "new_keeper" + "-" * 32,
"",
TransactionSettlementEvent.DONE,
),
),
)
def test_run(most_voted_payload: str, keepers: str,
exit_event: TransactionSettlementEvent) -> None
Run tests.
TestSelectKeeperTransactionSubmissionBRound Objects
class TestSelectKeeperTransactionSubmissionBRound(
TestSelectKeeperTransactionSubmissionARound)
Test SelectKeeperTransactionSubmissionBRound.
test_run
@pytest.mark.parametrize(
"most_voted_payload, keepers, exit_event",
(
(
int(1).to_bytes(32, "big").hex() + "new_keeper" + "-" * 32,
"",
TransactionSettlementEvent.DONE,
),
(
int(1).to_bytes(32, "big").hex() + "new_keeper" + "-" * 32,
int(1).to_bytes(32, "big").hex() + "".join([
keeper + "-" * 30
for keeper in ("test_keeper1", "test_keeper2")
]),
TransactionSettlementEvent.DONE,
),
),
)
def test_run(most_voted_payload: str, keepers: str,
exit_event: TransactionSettlementEvent) -> None
Run tests.
TestSelectKeeperTransactionSubmissionBAfterTimeoutRound Objects
class TestSelectKeeperTransactionSubmissionBAfterTimeoutRound(
TestSelectKeeperTransactionSubmissionBRound)
Test SelectKeeperTransactionSubmissionBAfterTimeoutRound.
test_run
@mock.patch.object(
TransactionSettlementSynchronizedSata,
"keepers_threshold_exceeded",
new_callable=mock.PropertyMock,
)
@pytest.mark.parametrize(
"keepers", (f"{int(1).to_bytes(32, 'big').hex()}keeper" + "-" * 36, ))
@pytest.mark.parametrize(
"attrs, threshold_exceeded, exit_event",
(
(
{
"tx_hashes_history": "t" * 66,
"missed_messages": {
f"keeper{'-' * 36}": 10
},
},
True,
# Since the threshold has been exceeded, we should return a `CHECK_HISTORY` event.
TransactionSettlementEvent.CHECK_HISTORY,
),
(
{
"missed_messages": {
f"keeper{'-' * 36}": 10
},
},
True,
TransactionSettlementEvent.CHECK_LATE_ARRIVING_MESSAGE,
),
(
{
"missed_messages": {
f"keeper{'-' * 36}": 10
},
},
False,
TransactionSettlementEvent.DONE,
),
),
)
def test_run(threshold_exceeded_mock: mock.PropertyMock, keepers: str,
attrs: Dict[str, Union[str, int]], threshold_exceeded: bool,
exit_event: TransactionSettlementEvent) -> None
Test SelectKeeperTransactionSubmissionBAfterTimeoutRound.
TestFinalizationRound Objects
class TestFinalizationRound(BaseOnlyKeeperSendsRoundTest)
Test FinalizationRound.
test_finalization_round
@pytest.mark.parametrize(
"tx_hashes_history, tx_digest, missed_messages, status, exit_event",
(
(
"",
"",
{
"test": 1
},
VerificationStatus.ERROR.value,
TransactionSettlementEvent.CHECK_LATE_ARRIVING_MESSAGE,
),
(
"",
"",
{},
VerificationStatus.ERROR.value,
TransactionSettlementEvent.FINALIZATION_FAILED,
),
(
"t" * 66,
"",
{},
VerificationStatus.VERIFIED.value,
TransactionSettlementEvent.CHECK_HISTORY,
),
(
"t" * 66,
"",
{},
VerificationStatus.ERROR.value,
TransactionSettlementEvent.CHECK_HISTORY,
),
(
"",
"",
{},
VerificationStatus.PENDING.value,
TransactionSettlementEvent.FINALIZATION_FAILED,
),
(
"",
"tx_digest" + "t" * 57,
{},
VerificationStatus.PENDING.value,
TransactionSettlementEvent.DONE,
),
(
"t" * 66,
"tx_digest" + "t" * 57,
{},
VerificationStatus.PENDING.value,
TransactionSettlementEvent.DONE,
),
(
"t" * 66,
"",
{},
VerificationStatus.INSUFFICIENT_FUNDS.value,
TransactionSettlementEvent.INSUFFICIENT_FUNDS,
),
),
)
def test_finalization_round(tx_hashes_history: str, tx_digest: str,
missed_messages: int, status: int,
exit_event: TransactionSettlementEvent) -> None
Runs tests.
test_finalization_round_no_tx_data
def test_finalization_round_no_tx_data() -> None
Test finalization round when tx_data is None.
TestCollectSignatureRound Objects
class TestCollectSignatureRound(BaseCollectDifferentUntilThresholdRoundTest)
Test CollectSignatureRound.
test_run
def test_run() -> None
Runs tests.
TestValidateTransactionRound Objects
class TestValidateTransactionRound(BaseValidateRoundTest)
Test ValidateRound.
TestCheckTransactionHistoryRound Objects
class TestCheckTransactionHistoryRound(BaseCollectSameUntilThresholdRoundTest)
Test CheckTransactionHistoryRound
test_run
@pytest.mark.parametrize(
"expected_status, expected_tx_hash, missed_messages, expected_event",
(
(
"0000000000000000000000000000000000000000000000000000000000000001",
"b0e6add595e00477cf347d09797b156719dc5233283ac76e4efce2a674fe72d9",
{},
TransactionSettlementEvent.DONE,
),
(
"0000000000000000000000000000000000000000000000000000000000000002",
"b0e6add595e00477cf347d09797b156719dc5233283ac76e4efce2a674fe72d9",
{},
TransactionSettlementEvent.NEGATIVE,
),
(
"0000000000000000000000000000000000000000000000000000000000000003",
"b0e6add595e00477cf347d09797b156719dc5233283ac76e4efce2a674fe72d9",
{},
TransactionSettlementEvent.NONE,
),
(
"0000000000000000000000000000000000000000000000000000000000000007",
"b0e6add595e00477cf347d09797b156719dc5233283ac76e4efce2a674fe72d9",
{},
TransactionSettlementEvent.NONE,
),
(
"0000000000000000000000000000000000000000000000000000000000000002",
"b0e6add595e00477cf347d09797b156719dc5233283ac76e4efce2a674fe72d9",
{
"test": 1
},
TransactionSettlementEvent.CHECK_LATE_ARRIVING_MESSAGE,
),
),
)
def test_run(expected_status: str, expected_tx_hash: str, missed_messages: int,
expected_event: TransactionSettlementEvent) -> None
Run tests.
TestSynchronizeLateMessagesRound Objects
class TestSynchronizeLateMessagesRound(BaseCollectNonEmptyUntilThresholdRound)
Test SynchronizeLateMessagesRound.
test_runs
@pytest.mark.parametrize(
"missed_messages, expected_event",
(
(
{
f"agent_{i}": 0
for i in range(4)
},
TransactionSettlementEvent.SUSPICIOUS_ACTIVITY,
),
({
f"agent_{i}": 2
for i in range(4)
}, TransactionSettlementEvent.DONE),
),
)
def test_runs(missed_messages: int,
expected_event: TransactionSettlementEvent) -> None
Runs tests.
test_payload_validates_hash_length
@pytest.mark.parametrize("correct_serialization", (True, False))
def test_payload_validates_hash_length(correct_serialization: bool) -> None
Reject tx_hashes that do not align to TX_HASH_LENGTH.
test_malformed_payload_rejected_via_from_json
def test_malformed_payload_rejected_via_from_json() -> None
from_json re-raises TransactionNotValidError on bad tx_hashes.
The abci check_tx/deliver_tx handlers decode a remote payload
via BaseTxPayload.from_json, which calls the dataclass constructor
and triggers __post_init__. Build a valid payload first to obtain
the on-wire dict shape, mutate tx_hashes to a length that violates
TX_HASH_LENGTH, then attempt to round-trip through from_json.
The framework relies on this raising a type the handler already
catches.
test_synchronized_datas
def test_synchronized_datas() -> None
Test SynchronizedData.
TestResetRound Objects
class TestResetRound(BaseCollectSameUntilThresholdRoundTest)
Test ResetRound.
test_runs
def test_runs() -> None
Runs tests.