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
 | set_agent_context(cls, agent_name: str)

Set the current agent context.

unset_agent_context

 | @classmethod
 | unset_agent_context(cls)

Unset the current agent context.

set_config

 | @classmethod
 | set_config(cls, dotted_path: str, value: Any, type_: str = "str") -> 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
 | nested_set_config(cls, dotted_path: str, value: Any) -> None

Force set config.

disable_aea_logging

 | @classmethod
 | disable_aea_logging(cls) -> None

Disable AEA logging of specific agent.

Run from agent's directory.

Returns:

None

run_cli_command

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

Run AEA CLI command.

Arguments:

• args: CLI args
• cwd: the working directory from where to run the command.

Raises:

• AEATestingException: if command fails.

Returns:

Result

start_subprocess

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

Run python with args as subprocess.

Arguments:

• args: CLI args

Returns:

subprocess object.

start_thread

 | @classmethod
 | start_thread(cls, target: Callable, **kwargs) -> Thread

Start python Thread.

Arguments:

• target: target method.
• process: subprocess passed to thread args.

Returns:

None.

create_agents

 | @classmethod
 | 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.
• empty: optional boolean flag for skip adding default dependencies.

Returns:

None

fetch_agent

 | @classmethod
 | 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
• agents_name: str agent name.
• is_local: a flag for local folder add True by default.

Returns:

None

difference_to_fetched_agent

 | @classmethod
 | 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
• agents_name: str agent name.

Returns:

list of files differing in the projects

delete_agents

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

Delete agents in current working directory.

Arguments:

• agents_names: str agent names.

Returns:

None

run_agent

 | @classmethod
 | 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
 | run_interaction(cls) -> subprocess.Popen

Run interaction as subprocess.

Run from agent's directory.

Arguments:

• args: CLI args

Returns:

subprocess object.

terminate_agents

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

Check if all subprocesses terminated successfully.

initialize_aea

 | @classmethod
 | initialize_aea(cls, author) -> None

Initialize AEA locally with author name.

Returns:

None

add_item

 | @classmethod
 | 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
 | 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
 | scaffold_item(cls, item_type: str, name: str) -> Result

Scaffold an item for the agent.

Run from agent's directory.

Arguments:

• item_type: str item type.
• name: name of the item.

Returns:

Result

fingerprint_item

 | @classmethod
 | 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.
• name: public id of the item.

Returns:

Result

eject_item

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

Eject an item in the agent.

Run from agent's directory.

Arguments:

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

Returns:

None

run_install

 | @classmethod
 | run_install(cls) -> Result

Execute AEA CLI install command.

Run from agent's directory.

Returns:

Result

generate_private_key

 | @classmethod
 | generate_private_key(cls, ledger_api_id: str = DEFAULT_LEDGER, private_key_file: 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.

Returns:

Result

add_private_key

 | @classmethod
 | add_private_key(cls, ledger_api_id: str = DEFAULT_LEDGER, private_key_filepath: str = DEFAULT_PRIVATE_KEY_FILE, connection: bool = False) -> 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.

Returns:

Result

replace_private_key_in_file

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

Returns:

None :raises: exception if file does not exist

generate_wealth

 | @classmethod
 | generate_wealth(cls, ledger_api_id: str = DEFAULT_LEDGER) -> Result

Generate wealth with CLI command.

Run from agent's directory.

Arguments:

• ledger_api_id: ledger API ID.

Returns:

Result

get_wealth

 | @classmethod
 | get_wealth(cls, ledger_api_id: str = DEFAULT_LEDGER) -> str

Get wealth with CLI command.

Run from agent's directory.

Arguments:

• ledger_api_id: ledger API ID.

Returns:

command line output

replace_file_content

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

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

Arguments:

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

Returns:

None

change_directory

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

Change current working directory.

Arguments:

• path: path to the new working directory.

Returns:

None

send_envelope_to_agent

 | @classmethod
 | send_envelope_to_agent(cls, envelope: Envelope, agent: str)

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

read_envelope_from_agent

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

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

missing_from_output

 | @classmethod
 | 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
 | is_running(cls, process: subprocess.Popen, timeout: int = DEFAULT_LAUNCH_TIMEOUT)

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

setup_class

 | @classmethod
 | setup_class(cls)

Set up the test class.

teardown_class

 | @classmethod
 | teardown_class(cls)

Teardown the test.

UseOef Objects

@pytest.mark.integration
class UseOef()

Inherit from this class to launch an OEF node.

AEATestCaseEmpty Objects

class AEATestCaseEmpty(BaseAEATestCase)

Test case for a default AEA project.

This test case will create a default AEA project.

setup_class

 | @classmethod
 | setup_class(cls)

Set up the test class.

AEATestCaseMany Objects

class AEATestCaseMany(BaseAEATestCase)

Test case for many AEA projects.

setup_class

 | @classmethod
 | setup_class(cls)

Set up the test class.

teardown_class

 | @classmethod
 | teardown_class(cls)

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
 | setup_class(cls)

Set up the test class.

teardown_class

 | @classmethod
 | teardown_class(cls)

Teardown the test class.

Previous:
Generic Next:
Test Skill