Skip to content

aea.test_tools.test_cases

This module contains test case classes based on pytest for AEA end-to-end testing.

BaseAEATestCase Objects

class BaseAEATestCase(ABC)

Base class for AEA test cases.

set_capfd_on_cli_runner

@pytest.fixture(autouse=True)
def set_capfd_on_cli_runner(capfd: CaptureFixture) -> None

Set pytest capfd on CLI runner

runner

CLI runner

author

author

subprocesses

list of launched subprocesses

threads

list of started threads

old_cwd

current working directory path

t

temporary directory path

current_agent_context

the name of the current agent

agents

the set of created agents

stdout

dict of process.pid: string stdout

stderr

dict of process.pid: string stderr

set_agent_context

@classmethod
def set_agent_context(cls, agent_name: str) -> None

Set the current agent context.

unset_agent_context

@classmethod
def unset_agent_context(cls) -> None

Unset the current agent context.

set_config

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

Set a config.

Run from agent's directory.

Arguments:

  • dotted_path: str dotted path to config param.
  • value: a new value to set.
  • type_: the type
  • aev: use the environment variables

Returns:

Result

nested_set_config

@classmethod
def nested_set_config(cls, dotted_path: str, value: Any) -> None

Force set config.

disable_aea_logging

@classmethod
def disable_aea_logging(cls) -> None

Disable AEA logging of specific agent.

Run from agent's directory.

run_cli_command

@classmethod
def run_cli_command(cls, *args: str, cwd: str = ".", **kwargs: str) -> Result

Run AEA CLI command.

Arguments:

  • args: CLI args
  • cwd: the working directory from where to run the command.
  • kwargs: other keyword arguments to click.CliRunner.invoke.

Raises:

  • AEATestingException: if command fails.

Returns:

Result

start_subprocess

@classmethod
def start_subprocess(cls, *args: str, cwd: str = ".") -> subprocess.Popen

Run python with args as subprocess.

Arguments:

  • args: CLI args
  • cwd: the current working directory

Returns:

subprocess object.

start_thread

@classmethod
def start_thread(cls, target: Callable, **kwargs: subprocess.Popen) -> Thread

Start python Thread.

Arguments:

  • target: target method.
  • kwargs: thread keyword arguments

Returns:

thread

create_agents

@classmethod
def create_agents(cls,
                  *agents_names: str,
                  is_local: bool = True,
                  is_empty: bool = False) -> None

Create agents in current working directory.

Arguments:

  • agents_names: str agent names.
  • is_local: a flag for local folder add True by default.
  • is_empty: optional boolean flag for skip adding default dependencies.

fetch_agent

@classmethod
def fetch_agent(cls,
                public_id: str,
                agent_name: str,
                is_local: bool = True) -> None

Create agents in current working directory.

Arguments:

  • public_id: str public id
  • agent_name: str agent name.
  • is_local: a flag for local folder add True by default.

difference_to_fetched_agent

@classmethod
def difference_to_fetched_agent(cls, public_id: str,
                                agent_name: str) -> List[str]

Compare agent against the one fetched from public id.

Arguments:

  • public_id: str public id
  • agent_name: str agent name.

Returns:

list of files differing in the projects

delete_agents

@classmethod
def delete_agents(cls, *agents_names: str) -> None

Delete agents in current working directory.

Arguments:

  • agents_names: str agent names.

run_agent

@classmethod
def run_agent(cls, *args: str) -> subprocess.Popen

Run agent as subprocess.

Run from agent's directory.

Arguments:

  • args: CLI args

Returns:

subprocess object.

terminate_agents

@classmethod
def terminate_agents(cls,
                     *subprocesses: subprocess.Popen,
                     timeout: int = TERMINATION_TIMEOUT) -> None

Terminate agent subprocesses.

Run from agent's directory.

Arguments:

  • subprocesses: the subprocesses running the agents
  • timeout: the timeout for interruption

is_successfully_terminated

@classmethod
def is_successfully_terminated(cls, *subprocesses: subprocess.Popen) -> bool

Check if all subprocesses terminated successfully.

initialize_aea

@classmethod
def initialize_aea(cls, author: str) -> None

Initialize AEA locally with author name.

add_item

@classmethod
def add_item(cls,
             item_type: str,
             public_id: str,
             local: bool = True) -> Result

Add an item to the agent.

Run from agent's directory.

Arguments:

  • item_type: str item type.
  • public_id: public id of the item.
  • local: a flag for local folder add True by default.

Returns:

Result

remove_item

@classmethod
def remove_item(cls, item_type: str, public_id: str) -> Result

Remove an item from the agent.

Run from agent's directory.

Arguments:

  • item_type: str item type.
  • public_id: public id of the item.

Returns:

Result

scaffold_item

@classmethod
def scaffold_item(cls,
                  item_type: str,
                  name: str,
                  skip_consistency_check: bool = False) -> Result

Scaffold an item for the agent.

Run from agent's directory.

Arguments:

  • item_type: str item type.
  • name: name of the item.
  • skip_consistency_check: if True, skip consistency check.

Returns:

Result

fingerprint_item

@classmethod
def fingerprint_item(cls, item_type: str, public_id: str) -> Result

Fingerprint an item for the agent.

Run from agent's directory.

Arguments:

  • item_type: str item type.
  • public_id: public id of the item.

Returns:

Result

eject_item

@classmethod
def eject_item(cls, item_type: str, public_id: str) -> Result

