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

class BaseAEATestCase(ABC)

Base class for AEA test cases.

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

Set the current agent context.

@classmethod
def unset_agent_context(cls) -> None

Unset the current agent context.

@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

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

Force set config.

@classmethod
def disable_aea_logging(cls) -> None

Disable AEA logging of specific agent.

Run from agent's directory.

@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

@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.

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

Start python Thread.

Arguments:

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

Returns:

thread

@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.

@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.

@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

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

Delete agents in current working directory.

Arguments:

• agents_names: str agent names.

@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.

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

Run interaction as subprocess.

Run from agent's directory.

Returns:

subprocess object.

@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

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

Check if all subprocesses terminated successfully.

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

Initialize AEA locally with author name.

@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

@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

@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

@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

@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

@classmethod
def run_install(cls) -> Result

Execute AEA CLI install command.

Run from agent's directory.

Returns:

Result

@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

@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

@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

@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

@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

@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

@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

@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.

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

Change current working directory.

Arguments:

• path: path to the new working directory.

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

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

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

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

@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.

@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

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

Call the cli command.

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

Load agent configuration.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test.

class AEATestCaseEmpty(BaseAEATestCase)

Test case for a default AEA project.

This test case will create a default AEA project.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

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.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

class AEATestCaseMany(BaseAEATestCase)

Test case for many AEA projects.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

class AEATestCaseManyFlaky(AEATestCaseMany)

Test case for many AEA projects which are flaky.

Use for flaky tests with the flaky decorator.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test class.

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.

@classmethod
def setup_class(cls) -> None

Set up the test class.

@classmethod
def teardown_class(cls) -> None

Teardown the test class.