Skip to content

plugins.aea-test-autonomy.aea_test_autonomy.base_test_classes.agents

End2end tests base class.

RoundChecks Objects

@dataclass
class RoundChecks()

Class for the necessary checks of a round during the tests.

name: is the name of the round for which the checks should be performed. event: is the name of the event that is considered as successful. n_periods: is the number of periods this event should appear for the check to be considered successful.

BaseTestEnd2End Objects

@pytest.mark.e2e

@pytest.mark.integration
class BaseTestEnd2End(AEATestCaseMany,  UseFlaskTendermintNode)

Base class for end-to-end tests of agents with a skill extending the abstract_abci_round skill.

The setup test function of this class will configure a set of 'n' agents with the configured (agent_package) agent, and a Tendermint network of 'n' nodes, one for each agent.

Test subclasses must set agent_package, wait_to_finish and check_strings.

set_config

@classmethod
def set_config(cls, dotted_path: str, value: Any, type_: Optional[str] = None, aev: bool = True) -> Result

Set config value.

prepare

def prepare(nb_nodes: int) -> None

Set up the agents.

prepare_and_launch

def prepare_and_launch(nb_nodes: int) -> None

Prepare and launch the agents.

terminate_processes

def terminate_processes() -> None

Terminate processes

missing_from_output

@classmethod
def missing_from_output(cls, happy_path: Tuple[RoundChecks, ...] = (), strict_check_strings: Tuple[str, ...] = (), period: int = 1, is_terminating: bool = True, **kwargs: Any, ,) -> Tuple[List[str], List[str]]

Check if strings are present in process output.

Read process stdout in thread and terminate when all strings are present or timeout expired.

Arguments:

  • happy_path: the happy path of the testing FSM.
  • strict_check_strings: tuple of strings expected to appear in output as is.
  • period: period of checking.
  • is_terminating: whether the agents are terminated if any of the check strings do not appear in the logs.
  • kwargs: the kwargs of the overridden method.

Returns:

tuple with two lists of missed strings, the strict and the round respectively.

check_aea_messages

def check_aea_messages() -> None

Check that each AEA prints these messages.

First failing check will cause assertion error and test tear down.

BaseTestEnd2EndExecution Objects

class BaseTestEnd2EndExecution(BaseTestEnd2End)

Test that an agent that is launched later can synchronize with the rest of the network

  • each agent starts, and sets up the ABCI connection, which in turn spawns both an ABCI server and a local Tendermint node (using the configuration folders we set up previously). The Tendermint node is unique for each agent
  • when we will stop one agent, also the ABCI server created by the ABCI connection will stop, and in turn the Tendermint node will stop. In particular, it does not keep polling the endpoint until it is up again, it just stops.
  • when we will restart the previously stopped agent, the ABCI connection will set up again both the server and the Tendermint node. The node will automatically connect to the rest of the Tendermint network, loads the entire blockchain bulit so far by the others, and starts sending ABCI requests to the agent (begin_block; deliver_tx*; end_block), plus other auxiliary requests like info , flush etc. The agent which is already processing incoming messages, forwards the ABCI requests to the ABCIHandler, which produces ABCI responses that are forwarded again via the ABCI connection such that the Tendermint node can receive the responses

check

def check() -> None

Check pre-conditions of the test

test_run

def test_run(nb_nodes: int) -> None

Run the test.

Back to top