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`_`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) -> 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

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.

run`_`interaction¶

@classmethod
def run_interaction(cls) -> subprocess.Popen

Run interaction as subprocess.

Run from agent's directory.

Returns:

subprocess object.

terminate`_`agents¶

@classmethod
def terminate_agents(cls,
                     *subprocesses: subprocess.Popen,
                     timeout: int = 20) -> 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

Raises:

None: exception if file does not exist

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.

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.

aea.test_tools.test_cases¶

BaseAEATestCase Objects¶

set_agent_context¶

unset_agent_context¶

set_config¶

nested_set_config¶

disable_aea_logging¶

run_cli_command¶

start_subprocess¶

start_thread¶

create_agents¶

fetch_agent¶

difference_to_fetched_agent¶

delete_agents¶

run_agent¶

run_interaction¶

terminate_agents¶

is_successfully_terminated¶

initialize_aea¶

add_item¶

remove_item¶

scaffold_item¶

fingerprint_item¶

eject_item¶

run_install¶

generate_private_key¶

add_private_key¶

remove_private_key¶

replace_private_key_in_file¶

generate_wealth¶

get_wealth¶

get_address¶

replace_file_content¶

change_directory¶

send_envelope_to_agent¶

read_envelope_from_agent¶

missing_from_output¶

is_running¶

invoke¶

load_agent_config¶

setup_class¶

teardown_class¶

AEATestCaseEmpty Objects¶

setup_class¶

teardown_class¶

AEATestCaseEmptyFlaky Objects¶

setup_class¶

teardown_class¶

AEATestCaseMany Objects¶

setup_class¶

teardown_class¶

AEATestCaseManyFlaky Objects¶

setup_class¶

teardown_class¶

AEATestCase Objects¶

setup_class¶

teardown_class¶

aea.test`_`tools.test`_`cases¶

set`_`agent`_`context¶

unset`_`agent`_`context¶

set`_`config¶

nested`_`set`_`config¶

disable`_`aea`_`logging¶

run`_`cli`_`command¶

start`_`subprocess¶

start`_`thread¶

create`_`agents¶

fetch`_`agent¶

difference`_`to`_`fetched`_`agent¶

delete`_`agents¶

run`_`agent¶

run`_`interaction¶

terminate`_`agents¶

is`_`successfully`_`terminated¶

initialize`_`aea¶

add`_`item¶

remove`_`item¶

scaffold`_`item¶

fingerprint`_`item¶

eject`_`item¶

run`_`install¶

generate`_`private`_`key¶

add`_`private`_`key¶

remove`_`private`_`key¶

replace`_`private`_`key`_`in`_`file¶

generate`_`wealth¶

get`_`wealth¶

get`_`address¶

replace`_`file`_`content¶

change`_`directory¶

send`_`envelope`_`to`_`agent¶

read`_`envelope`_`from`_`agent¶

missing`_`from`_`output¶

is`_`running¶

load`_`agent`_`config¶

setup`_`class¶

teardown`_`class¶

setup`_`class¶

teardown`_`class¶

setup`_`class¶

teardown`_`class¶

setup`_`class¶

teardown`_`class¶

setup`_`class¶

teardown`_`class¶

setup`_`class¶

teardown`_`class¶