Python Quick Start

run_test_suite(
  id="my-test-suite",
  test_cases=gen_test_cases(),
  evaluators=[HasAllSubstrings(), IsFriendly()],
  fn=test_fn,
)

poetry add autoblocksai
# or
pip install autoblocksai

import dataclasses

from autoblocks.testing.models import BaseTestCase
from autoblocks.testing.util import md5

@dataclasses.dataclass
class MyTestCase(BaseTestCase):
    """
    A test case can be any class that subclasses BaseTestCase.

    This example is a dataclass, but it could also be a pydantic model,
    plain Python class, etc.
    """
    input: str
    expected_substrings: list[str]

    def hash(self) -> str:
        """
        This hash serves as a unique identifier for a test case throughout its lifetime.

        Required to be implemented by subclasses of BaseTestCase.
        """
        return md5(self.input)

import random
import asyncio

async def test_fn(test_case: MyTestCase) -> str:
    """ This could also be a synchronous function. """
    # Simulate doing work
    await asyncio.sleep(random.random())

    substrings = test_case.input.split("-")
    if random.random() < 0.2:
        # Remove a substring randomly. This will cause about 20% of the test cases to fail
        # the "has-all-substrings" evaluator.
        substrings.pop()

    return "-".join(substrings)

import random
import asyncio

from autoblocks.testing.models import BaseTestEvaluator
from autoblocks.testing.models import Evaluation
from autoblocks.testing.models import Threshold

class HasAllSubstrings(BaseTestEvaluator):
    """
    An evaluator is a class that subclasses BaseTestEvaluator.

    It must specify an ID, which is a unique identifier for the evaluator.
    """
    id = "has-all-substrings"

    def evaluate_test_case(self, test_case: MyTestCase, output: str) -> Evaluation:
        """
        Evaluates the output of a test case.

        Required to be implemented by subclasses of BaseTestEvaluator.
        This method can be synchronous or asynchronous.
        """
        missing_substrings = [s for s in test_case.expected_substrings if s not in output]
        score = 0 if missing_substrings else 1
        return Evaluation(
            score=score,
            # If the score is not greater than or equal to 1,
            # this evaluation will be marked as a failure.
            threshold=Threshold(gte=1),
            metadata=dict(
                # Include the missing substrings as metadata
                # so that we can easily see which strings were
                # missing when viewing a failed evaluation
                # in the Autoblocks UI.
                missing_substrings=missing_substrings,
            ),
        )

class IsFriendly(BaseTestEvaluator):
    id = "is-friendly"

    # The maximum number of concurrent calls to `evaluate_test_case` allowed for this evaluator.
    # Useful to avoid rate limiting from external services, such as an LLM provider.
    max_concurrency = 5  

    async def get_score(self, output: str) -> float:
        # Simulate doing work
        await asyncio.sleep(random.random())

        # Simulate a friendliness score, e.g. as determined by an LLM.
        return random.random()

    async def evaluate_test_case(self, test_case: BaseTestCase, output: str) -> Evaluation:
        """
        This can also be an async function. This is useful if you are interacting
        with an external service that requires async calls, such as OpenAI,
        or if the evaluation you are performing could benefit from concurrency.
        """
        score = await self.get_score(output)
        return Evaluation(
            score=score,
            # Evaluations don't need thresholds attached to them.
            # In this case, the evaluation will just consist of the score.
        )

import uuid

from autoblocks.testing.v2.run import run_test_suite

def gen_test_cases(n: int) -> list[MyTestCase]:
    test_cases = []
    for _ in range(n):
        random_id = str(uuid.uuid4())
        test_cases.append(
            MyTestCase(
                input=random_id,
                expected_substrings=random_id.split("-"),
            ),
        )
    return test_cases

run_test_suite(
    id="my-test-suite",
    fn=test_fn,
    test_cases=gen_test_cases(400),
    evaluators=[
        HasAllSubstrings(),
        IsFriendly(),
    ],
    # The maximum number of test cases that can be running
    # concurrently through `fn`. Useful to avoid rate limiting
    # from external services, such as an LLM provider.
    max_test_case_concurrency=10,
)

export AUTOBLOCKS_V2_API_KEY=...

# Assumes you've saved the above code in a file called run.py
npx autoblocks testing exec -m "my first run" -- python3 run.py