Skip to Content
ContributeWorkerTest Execution

Test Execution System

Tests run against endpoints through executors selected by test type, coordinated by an async batch engine. This page covers the executor architecture; for single-turn vs. multi-turn tests see Test Types, and for parallel vs. sequential runs see Execution Modes.

Architecture

A factory routes each test to an executor based on its type:

code.txt
Test → Factory → Executor → Results
       ↓
  SingleTurnExecutor (traditional tests)
  MultiTurnExecutor (Penelope-based)
  [Future: ImageTestExecutor, etc.]

Key Components

  1. Test Execution Entry Point (test_execution.py)

    • Main execute_test() function
    • Routes to appropriate executor via factory pattern
    • Maintains backward compatibility

  2. Executors (executors/)

    • BaseTestExecutor - Abstract base class
    • SingleTurnTestExecutor - Traditional request/response tests
    • MultiTurnTestExecutor - Agentic multi-turn tests using Penelope
    • factory.py - Routing logic based on test type

  3. Shared Utilities (executors/shared.py)

    • Common helper functions
    • Data retrieval and validation
    • Result storage

Test Flow

Standard Execution Flow

code.txt
Test Configuration
  ↓
Retrieve Test
  ↓
Determine Test Type
  ↓
  ├─→ Single-Turn → SingleTurnExecutor
  │
  └─→ Multi-Turn → MultiTurnExecutor
  ↓
Store Results
  ↓
Return Summary

Parallel vs Sequential

Test configurations can execute multiple tests in two modes:

  • Parallel (default): one Celery task runs an internal async batch engine
  • Sequential: tests run one after another

See Execution Modes for details.

Executor Pattern

Creating an Executor

All executors must implement the BaseTestExecutor interface:

custom_executor.py
from rhesis.backend.tasks.execution.executors.base import BaseTestExecutor

class CustomTestExecutor(BaseTestExecutor):
    def execute(
        self,
        db: Session,
        test_config_id: str,
        test_run_id: str,
        test_id: str,
        endpoint_id: str,
        organization_id: Optional[str] = None,
        user_id: Optional[str] = None,
        model: Optional[Any] = None,
    ) -> Dict[str, Any]:
        # Your execution logic here
        return {
            "test_id": test_id,
            "execution_time": execution_time_ms,
            "metrics": metrics_dict,
        }

Return Format

All executors must return a dictionary with:

  • test_id (str): The test identifier
  • execution_time (float): Execution time in milliseconds
  • metrics (Dict[str, Any]): Metric evaluation results

Usage Examples

Running a Test

run_test.py
from rhesis.backend.tasks.execution.test_execution import execute_test

# Execute test (automatically routes to correct executor)
result = execute_test(
    db=db,
    test_config_id="config-uuid",
    test_run_id="run-uuid",
    test_id="test-uuid",
    endpoint_id="endpoint-uuid",
    organization_id="org-uuid",
    user_id="user-uuid"
)

print(f"Execution time: {result['execution_time']}ms")
print(f"Metrics: {result['metrics']}")

Using Executors Directly

direct_executor.py
from rhesis.backend.tasks.execution.executors import (
    create_executor,
    SingleTurnTestExecutor,
    MultiTurnTestExecutor,
)

# Get test and create appropriate executor
test = get_test(db, test_id)
executor = create_executor(test)

# Or use a specific executor directly
executor = MultiTurnTestExecutor()
result = executor.execute(db, test_config_id, ...)

Extending the System

Adding a New Test Type

  1. Create the Executor
executors/image_test_executor.py
# executors/image_test_executor.py
from rhesis.backend.tasks.execution.executors.base import BaseTestExecutor

class ImageTestExecutor(BaseTestExecutor):
    def execute(self, ...):
        # Image-specific test logic
        pass
  1. Update the Factory
executors/factory.py
# executors/factory.py
def create_executor(test: Test) -> BaseTestExecutor:
    test_type = get_test_type(test)

    if test_type == TestType.IMAGE:
        from .image_test_executor import ImageTestExecutor
        return ImageTestExecutor()
    elif test_type == TestType.MULTI_TURN:
        # ...existing code
  1. Add Test Type Enum
tasks/enums.py
# tasks/enums.py
class TestType(str, Enum):
    SINGLE_TURN = "Single-Turn"
    MULTI_TURN = "Multi-Turn"
    IMAGE = "Image"  # New type

Async Batch Engine

The parallel path uses an internal async batch runner rather than Celery chord fan-out:

  • execute_test_configuration creates a single worker task per test run
  • execute_tests_as_batch prefetches execution context and runs run_batch(...)
  • run_batch runs each test as an asyncio task with semaphore-limited concurrency
  • cancellation is cooperative via Celery revoke checks and task cancellation
  • per-test timeout defaults to 1800 seconds (30 minutes); batch concurrency defaults to 10

Module Location

The test execution code is located in the backend repository but is primarily executed by workers:

code.txt
apps/backend/src/rhesis/backend/tasks/execution/
├── executors/
│   ├── __init__.py
│   ├── base.py                    # BaseTestExecutor ABC
│   ├── data.py                    # Data retrieval utilities
│   ├── metrics.py                 # Metrics processing
│   ├── results.py                 # Result storage
│   ├── runners.py                 # Core execution logic (shared)
│   ├── single_turn.py             # Single-turn executor
│   ├── multi_turn.py              # Multi-turn executor (Penelope)
│   └── factory.py                 # Routing logic
├── test_execution.py              # Main entry point
├── modes.py                       # Execution mode utilities
├── orchestration.py               # Task orchestration
├── batch/                         # Async batch execution engine
│   ├── __init__.py                # execute_tests_as_batch
│   ├── context.py                 # prefetch + ExecutionContext
│   └── runner.py                  # run_batch + cancellation watchdog
├── sequential.py                  # Sequential execution
└── penelope_target.py             # Backend target for Penelope
Trace Ingestion PipelineTest Types