Harness

Test harness for workflow testing.

Classes

class arshai.testing.harness.MockMemoryManager

Bases: object

Mock memory manager for testing agents with caching.

Example

agent = MyAgent() agent.memory = MockMemoryManager()

# Test caching behavior result = await agent.cached_method(data)

__init__()

async get(key)

Get value from mock storage.

Return type:

Optional[bytes]

async set(key, value, ttl=300)

Set value in mock storage.

Return type:

None

get_sync(key)

Synchronous get for testing.

Return type:

Optional[bytes]

set_sync(key, value, ttl=300)

Synchronous set for testing.

Return type:

None

clear()

Clear all stored values.

delete_pattern(pattern)

Delete keys matching pattern.

class arshai.testing.harness.WorkflowTestHarness

Bases: object

Test harness for workflow testing.

Provides utilities for testing workflows with mocked nodes, execution tracking, and result verification.

Example

harness = WorkflowTestHarness()

# Test with mocked nodes result = await harness.test_workflow(

workflow=my_workflow, input_state=test_state, mock_nodes={

“external_api”: {“status”: “success”}, “database”: {“data”: [1, 2, 3]}

}

)

# Verify execution path assert harness.executed_nodes == [“input”, “external_api”, “database”, “output”]

__init__()

async test_workflow(workflow, input_state, mock_nodes=None, record_execution=True)

Test workflow with optional mocked nodes.

Parameters:

• workflow (BaseWorkflowOrchestrator) – Workflow to test

• input_state (IWorkflowState) – Input state

• mock_nodes (Optional[Dict[str, Any]]) – Dict of node names to mock outputs

• record_execution (bool) – Record execution details

Return type:

IWorkflowState

Returns:

Workflow result

create_mock_agent(name, response='mock response')

Create a mock agent for testing.

Parameters:

• name (str) – Agent name

• response (str) – Mock response string

Return type:

BaseAgent

Returns:

Mock agent instance

assert_node_executed(node_name)

Assert that a specific node was executed.

Parameters:

node_name (str) – Name of the node to check

Raises:

AssertionError – If node was not executed

assert_execution_order(expected_order)

Assert nodes were executed in specific order.

Parameters:

expected_order (List[str]) – Expected execution order

Raises:

AssertionError – If execution order doesn’t match

assert_node_not_executed(node_name)

Assert that a specific node was NOT executed.

Parameters:

node_name (str) – Name of the node to check

Raises:

AssertionError – If node was executed

get_node_input(node_name)

Get the input that was passed to a specific node.

Parameters:

node_name (str) – Name of the node

Return type:

Any

Returns:

Input data passed to the node

get_node_output(node_name)

Get the output from a specific node.

Parameters:

node_name (str) – Name of the node

Return type:

Any

Returns:

Output data from the node

reset()

Reset the test harness for a new test.

get_execution_stats()

Get statistics about the workflow execution.

Return type:

dict

Returns:

Dictionary with execution statistics