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_: 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
| 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 = ".", **kwargs) -> Result
Run AEA CLI command.
Arguments:
args
: CLI argscwd
: 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
| 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 idagents_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 idagents_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 agentstimeout
: 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 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
| 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
remove_
private_
key
| @classmethod
| 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
| 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 keyprivate_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
invoke
| @classmethod
| invoke(cls, *args)
Call the cli command.
load_
agent_
config
| @classmethod
| load_agent_config(cls, agent_name: str) -> AgentConfig
Load agent configuration.
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.