Eject an item in the agent in quiet mode (i.e. no interaction).

Run from agent's directory.

Arguments:

  • item_type: str item type.
  • public_id: public id of the item.

Returns:

None

run_install

@classmethod
def run_install(cls) -> Result

Execute AEA CLI install command.

Run from agent's directory.

Returns:

Result

generate_private_key

@classmethod
def generate_private_key(cls,
                         ledger_api_id: str = DEFAULT_LEDGER,
                         private_key_file: Optional[str] = None,
                         password: Optional[str] = None) -> Result

Generate AEA private key with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • private_key_file: the private key file.
  • password: the password.

Returns:

Result

add_private_key

@classmethod
def add_private_key(cls,
                    ledger_api_id: str = DEFAULT_LEDGER,
                    private_key_filepath: str = DEFAULT_PRIVATE_KEY_FILE,
                    connection: bool = False,
                    password: Optional[str] = None) -> Result

Add private key with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • private_key_filepath: private key filepath.
  • connection: whether or not the private key filepath is for a connection.
  • password: the password to encrypt private keys.

Returns:

Result

remove_private_key

@classmethod
def remove_private_key(cls,
                       ledger_api_id: str = DEFAULT_LEDGER,
                       connection: bool = False) -> Result

Remove private key with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • connection: whether or not the private key filepath is for a connection.

Returns:

Result

replace_private_key_in_file

@classmethod
def replace_private_key_in_file(
        cls,
        private_key: str,
        private_key_filepath: str = DEFAULT_PRIVATE_KEY_FILE) -> None

Replace the private key in the provided file with the provided key.

Arguments:

  • private_key: the private key
  • private_key_filepath: the filepath to the private key file

generate_wealth

@classmethod
def generate_wealth(cls,
                    ledger_api_id: str = DEFAULT_LEDGER,
                    password: Optional[str] = None) -> Result

Generate wealth with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • password: the password.

Returns:

Result

get_wealth

@classmethod
def get_wealth(cls,
               ledger_api_id: str = DEFAULT_LEDGER,
               password: Optional[str] = None) -> str

Get wealth with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • password: the password to encrypt/decrypt private keys.

Returns:

command line output

get_address

@classmethod
def get_address(cls,
                ledger_api_id: str = DEFAULT_LEDGER,
                password: Optional[str] = None) -> str

Get address with CLI command.

Run from agent's directory.

Arguments:

  • ledger_api_id: ledger API ID.
  • password: the password to encrypt/decrypt private keys.

Returns:

command line output

replace_file_content

@classmethod
def replace_file_content(cls, src: Path, dest: Path) -> None

Replace the content of the source file to the destination file.

Arguments:

  • src: the source file.
  • dest: the destination file.

change_directory

@classmethod
def change_directory(cls, path: Path) -> None

Change current working directory.

Arguments:

  • path: path to the new working directory.

send_envelope_to_agent

@classmethod
def send_envelope_to_agent(cls, envelope: Envelope, agent: str) -> None

Send an envelope to an agent, using the stub connection.

read_envelope_from_agent

@classmethod
def read_envelope_from_agent(cls, agent: str) -> Envelope

Read an envelope from an agent, using the stub connection.

missing_from_output

@classmethod
def missing_from_output(cls,
                        process: subprocess.Popen,
                        strings: Sequence[str],
                        timeout: int = DEFAULT_PROCESS_TIMEOUT,
                        period: int = 1,
                        is_terminating: bool = True) -> 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:

  • process: agent subprocess.
  • strings: tuple of strings expected to appear in output.
  • timeout: int amount of seconds before stopping check.
  • period: int period of checking.
  • is_terminating: whether or not the agents are terminated

Returns:

list of missed strings.

is_running

@classmethod
def is_running(cls,
               process: subprocess.Popen,
               timeout: int = DEFAULT_LAUNCH_TIMEOUT) -> bool

Check if the AEA is launched and running (ready to process messages).

Arguments:

  • process: agent subprocess.
  • timeout: the timeout to wait for launch to complete

Returns:

bool indicating status

invoke

@classmethod
def invoke(cls, *args: str) -> Result

Call the cli command.

load_agent_config

@classmethod
def load_agent_config(cls, agent_name: str) -> AgentConfig

Load agent configuration.

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test.

AEATestCaseEmpty Objects

class AEATestCaseEmpty(BaseAEATestCase)

Test case for a default AEA project.

This test case will create a default AEA project.

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

AEATestCaseEmptyFlaky Objects

class AEATestCaseEmptyFlaky(AEATestCaseEmpty)

Test case for a default AEA project.

This test case will create a default AEA project.

Use for flaky tests with the flaky decorator.

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

AEATestCaseMany Objects

class AEATestCaseMany(BaseAEATestCase)

Test case for many AEA projects.

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

AEATestCaseManyFlaky Objects

class AEATestCaseManyFlaky(AEATestCaseMany)

Test case for many AEA projects which are flaky.

Use for flaky tests with the flaky decorator.

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

AEATestCase Objects

class AEATestCase(BaseAEATestCase)

Test case from an existing AEA project.

Subclass this class and set path_to_aea properly. By default, it is assumed the project is inside the current working directory.

t

temporary directory path

setup_class

@classmethod
def setup_class(cls) -> None

Set up the test class.

teardown_class

@classmethod
def teardown_class(cls) -> None

Teardown the test class.