packages.valory.skills.abstract_round_abci.tests.test_models

Test the models.py module of the skill.

TestApiSpecsModel Objects

class TestApiSpecsModel()

Test ApiSpecsModel.

setup_method

def setup_method() -> None

Setup test.

test_init

def test_init() -> None

Test initialization.

test_suggested_sleep_time

@pytest.mark.parametrize(
    "retries, expected",
    [
        (0, 1.0),
        (5, 32.0),
        (9, 512.0),
        (11, 2048.0),
        # 2**12 = 4096 > 3600 = MAX_SUGGESTED_SLEEP_TIME, so the cap engages
        (12, MAX_SUGGESTED_SLEEP_TIME),
        (50, MAX_SUGGESTED_SLEEP_TIME),
    ],
)
def test_suggested_sleep_time(retries: int, expected: float) -> None

Test that suggested_sleep_time is capped at MAX_SUGGESTED_SLEEP_TIME.

test_suggested_sleep_time_warning_fires_once_per_sequence

def test_suggested_sleep_time_warning_fires_once_per_sequence(
        caplog: pytest.LogCaptureFixture) -> None

The cap warning fires exactly once per retry sequence.

At retries=12 (the first retry where 2**retries > MAX) the warning fires once. Subsequent reads at higher retries within the same sequence do not refire. After reset_retries lowers the raw sleep time back under the cap, a new sequence can warn again.

test_retries

def test_retries() -> None

Tests for retries.

test_get_spec

def test_get_spec() -> None

Test get_spec method.

test_process_response

@pytest.mark.parametrize(
    "api_specs_config, message, expected_res, expected_error",
    (
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="value",
                response_index=None,
                response_type="float",
                error_key=None,
                error_index=None,
                error_data=None,
            ),
            MagicMock(body=b'{"value": "10.232"}', status_code=200),
            10.232,
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="test:response:key",
                response_index=2,
                response_type="dict",
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(
                body=
                b'{"test": {"response": {"key": ["does_not_matter", "does_not_matter", {"this": "matters"}]}}}',
                status_code=200,
            ),
            {
                "this": "matters"
            },
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="test:response:key",
                response_index=2,
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(body=b'{"cannot be parsed', status_code=200),
            None,
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="test:response:key",
                response_index=2,
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(
                # the null will raise `TypeError` and we test that it is handled
                body=
                b'{"test": {"response": {"key": ["does_not_matter", "does_not_matter", null]}}}',
                status_code=200,
            ),
            "None",
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="test:response:key",
                response_index=
                2,  # this will raise `IndexError` and we test that it is handled
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(
                body=
                b'{"test": {"response": {"key": ["does_not_matter", "does_not_matter"]}}}',
                status_code=200,
            ),
            None,
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key=
                "test:response:key",  # this will raise `KeyError` and we test that it is handled
                response_index=2,
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(
                body=
                b'{"test": {"response": {"key_does_not_match": ["does_not_matter", "does_not_matter"]}}}',
                status_code=200,
            ),
            None,
            None,
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="test:response:key",
                response_index=2,
                error_key="error:key",
                error_index=3,
                error_type="str",
                error_data=None,
            ),
            MagicMock(
                body=
                b'{"test": {"response": {"key_does_not_match": ["does_not_matter", "does_not_matter"]}}, '
                b'"error": {"key": [0, 1, 2, "test that the error is being parsed correctly"]}}',
                status_code=200,
            ),
            None,
            "test that the error is being parsed correctly",
        ),
    ),
)
def test_process_response(api_specs_config: dict, message: MagicMock,
                          expected_res: Any, expected_error: Any) -> None

Test process_response method.

test_process_response_returns_none_on_unparseable_body

