packages.valory.skills.abstract_round_abci.tests.test_base
Test the base.py module of the skill.
hypothesis_cleanup
@pytest.fixture(scope="session", autouse=True)
def hypothesis_cleanup() -> Generator
Fixture to remove hypothesis directory after tests.
BasePayload Objects
class BasePayload(BaseTxPayload, ABC)
Base payload class for testing.
PayloadA Objects
@dataclass(frozen=True)
class PayloadA(BasePayload)
Payload class for payload type 'A'.
PayloadB Objects
@dataclass(frozen=True)
class PayloadB(BasePayload)
Payload class for payload type 'B'.
PayloadC Objects
@dataclass(frozen=True)
class PayloadC(BasePayload)
Payload class for payload type 'C'.
PayloadD Objects
@dataclass(frozen=True)
class PayloadD(BasePayload)
Payload class for payload type 'D'.
DummyPayload Objects
@dataclass(frozen=True)
class DummyPayload(BasePayload)
Dummy payload class.
TooBigPayload Objects
@dataclass(frozen=True)
class TooBigPayload(BaseTxPayload)
Base payload class for testing.
ObjectImitator Objects
class ObjectImitator()
For custom eq implementation testing
__init__
def __init__(other: Any)
Copying references to class attr, and instance attr
test_base_tx_payload
def test_base_tx_payload() -> None
Test BaseTxPayload.
test_meta_round_abstract_round_when_instance_not_subclass_of_abstract_round
def test_meta_round_abstract_round_when_instance_not_subclass_of_abstract_round(
) -> (None)
Test instantiation of meta class when instance not a subclass of abstract round.
test_abstract_round_instantiation_without_attributes_raises_error
def test_abstract_round_instantiation_without_attributes_raises_error(
) -> None
Test that definition of concrete subclass of AbstractRound without attributes raises error.
test_specific_round_instantiation_without_extended_requirements_raises_error
def test_specific_round_instantiation_without_extended_requirements_raises_error(
) -> (None)
Test that definition of concrete subclass of CollectSameUntilThresholdRound without extended raises error.
TestTransactions Objects
class TestTransactions()
Test Transactions class.
setup_method
def setup_method() -> None
Set up the test.
test_encode_decode
def test_encode_decode() -> None
Test encoding and decoding of payloads.
test_encode_decode_transaction
def test_encode_decode_transaction() -> None
Test encode/decode of a transaction.
test_encode_too_big_payload
def test_encode_too_big_payload() -> None
Test encode of a too big payload.
test_encode_too_big_transaction
def test_encode_too_big_transaction() -> None
Test encode of a too big transaction.
test_sign_verify_transaction
def test_sign_verify_transaction() -> None
Test sign/verify transaction.
test_payload_not_equal_lookalike
def test_payload_not_equal_lookalike() -> None
Test payload eq reflection via NotImplemented
test_transaction_not_equal_lookalike
def test_transaction_not_equal_lookalike() -> None
Test transaction eq reflection via NotImplemented
teardown_method
def teardown_method() -> None
Tear down the test.
test_verify_transaction_negative_case
@mock.patch("aea.crypto.ledger_apis.LedgerApis.recover_message",
return_value={"wrong_sender"})
def test_verify_transaction_negative_case(*_mocks: Any) -> None
Test verify() of transaction, negative case.
SomeClass Objects
@dataclass(frozen=True)
class SomeClass(BaseTxPayload)
Test class.
test_payload_serializer_is_deterministic
@given(
dictionaries(
keys=text(),
values=one_of(floats(allow_nan=False, allow_infinity=False),
booleans()),
))
def test_payload_serializer_is_deterministic(obj: Any) -> None
Test that 'DictProtobufStructSerializer' is deterministic.
test_initialize_block
def test_initialize_block() -> None
Test instantiation of a Block instance.
TestBlockchain Objects
class TestBlockchain()
Test a blockchain object.
setup_method
def setup_method() -> None
Set up the test.
test_height
def test_height() -> None
Test the 'height' property getter.
test_len
def test_len() -> None
Test the 'length' property getter.
test_add_block_positive
def test_add_block_positive() -> None
Test 'add_block', success.
test_add_block_negative_wrong_height
def test_add_block_negative_wrong_height() -> None
Test 'add_block', wrong height.
test_add_block_before_initial_height
def test_add_block_before_initial_height() -> None
Test 'add_block', too old height.
test_blocks
def test_blocks() -> None
Test 'blocks' property getter.
TestBlockBuilder Objects
class TestBlockBuilder()
Test block builder.
setup_method
def setup_method() -> None
Set up the method.
test_get_header_positive
def test_get_header_positive() -> None
Test header property getter, positive.
test_get_header_negative
def test_get_header_negative() -> None
Test header property getter, negative.
test_set_header_positive
def test_set_header_positive() -> None
Test header property setter, positive.
test_set_header_negative
def test_set_header_negative() -> None
Test header property getter, negative.
test_transitions_getter
def test_transitions_getter() -> None
Test 'transitions' property getter.
test_add_transitions
def test_add_transitions() -> None
Test 'add_transition'.
test_get_block_negative_header_not_set_yet
def test_get_block_negative_header_not_set_yet() -> None
Test 'get_block', negative case (header not set yet).
test_get_block_positive
def test_get_block_positive() -> None
Test 'get_block', positive case.
TestAbciAppDB Objects
class TestAbciAppDB()
Test 'AbciAppDB' class.
setup_method
def setup_method() -> None
Set up the tests.
test_init
@pytest.mark.parametrize(
"data, setup_data",
(
({
"participants": ["a", "b"]
}, {
"participants": ["a", "b"]
}),
({
"participants": []
}, {}),
({
"participants": None
}, None),
("participants", None),
(1, None),
(object(), None),
(["participants"], None),
({
"participants": [],
"other": [1, 2]
}, {
"other": [1, 2]
}),
),
)
@pytest.mark.parametrize(
"cross_period_persisted_keys, expected_cross_period_persisted_keys",
((None, set()), (set(), set()), ({"test"}, {"test"})),
)
def test_init(data: Dict, setup_data: Optional[Dict],
cross_period_persisted_keys: Optional[Set[str]],
expected_cross_period_persisted_keys: Set[str]) -> None
Test constructor.
EnumTest Objects
class EnumTest(Enum)
A test Enum class
test_normalize
@pytest.mark.parametrize(
"data_in, expected_output",
(
(0, 0),
([], []),
({
"test": 2
}, {
"test": 2
}),
(EnumTest.test, 10),
(b"test", b"test".hex()),
({3, 4}, "[3, 4]"),
(object(), None),
),
)
def test_normalize(data_in: Any, expected_output: Any) -> None
Test normalize.
test_reset_index
@pytest.mark.parametrize("data", {0: [{"test": 2}]})
def test_reset_index(data: Dict) -> None
Test reset_index.
test_round_count_setter
def test_round_count_setter() -> None
Tests the round count setter.
test_try_alter_init_data
def test_try_alter_init_data() -> None
Test trying to alter the init data.
test_cross_period_persisted_keys
def test_cross_period_persisted_keys() -> None
Test cross_period_persisted_keys property
test_get
def test_get() -> None
Test getters.
test_increment_round_count
def test_increment_round_count() -> None
Test increment_round_count.
test_validate
@mock.patch.object(
abci_base,
"is_json_serializable",
return_value=False,
)
def test_validate(_: mock._patch) -> None
Test validate method.
test_update
@pytest.mark.parametrize(
"setup_data, update_data, expected_data",
(
(dict(), {
"dummy_key": "dummy_value"
}, {
0: {
"dummy_key": ["dummy_value"]
}
}),
(
dict(),
{
"dummy_key": ["dummy_value1", "dummy_value2"]
},
{
0: {
"dummy_key": [["dummy_value1", "dummy_value2"]]
}
},
),
(
{
"test": ["test"]
},
{
"dummy_key": "dummy_value"
},
{
0: {
"dummy_key": ["dummy_value"],
"test": ["test"]
}
},
),
(
{
"test": ["test"]
},
{
"test": "dummy_value"
},
{
0: {
"test": ["test", "dummy_value"]
}
},
),
(
{
"test": [["test"]]
},
{
"test": ["dummy_value1", "dummy_value2"]
},
{
0: {
"test": [["test"], ["dummy_value1", "dummy_value2"]]
}
},
),
(
{
"test": ["test"]
},
{
"test": ["dummy_value1", "dummy_value2"]
},
{
0: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
}
},
),
),
)
def test_update(setup_data: Dict, update_data: Dict,
expected_data: Dict[int, Dict]) -> None
Test update db.
test_create
@pytest.mark.parametrize(
"replacement_value, expected_replacement",
(
(132, 132),
("test", "test"),
(set("132"), ("1", "2", "3")),
({"132"}, ("132", )),
(frozenset("231"), ("1", "2", "3")),
(frozenset({"231"}), ("231", )),
(("1", "3", "2"), ("1", "3", "2")),
(["1", "5", "3"], ["1", "5", "3"]),
),
)
@pytest.mark.parametrize(
"setup_data, cross_period_persisted_keys",
(
(dict(), frozenset()),
({
"test": [["test"]]
}, frozenset()),
({
"test": [["test"]]
}, frozenset({"test"})),
({
"test": ["test"]
}, frozenset({"test"})),
),
)
def test_create(replacement_value: Any, expected_replacement: Any,
setup_data: Dict,
cross_period_persisted_keys: FrozenSet[str]) -> None
Test create db.
test_create_key_not_in_db
def test_create_key_not_in_db() -> None
Test the create method when a given or a cross-period key does not exist in the db.
test_cleanup
@pytest.mark.parametrize(
"existing_data, cleanup_history_depth, cleanup_history_depth_current, expected",
(
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
}
},
0,
None,
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
}
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": [0]
},
},
0,
None,
{
2: {
"test": [0]
}
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": [0, 1, 2]
},
},
0,
0,
{
2: {
"test": [0, 1, 2]
}
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": [0, 1, 2]
},
},
0,
1,
{
2: {
"test": [2]
}
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": list(range(5))
},
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15, 20))
},
},
3,
0,
{
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15, 20))
},
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": list(range(5))
},
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15, 20))
},
},
5,
3,
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": list(range(5))
},
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15 + 2, 20))
},
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": list(range(5))
},
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15, 20))
},
},
2,
3,
{
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15 + 2, 20))
},
},
),
(
{
1: {
"test": ["test", ["dummy_value1", "dummy_value2"]]
},
2: {
"test": list(range(5))
},
3: {
"test": list(range(5, 10))
},
4: {
"test": list(range(10, 15))
},
5: {
"test": list(range(15, 20))
},
},
0,
1,
{
5: {
"test": [19]
},
},
),
),
)
def test_cleanup(existing_data: Dict[int, Dict[str, List[Any]]],
cleanup_history_depth: int,
cleanup_history_depth_current: Optional[int],
expected: Dict[int, Dict[str, List[Any]]]) -> None
Test cleanup db.
test_serialize
def test_serialize() -> None
Test serialize method.
test_sync
@pytest.mark.parametrize(
"_data",
({
"db_data": {
0: {
"test": [0]
}
},
"slashing_config": "serialized_config"
}, ),
)
def test_sync(_data: Dict[str, Dict[int, Dict[str, List[Any]]]]) -> None
Test sync method.
test_sync_incorrect_data
@pytest.mark.parametrize(
"serialized_data, match",
(
(b"", "Could not decode data using "),
(
json.dumps({"both_mandatory_keys_missing": {}}),
"internal error: Mandatory keys `db_data`, `slashing_config` are missing from the deserialized data: "
"{'both_mandatory_keys_missing': {}}\nThe following serialized data were given: "
'{"both_mandatory_keys_missing": {}}',
),
(
json.dumps({"db_data": {}}),
"internal error: Mandatory keys `db_data`, `slashing_config` are missing from the deserialized data: "
"{'db_data': {}}\nThe following serialized data were given: {\"db_data\": {}}",
),
(
json.dumps({"slashing_config": {}}),
"internal error: Mandatory keys `db_data`, `slashing_config` are missing from the deserialized data: "
"{'slashing_config': {}}\nThe following serialized data were given: {\"slashing_config\": {}}",
),
(
json.dumps({
"db_data": {
"invalid_index": {}
},
"slashing_config": "anything"
}),
"An invalid index was found while trying to sync the db using data: ",
),
(
json.dumps({
"db_data": "invalid",
"slashing_config": "anything"
}),
"Could not decode db data with an invalid format: ",
),
),
)
def test_sync_incorrect_data(serialized_data: Any, match: str) -> None
Test sync method with incorrect data.
test_hash
def test_hash() -> None
Test hash method.
TestBaseSynchronizedData Objects
class TestBaseSynchronizedData()
Test 'BaseSynchronizedData' class.
setup_method
def setup_method() -> None
Set up the tests.
test_slashing_config
@given(text())
def test_slashing_config(slashing_config: str) -> None
Test the slashing_config property.
test_participants_getter_positive
def test_participants_getter_positive() -> None
Test 'participants' property getter.
test_nb_participants_getter
def test_nb_participants_getter() -> None
Test 'participants' property getter.
test_participants_getter_negative
def test_participants_getter_negative() -> None
Test 'participants' property getter, negative case.
test_update
def test_update() -> None
Test the 'update' method.
test_create
def test_create() -> None
Test the 'create' method.
test_repr
def test_repr() -> None
Test the 'repr' magic method.
test_participants_list_is_empty
def test_participants_list_is_empty() -> None
Tets when participants list is set to zero.
test_all_participants_list_is_empty
def test_all_participants_list_is_empty() -> None
Tets when participants list is set to zero.
test_consensus_threshold
@pytest.mark.parametrize(
"n_participants, given_threshold, expected_threshold",
(
(1, None, 1),
(5, None, 4),
(10, None, 7),
(345, None, 231),
(246236, None, 164158),
(1, 1, 1),
(5, 5, 5),
(10, 7, 7),
(10, 8, 8),
(10, 9, 9),
(10, 10, 10),
(345, 300, 300),
(246236, 194158, 194158),
),
)
def test_consensus_threshold(n_participants: int, given_threshold: int,
expected_threshold: int) -> None
Test the consensus_threshold property.
test_consensus_threshold_incorrect
@pytest.mark.parametrize(
"n_participants, given_threshold",
(
(1, 2),
(5, 2),
(10, 4),
(10, 11),
(10, 18),
(345, 200),
(246236, 164157),
(246236, 246237),
),
)
def test_consensus_threshold_incorrect(n_participants: int,
given_threshold: int) -> None
Test the consensus_threshold property when an incorrect threshold value has been inserted to the db.
test_properties
def test_properties() -> None
Test several properties
DummyConcreteRound Objects
class DummyConcreteRound(AbstractRound)
A dummy concrete round's implementation.
end_block
def end_block() -> Optional[Tuple[BaseSynchronizedData, EventType]]
A dummy end_block implementation.
check_payload
def check_payload(payload: BaseTxPayload) -> None
A dummy check_payload implementation.
process_payload
def process_payload(payload: BaseTxPayload) -> None
A dummy process_payload implementation.
check_majority_possible_with_new_vote
def check_majority_possible_with_new_vote(
votes_by_participant: Dict[str, BaseTxPayload],
new_voter: str,
new_vote: BaseTxPayload,
nb_participants: int,
exception_cls: Type[ABCIAppException] = ABCIAppException) -> None
A dummy implementation for testing.
TestAbstractRound Objects
class TestAbstractRound()
Test the 'AbstractRound' class.
setup_method
def setup_method() -> None
Set up the tests.
test_auto_round_id
def test_auto_round_id() -> None
Test that the 'auto_round_id()' method works as expected.
test_must_not_set_round_id
def test_must_not_set_round_id() -> None
Test that the 'round_id' must be set in concrete classes.
test_must_set_payload_class_type
def test_must_set_payload_class_type() -> None
Test that the 'payload_class' must be set in concrete classes.
test_check_payload_type_with_previous_round_transaction
def test_check_payload_type_with_previous_round_transaction() -> None
Test check 'check_payload_type'.
test_check_payload_type
def test_check_payload_type() -> None
Test check 'check_payload_type'.
test_synchronized_data_getter
def test_synchronized_data_getter() -> None
Test 'synchronized_data' property getter.
test_check_transaction_unknown_payload
def test_check_transaction_unknown_payload() -> None
Test 'check_transaction' method, with unknown payload type.
test_check_transaction_known_payload
def test_check_transaction_known_payload() -> None
Test 'check_transaction' method, with known payload type.
test_process_transaction_negative_unknown_payload
def test_process_transaction_negative_unknown_payload() -> None
Test 'process_transaction' method, with unknown payload type.
test_process_transaction_negative_check_transaction_fails
def test_process_transaction_negative_check_transaction_fails() -> None
Test 'process_transaction' method, with 'check_transaction' failing.
test_process_transaction_positive
def test_process_transaction_positive() -> None
Test 'process_transaction' method, positive case.
test_check_majority_possible_raises_error_when_nb_participants_is_0
def test_check_majority_possible_raises_error_when_nb_participants_is_0(
) -> None
Check that 'check_majority_possible' raises error when nb_participants=0.
test_check_majority_possible_passes_when_vote_set_is_empty
def test_check_majority_possible_passes_when_vote_set_is_empty() -> None
Check that 'check_majority_possible' passes when the set of votes is empty.
test_check_majority_possible_passes_when_vote_set_nonempty_and_check_passes
def test_check_majority_possible_passes_when_vote_set_nonempty_and_check_passes(
) -> None
Check that 'check_majority_possible' passes when set of votes is non-empty.
The check passes because: - the threshold is 2 - the other voter can vote for the same item of the first voter
test_check_majority_possible_passes_when_payload_attributes_majority_match
def test_check_majority_possible_passes_when_payload_attributes_majority_match(
) -> None
Test 'check_majority_possible' when set of votes is non-empty and the majority of the attribute values match.
The check passes because: - the threshold is 3 (participants are 4) - 3 voters have the same attribute value in their payload
test_check_majority_possible_passes_when_vote_set_nonempty_and_check_doesnt_pass
def test_check_majority_possible_passes_when_vote_set_nonempty_and_check_doesnt_pass(
) -> None
Check that 'check_majority_possible' doesn't pass when set of votes is non-empty.
the check does not pass because: - the threshold is 2 - both voters have already voted for different items
test_is_majority_possible_positive_case
def test_is_majority_possible_positive_case() -> None
Test 'is_majority_possible', positive case.
test_is_majority_possible_negative_case
def test_is_majority_possible_negative_case() -> None
Test 'is_majority_possible', negative case.
test_check_majority_possible_raises_error_when_new_voter_already_voted
def test_check_majority_possible_raises_error_when_new_voter_already_voted(
) -> None
Test 'check_majority_possible_with_new_vote' raises when new voter already voted.
test_check_majority_possible_raises_error_when_nb_participants_inconsistent
def test_check_majority_possible_raises_error_when_nb_participants_inconsistent(
) -> None
Test 'check_majority_possible_with_new_vote' raises when 'nb_participants' inconsistent with other args.
test_check_majority_possible_when_check_passes
def test_check_majority_possible_when_check_passes() -> None
Test 'check_majority_possible_with_new_vote' when the check passes.
The test passes because: - the number of participants is 2, and so the threshold is 2 - the new voter votes for the same item already voted by voter 1.
TestTimeouts Objects
class TestTimeouts()
Test the 'Timeouts' class.
setup_method
def setup_method() -> None
Set up the test.
test_size
def test_size() -> None
Test the 'size' property.
test_add_timeout
def test_add_timeout() -> None
Test the 'add_timeout' method.
test_cancel_timeout
def test_cancel_timeout() -> None
Test the 'cancel_timeout' method.
test_pop_earliest_cancelled_timeouts
def test_pop_earliest_cancelled_timeouts() -> None
Test the 'pop_earliest_cancelled_timeouts' method.
test_get_earliest_timeout_a
def test_get_earliest_timeout_a() -> None
Test the 'get_earliest_timeout' method.
test_get_earliest_timeout_b
def test_get_earliest_timeout_b() -> None
Test the 'get_earliest_timeout' method.
test_pop_timeout
def test_pop_timeout() -> None
Test the 'pop_timeout' method.
TestAbciApp Objects
class TestAbciApp()
Test the 'AbciApp' class.
setup_method
def setup_method() -> None
Set up the test.
teardown_method
def teardown_method() -> None
Teardown the test.
test_is_abstract
@pytest.mark.parametrize("flag", (True, False))
def test_is_abstract(flag: bool) -> None
Test is_abstract property.
test_initial_round_cls_not_set
def test_initial_round_cls_not_set() -> None
Test when 'initial_round_cls' is not set.
test_transition_function_not_set
def test_transition_function_not_set() -> None
Test when 'transition_function' is not set.
test_last_timestamp_negative
def test_last_timestamp_negative() -> None
Test the 'last_timestamp' property, negative case.
test_last_timestamp_positive
def test_last_timestamp_positive() -> None
Test the 'last_timestamp' property, positive case.
test_get_synced_value
@pytest.mark.parametrize(
"db_key, sync_classes, default, property_found",
(
("", set(), "default", False),
("non_existing_key", {BaseSynchronizedData}, True, False),
("participants", {BaseSynchronizedData}, {}, False),
("is_keeper_set", {BaseSynchronizedData}, True, True),
),
)
def test_get_synced_value(db_key: str,
sync_classes: Set[Type[BaseSynchronizedData]],
default: Any, property_found: bool) -> None
Test the _get_synced_value method.
test_process_event
def test_process_event() -> None
Test the 'process_event' method, positive case, with timeout events.
test_process_event_negative_case
def test_process_event_negative_case() -> None
Test the 'process_event' method, negative case.
test_update_time
def test_update_time() -> None
Test the 'update_time' method.
test_get_all_events
def test_get_all_events() -> None
Test the all events getter.
test_get_all_rounds_classes
@pytest.mark.parametrize("include_background_rounds", (True, False))
def test_get_all_rounds_classes(include_background_rounds: bool) -> None
Test the get all rounds getter.
test_get_all_rounds_classes_bg_ever_running
def test_get_all_rounds_classes_bg_ever_running() -> None
Test the get all rounds when the background round is of an ever running type.
test_add_background_app
def test_add_background_app() -> None
Tests the add method for the background apps.
test_bg_apps_prioritized_independent_groups
def test_bg_apps_prioritized_independent_groups() -> None
Test that bg_apps_prioritized returns independent lists per group.
test_cleanup
def test_cleanup() -> None
Test the cleanup method.
test_check_transaction_for_termination_round
@mock.patch.object(ConcreteBackgroundRound, "check_transaction")
@pytest.mark.parametrize(
"transaction",
[mock.MagicMock(payload=DUMMY_CONCRETE_BACKGROUND_PAYLOAD)],
)
def test_check_transaction_for_termination_round(
check_transaction_mock: mock.Mock, transaction: Transaction) -> None
Tests process_transaction when it's a transaction meant for the termination app.
test_process_transaction_for_termination_round
@mock.patch.object(ConcreteBackgroundRound, "process_transaction")
@pytest.mark.parametrize(
"transaction",
[mock.MagicMock(payload=DUMMY_CONCRETE_BACKGROUND_PAYLOAD)],
)
def test_process_transaction_for_termination_round(
process_transaction_mock: mock.Mock, transaction: Transaction) -> None
Tests process_transaction when it's a transaction meant for the termination app.
TestOffenceTypeFns Objects
class TestOffenceTypeFns()
Test OffenceType-related functions.
test_light_offences
@staticmethod
def test_light_offences() -> None
Test light_offences function.
test_serious_offences
@staticmethod
def test_serious_offences() -> None
Test serious_offences function.
availability_window_data
@composite
def availability_window_data(draw: DrawFn) -> Dict[str, int]
A strategy for building valid availability window data.
TestAvailabilityWindow Objects
class TestAvailabilityWindow()
Test AvailabilityWindow.
test_not_equal
@staticmethod
@given(integers(min_value=1, max_value=100))
def test_not_equal(max_length: int) -> None
Test the add method.
test_add
@staticmethod
@given(integers(min_value=0, max_value=100), data())
def test_add(max_length: int, hypothesis_data: Any) -> None
Test the add method.
test_to_dict
@staticmethod
@given(
max_length=integers(min_value=1, max_value=30_000),
num_positive=integers(min_value=0),
num_negative=integers(min_value=0),
)
@pytest.mark.parametrize(
"window, expected_serialization",
(
(deque(()), 0),
(deque((False, False, False)), 0),
(deque((True, False, True)), 5),
(deque((True for _ in range(3))), 7),
(
deque((True for _ in range(1000))),
int("10715086071862673209484250490600018105614048117055336074437503883703510511249361224931983788156958"
"58127594672917553146825187145285692314043598457757469857480393456777482423098542107460506237114187"
"79541821530464749835819412673987675591655439460770629145711964776865421676604298316526243868372056"
"68069375"),
),
),
)
def test_to_dict(max_length: int, num_positive: int, num_negative: int,
window: Deque, expected_serialization: int) -> None
Test to_dict method.
test_validate_key
@staticmethod
@pytest.mark.parametrize(
"data_, key, validator, expected_error",
(
({
"a": 1,
"b": 2,
"c": 3
}, "a", lambda x: x > 0, None),
(
{
"a": 1,
"b": 2,
"c": 3
},
"d",
lambda x: x > 0,
r"Missing required key: d\.",
),
(
{
"a": "1",
"b": 2,
"c": 3
},
"a",
lambda x: x > 0,
r"a must be of type int\.",
),
(
{
"a": -1,
"b": 2,
"c": 3
},
"a",
lambda x: x > 0,
r"a has invalid value -1\.",
),
),
)
def test_validate_key(data_: dict, key: str, validator: Callable,
expected_error: Optional[str]) -> None
Test the _validate_key method.
test_validate_negative
@staticmethod
@pytest.mark.parametrize(
"data_, error_regex",
(
("not a dict", r"Expected dict, got"),
(
{
"max_length": -1,
"array": 42,
"num_positive": 10,
"num_negative": 0
},
r"max_length",
),
(
{
"max_length": 2,
"array": 4,
"num_positive": 10,
"num_negative": 0
},
r"array",
),
(
{
"max_length": 8,
"array": 42,
"num_positive": -1,
"num_negative": 0
},
r"num_positive",
),
(
{
"max_length": 8,
"array": 42,
"num_positive": 10,
"num_negative": -1
},
r"num_negative",
),
),
)
def test_validate_negative(data_: dict, error_regex: str) -> None
Negative tests for the _validate method.
test_validate_positive
@staticmethod
@given(availability_window_data())
def test_validate_positive(data_: Dict[str, int]) -> None
Positive tests for the _validate method.
test_from_dict
@staticmethod
@given(availability_window_data())
def test_from_dict(data_: Dict[str, int]) -> None
Test from_dict method.
test_to_dict_and_back
@staticmethod
@given(availability_window_data())
def test_to_dict_and_back(data_: Dict[str, int]) -> None
Test that the from_dict produces an object that generates the input data again when calling to_dict.
TestOffenceStatus Objects
class TestOffenceStatus()
Test the OffenceStatus dataclass.
test_slash_amount
@staticmethod
@pytest.mark.parametrize("custom_amount", (0, 5))
@pytest.mark.parametrize("light_unit_amount, serious_unit_amount", ((1, 2), ))
@pytest.mark.parametrize(
"validator_downtime, invalid_payload, blacklisted, suspected, "
"num_unknown_offenses, num_double_signed, num_light_client_attack, expected",
(
(False, False, False, False, 0, 0, 0, 0),
(True, False, False, False, 0, 0, 0, 1),
(False, True, False, False, 0, 0, 0, 1),
(False, False, True, False, 0, 0, 0, 1),
(False, False, False, True, 0, 0, 0, 1),
(False, False, False, False, 1, 0, 0, 2),
(False, False, False, False, 0, 1, 0, 2),
(False, False, False, False, 0, 0, 1, 2),
(False, False, False, False, 0, 2, 1, 6),
(False, True, False, True, 5, 2, 1, 18),
(True, True, True, True, 5, 2, 1, 20),
),
)
def test_slash_amount(custom_amount: int, light_unit_amount: int,
serious_unit_amount: int, validator_downtime: bool,
invalid_payload: bool, blacklisted: bool,
suspected: bool, num_unknown_offenses: int,
num_double_signed: int, num_light_client_attack: int,
expected: int) -> None
Test the slash_amount method.
offence_tracking
@composite
def offence_tracking(draw: DrawFn) -> Tuple[Evidences, LastCommitInfo]
A strategy for building offences reported by Tendermint.
offence_status
@composite
def offence_status(draw: DrawFn) -> OffenceStatus
Build an offence status instance.
TestOffenseStatusEncoderDecoder Objects
class TestOffenseStatusEncoderDecoder()
Test the OffenseStatusEncoder and the OffenseStatusDecoder.
test_encode_decode_offense_status
@staticmethod
@given(dictionaries(keys=text(), values=offence_status(), min_size=1))
def test_encode_decode_offense_status(offense_status: str) -> None
Test encoding an offense status mapping and then decoding it by using the custom encoder/decoder.
test_encode_unknown
def test_encode_unknown() -> None
Test the encoder with an unknown input.
TestRoundSequence Objects
class TestRoundSequence()
Test the RoundSequence class.
setup_method
def setup_method() -> None
Set up the test.
test_slashing_properties
@pytest.mark.parametrize(
"property_name, set_twice_exc, config_exc",
((
"validator_to_agent",
"The mapping of the validators' addresses to their agent addresses can only be set once. "
"Attempted to set with {new_content_attempt} but it has content already: {value}.",
"The mapping of the validators' addresses to their agent addresses has not been set.",
), ),
)
@given(data())
def test_slashing_properties(property_name: str, set_twice_exc: str,
config_exc: str, _data: Any) -> None
Test validator_to_agent getter and setter.
test_sync_db_and_slashing
@mock.patch("json.loads", return_value="json_serializable")
@pytest.mark.parametrize("slashing_config", (None, "", "test"))
def test_sync_db_and_slashing(mock_loads: mock.MagicMock,
slashing_config: str) -> None
Test the sync_db_and_slashing method.
test_store_offence_status
@mock.patch("json.dumps")
@pytest.mark.parametrize("slashing_enabled", (True, False))
def test_store_offence_status(mock_dumps: mock.MagicMock,
slashing_enabled: bool) -> None
Test the store_offence_status method.
test_get_agent_address
@given(
validator=builds(Validator, address=binary(), power=integers()),
agent_address=text(),
)
def test_get_agent_address(validator: Validator, agent_address: str) -> None
Test get_agent_address method.
test_height
@pytest.mark.parametrize("offset", tuple(range(5)))
@pytest.mark.parametrize("n_blocks", (0, 1, 10))
def test_height(n_blocks: int, offset: int) -> None
Test 'height' property.
test_is_finished
def test_is_finished() -> None
Test 'is_finished' property.
test_last_round
def test_last_round() -> None
Test 'last_round' property.
test_last_timestamp_none
def test_last_timestamp_none() -> None
Test 'last_timestamp' property.
The property is None because there are no blocks.
test_last_timestamp
def test_last_timestamp() -> None
Test 'last_timestamp' property, positive case.
test_abci_app_negative
def test_abci_app_negative() -> None
Test 'abci_app' property, negative case.
test_check_is_finished_negative
def test_check_is_finished_negative() -> None
Test 'check_is_finished', negative case.
test_current_round_positive
def test_current_round_positive() -> None
Test 'current_round' property getter, positive case.
test_current_round_negative_current_round_not_set
def test_current_round_negative_current_round_not_set() -> None
Test 'current_round' property getter, negative case (current round not set).
test_current_round_id
def test_current_round_id() -> None
Test 'current_round_id' property getter
test_latest_result
def test_latest_result() -> None
Test 'latest_result' property getter.
test_last_round_transition_timestamp
@pytest.mark.parametrize("committed", (True, False))
def test_last_round_transition_timestamp(committed: bool) -> None
Test 'last_round_transition_timestamp' method.
test_last_round_transition_height
@pytest.mark.parametrize("committed", (True, False))
def test_last_round_transition_height(committed: bool) -> None
Test 'last_round_transition_height' method.
test_block_before_blockchain_is_init
def test_block_before_blockchain_is_init(caplog: LogCaptureFixture) -> None
Test block received before blockchain initialized.
test_last_round_transition_root_hash
@pytest.mark.parametrize("last_round_transition_root_hash", (b"", b"test"))
def test_last_round_transition_root_hash(
last_round_transition_root_hash: bytes) -> None
Test 'last_round_transition_root_hash' method.
test_last_round_transition_tm_height
@pytest.mark.parametrize("tm_height", (None, 1, 5))
def test_last_round_transition_tm_height(tm_height: Optional[int]) -> None
Test 'last_round_transition_tm_height' method.
test_tm_height
@given(one_of(none(), integers()))
def test_tm_height(tm_height: int) -> None
Test tm_height getter and setter.
test_block_stall_deadline_expired
@given(one_of(none(), datetimes()))
def test_block_stall_deadline_expired(
block_stall_deadline: datetime.datetime) -> None
Test 'block_stall_deadline_expired' method.
test_init_chain
@pytest.mark.parametrize("begin_height", tuple(range(0, 50, 10)))
@pytest.mark.parametrize("initial_height", tuple(range(0, 11, 5)))
def test_init_chain(begin_height: int, initial_height: int) -> None
Test 'init_chain' method.
test_track_tm_offences
@given(offence_tracking())
@settings(suppress_health_check=[HealthCheck.too_slow])
def test_track_tm_offences(offences: Tuple[Evidences, LastCommitInfo]) -> None
Test _track_tm_offences method.
test_track_app_offences
@mock.patch.object(abci_base, "ADDRESS_LENGTH", len("agent_i"))
def test_track_app_offences() -> None
Test _track_app_offences method.
test_handle_slashing_not_configured
@given(builds(SlashingNotConfiguredError, text()))
def test_handle_slashing_not_configured(
exc: SlashingNotConfiguredError) -> None
Test _handle_slashing_not_configured method.
test_try_track_offences
@pytest.mark.parametrize("_track_offences_raises", (True, False))
def test_try_track_offences(_track_offences_raises: bool) -> None
Test _try_track_offences method.
test_begin_block_negative_is_finished
def test_begin_block_negative_is_finished() -> None
Test 'begin_block' method, negative case (round sequence is finished).
test_begin_block_negative_wrong_phase
def test_begin_block_negative_wrong_phase() -> None
Test 'begin_block' method, negative case (wrong phase).
test_begin_block_positive
def test_begin_block_positive() -> None
Test 'begin_block' method, positive case.
test_deliver_tx_negative_wrong_phase
def test_deliver_tx_negative_wrong_phase() -> None
Test 'begin_block' method, negative (wrong phase).
test_deliver_tx_positive_not_valid
def test_deliver_tx_positive_not_valid() -> None
Test 'begin_block' method, positive (not valid).
test_end_block_negative_wrong_phase
def test_end_block_negative_wrong_phase() -> None
Test 'end_block' method, negative case (wrong phase).
test_end_block_positive
def test_end_block_positive() -> None
Test 'end_block' method, positive case.
test_commit_negative_wrong_phase
def test_commit_negative_wrong_phase() -> None
Test 'end_block' method, negative case (wrong phase).
test_commit_negative_exception
def test_commit_negative_exception() -> None
Test 'end_block' method, negative case (raise exception).
test_commit_positive_no_change_round
def test_commit_positive_no_change_round() -> None
Test 'end_block' method, positive (no change round).
test_commit_positive_with_change_round
def test_commit_positive_with_change_round() -> None
Test 'end_block' method, positive (with change round).
test_reset_blockchain
@pytest.mark.parametrize("is_replay", (True, False))
def test_reset_blockchain(is_replay: bool) -> None
Test reset_blockchain method.
last_round_values_updated
def last_round_values_updated(any_: bool = True) -> bool
Check if the values for the last round-related attributes have been updated.
test_update_round
@mock.patch.object(AbciApp, "process_event")
@mock.patch.object(RoundSequence, "serialized_offence_status")
@pytest.mark.parametrize("end_block_res", (None, (MagicMock(), MagicMock())))
@pytest.mark.parametrize(
"slashing_enabled, offence_status_",
(
(
False,
False,
),
(
False,
True,
),
(
False,
False,
),
(
True,
True,
),
),
)
def test_update_round(serialized_offence_status_mock: mock.Mock,
process_event_mock: mock.Mock,
end_block_res: Optional[Tuple[BaseSynchronizedData,
Any]],
slashing_enabled: bool, offence_status_: dict) -> None
Test '_update_round' method.
test_update_round_when_termination_returns
@mock.patch.object(AbciApp, "process_event")
@pytest.mark.parametrize(
"termination_round_result, current_round_result",
[
(None, None),
(None, (MagicMock(), MagicMock())),
((MagicMock(), MagicMock()), None),
((MagicMock(), MagicMock()), (MagicMock(), MagicMock())),
],
)
def test_update_round_when_termination_returns(
process_event_mock: mock.Mock,
termination_round_result: Optional[Tuple[BaseSynchronizedData, Any]],
current_round_result: Optional[Tuple[BaseSynchronizedData,
Any]]) -> None
Test '_update_round' method.
test_reset_state
@pytest.mark.parametrize("restart_from_round", (ConcreteRoundA, MagicMock()))
@pytest.mark.parametrize("serialized_db_state", (None, "serialized state"))
@given(integers())
def test_reset_state(restart_from_round: AbstractRound,
serialized_db_state: str, round_count: int) -> None
Tests reset_state
test_reset_to_default_params
def test_reset_to_default_params() -> None
Tests _reset_to_default_params.
test_add_pending_offence
def test_add_pending_offence() -> None
Tests add_pending_offence.
test_meta_abci_app_when_instance_not_subclass_of_abstract_round
def test_meta_abci_app_when_instance_not_subclass_of_abstract_round() -> None
Test instantiation of meta-class when instance not a subclass of AbciApp.
Since the class is not a subclass of AbciApp, the checks performed by the meta-class should not apply.
test_meta_abci_app_when_final_round_not_subclass_of_degenerate_round
def test_meta_abci_app_when_final_round_not_subclass_of_degenerate_round(
) -> None
Test instantiation of meta-class when a final round is not a subclass of DegenerateRound.
test_synchronized_data_type_on_abci_app_init
def test_synchronized_data_type_on_abci_app_init(
caplog: LogCaptureFixture) -> None
Test synchronized data access
test_get_name
def test_get_name() -> None
Test the get_name method.
test_pending_offences_payload
@pytest.mark.parametrize(
"sender, accused_agent_address, offense_round, offense_type_value, last_transition_timestamp, time_to_live, custom_amount",
((
"sender",
"test_address",
90,
3,
10,
2,
10,
), ),
)
def test_pending_offences_payload(sender: str, accused_agent_address: str,
offense_round: int, offense_type_value: int,
last_transition_timestamp: int,
time_to_live: int,
custom_amount: int) -> None
Test PendingOffencesPayload
TestPendingOffencesRound Objects
class TestPendingOffencesRound(BaseRoundTestClass)
Tests for PendingOffencesRound.
test_run
@given(
accused_agent_address=sampled_from(list(get_participants())),
offense_round=integers(min_value=0),
offense_type_value=sampled_from(
[value.value for value in OffenseType.__members__.values()]),
last_transition_timestamp=floats(
min_value=timegm(datetime.datetime(1971, 1, 1).utctimetuple()),
max_value=timegm(datetime.datetime(8000, 1, 1).utctimetuple()) - 2000,
),
time_to_live=floats(min_value=1, max_value=2000),
custom_amount=integers(min_value=0),
)
def test_run(accused_agent_address: str, offense_round: int,
offense_type_value: int, last_transition_timestamp: float,
time_to_live: float, custom_amount: int) -> None
Run tests.