@pytest.mark.parametrize(
    "api_specs_config, message",
    (
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="value",
                response_index=None,
                response_type="float",
                error_key=None,
                error_index=None,
                error_data=None,
            ),
            MagicMock(body=b"\xff\xfe\xfd", status_code=200),
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="value",
                response_index=None,
                response_type="int",
                error_key=None,
                error_index=None,
                error_data=None,
            ),
            MagicMock(body=b'{"value": "N/A"}', status_code=200),
        ),
        (
            dict(
                **BASE_DUMMY_SPECS_CONFIG,
                response_key="value",
                response_index=None,
                response_type="float",
                error_key=None,
                error_index=None,
                error_data=None,
            ),
            MagicMock(body=b'{"value": "null"}', status_code=200),
        ),
    ),
)
def test_process_response_returns_none_on_unparseable_body(
        api_specs_config: dict, message: MagicMock) -> None

Return None on bodies that cannot be decoded or coerced.

test_process_response_returns_none_on_server_error_status

@pytest.mark.parametrize(
    "status_code",
    (500, 502, 503, 504, 600),
)
def test_process_response_returns_none_on_server_error_status(
        status_code: int) -> None

5xx (including the http_client transport-failure synthetic 600) short-circuit JSON parse.

test_attribute_manipulation

def test_attribute_manipulation() -> None

Test manipulating the attributes.

ConcreteRound Objects

class ConcreteRound(AbstractRound)

A ConcreteRoundA for testing purposes.

end_block

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

Handle the end of the block.

SharedState Objects

class SharedState(BaseSharedState)

Shared State for testing purposes.

TestSharedState Objects

class TestSharedState()

Test SharedState(Model) class.

test_initialization

def test_initialization(*_: Any) -> None

Test the initialization of the shared state.

dummy_state_setup

@staticmethod
def dummy_state_setup(shared_state: SharedState) -> None

Setup a shared state instance with dummy params.

test_setup_slashing

@pytest.mark.parametrize(
    "acn_configured_agents, validator_to_agent, raises",
    (
        (
            {i
             for i in range(4)},
            {
                f"validator_address_{i}": i
                for i in range(4)
            },
            False,
        ),
        (
            {i
             for i in range(5)},
            {
                f"validator_address_{i}": i
                for i in range(4)
            },
            True,
        ),
        (
            {i
             for i in range(4)},
            {
                f"validator_address_{i}": i
                for i in range(5)
            },
            True,
        ),
    ),
)
def test_setup_slashing(acn_configured_agents: Set[str],
                        validator_to_agent: Dict[str,
                                                 str], raises: bool) -> None

Test the validator_to_agent properties.

test_setup

def test_setup(*_: Any) -> None

Test setup method.

test_get_validator_address

@pytest.mark.parametrize(
    "initial_tm_configs, address_input, exception, expected",
    (
        (
            {},
            "0x1",
            "The validator address of non-participating agent `0x1` was requested.",
            None,
        ),
        ({}, "0x0", "SharedState's setup was not performed successfully.",
         None),
        (
            {
                "0x0": None
            },
            "0x0",
            "ACN registration has not been successfully performed for agent `0x0`. "
            "Have you set the `share_tm_config_on_startup` flag to `true` in the configuration?",
            None,
        ),
        (
            {
                "0x0": {}
            },
            "0x0",
            "The tendermint configuration for agent `0x0` is invalid: `{}`.",
            None,
        ),
        (
            {
                "0x0": {
                    "address": None
                }
            },
            "0x0",
            "The tendermint configuration for agent `0x0` is invalid: `{'address': None}`.",
            None,
        ),
        (
            {
                "0x0": {
                    "address": "test_validator_address"
                }
            },
            "0x0",
            None,
            "test_validator_address",
        ),
    ),
)
def test_get_validator_address(initial_tm_configs: Dict[str,
                                                        Optional[Dict[str,
                                                                      Any]]],
                               address_input: str, exception: Optional[str],
                               expected: Optional[str]) -> None

Test get_validator_address method.

test_acn_container

@pytest.mark.parametrize("self_idx", (range(4)))
def test_acn_container(self_idx: int) -> None

Test the acn_container method.

test_synchronized_data_negative_not_available

def test_synchronized_data_negative_not_available(*_: Any) -> None

Test 'synchronized_data' property getter, negative case (not available).

test_synchronized_data_positive

def test_synchronized_data_positive(*_: Any) -> None

Test 'synchronized_data' property getter, negative case (not available).

test_synchronized_data_db

def test_synchronized_data_db(*_: Any) -> None

Test 'synchronized_data' AbciAppDB.

test_get_acn_result

@pytest.mark.parametrize(
    "address_to_acn_deliverable, n_participants, expected",
    (
        ({}, 4, None),
        ({
            i: "test"
            for i in range(4)
        }, 4, "test"),
        (
            {
                i: TendermintRecoveryParams("test")
                for i in range(4)
            },
            4,
            TendermintRecoveryParams("test"),
        ),
        ({
            1: "test",
            2: "non-matching",
            3: "test",
            4: "test"
        }, 4, "test"),
        ({
            i: "test"
            for i in range(4)
        }, 4, "test"),
        ({
            1: "no",
            2: "result",
            3: "matches",
            4: ""
        }, 4, None),
    ),
)
def test_get_acn_result(address_to_acn_deliverable: Dict[str, Any],
                        n_participants: int, expected: Optional[str]) -> None

Test get_acn_result.

test_recovery_params_on_init

def test_recovery_params_on_init() -> None

Test that tm_recovery_params get initialized correctly.

test_set_last_reset_params

def test_set_last_reset_params() -> None

Test that last_reset_params get set correctly.

TestBenchmarkTool Objects

class TestBenchmarkTool()

Test BenchmarkTool

test_end_2_end

def test_end_2_end() -> None

Test end 2 end of the tool.

test_requests_model_initialization

def test_requests_model_initialization() -> None

Test initialization of the 'Requests(Model)' class.

test_base_params_model_initialization

def test_base_params_model_initialization() -> None

Test initialization of the 'BaseParams(Model)' class.

test_incorrect_setup

@pytest.mark.parametrize(
    "setup, error_text",
    (
        ({}, "`setup` params contain no values!"),
        (
            {
                "a": "b"
            },
            "Value for `safe_contract_address` missing from the `setup` params.",
        ),
    ),
)
def test_incorrect_setup(setup: Dict[str, Any], error_text: str) -> None

Test BaseParams model initialization with incorrect setup data.

test_genesis_block

def test_genesis_block() -> None

Test genesis block methods.

test_genesis_evidence

def test_genesis_evidence() -> None

Test genesis evidence methods.

test_genesis_validator

def test_genesis_validator() -> None

Test genesis validator methods.

test_genesis_consensus_params

def test_genesis_consensus_params() -> None

Test genesis consensus params methods.

test_genesis_config

def test_genesis_config() -> None

Test genesis config methods.

test_meta_shared_state_when_instance_not_subclass_of_shared_state

def test_meta_shared_state_when_instance_not_subclass_of_shared_state(
) -> None

Test instantiation of meta class when instance not a subclass of shared state.

test_shared_state_instantiation_without_attributes_raises_error

def test_shared_state_instantiation_without_attributes_raises_error() -> None

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

A Objects

@dataclass
class A()

Class for testing.

B Objects

@dataclass
class B()

Class for testing.

C Objects

class C(TypedDict)

Class for testing.

D Objects

class D(TypedDict)

Class for testing.

test_type_check_positive

@pytest.mark.parametrize("name,value,type_hint", testdata_positive)
def test_type_check_positive(name: str, value: Any, type_hint: Any) -> None

Test the type check mixin.

test_type_check_negative

@pytest.mark.parametrize("name,value,type_hint", testdata_negative)
def test_type_check_negative(name: str, value: Any, type_hint: Any) -> None

Test the type check mixin.