Ethereum Test Specs package

Test spec definitions and utilities.

BaseTest

Bases: BaseModel

Represents a base Ethereum test which must return a single test fixture.

Source code in packages/testing/src/execution_testing/specs/base.py

class BaseTest(BaseModel):
    """
    Represents a base Ethereum test which must return a single test fixture.
    """

    model_config = ConfigDict(extra="forbid")

    fork: Fork | TransitionFork = (
        BaseFork
        # default to BaseFork to allow the filler to set it,
        # instead of each test having to set it
    )
    operation_mode: OpMode | None = None
    gas_optimization_max_gas_limit: int | None = None
    expected_benchmark_gas_used: int | None = None
    skip_gas_used_validation: bool = False
    expected_receipt_status: int | None = None
    is_tx_gas_heavy_test: bool = False
    is_exception_test: bool = False
    is_inclusion_test: bool = False

    # Class variables, to be set by subclasses
    spec_types: ClassVar[Dict[str, Type["BaseTest"]]] = {}
    supported_fixture_formats: ClassVar[
        Sequence[FixtureFormat | LabeledFixtureFormat]
    ] = []
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = []

    supported_markers: ClassVar[Dict[str, str]] = {}

    def model_post_init(self, __context: Any, /) -> None:
        """
        Model post-init to assert that the custom pre-allocation was
        provided and the default was not used.
        """
        super().model_post_init(__context)
        assert self.fork != BaseFork, (
            "Fork was not provided by the filler/executor."
        )

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard a fixture format from filling if the appropriate marker is
        used.

        The format keeps its label: comparing it against a plain format
        matches every label of it, `format_id()` identifies a single one.
        """
        del fixture_format, markers
        return False

    @classmethod
    def fixture_format_parameters(
        cls,
        *,
        markers: List[pytest.Mark],
    ) -> List[Tuple[FixtureFormat | LabeledFixtureFormat, ParameterSet]]:
        """
        Return one pytest parameter per fixture format this spec type fills,
        paired with the format it fills, so the caller can narrow the formats
        down to the ones its session generates.

        The first format not vetoed by `discard_fixture_format_by_marks` is
        the primary and is marked `primary_format`, so `-m primary_format`
        fills each test once. Only sessions that generate a single format per
        test can filter that primary out, and there the marker is moot.

        An override that parametrizes further must leave exactly one parameter
        marked `primary_format`.
        """
        parameters: List[
            Tuple[FixtureFormat | LabeledFixtureFormat, ParameterSet]
        ] = []
        for format_with_or_without_label in cls.supported_fixture_formats:
            if cls.discard_fixture_format_by_marks(
                format_with_or_without_label,
                markers,
            ):
                continue
            parameter = labeled_format_parameter_set(
                format_with_or_without_label,
                primary_format=not parameters,
            )
            parameters.append((format_with_or_without_label, parameter))
        return parameters

    @classmethod
    def execute_format_parameters(
        cls,
    ) -> List[Tuple[LabeledExecuteFormat, ParameterSet]]:
        """
        Return one pytest parameter per execute format this spec type runs.

        Each element pairs the labeled format with the parameter that executes
        it, so the caller can inspect what the format requires of the session
        (e.g. an Engine RPC) and skip or select it accordingly.

        Formats are not vetoed here the way `fixture_format_parameters` vetoes
        them: `discard_execute_format_by_marks` needs the fork, which is only
        parametrized later, so the caller applies it during collection.

        Subclasses may override this to parametrize further, e.g. to execute a
        single execute format more than once.
        """
        return [
            (
                labeled_execute_format,
                labeled_format_parameter_set(labeled_execute_format),
            )
            for labeled_execute_format in cls.supported_execute_formats
        ]

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        """
        Register all subclasses of BaseFixture with a fixture format name set
        as possible fixture formats.
        """
        super().__pydantic_init_subclass__(**kwargs)

        # Don't register dynamically generated wrapper classes
        if getattr(cls, "__is_base_test_wrapper__", False):
            return

        if cls.pytest_parameter_name():
            # Register the new fixture format
            BaseTest.spec_types[cls.pytest_parameter_name()] = cls

    @classmethod
    def from_test(
        cls,
        *,
        base_test: "BaseTest",
        **kwargs: Any,
    ) -> Self:
        """Create a test in a different format from a base test."""
        for k in BaseTest.model_fields.keys():
            if k not in kwargs and k in base_test.model_fields_set:
                kwargs[k] = getattr(base_test, k)
        return cls(**kwargs)

    @classmethod
    def discard_execute_format_by_marks(
        cls,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
        fork: Fork | TransitionFork,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard an execute format from executing if the appropriate marker is
        used.
        """
        del execute_format, fork, markers
        return False

    @abstractmethod
    def generate(
        self,
        *,
        t8n: TransitionTool,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the test fixture using the given fixture format."""
        pass

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        raise Exception(f"Unsupported execute format: {execute_format}")

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Must return the name of the parameter used in pytest to select this
        spec type as filler for the test.

        By default, it returns the underscore separated name of the class.
        """
        if cls == BaseTest:
            return ""
        return reduce(
            lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__
        ).lower()

    def check_exception_test(
        self,
        *,
        exception: bool,
    ) -> None:
        """Compare the test marker against the outcome of the test."""
        if self.is_exception_test != exception:
            if exception:
                raise Exception(
                    "Test produced an invalid block or transaction but was "
                    "not marked with the `exception_test` marker. Add the "
                    "`@pytest.mark.exception_test` decorator to the test."
                )
            else:
                raise Exception(
                    "Test didn't produce an invalid block or transaction but "
                    "was marked with the `exception_test` marker. Remove the "
                    "`@pytest.mark.exception_test` decorator from the test."
                )

    def get_genesis_environment(self) -> Environment:
        """
        Get the genesis environment for pre-allocation groups.

        Must be implemented by subclasses to provide the appropriate
        environment.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} must implement genesis environment "
            "access for use with pre-allocation groups."
        )

    def validate_benchmark_gas(
        self,
        *,
        benchmark_gas_used: int | None,
        gas_benchmark_value: int,
        benchmark_block_gas_used: int | None = None,
    ) -> None:
        """
        Validates the total consumed gas of the last block in the test matches
        the expectation of the benchmark test.

        ``benchmark_gas_used`` is the combined gas across all dimensions (the
        receipt ``cumulativeGasUsed``) and is checked against
        ``expected_benchmark_gas_used``. ``benchmark_block_gas_used`` is the
        block-header gas, i.e. the maximum across the independent gas
        dimensions (EIP-8037); it is what must stay within the block gas
        limit, because the combined value can legitimately exceed it. When it
        is not available (e.g. execute mode), the combined value is used for
        the ceiling check instead.

        Requires the following fields to be set:
        - expected_benchmark_gas_used
        - operation_mode
        """
        if self.operation_mode != OpMode.BENCHMARKING:
            return
        assert benchmark_gas_used is not None, "_benchmark_gas_used is not set"
        # Perform gas validation if required for benchmarking.
        # Ensures benchmark tests consume exactly the expected gas.
        if not self.skip_gas_used_validation:
            # Verify that the total gas consumed in the last block
            # matches expectations
            expected_benchmark_gas_used = self.expected_benchmark_gas_used
            if expected_benchmark_gas_used is None:
                expected_benchmark_gas_used = gas_benchmark_value
            diff = benchmark_gas_used - expected_benchmark_gas_used
            assert benchmark_gas_used == expected_benchmark_gas_used, (
                f"Total gas used ({benchmark_gas_used}) does not "
                "match expected benchmark gas "
                f"({expected_benchmark_gas_used}), "
                f"difference: {diff}"
            )
        # No single gas dimension may exceed the block gas limit. The
        # block-header gas is the max across dimensions; the combined
        # execution+state gas may exceed the target under EIP-8037, so the
        # ceiling is checked against the header value when available.
        block_gas_used = (
            benchmark_block_gas_used
            if benchmark_block_gas_used is not None
            else benchmark_gas_used
        )
        assert block_gas_used <= gas_benchmark_value, (
            f"benchmark block gas used ({block_gas_used}) exceeds maximum "
            "benchmark gas allowed for this configuration: "
            f"{gas_benchmark_value}"
        )

    def validate_receipt_status(
        self,
        *,
        receipts: List[TransactionReceipt],
        block_number: int,
    ) -> None:
        """
        Validate receipt status for every transaction in a block.

        When expected_receipt_status is set, verify that all
        receipts match. Catches silent OOG failures that roll
        back state and invalidate benchmarks.
        """
        if self.expected_receipt_status is None:
            return
        for i, receipt in enumerate(receipts):
            if receipt.status is not None and (
                int(receipt.status) != self.expected_receipt_status
            ):
                raise Exception(
                    f"Transaction {i} in block "
                    f"{block_number} has receipt "
                    f"status {int(receipt.status)}, "
                    f"expected "
                    f"{self.expected_receipt_status}."
                )

model_post_init(__context)

Model post-init to assert that the custom pre-allocation was provided and the default was not used.

Source code in packages/testing/src/execution_testing/specs/base.py

def model_post_init(self, __context: Any, /) -> None:
    """
    Model post-init to assert that the custom pre-allocation was
    provided and the default was not used.
    """
    super().model_post_init(__context)
    assert self.fork != BaseFork, (
        "Fork was not provided by the filler/executor."
    )

discard_fixture_format_by_marks(fixture_format, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

The format keeps its label: comparing it against a plain format matches every label of it, format_id() identifies a single one.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard a fixture format from filling if the appropriate marker is
    used.

    The format keeps its label: comparing it against a plain format
    matches every label of it, `format_id()` identifies a single one.
    """
    del fixture_format, markers
    return False

fixture_format_parameters(*, markers) classmethod

Return one pytest parameter per fixture format this spec type fills, paired with the format it fills, so the caller can narrow the formats down to the ones its session generates.

The first format not vetoed by discard_fixture_format_by_marks is the primary and is marked primary_format, so -m primary_format fills each test once. Only sessions that generate a single format per test can filter that primary out, and there the marker is moot.

An override that parametrizes further must leave exactly one parameter marked primary_format.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def fixture_format_parameters(
    cls,
    *,
    markers: List[pytest.Mark],
) -> List[Tuple[FixtureFormat | LabeledFixtureFormat, ParameterSet]]:
    """
    Return one pytest parameter per fixture format this spec type fills,
    paired with the format it fills, so the caller can narrow the formats
    down to the ones its session generates.

    The first format not vetoed by `discard_fixture_format_by_marks` is
    the primary and is marked `primary_format`, so `-m primary_format`
    fills each test once. Only sessions that generate a single format per
    test can filter that primary out, and there the marker is moot.

    An override that parametrizes further must leave exactly one parameter
    marked `primary_format`.
    """
    parameters: List[
        Tuple[FixtureFormat | LabeledFixtureFormat, ParameterSet]
    ] = []
    for format_with_or_without_label in cls.supported_fixture_formats:
        if cls.discard_fixture_format_by_marks(
            format_with_or_without_label,
            markers,
        ):
            continue
        parameter = labeled_format_parameter_set(
            format_with_or_without_label,
            primary_format=not parameters,
        )
        parameters.append((format_with_or_without_label, parameter))
    return parameters

execute_format_parameters() classmethod

Return one pytest parameter per execute format this spec type runs.

Each element pairs the labeled format with the parameter that executes it, so the caller can inspect what the format requires of the session (e.g. an Engine RPC) and skip or select it accordingly.

Formats are not vetoed here the way fixture_format_parameters vetoes them: discard_execute_format_by_marks needs the fork, which is only parametrized later, so the caller applies it during collection.

Subclasses may override this to parametrize further, e.g. to execute a single execute format more than once.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def execute_format_parameters(
    cls,
) -> List[Tuple[LabeledExecuteFormat, ParameterSet]]:
    """
    Return one pytest parameter per execute format this spec type runs.

    Each element pairs the labeled format with the parameter that executes
    it, so the caller can inspect what the format requires of the session
    (e.g. an Engine RPC) and skip or select it accordingly.

    Formats are not vetoed here the way `fixture_format_parameters` vetoes
    them: `discard_execute_format_by_marks` needs the fork, which is only
    parametrized later, so the caller applies it during collection.

    Subclasses may override this to parametrize further, e.g. to execute a
    single execute format more than once.
    """
    return [
        (
            labeled_execute_format,
            labeled_format_parameter_set(labeled_execute_format),
        )
        for labeled_execute_format in cls.supported_execute_formats
    ]

__pydantic_init_subclass__(**kwargs) classmethod

Register all subclasses of BaseFixture with a fixture format name set as possible fixture formats.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
    """
    Register all subclasses of BaseFixture with a fixture format name set
    as possible fixture formats.
    """
    super().__pydantic_init_subclass__(**kwargs)

    # Don't register dynamically generated wrapper classes
    if getattr(cls, "__is_base_test_wrapper__", False):
        return

    if cls.pytest_parameter_name():
        # Register the new fixture format
        BaseTest.spec_types[cls.pytest_parameter_name()] = cls

from_test(*, base_test, **kwargs) classmethod

Create a test in a different format from a base test.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def from_test(
    cls,
    *,
    base_test: "BaseTest",
    **kwargs: Any,
) -> Self:
    """Create a test in a different format from a base test."""
    for k in BaseTest.model_fields.keys():
        if k not in kwargs and k in base_test.model_fields_set:
            kwargs[k] = getattr(base_test, k)
    return cls(**kwargs)

discard_execute_format_by_marks(execute_format, fork, markers) classmethod

Discard an execute format from executing if the appropriate marker is used.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def discard_execute_format_by_marks(
    cls,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
    fork: Fork | TransitionFork,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard an execute format from executing if the appropriate marker is
    used.
    """
    del execute_format, fork, markers
    return False

generate(*, t8n, fixture_format) abstractmethod

Generate the test fixture using the given fixture format.

Source code in packages/testing/src/execution_testing/specs/base.py

@abstractmethod
def generate(
    self,
    *,
    t8n: TransitionTool,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the test fixture using the given fixture format."""
    pass

execute(*, execute_format)

Generate the list of test fixtures.

Source code in packages/testing/src/execution_testing/specs/base.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    raise Exception(f"Unsupported execute format: {execute_format}")

pytest_parameter_name() classmethod

Must return the name of the parameter used in pytest to select this spec type as filler for the test.

By default, it returns the underscore separated name of the class.

Source code in packages/testing/src/execution_testing/specs/base.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Must return the name of the parameter used in pytest to select this
    spec type as filler for the test.

    By default, it returns the underscore separated name of the class.
    """
    if cls == BaseTest:
        return ""
    return reduce(
        lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__
    ).lower()

check_exception_test(*, exception)

Compare the test marker against the outcome of the test.

Source code in packages/testing/src/execution_testing/specs/base.py

def check_exception_test(
    self,
    *,
    exception: bool,
) -> None:
    """Compare the test marker against the outcome of the test."""
    if self.is_exception_test != exception:
        if exception:
            raise Exception(
                "Test produced an invalid block or transaction but was "
                "not marked with the `exception_test` marker. Add the "
                "`@pytest.mark.exception_test` decorator to the test."
            )
        else:
            raise Exception(
                "Test didn't produce an invalid block or transaction but "
                "was marked with the `exception_test` marker. Remove the "
                "`@pytest.mark.exception_test` decorator from the test."
            )

get_genesis_environment()

Get the genesis environment for pre-allocation groups.

Must be implemented by subclasses to provide the appropriate environment.

Source code in packages/testing/src/execution_testing/specs/base.py

def get_genesis_environment(self) -> Environment:
    """
    Get the genesis environment for pre-allocation groups.

    Must be implemented by subclasses to provide the appropriate
    environment.
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} must implement genesis environment "
        "access for use with pre-allocation groups."
    )

validate_benchmark_gas(*, benchmark_gas_used, gas_benchmark_value, benchmark_block_gas_used=None)

Validates the total consumed gas of the last block in the test matches the expectation of the benchmark test.

benchmark_gas_used is the combined gas across all dimensions (the receipt cumulativeGasUsed) and is checked against expected_benchmark_gas_used. benchmark_block_gas_used is the block-header gas, i.e. the maximum across the independent gas dimensions (EIP-8037); it is what must stay within the block gas limit, because the combined value can legitimately exceed it. When it is not available (e.g. execute mode), the combined value is used for the ceiling check instead.

Requires the following fields to be set: - expected_benchmark_gas_used - operation_mode

Source code in packages/testing/src/execution_testing/specs/base.py

def validate_benchmark_gas(
    self,
    *,
    benchmark_gas_used: int | None,
    gas_benchmark_value: int,
    benchmark_block_gas_used: int | None = None,
) -> None:
    """
    Validates the total consumed gas of the last block in the test matches
    the expectation of the benchmark test.

    ``benchmark_gas_used`` is the combined gas across all dimensions (the
    receipt ``cumulativeGasUsed``) and is checked against
    ``expected_benchmark_gas_used``. ``benchmark_block_gas_used`` is the
    block-header gas, i.e. the maximum across the independent gas
    dimensions (EIP-8037); it is what must stay within the block gas
    limit, because the combined value can legitimately exceed it. When it
    is not available (e.g. execute mode), the combined value is used for
    the ceiling check instead.

    Requires the following fields to be set:
    - expected_benchmark_gas_used
    - operation_mode
    """
    if self.operation_mode != OpMode.BENCHMARKING:
        return
    assert benchmark_gas_used is not None, "_benchmark_gas_used is not set"
    # Perform gas validation if required for benchmarking.
    # Ensures benchmark tests consume exactly the expected gas.
    if not self.skip_gas_used_validation:
        # Verify that the total gas consumed in the last block
        # matches expectations
        expected_benchmark_gas_used = self.expected_benchmark_gas_used
        if expected_benchmark_gas_used is None:
            expected_benchmark_gas_used = gas_benchmark_value
        diff = benchmark_gas_used - expected_benchmark_gas_used
        assert benchmark_gas_used == expected_benchmark_gas_used, (
            f"Total gas used ({benchmark_gas_used}) does not "
            "match expected benchmark gas "
            f"({expected_benchmark_gas_used}), "
            f"difference: {diff}"
        )
    # No single gas dimension may exceed the block gas limit. The
    # block-header gas is the max across dimensions; the combined
    # execution+state gas may exceed the target under EIP-8037, so the
    # ceiling is checked against the header value when available.
    block_gas_used = (
        benchmark_block_gas_used
        if benchmark_block_gas_used is not None
        else benchmark_gas_used
    )
    assert block_gas_used <= gas_benchmark_value, (
        f"benchmark block gas used ({block_gas_used}) exceeds maximum "
        "benchmark gas allowed for this configuration: "
        f"{gas_benchmark_value}"
    )

validate_receipt_status(*, receipts, block_number)

Validate receipt status for every transaction in a block.

When expected_receipt_status is set, verify that all receipts match. Catches silent OOG failures that roll back state and invalidate benchmarks.

Source code in packages/testing/src/execution_testing/specs/base.py

def validate_receipt_status(
    self,
    *,
    receipts: List[TransactionReceipt],
    block_number: int,
) -> None:
    """
    Validate receipt status for every transaction in a block.

    When expected_receipt_status is set, verify that all
    receipts match. Catches silent OOG failures that roll
    back state and invalidate benchmarks.
    """
    if self.expected_receipt_status is None:
        return
    for i, receipt in enumerate(receipts):
        if receipt.status is not None and (
            int(receipt.status) != self.expected_receipt_status
        ):
            raise Exception(
                f"Transaction {i} in block "
                f"{block_number} has receipt "
                f"status {int(receipt.status)}, "
                f"expected "
                f"{self.expected_receipt_status}."
            )

BenchmarkTest

Bases: BaseTest

Test type designed specifically for benchmark test cases.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

class BenchmarkTest(BaseTest):
    """Test type designed specifically for benchmark test cases."""

    model_config = ConfigDict(extra="forbid")

    pre: Alloc = Field(default_factory=Alloc)
    post: Alloc = Field(default_factory=Alloc)
    tx: Transaction | None = None
    setup_blocks: List[Block] = Field(default_factory=list)
    blocks: List[Block] | None = None
    block_exception: (
        List[TransactionException | BlockException]
        | TransactionException
        | BlockException
        | None
    ) = None
    env: Environment = Field(default_factory=Environment)
    expected_benchmark_gas_used: int | None = None
    gas_benchmark_value: int = Field(
        default_factory=lambda: int(Environment().gas_limit)
    )
    fixed_opcode_count: float | None = None
    expected_opcode_count: int | None = None
    target_opcode: Op | OpcodeTarget | None = None
    code_generator: BenchmarkCodeGenerator | None = None
    # By default, benchmark tests require neither of these
    include_full_post_state_in_output: bool = False
    include_tx_receipts_in_output: bool = False

    supported_fixture_formats: ClassVar[
        Sequence[FixtureFormat | LabeledFixtureFormat]
    ] = [
        BlockchainFixture,
        BlockchainEngineFixture,
        BlockchainEngineXFixture,
        BlockchainEngineStatefulFixture,
    ]

    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "benchmark_test",
            "An execute test derived from a benchmark test",
        ),
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "blockchain_test_engine_only": (
            "Only generate a blockchain test engine fixture"
        ),
        "blockchain_test_only": "Only generate a blockchain test fixture",
        "repricing": "Mark test as reference test for gas repricing analysis",
    }

    def model_post_init(self, __context: Any, /) -> None:
        """
        Model post-init to assert that the custom pre-allocation was
        provided and the default was not used.
        """
        super().model_post_init(__context)
        assert "pre" in self.model_fields_set, (
            "pre allocation was not provided"
        )

        set_props = [
            name
            for name, val in [
                ("code_generator", self.code_generator),
                ("blocks", self.blocks),
                ("tx", self.tx),
            ]
            if val is not None
        ]

        if len(set_props) != 1:
            raise ValueError(
                f"Exactly one must be set, but got {len(set_props)}: "
                f"{', '.join(set_props)}"
            )

        blocks: List[Block] = self.setup_blocks

        if self.fixed_opcode_count is not None and self.code_generator is None:
            pytest.skip(
                "Cannot run fixed opcode count tests without a code generator"
            )

        if self.code_generator is not None:
            # Inject fixed_opcode_count into the code generator if provided
            self.code_generator.fixed_opcode_count = self.fixed_opcode_count

            # In fixed opcode count mode, skip gas validation since we're
            # measuring performance by operation count, not gas usage
            if self.fixed_opcode_count is not None:
                self.skip_gas_used_validation = True
                generated_blocks = (
                    self.generate_fixed_opcode_count_transactions()
                )
            else:
                generated_blocks = self.generate_blocks_from_code_generator()
            blocks += generated_blocks

        elif self.blocks is not None:
            blocks += self.blocks

        elif self.tx is not None:
            gas_limit = (
                self.fork.fork_at(
                    block_number=0, timestamp=0
                ).transaction_gas_limit_cap()
                or self.gas_benchmark_value
            )

            transactions = self.split_transaction(self.tx, gas_limit)

            blocks.append(Block(txs=transactions))

        else:
            raise ValueError(
                "Cannot create BlockchainTest without a code generator, "
                "transactions, or blocks"
            )

        self.blocks = blocks

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Return the parameter name used in pytest
        to select this spec type.
        """
        return "benchmark_test"

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard a fixture format from filling if the
        appropriate marker is used.
        """
        if "blockchain_test_only" in [m.name for m in markers]:
            return fixture_format != BlockchainFixture
        if "blockchain_test_engine_only" in [m.name for m in markers]:
            return fixture_format != BlockchainEngineFixture
        return False

    def get_genesis_environment(self) -> Environment:
        """Get the genesis environment for this benchmark test."""
        return self.generate_blockchain_test().get_genesis_environment()

    def split_transaction(
        self, tx: Transaction, gas_limit_cap: int | None
    ) -> List[Transaction]:
        """
        Split a transaction that exceeds the gas
        limit cap into multiple transactions.
        """
        if gas_limit_cap is None:
            tx.gas_limit = HexNumber(self.gas_benchmark_value)
            return [tx]

        if gas_limit_cap >= self.gas_benchmark_value:
            tx.gas_limit = HexNumber(self.gas_benchmark_value)
            return [tx]

        num_splits = math.ceil(self.gas_benchmark_value / gas_limit_cap)
        remaining_gas = self.gas_benchmark_value

        split_transactions = []
        for i in range(num_splits):
            split_tx = tx.model_copy()
            split_tx.gas_limit = HexNumber(
                remaining_gas if i == num_splits - 1 else gas_limit_cap
            )
            remaining_gas -= gas_limit_cap
            split_tx.nonce = HexNumber(tx.nonce + i)
            split_transactions.append(split_tx)

        return split_transactions

    def generate_blocks_from_code_generator(self) -> List[Block]:
        """Generate blocks using the code generator."""
        if self.code_generator is None:
            raise Exception("Code generator is not set")
        self.code_generator.deploy_contracts(
            pre=self.pre, fork=self.fork.fork_at(block_number=0, timestamp=0)
        )
        gas_limit = (
            self.fork.fork_at(
                block_number=0, timestamp=0
            ).transaction_gas_limit_cap()
            or self.gas_benchmark_value
        )
        benchmark_tx = self.code_generator.generate_transaction(
            pre=self.pre, gas_benchmark_value=gas_limit
        )

        execution_txs = self.split_transaction(benchmark_tx, gas_limit)
        execution_block = Block(txs=execution_txs)

        return [execution_block]

    def generate_fixed_opcode_count_transactions(self) -> List[Block]:
        """Generate transactions with a fixed opcode count."""
        if self.code_generator is None:
            raise Exception("Code generator is not set")
        self.code_generator.deploy_fix_count_contracts(
            pre=self.pre, fork=self.fork.fork_at(block_number=0, timestamp=0)
        )
        gas_limit = (
            self.fork.fork_at(
                block_number=0, timestamp=0
            ).transaction_gas_limit_cap()
            or self.gas_benchmark_value
        )
        benchmark_tx = self.code_generator.generate_transaction(
            pre=self.pre, gas_benchmark_value=gas_limit
        )
        execution_block = Block(txs=[benchmark_tx])
        return [execution_block]

    def generate_blockchain_test(self) -> BlockchainTest:
        """Create a BlockchainTest from this BenchmarkTest."""
        return BlockchainTest.from_test(
            base_test=self,
            genesis_environment=self.env,
            pre=self.pre,
            post=self.post,
            blocks=self.blocks,
            include_full_post_state_in_output=self.include_full_post_state_in_output,
            include_tx_receipts_in_output=self.include_tx_receipts_in_output,
        )

    @staticmethod
    def _verify_target_opcode_count(
        *,
        target_opcode: Op | OpcodeTarget,
        opcode_count: OpcodeCount,
        expected_opcode_count: int,
        tolerance: float = 0.05,
    ) -> None:
        """Verify target opcode was executed the expected number of times."""
        count_opcode = (
            target_opcode.opcode
            if isinstance(target_opcode, OpcodeTarget)
            else target_opcode
        )
        actual = opcode_count.root.get(count_opcode, 0)
        max_difference = expected_opcode_count * tolerance  # 5% tolerance

        if abs(actual - expected_opcode_count) > max_difference:
            raise ValueError(
                f"Target opcode {target_opcode} count mismatch: "
                f"expected ~{expected_opcode_count} "
                f"(±{tolerance * 100}%), got {actual}"
            )

    def generate(
        self,
        t8n: TransitionTool,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the blockchain test fixture."""
        self.check_exception_test(
            exception=self.tx.error is not None if self.tx else False
        )
        if fixture_format in BlockchainTest.supported_fixture_formats:
            fill_result = self.generate_blockchain_test().generate(
                t8n=t8n, fixture_format=fixture_format
            )

            # Verify target opcode count if specified
            if (
                self.fixed_opcode_count is not None
                or self.expected_opcode_count is not None
            ):
                target_opcode = self.target_opcode
                assert target_opcode is not None, "target_opcode is not set"
                opcode_count = fill_result.benchmark_opcode_count
                if opcode_count is not None:
                    if self.fixed_opcode_count is not None:
                        self._verify_target_opcode_count(
                            target_opcode=target_opcode,
                            opcode_count=opcode_count,
                            expected_opcode_count=int(
                                # fixed_opcode_count is in thousands units
                                self.fixed_opcode_count * 1000
                            ),
                        )
                    elif self.expected_opcode_count is not None:
                        self._verify_target_opcode_count(
                            target_opcode=target_opcode,
                            opcode_count=opcode_count,
                            expected_opcode_count=self.expected_opcode_count,
                        )
                    else:
                        raise Exception(
                            "Opcode verification needs either "
                            "`fixed_opcode_count` or `expected_opcode_count` "
                            "to be set."
                        )

            if self.target_opcode is not None:
                fill_result.metadata["target_opcode"] = str(self.target_opcode)

            return fill_result
        else:
            raise Exception(f"Unsupported fixture format: {fixture_format}")

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Execute the benchmark test by sending it to the live network."""
        if execute_format == TransactionPost:
            assert self.blocks is not None
            return TransactionPost(
                blocks=[block.txs for block in self.blocks],
                post=self.post,
                benchmark_mode=True,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

model_post_init(__context)

Model post-init to assert that the custom pre-allocation was provided and the default was not used.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def model_post_init(self, __context: Any, /) -> None:
    """
    Model post-init to assert that the custom pre-allocation was
    provided and the default was not used.
    """
    super().model_post_init(__context)
    assert "pre" in self.model_fields_set, (
        "pre allocation was not provided"
    )

    set_props = [
        name
        for name, val in [
            ("code_generator", self.code_generator),
            ("blocks", self.blocks),
            ("tx", self.tx),
        ]
        if val is not None
    ]

    if len(set_props) != 1:
        raise ValueError(
            f"Exactly one must be set, but got {len(set_props)}: "
            f"{', '.join(set_props)}"
        )

    blocks: List[Block] = self.setup_blocks

    if self.fixed_opcode_count is not None and self.code_generator is None:
        pytest.skip(
            "Cannot run fixed opcode count tests without a code generator"
        )

    if self.code_generator is not None:
        # Inject fixed_opcode_count into the code generator if provided
        self.code_generator.fixed_opcode_count = self.fixed_opcode_count

        # In fixed opcode count mode, skip gas validation since we're
        # measuring performance by operation count, not gas usage
        if self.fixed_opcode_count is not None:
            self.skip_gas_used_validation = True
            generated_blocks = (
                self.generate_fixed_opcode_count_transactions()
            )
        else:
            generated_blocks = self.generate_blocks_from_code_generator()
        blocks += generated_blocks

    elif self.blocks is not None:
        blocks += self.blocks

    elif self.tx is not None:
        gas_limit = (
            self.fork.fork_at(
                block_number=0, timestamp=0
            ).transaction_gas_limit_cap()
            or self.gas_benchmark_value
        )

        transactions = self.split_transaction(self.tx, gas_limit)

        blocks.append(Block(txs=transactions))

    else:
        raise ValueError(
            "Cannot create BlockchainTest without a code generator, "
            "transactions, or blocks"
        )

    self.blocks = blocks

pytest_parameter_name() classmethod

Return the parameter name used in pytest to select this spec type.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Return the parameter name used in pytest
    to select this spec type.
    """
    return "benchmark_test"

discard_fixture_format_by_marks(fixture_format, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard a fixture format from filling if the
    appropriate marker is used.
    """
    if "blockchain_test_only" in [m.name for m in markers]:
        return fixture_format != BlockchainFixture
    if "blockchain_test_engine_only" in [m.name for m in markers]:
        return fixture_format != BlockchainEngineFixture
    return False

get_genesis_environment()

Get the genesis environment for this benchmark test.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def get_genesis_environment(self) -> Environment:
    """Get the genesis environment for this benchmark test."""
    return self.generate_blockchain_test().get_genesis_environment()

split_transaction(tx, gas_limit_cap)

Split a transaction that exceeds the gas limit cap into multiple transactions.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def split_transaction(
    self, tx: Transaction, gas_limit_cap: int | None
) -> List[Transaction]:
    """
    Split a transaction that exceeds the gas
    limit cap into multiple transactions.
    """
    if gas_limit_cap is None:
        tx.gas_limit = HexNumber(self.gas_benchmark_value)
        return [tx]

    if gas_limit_cap >= self.gas_benchmark_value:
        tx.gas_limit = HexNumber(self.gas_benchmark_value)
        return [tx]

    num_splits = math.ceil(self.gas_benchmark_value / gas_limit_cap)
    remaining_gas = self.gas_benchmark_value

    split_transactions = []
    for i in range(num_splits):
        split_tx = tx.model_copy()
        split_tx.gas_limit = HexNumber(
            remaining_gas if i == num_splits - 1 else gas_limit_cap
        )
        remaining_gas -= gas_limit_cap
        split_tx.nonce = HexNumber(tx.nonce + i)
        split_transactions.append(split_tx)

    return split_transactions

generate_blocks_from_code_generator()

Generate blocks using the code generator.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def generate_blocks_from_code_generator(self) -> List[Block]:
    """Generate blocks using the code generator."""
    if self.code_generator is None:
        raise Exception("Code generator is not set")
    self.code_generator.deploy_contracts(
        pre=self.pre, fork=self.fork.fork_at(block_number=0, timestamp=0)
    )
    gas_limit = (
        self.fork.fork_at(
            block_number=0, timestamp=0
        ).transaction_gas_limit_cap()
        or self.gas_benchmark_value
    )
    benchmark_tx = self.code_generator.generate_transaction(
        pre=self.pre, gas_benchmark_value=gas_limit
    )

    execution_txs = self.split_transaction(benchmark_tx, gas_limit)
    execution_block = Block(txs=execution_txs)

    return [execution_block]

generate_fixed_opcode_count_transactions()

Generate transactions with a fixed opcode count.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def generate_fixed_opcode_count_transactions(self) -> List[Block]:
    """Generate transactions with a fixed opcode count."""
    if self.code_generator is None:
        raise Exception("Code generator is not set")
    self.code_generator.deploy_fix_count_contracts(
        pre=self.pre, fork=self.fork.fork_at(block_number=0, timestamp=0)
    )
    gas_limit = (
        self.fork.fork_at(
            block_number=0, timestamp=0
        ).transaction_gas_limit_cap()
        or self.gas_benchmark_value
    )
    benchmark_tx = self.code_generator.generate_transaction(
        pre=self.pre, gas_benchmark_value=gas_limit
    )
    execution_block = Block(txs=[benchmark_tx])
    return [execution_block]

generate_blockchain_test()

Create a BlockchainTest from this BenchmarkTest.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def generate_blockchain_test(self) -> BlockchainTest:
    """Create a BlockchainTest from this BenchmarkTest."""
    return BlockchainTest.from_test(
        base_test=self,
        genesis_environment=self.env,
        pre=self.pre,
        post=self.post,
        blocks=self.blocks,
        include_full_post_state_in_output=self.include_full_post_state_in_output,
        include_tx_receipts_in_output=self.include_tx_receipts_in_output,
    )

generate(t8n, fixture_format)

Generate the blockchain test fixture.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def generate(
    self,
    t8n: TransitionTool,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the blockchain test fixture."""
    self.check_exception_test(
        exception=self.tx.error is not None if self.tx else False
    )
    if fixture_format in BlockchainTest.supported_fixture_formats:
        fill_result = self.generate_blockchain_test().generate(
            t8n=t8n, fixture_format=fixture_format
        )

        # Verify target opcode count if specified
        if (
            self.fixed_opcode_count is not None
            or self.expected_opcode_count is not None
        ):
            target_opcode = self.target_opcode
            assert target_opcode is not None, "target_opcode is not set"
            opcode_count = fill_result.benchmark_opcode_count
            if opcode_count is not None:
                if self.fixed_opcode_count is not None:
                    self._verify_target_opcode_count(
                        target_opcode=target_opcode,
                        opcode_count=opcode_count,
                        expected_opcode_count=int(
                            # fixed_opcode_count is in thousands units
                            self.fixed_opcode_count * 1000
                        ),
                    )
                elif self.expected_opcode_count is not None:
                    self._verify_target_opcode_count(
                        target_opcode=target_opcode,
                        opcode_count=opcode_count,
                        expected_opcode_count=self.expected_opcode_count,
                    )
                else:
                    raise Exception(
                        "Opcode verification needs either "
                        "`fixed_opcode_count` or `expected_opcode_count` "
                        "to be set."
                    )

        if self.target_opcode is not None:
            fill_result.metadata["target_opcode"] = str(self.target_opcode)

        return fill_result
    else:
        raise Exception(f"Unsupported fixture format: {fixture_format}")

execute(*, execute_format)

Execute the benchmark test by sending it to the live network.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Execute the benchmark test by sending it to the live network."""
    if execute_format == TransactionPost:
        assert self.blocks is not None
        return TransactionPost(
            blocks=[block.txs for block in self.blocks],
            post=self.post,
            benchmark_mode=True,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

OpcodeTarget dataclass

Map a display name to an underlying opcode for count validation.

Use when the fixture metadata should show a descriptive label (e.g. a precompile name) while opcode-count validation targets the real EVM opcode that gets executed (e.g. STATICCALL).

Source code in packages/testing/src/execution_testing/specs/benchmark.py

@dataclass(frozen=True)
class OpcodeTarget:
    """
    Map a display name to an underlying opcode for count validation.

    Use when the fixture metadata should show a descriptive label (e.g. a
    precompile name) while opcode-count validation targets the real EVM
    opcode that gets executed (e.g. STATICCALL).
    """

    name: str
    opcode: Op

    def __str__(self) -> str:
        """Return the display name."""
        return self.name

__str__()

Return the display name.

Source code in packages/testing/src/execution_testing/specs/benchmark.py

def __str__(self) -> str:
    """Return the display name."""
    return self.name

BlobsTest

Bases: BaseTest

Test specification for blob tests.

Source code in packages/testing/src/execution_testing/specs/blobs.py

class BlobsTest(BaseTest):
    """Test specification for blob tests."""

    pre: Alloc
    txs: List[NetworkWrappedTransaction | Transaction]
    nonexisting_blob_hashes: List[Hash] | None = None
    interleave_nonexisting_blob_hashes: bool = False
    get_blobs_version: int | None = None
    cell_mask: int | None = None
    custody_columns: bytes | None = None

    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            BlobTransaction,
            "blob_transaction_test",
            "A test that executes a blob transaction",
        ),
    ]

    def generate(
        self,
        *,
        t8n: TransitionTool,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the list of test fixtures."""
        del t8n
        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == BlobTransaction:
            return BlobTransaction(
                txs=self.txs,
                nonexisting_blob_hashes=self.nonexisting_blob_hashes,
                interleave_nonexisting_blob_hashes=(
                    self.interleave_nonexisting_blob_hashes
                ),
                get_blobs_version=self.get_blobs_version,
                cell_mask=self.cell_mask,
                custody_columns=self.custody_columns,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

generate(*, t8n, fixture_format)

Generate the list of test fixtures.

Source code in packages/testing/src/execution_testing/specs/blobs.py

def generate(
    self,
    *,
    t8n: TransitionTool,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the list of test fixtures."""
    del t8n
    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, execute_format)

Generate the list of test fixtures.

Source code in packages/testing/src/execution_testing/specs/blobs.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == BlobTransaction:
        return BlobTransaction(
            txs=self.txs,
            nonexisting_blob_hashes=self.nonexisting_blob_hashes,
            interleave_nonexisting_blob_hashes=(
                self.interleave_nonexisting_blob_hashes
            ),
            get_blobs_version=self.get_blobs_version,
            cell_mask=self.cell_mask,
            custody_columns=self.custody_columns,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

Block

Bases: Header

Block type used to describe block properties in test specs.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

class Block(Header):
    """Block type used to describe block properties in test specs."""

    header_verify: Header | None = None
    # If set, the block header will be verified against the specified values.
    rlp_modifier: Header | None = None
    """
    An RLP modifying header which values would be used to override the ones
    returned by the `ethereum_clis.TransitionTool`.
    """
    expected_block_access_list: BlockAccessListExpectation | None = None
    """
    If set, the block access list will be verified and potentially corrupted
    for invalid tests.
    """
    exception: BLOCK_EXCEPTION_TYPE = None
    # If set, the block is expected to be rejected by the client.
    skip_exception_verification: bool = False
    """
    Skip verifying that the exception is returned by the transition tool. This
    could be because the exception is inserted in the block after the
    transition tool evaluates it.
    """
    engine_api_error_code: EngineAPIError | None = None
    """
    If set, the block is expected to produce an error response from the Engine
    API.
    """
    include_receipts_in_output: bool | None = None
    """
    If set to `True`, the block’s output fixture representation will include
    full transaction receipts. If unset, the test-level value is used.
    """
    txs: List[Transaction] = Field(default_factory=list)
    """List of transactions included in the block."""
    ommers: List[Header] | None = None
    """List of ommer headers included in the block."""
    withdrawals: List[Withdrawal] | None = None
    """List of withdrawals to perform for this block."""
    requests: List[Bytes] | None = None
    """Custom list of requests to embed in this block."""
    expected_post_state: Alloc | None = None
    """Post state for verification after block execution in BlockchainTest"""
    block_access_list: Bytes | None = Field(None)
    """EIP-7928: Block-level access lists (serialized)."""
    engine_new_payload_block_access_list: Bytes | None = None
    """EIP-7928: override only the engine newPayload blockAccessList field."""
    engine_new_payload_slot_number: HexNumber | None = None
    """EIP-7843: override only the engine payload slotNumber field."""
    expected_gas_used: int | None = None
    """Expected gas used for the block."""

    @property
    def phase(self) -> TestPhase | None:
        """
        Return the single phase shared by all txs, or ``None`` when the
        block has no phase-tagged txs.

        Mixed-phase blocks must be split via ``_split_blocks_by_phase``
        before this property is read — they would otherwise need an
        arbitrary tiebreaker, which is a bug, not a default.
        """
        phases = {_tx_phase(tx) for tx in self.txs}
        phases.discard(None)
        if not phases:
            return None
        if len(phases) == 1:
            return next(iter(phases))
        raise AssertionError(
            f"Block.phase called on mixed-phase block (phases={phases}); "
            "split via _split_blocks_by_phase first."
        )

    def engine_payload_only_overrides(self) -> List[str]:
        """
        Return the names of the settings that only reach the engine payload.

        An RLP block cannot carry them, so a fixture that delivers blocks as
        RLP would present a valid block while claiming the exception they
        are meant to cause.
        """
        overrides: List[str] = []
        if self.engine_new_payload_block_access_list is not None:
            overrides.append("engine_new_payload_block_access_list")
        if self.engine_new_payload_slot_number is not None:
            overrides.append("engine_new_payload_slot_number")
        if (
            self.expected_block_access_list is not None
            and self.expected_block_access_list.has_rlp_modifier
        ):
            overrides.append("expected_block_access_list.modify_rlp")
        return overrides

    def set_environment(self, env: Environment) -> Environment:
        """
        Create copy of the environment with the characteristics of this
        specific block.
        """
        new_env_values: Dict[str, Any] = {}

        """
        Values that need to be set in the environment and are `None` for this
        block need to be set to their defaults.
        """
        new_env_values["difficulty"] = self.difficulty
        new_env_values["prev_randao"] = self.prev_randao
        new_env_values["fee_recipient"] = (
            self.fee_recipient
            if self.fee_recipient is not None
            else Environment().fee_recipient
        )
        new_env_values["gas_limit"] = (
            self.gas_limit or env.parent_gas_limit or Environment().gas_limit
        )
        if not isinstance(self.base_fee_per_gas, Removable):
            new_env_values["base_fee_per_gas"] = self.base_fee_per_gas
        new_env_values["withdrawals"] = self.withdrawals
        if not isinstance(self.excess_blob_gas, Removable):
            new_env_values["excess_blob_gas"] = self.excess_blob_gas
        if not isinstance(self.blob_gas_used, Removable):
            new_env_values["blob_gas_used"] = self.blob_gas_used
        if not isinstance(self.parent_beacon_block_root, Removable):
            new_env_values["parent_beacon_block_root"] = (
                self.parent_beacon_block_root
            )
        if (
            not isinstance(self.requests_hash, Removable)
            and self.block_access_list is not None
        ):
            new_env_values["block_access_list_hash"] = (
                self.block_access_list.keccak256()
            )
            new_env_values["block_access_list"] = self.block_access_list
        if (
            not isinstance(self.block_access_list, Removable)
            and self.block_access_list is not None
        ):
            new_env_values["block_access_list"] = self.block_access_list
        if not isinstance(self.slot_number, Removable):
            new_env_values["slot_number"] = self.slot_number
        """
        These values are required, but they depend on the previous environment,
        so they can be calculated here.
        """
        if self.number is not None:
            new_env_values["number"] = self.number
        else:
            # calculate the next block number for the environment
            if len(env.block_hashes) == 0:
                new_env_values["number"] = 0
            else:
                new_env_values["number"] = (
                    max([Number(n) for n in env.block_hashes.keys()]) + 1
                )

        if self.timestamp is not None:
            new_env_values["timestamp"] = self.timestamp
        else:
            assert env.parent_timestamp is not None
            new_env_values["timestamp"] = int(
                Number(env.parent_timestamp) + 12
            )

        return env.copy(**new_env_values)

rlp_modifier = None class-attribute instance-attribute

An RLP modifying header which values would be used to override the ones returned by the ethereum_clis.TransitionTool.

expected_block_access_list = None class-attribute instance-attribute

If set, the block access list will be verified and potentially corrupted for invalid tests.

skip_exception_verification = False class-attribute instance-attribute

Skip verifying that the exception is returned by the transition tool. This could be because the exception is inserted in the block after the transition tool evaluates it.

engine_api_error_code = None class-attribute instance-attribute

If set, the block is expected to produce an error response from the Engine API.

include_receipts_in_output = None class-attribute instance-attribute

If set to True, the block’s output fixture representation will include full transaction receipts. If unset, the test-level value is used.

txs = Field(default_factory=list) class-attribute instance-attribute

List of transactions included in the block.

ommers = None class-attribute instance-attribute

List of ommer headers included in the block.

withdrawals = None class-attribute instance-attribute

List of withdrawals to perform for this block.

requests = None class-attribute instance-attribute

Custom list of requests to embed in this block.

expected_post_state = None class-attribute instance-attribute

Post state for verification after block execution in BlockchainTest

block_access_list = Field(None) class-attribute instance-attribute

EIP-7928: Block-level access lists (serialized).

engine_new_payload_block_access_list = None class-attribute instance-attribute

EIP-7928: override only the engine newPayload blockAccessList field.

engine_new_payload_slot_number = None class-attribute instance-attribute

EIP-7843: override only the engine payload slotNumber field.

expected_gas_used = None class-attribute instance-attribute

Expected gas used for the block.

phase property

Return the single phase shared by all txs, or None when the block has no phase-tagged txs.

Mixed-phase blocks must be split via _split_blocks_by_phase before this property is read — they would otherwise need an arbitrary tiebreaker, which is a bug, not a default.

engine_payload_only_overrides()

Return the names of the settings that only reach the engine payload.

An RLP block cannot carry them, so a fixture that delivers blocks as RLP would present a valid block while claiming the exception they are meant to cause.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def engine_payload_only_overrides(self) -> List[str]:
    """
    Return the names of the settings that only reach the engine payload.

    An RLP block cannot carry them, so a fixture that delivers blocks as
    RLP would present a valid block while claiming the exception they
    are meant to cause.
    """
    overrides: List[str] = []
    if self.engine_new_payload_block_access_list is not None:
        overrides.append("engine_new_payload_block_access_list")
    if self.engine_new_payload_slot_number is not None:
        overrides.append("engine_new_payload_slot_number")
    if (
        self.expected_block_access_list is not None
        and self.expected_block_access_list.has_rlp_modifier
    ):
        overrides.append("expected_block_access_list.modify_rlp")
    return overrides

set_environment(env)

Create copy of the environment with the characteristics of this specific block.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def set_environment(self, env: Environment) -> Environment:
    """
    Create copy of the environment with the characteristics of this
    specific block.
    """
    new_env_values: Dict[str, Any] = {}

    """
    Values that need to be set in the environment and are `None` for this
    block need to be set to their defaults.
    """
    new_env_values["difficulty"] = self.difficulty
    new_env_values["prev_randao"] = self.prev_randao
    new_env_values["fee_recipient"] = (
        self.fee_recipient
        if self.fee_recipient is not None
        else Environment().fee_recipient
    )
    new_env_values["gas_limit"] = (
        self.gas_limit or env.parent_gas_limit or Environment().gas_limit
    )
    if not isinstance(self.base_fee_per_gas, Removable):
        new_env_values["base_fee_per_gas"] = self.base_fee_per_gas
    new_env_values["withdrawals"] = self.withdrawals
    if not isinstance(self.excess_blob_gas, Removable):
        new_env_values["excess_blob_gas"] = self.excess_blob_gas
    if not isinstance(self.blob_gas_used, Removable):
        new_env_values["blob_gas_used"] = self.blob_gas_used
    if not isinstance(self.parent_beacon_block_root, Removable):
        new_env_values["parent_beacon_block_root"] = (
            self.parent_beacon_block_root
        )
    if (
        not isinstance(self.requests_hash, Removable)
        and self.block_access_list is not None
    ):
        new_env_values["block_access_list_hash"] = (
            self.block_access_list.keccak256()
        )
        new_env_values["block_access_list"] = self.block_access_list
    if (
        not isinstance(self.block_access_list, Removable)
        and self.block_access_list is not None
    ):
        new_env_values["block_access_list"] = self.block_access_list
    if not isinstance(self.slot_number, Removable):
        new_env_values["slot_number"] = self.slot_number
    """
    These values are required, but they depend on the previous environment,
    so they can be calculated here.
    """
    if self.number is not None:
        new_env_values["number"] = self.number
    else:
        # calculate the next block number for the environment
        if len(env.block_hashes) == 0:
            new_env_values["number"] = 0
        else:
            new_env_values["number"] = (
                max([Number(n) for n in env.block_hashes.keys()]) + 1
            )

    if self.timestamp is not None:
        new_env_values["timestamp"] = self.timestamp
    else:
        assert env.parent_timestamp is not None
        new_env_values["timestamp"] = int(
            Number(env.parent_timestamp) + 12
        )

    return env.copy(**new_env_values)

BlockchainTest

Bases: BaseTest

Filler type that tests multiple blocks (valid or invalid) in a chain.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

class BlockchainTest(BaseTest):
    """Filler type that tests multiple blocks (valid or invalid) in a chain."""

    pre: Alloc
    post: Alloc
    blocks: List[Block]
    genesis_environment: Environment = Field(default_factory=Environment)
    chain_id: int = Field(
        default_factory=lambda: ChainConfigDefaults.chain_id,
        validate_default=True,
    )
    include_full_post_state_in_output: bool = True
    """
    Include the post state in the fixture output. Otherwise, the state
    verification is only performed based on the state root.
    """
    include_tx_receipts_in_output: bool = True
    """
    Include transaction receipts in the fixture output.
    """

    supported_fixture_formats: ClassVar[
        Sequence[FixtureFormat | LabeledFixtureFormat]
    ] = [
        BlockchainFixture,
        BlockchainEngineFixture,
        BlockchainEngineSyncFixture,
        BlockchainEngineXFixture,
        BlockchainEngineStatefulFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "blockchain_test",
            "An execute test derived from a blockchain test",
        ),
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "blockchain_test_engine_only": (
            "Only generate a blockchain test engine fixture"
        ),
        "blockchain_test_only": "Only generate a blockchain test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard a fixture format from filling if the appropriate marker is
        used.
        """
        marker_names = [m.name for m in markers]
        if (
            fixture_format != BlockchainFixture
            and "blockchain_test_only" in marker_names
        ):
            return True
        if (
            fixture_format
            not in [BlockchainEngineFixture, BlockchainEngineXFixture]
            and "blockchain_test_engine_only" in marker_names
        ):
            return True
        return False

    def model_post_init(self, __context: Any, /) -> None:
        """
        Model post-init to assert static (pre-fill/execute) checks.
        """
        super().model_post_init(__context)
        if self.is_inclusion_test:
            # Verify that the blockchain contains at most one invalid
            # transaction which must be located at the end of the last block.
            for i, block in enumerate(self.blocks):
                if i != (len(self.blocks) - 1):
                    valid_tx_range = block.txs[:]
                else:
                    valid_tx_range = block.txs[:-1]
                if any(tx.error is not None for tx in valid_tx_range):
                    raise Exception(
                        "test correctness: in an inclusion test the only "
                        "transaction allowed to produce an exception is the "
                        "last transaction of the last block, but block "
                        f"{i} contains an invalid transaction elsewhere"
                    )
        for i, block in enumerate(self.blocks):
            expectation = block.expected_block_access_list
            if (
                expectation is not None
                and expectation.has_modifier
                and not block.exception
                and block.engine_api_error_code is None
            ):
                raise Exception(
                    f"test correctness: block {i} modifies its block access "
                    "list or its encoding but declares no `exception` or "
                    "`engine_api_error_code`, so the corrupted block access "
                    "list would be filled as valid. Declare the exception "
                    "the modified block access list must cause, or drop the "
                    "modifier."
                )
            if (
                block.engine_new_payload_block_access_list is not None
                and expectation is not None
                and expectation.has_rlp_modifier
            ):
                raise Exception(
                    f"test correctness: block {i} sets "
                    "`engine_new_payload_block_access_list` and re-encodes "
                    "the block access list with `modify_rlp`; the explicit "
                    "payload override would discard the re-encoding. Keep "
                    "one: `engine_new_payload_block_access_list` delivers "
                    "arbitrary bytes, `modify_rlp` re-encodes the list the "
                    "transition tool produced."
                )

    def get_genesis_environment(self) -> Environment:
        """Get the genesis environment for pre-allocation groups."""
        # Checked before the defaults below add the mix hash under
        # `prev_randao`, which every genesis header carries.
        self.genesis_environment.check_fork_fields(
            self.fork.transitions_from()
        )
        modified_values = self.genesis_environment.set_fork_requirements(
            self.fork.transitions_from()
        ).model_dump(exclude_unset=True)
        return Environment(**(GENESIS_ENVIRONMENT_DEFAULTS | modified_values))

    def make_genesis(
        self, *, apply_pre_allocation_blockchain: bool
    ) -> Tuple[Alloc, FixtureBlock]:
        """Create a genesis block from the blockchain test definition."""
        env = self.get_genesis_environment()
        assert env.withdrawals is None or len(env.withdrawals) == 0, (
            "withdrawals must be empty at genesis"
        )
        assert (
            env.parent_beacon_block_root is None
            or env.parent_beacon_block_root == Hash(0)
        ), "parent_beacon_block_root must be empty at genesis"

        pre_alloc = self.pre
        if apply_pre_allocation_blockchain:
            pre_alloc = Alloc.merge(
                Alloc.model_validate(
                    self.fork.transitions_to().pre_allocation_blockchain()
                ),
                pre_alloc,
            )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")
        if pre_alloc.state_commitment() is None:
            pre_alloc.migrate_state_commitment(
                self.fork.transitions_from().state_commitment()
            )
        state_root = pre_alloc.state_root()
        genesis = FixtureHeader.genesis(
            self.fork.transitions_from(), env, state_root
        )

        return (
            pre_alloc,
            FixtureBlockBase(
                header=genesis,
                withdrawals=None if env.withdrawals is None else [],
            ).with_rlp(txs=[]),
        )

    def generate_block_data(
        self,
        t8n: FillerBackend,
        block: Block,
        previous_env: Environment,
        previous_alloc: Alloc | LazyAlloc,
    ) -> BuiltBlock:
        """
        Generate common block data for both make_fixture and make_hive_fixture.

        ``t8n`` is any backend satisfying ``FillerBackend``. The
        default compute path passes a concrete ``TransitionTool``; stateful
        filling will pass an ``ClientBackend`` that drives
        ``testing_buildBlockV1`` against a live client.
        """
        env = block.set_environment(previous_env)
        fork = self.fork.fork_at(
            block_number=env.number, timestamp=env.timestamp
        )
        env = env.set_fork_requirements(fork)
        env.check_fork_fields(fork)
        txs = block.txs[:]
        if any("gas_limit" not in tx.model_fields_set for tx in block.txs):
            max_tx_gas_limit = Transaction.calculate_max_gas_limit(
                txs=txs,
                env_gas_limit=int(env.gas_limit),
                transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
                state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
            )
            if max_tx_gas_limit == 0:
                raise Exception(
                    "test correctness: unable to automatically calculate gas "
                    "limit for transactions (No remaining gas)."
                )
            txs = [
                tx.with_gas_limit(
                    max_gas_limit=max_tx_gas_limit,
                    transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
                    state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
                )
                for tx in txs
            ]
        txs = [tx.with_signature_and_sender() for tx in txs]

        if (failing_tx_count := len([tx for tx in txs if tx.error])) > 0:
            if failing_tx_count > 1:
                raise Exception(
                    "test correctness: only one transaction can produce "
                    "an exception in a block"
                )
            if not txs[-1].error:
                raise Exception(
                    "test correctness: the transaction that produces an "
                    "exception must be the last transaction in the block"
                )

        transition_tool_output = t8n.evaluate(
            transition_tool_data=TransitionTool.TransitionToolData(
                alloc=previous_alloc,
                txs=txs,
                env=env,
                fork=fork,
                chain_id=self.chain_id,
                reward=fork.get_reward(),
                blob_schedule=fork.blob_schedule(),
            ),
            slow_request=self.is_tx_gas_heavy_test,
        )

        # One special case of the invalid transactions is the blob gas used,
        # since this value is not included in the transition tool result, but
        # it is included in the block header, and some clients check it before
        # executing the block by simply counting the type-3 txs, we need to set
        # the correct value by default.
        blob_gas_used: int | None = None
        if fork.supports_blobs():
            if (blob_gas_per_blob := fork.blob_gas_per_blob()) > 0:
                blob_gas_used = blob_gas_per_blob * count_blobs(txs)

        # Prepare slot_number for header initialization
        slot_number_value: ZeroPaddedHexNumber | None = None
        if fork.header_slot_number_required():
            slot_number_value = ZeroPaddedHexNumber(
                int(env.slot_number) if env.slot_number is not None else 0
            )

        header = FixtureHeader(
            **(
                transition_tool_output.result.model_dump(
                    exclude_none=True,
                    exclude={"blob_gas_used", "transactions_trie"},
                )
                | env.model_dump(
                    exclude_none=True,
                    exclude={"blob_gas_used", "slot_number"},
                )
            ),
            blob_gas_used=blob_gas_used,
            transactions_trie=Transaction.list_root(txs),
            extra_data=(
                block.extra_data if block.extra_data is not None else b""
            ),
            slot_number=slot_number_value,
            fork=fork,
        )

        if block.header_verify is not None:
            # Verify the header after transition tool processing.
            try:
                block.header_verify.verify(header)
            except Exception as e:
                raise Exception(
                    f"Verification of block {int(env.number)} failed"
                ) from e

        if block.expected_gas_used is not None:
            gas_used = int(transition_tool_output.result.gas_used)
            assert gas_used == block.expected_gas_used, (
                f"gas_used ({gas_used}) does not match expected_gas_used "
                f"({block.expected_gas_used})"
                f", difference: {gas_used - block.expected_gas_used}"
            )

        requests_list: List[Bytes] | None = None
        if fork.header_requests_required():
            assert transition_tool_output.result.requests is not None, (
                "Requests are required for this block"
            )
            requests = Requests(
                requests_lists=list(transition_tool_output.result.requests)
            )

            if Hash(requests) != header.requests_hash:
                raise Exception(
                    "Requests root in header does not match the requests "
                    "root in the transition tool output: "
                    f"{header.requests_hash} != {Hash(requests)}"
                )

            requests_list = requests.requests_list

        if block.requests is not None:
            header.requests_hash = Hash(
                Requests(requests_lists=list(block.requests))
            )
            requests_list = block.requests

        # Decode BAL from RLP bytes provided by the transition tool.
        t8n_bal_rlp = transition_tool_output.result.block_access_list
        t8n_bal: BlockAccessList | None = None
        if t8n_bal_rlp is not None:
            t8n_bal = BlockAccessList.from_rlp(t8n_bal_rlp)

        if fork.header_bal_hash_required():
            assert t8n_bal is not None, (
                "Block access list is required for this block but was not "
                "provided by the transition tool"
            )

            if t8n.attests_block_access_list_hash:
                computed_block_access_list_hash = Hash(t8n_bal.rlp.keccak256())
                assert (
                    computed_block_access_list_hash
                    == header.block_access_list_hash
                ), (
                    "Block access list hash in header does not match the "
                    f"computed hash from BAL: {header.block_access_list_hash} "
                    f"!= {computed_block_access_list_hash}"
                )

        if block.rlp_modifier is not None:
            # Modify any parameter specified in the `rlp_modifier` after
            # transition tool processing.
            header = block.rlp_modifier.apply(header)
            header.fork = fork  # Deleted during `apply` because `exclude=True`

        # Process block access list - apply transformer if present for invalid
        # tests
        bal = t8n_bal
        bal_rlp_override: Bytes | None = None

        # Always validate BAL structural integrity (ordering, duplicates)
        # if present
        if t8n_bal is not None:
            t8n_bal.validate_structure()

        # If expected BAL is defined, verify against it
        if (
            block.expected_block_access_list is not None
            and t8n_bal is not None
        ):
            block.expected_block_access_list.verify_against(t8n_bal)

            bal = block.expected_block_access_list.modify_if_invalid_test(
                t8n_bal
            )
            if bal.rlp != t8n_bal.rlp:
                # Compare bytes, not contents: an encoding override also
                # moves the header commitment.
                header.block_access_list_hash = Hash(bal.rlp.keccak256())
            bal_rlp_override = block.expected_block_access_list.modified_rlp(
                bal
            )
            if (
                block.expected_block_access_list.has_modifier
                and bal.rlp == t8n_bal.rlp
                and (
                    bal_rlp_override is None or bal_rlp_override == t8n_bal.rlp
                )
            ):
                raise Exception(
                    f"test correctness: block number {int(env.number)}'s "
                    "block access list modifier left the list unchanged, so "
                    "the block would be labelled invalid for no reason. Make "
                    "the modifier change the list, or drop it along with the "
                    "exception."
                )
            if bal.has_rlp_override and (
                block.engine_new_payload_block_access_list is not None
                or bal_rlp_override is not None
            ):
                raise Exception(
                    f"test correctness: block number {int(env.number)} "
                    "re-encodes the block access list with `override_rlp` "
                    "and also replaces the payload bytes, so the payload "
                    "would not carry what the header commits to. Keep one: "
                    "`override_rlp` commits the header to its re-encoding, "
                    "`modify_rlp` re-encodes the payload only, "
                    "`engine_new_payload_block_access_list` delivers "
                    "arbitrary bytes."
                )

        built_block_kwargs: Dict[str, Any] = dict(
            header=header,
            alloc=transition_tool_output.alloc,
            state_root=transition_tool_output.result.state_root,
            env=env,
            txs=txs,
            ommers=[],
            withdrawals=env.withdrawals,
            requests=requests_list,
            result=transition_tool_output.result,
            expected_exception=block.exception,
            engine_api_error_code=block.engine_api_error_code,
            rlp_modifier=block.rlp_modifier,
            fork=fork,
            block_access_list=bal,
            engine_new_payload_block_access_list=(
                block.engine_new_payload_block_access_list
                if block.engine_new_payload_block_access_list is not None
                else bal_rlp_override
            ),
            engine_new_payload_slot_number=(
                block.engine_new_payload_slot_number
            ),
        )
        built_block: BuiltBlock
        if transition_tool_output.engine_payload is not None:
            built_block = TestingBuildBlock(
                **built_block_kwargs,
                engine_payload=transition_tool_output.engine_payload,
            )
        else:
            built_block = BuiltBlock(**built_block_kwargs)

        try:
            rejected_txs = built_block.verify_transactions(
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
            if (
                not rejected_txs
                and block.rlp_modifier is None
                and block.requests is None
                and not block.skip_exception_verification
                and block.engine_new_payload_block_access_list is None
                and block.engine_new_payload_slot_number is None
                and not (
                    block.expected_block_access_list is not None
                    and block.expected_block_access_list.has_modifier
                )
            ):
                # Only verify block level exception if: - No transaction
                # exception was raised, because these are not reported as block
                # exceptions. - No RLP modifier was specified, because the
                # modifier is what normally produces the block exception. - No
                # requests were specified, because modified requests are also
                # what normally produces the block exception. - No engine
                # payload BAL override was specified, because it corrupts only
                # the engine payload after the transition tool has run. - No
                # BAL modifier was specified, because a rewritten BAL, whether
                # in contents or in encoding, is applied after the transition
                # tool has run and is what produces the block exception.
                built_block.verify_block_exception(
                    transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
                )
            verify_result(transition_tool_output.result, env)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(previous_alloc)
            pprint(transition_tool_output.alloc.materialize())
            raise e

        if len(rejected_txs) > 0 and block.exception is None:
            print_traces(t8n.get_traces())
            raise Exception(
                "one or more transactions in `BlockchainTest` are "
                + "intrinsically invalid, but the block was not expected "
                + "to be invalid. Please verify whether the transaction "
                + "was indeed expected to fail and add the proper "
                + "`block.exception`"
            )

        return built_block

    def verify_post_state(
        self,
        t8n: FillerBackend,
        t8n_state: Alloc,
        expected_state: Alloc | None = None,
    ) -> None:
        """Verify post alloc after all block/s or payload/s are generated."""
        try:
            if expected_state:
                expected_state.verify_post_alloc(t8n_state)
            else:
                self.post.verify_post_alloc(t8n_state)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

    def make_fixture(
        self,
        t8n: FillerBackend,
    ) -> FillResult:
        """Create a fixture from the blockchain test definition."""
        for i, block in enumerate(self.blocks):
            overrides = block.engine_payload_only_overrides()
            if overrides:
                raise Exception(
                    f"test correctness: block {i} sets {', '.join(overrides)}"
                    ", which only reach the engine payload and cannot be "
                    "expressed in an RLP blockchain fixture. Mark the test "
                    "`blockchain_test_engine_only`."
                )

        fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

        pre, genesis = self.make_genesis(apply_pre_allocation_blockchain=True)

        alloc: Alloc | LazyAlloc = pre
        state_root = genesis.header.state_root
        env = environment_from_parent_header(genesis.header)
        head = genesis.header.block_hash
        invalid_blocks = 0
        benchmark_gas_used: int | None = None
        benchmark_block_gas_used: int | None = None
        benchmark_opcode_count: OpcodeCount | None = None
        for i, block in enumerate(self.blocks):
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            built_block = self.generate_block_data(
                t8n=t8n,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
            )
            block_number = int(built_block.header.number)
            if (
                built_block.block_access_list is not None
                and built_block.block_access_list.has_rlp_override
            ):
                raise Exception(
                    f"test correctness: block {i}'s block access list "
                    "modifier re-encodes the list, but block RLP does not "
                    "carry the block access list, so an RLP blockchain "
                    "fixture cannot deliver the re-encoded bytes. Mark the "
                    "test `blockchain_test_engine_only`."
                )
            is_last_block = block is self.blocks[-1]
            if is_last_block and self.operation_mode == OpMode.BENCHMARKING:
                benchmark_gas_used = built_block.cumulative_gas_used()
                benchmark_block_gas_used = built_block.block_gas_used()
                benchmark_opcode_count = built_block.result.opcode_count
            if built_block.result.receipts:
                self.validate_receipt_status(
                    receipts=built_block.result.receipts,
                    block_number=block_number,
                )
            include_receipts = (
                block.include_receipts_in_output
                if block.include_receipts_in_output is not None
                else self.include_tx_receipts_in_output
            )
            fixture_blocks.append(
                built_block.get_fixture_block(
                    include_receipts=include_receipts
                )
            )

            # BAL verification already done in to_fixture_bal() if
            # expected_block_access_list set

            if block.exception is None:
                # Update env, alloc and last block hash for the next block.
                alloc = built_block.alloc
                state_root = built_block.state_root
                env = apply_new_parent(built_block.env, built_block.header)
                head = built_block.header.block_hash
            else:
                invalid_blocks += 1

            if block.expected_post_state:
                self.verify_post_state(
                    t8n,
                    t8n_state=alloc.materialize()
                    if isinstance(alloc, LazyAlloc)
                    else alloc,
                    expected_state=block.expected_post_state,
                )
        self.check_exception_test(exception=invalid_blocks > 0)
        alloc = alloc.materialize() if isinstance(alloc, LazyAlloc) else alloc
        self.verify_post_state(t8n, t8n_state=alloc)
        fixture = BlockchainFixture(
            fork=self.fork,
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            post_state=alloc
            if self.include_full_post_state_in_output
            else None,
            post_state_hash=state_root
            if not self.include_full_post_state_in_output
            else None,
            config=FixtureConfig(
                fork=self.fork,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                    self.fork.transitions_to().blob_schedule()
                ),
                chain_id=self.chain_id,
            ),
        )
        return FillResult(
            fixture=fixture,
            gas_optimization=None,
            benchmark_gas_used=benchmark_gas_used,
            benchmark_block_gas_used=benchmark_block_gas_used,
            benchmark_opcode_count=benchmark_opcode_count,
            post_verifications=PostVerifications.from_alloc(self.post),
        )

    def make_hive_fixture(
        self,
        t8n: FillerBackend,
        fixture_format: FixtureFormat
        | LabeledFixtureFormat = BlockchainEngineFixture,
    ) -> FillResult:
        """Create a hive fixture from the blocktest definition."""
        fixture_payloads: List[FixtureEngineNewPayload] = []

        pre, genesis = self.make_genesis(
            apply_pre_allocation_blockchain=fixture_format
            != BlockchainEngineXFixture,
        )
        alloc: Alloc | LazyAlloc = pre
        state_root = genesis.header.state_root
        env = environment_from_parent_header(genesis.header)
        head_hash = genesis.header.block_hash
        invalid_blocks = 0
        benchmark_gas_used: int | None = None
        benchmark_block_gas_used: int | None = None
        benchmark_opcode_count: OpcodeCount | None = None
        for block in self.blocks:
            built_block = self.generate_block_data(
                t8n=t8n,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
            )
            block_number = int(built_block.header.number)
            is_last_block = block is self.blocks[-1]
            if is_last_block and self.operation_mode == OpMode.BENCHMARKING:
                benchmark_gas_used = built_block.cumulative_gas_used()
                benchmark_block_gas_used = built_block.block_gas_used()
                benchmark_opcode_count = built_block.result.opcode_count
            if built_block.result.receipts:
                self.validate_receipt_status(
                    receipts=built_block.result.receipts,
                    block_number=block_number,
                )
            fixture_payloads.append(
                built_block.get_fixture_engine_new_payload()
            )
            if block.exception is None:
                alloc = built_block.alloc
                state_root = built_block.state_root
                env = apply_new_parent(built_block.env, built_block.header)
                head_hash = built_block.header.block_hash
            else:
                invalid_blocks += 1

            if block.expected_post_state:
                self.verify_post_state(
                    t8n,
                    t8n_state=alloc.materialize()
                    if isinstance(alloc, LazyAlloc)
                    else alloc,
                    expected_state=block.expected_post_state,
                )
        self.check_exception_test(exception=invalid_blocks > 0)
        fcu_version = (
            self.fork.transitions_from().engine_forkchoice_updated_version()
        )
        assert fcu_version is not None, (
            "A hive fixture was requested but no forkchoice update is defined."
            " The framework should never try to execute this test case."
        )

        alloc = alloc.materialize() if isinstance(alloc, LazyAlloc) else alloc
        self.verify_post_state(t8n, t8n_state=alloc)

        # Create base fixture data, common to all fixture formats
        fixture_data: Dict[str, Any] = {
            "fork": self.fork,
            "genesis": genesis.header,
            "payloads": fixture_payloads,
            "last_block_hash": head_hash,
            "post_state_hash": state_root
            if not self.include_full_post_state_in_output
            else None,
            "config": FixtureConfig(
                fork=self.fork,
                chain_id=self.chain_id,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                    self.fork.transitions_to().blob_schedule()
                ),
            ),
        }

        # Add format-specific fields
        fixture: BaseFixture
        if fixture_format == BlockchainEngineXFixture:
            # For Engine X format, exclude pre (will be provided via shared
            # state) and prepare for state diff optimization
            pre_alloc_group_hash = self.pre.get_alloc_grouping_hash()
            if pre_alloc_group_hash is None:
                raise ValueError(
                    "Engine X fixtures require a pre-alloc group; was phase 1 "
                    "run?"
                )
            fixture_data["pre_hash"] = pre_alloc_group_hash
            fixture_data["post_state_diff"] = alloc.calculate_diff(self.pre)
        elif fixture_format == BlockchainEngineSyncFixture:
            # Sync fixture format
            assert genesis.header.block_hash != head_hash, (
                "Invalid payload tests negative test via sync is not "
                "supported yet."
            )
            # Most clients require the header to start the sync process, so we
            # create an empty block on top of the last block of the test to
            # send it as new payload and trigger the sync process.
            sync_built_block = self.generate_block_data(
                t8n=t8n,
                block=Block(),
                previous_env=env,
                previous_alloc=alloc,
            )
            fixture_data.update(
                {
                    "sync_payload": (
                        sync_built_block.get_fixture_engine_new_payload()
                    ),
                    "pre": pre,
                    "post_state": alloc
                    if self.include_full_post_state_in_output
                    else None,
                }
            )
        else:
            # Standard engine fixture
            fixture_data.update(
                {
                    "pre": pre,
                    "post_state": alloc
                    if self.include_full_post_state_in_output
                    else None,
                }
            )
        fixture = fixture_format.format_class()(**fixture_data)

        return FillResult(
            fixture=fixture,
            gas_optimization=None,
            benchmark_gas_used=benchmark_gas_used,
            benchmark_block_gas_used=benchmark_block_gas_used,
            benchmark_opcode_count=benchmark_opcode_count,
            post_verifications=PostVerifications.from_alloc(self.post),
        )

    def make_stateful_fixture(
        self,
        t8n: FillerBackend,
    ) -> FillResult:
        """
        Create a ``BlockchainEngineStatefulFixture`` against a live client.

        Differs from ``make_hive_fixture``:

        - No genesis building: the client already has warm state from a
          snapshot. The backend (typically ``ClientBackend``) owns
          ``snapshot_block`` / ``start_block`` captured at session start.
        - ``pre.fund_eoa`` / ``pre.deploy_contract`` calls are materialised
          into a synthetic setup block prepended to ``self.blocks``,
          instead of being baked into genesis alloc.
        - Payloads are partitioned by ``FixtureEngineNewPayload.phase``
          into ``setup_payloads`` (setup-phase txs) and ``payloads``
          (execution-phase txs).
        - ``post`` is verified against the live client (the oracle) via
          ``get_post_state_alloc``; there is no t8n post alloc to diff.
        """
        if not isinstance(t8n, ClientBackend):
            raise RuntimeError(
                "make_stateful_fixture requires a ClientBackend; got "
                f"{type(t8n).__name__}."
            )
        if t8n.snapshot_block is None or t8n.start_block is None:
            raise RuntimeError(
                "ClientBackend.snapshot_block / .start_block must be "
                "captured by the fill-stateful pre-run before fill."
            )
        snapshot_block = t8n.snapshot_block
        start_block = t8n.start_block

        # Mirror execute.py's pre-send flow so pending-tx funding amounts
        # materialise before we drain the queue: required-balances →
        # resolve deferred deploys/stubs/fund_addresses → run
        # minimum_balance_for_pending_transactions. Tests whose Alloc
        # does not expose ``pending_transactions`` skip this entirely.
        pending_getter = getattr(self.pre, "pending_transactions", None)
        resolve_deferred = getattr(self.pre, "resolve_deferred_checks", None)
        min_balance = getattr(
            self.pre, "minimum_balance_for_pending_transactions", None
        )
        if (
            callable(pending_getter)
            and callable(resolve_deferred)
            and callable(min_balance)
        ):
            execute_plan = self.execute(execute_format=TransactionPost)
            session_fork = self.fork.fork_at(block_number=0, timestamp=0)
            # Session fees pinned on the backend by the fill-stateful plugin.
            gas_price = t8n.gas_price
            max_fee_per_gas = t8n.max_fee_per_gas
            max_priority_fee_per_gas = t8n.max_priority_fee_per_gas
            max_fee_per_blob_gas = t8n.max_fee_per_blob_gas
            if not all(
                [
                    gas_price,
                    max_fee_per_gas,
                    max_priority_fee_per_gas,
                    max_fee_per_blob_gas,
                ]
            ):
                raise RuntimeError(
                    "make_stateful_fixture requires the backend to carry "
                    f"non-zero session fees; got gas_price={gas_price}, "
                    f"max_fee_per_gas={max_fee_per_gas}, "
                    f"max_priority_fee_per_gas={max_priority_fee_per_gas}, "
                    f"max_fee_per_blob_gas={max_fee_per_blob_gas}."
                )
            execute_plan.prepare_transactions(
                env=Environment(gas_limit=HexNumber(start_block["gasLimit"])),
                gas_price=gas_price,
                max_fee_per_gas=max_fee_per_gas,
                max_priority_fee_per_gas=max_priority_fee_per_gas,
                max_fee_per_blob_gas=max_fee_per_blob_gas,
                fork=session_fork,
            )
            required_balances = execute_plan.get_required_sender_balances(
                fork=session_fork,
            )
            resolve_deferred()
            min_balance(
                required_balances,
                gas_price=gas_price,
                max_fee_per_gas=max_fee_per_gas,
                max_priority_fee_per_gas=max_priority_fee_per_gas,
                max_fee_per_blob_gas=max_fee_per_blob_gas,
            )

        self.pre.verify_deployed_accounts(
            int(HexNumber(start_block["number"]))
        )

        # Materialise queued pre-alloc txs into a synthetic setup block.
        blocks_to_process: List[Block] = []
        if callable(pending_getter):
            setup_txs = pending_getter()
            if setup_txs:
                blocks_to_process.append(Block(txs=setup_txs))
        # Each block must be single-phase (Block.phase asserts otherwise);
        # mixed blocks (e.g. EIP-7702 authorization + benchmark exec) are
        # split into contiguous phase runs so benchmark gas isn't
        # swallowed into ``setupEngineNewPayloads``.
        blocks_to_process.extend(_split_blocks_by_phase(self.blocks))

        # Chain off the session start_block. We pull parent_* from a
        # FixtureHeader-validated copy of the client's block dict, but
        # seed block_hashes with the client's hash directly — FixtureHeader
        # recomputes block_hash from RLP and that diverges from the
        # client's authoritative hash unless every header byte is
        # reproduced exactly.
        start_block_number = int(HexNumber(start_block["number"]))
        start_block_hash = Hash(start_block["hash"])
        parent_header = FixtureHeader.model_validate(start_block)
        env = Environment(
            parent_difficulty=parent_header.difficulty,
            parent_timestamp=parent_header.timestamp,
            parent_base_fee_per_gas=parent_header.base_fee_per_gas,
            parent_blob_gas_used=parent_header.blob_gas_used,
            parent_excess_blob_gas=parent_header.excess_blob_gas,
            parent_gas_used=parent_header.gas_used,
            parent_gas_limit=parent_header.gas_limit,
            parent_ommers_hash=parent_header.ommers_hash,
            block_hashes={
                HexNumber(start_block_number): start_block_hash,
            },
        )

        setup_payloads: List[FixtureEngineNewPayload] = []
        execution_payloads: List[FixtureEngineNewPayload] = []
        # Aligned 1:1 with execution_payloads; None when no trace.
        execution_opcode_counts: List[Dict[str, int] | None] = []
        head_hash = start_block_hash
        benchmark_gas_used: int | None = None
        benchmark_block_gas_used: int | None = None
        benchmark_opcode_count: OpcodeCount | None = None
        # Alloc is not authoritative in stateful mode; pass self.pre as a
        # placeholder — ClientBackend ignores it.
        alloc: Alloc | LazyAlloc = self.pre
        for block in blocks_to_process:
            built_block = self.generate_block_data(
                t8n=t8n,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
            )
            assert isinstance(built_block, TestingBuildBlock), (
                "ClientBackend must return TestingBuildBlock; got "
                f"{type(built_block).__name__}"
            )
            payload = payload_metadata_to_fixture(
                built_block.engine_payload, phase=block.phase
            )
            # The client's authoritative block hash (the FixtureHeader RLP
            # hash diverges — the client picks fields like gas_limit).
            client_hash = Hash(
                built_block.engine_payload.payload_response.execution_payload.block_hash
            )
            if payload.phase == TestPhase.SETUP:
                setup_payloads.append(payload)
            else:
                execution_payloads.append(payload)
                block_opcode_count = t8n.extract_block_opcode_count(
                    client_hash
                )
                execution_opcode_counts.append(
                    block_opcode_count.model_dump()
                    if block_opcode_count is not None
                    else None
                )
                # Setup blocks (pre-alloc funding/deploys) are exempt:
                # ``expected_receipt_status`` describes the test's own
                # transactions, and setup txs always succeed.
                if built_block.result.receipts:
                    self.validate_receipt_status(
                        receipts=built_block.result.receipts,
                        block_number=int(built_block.header.number),
                    )
                if self.operation_mode == OpMode.BENCHMARKING:
                    benchmark_gas_used = built_block.cumulative_gas_used()
                    benchmark_block_gas_used = built_block.block_gas_used()
                    # Consumed by BenchmarkTest's opcode-count verification.
                    benchmark_opcode_count = block_opcode_count
            # apply_new_parent records the RLP hash; the next block's
            # parent_hash must point at what the client actually built.
            env = apply_new_parent(built_block.env, built_block.header)
            env = env.copy(
                block_hashes={
                    **env.block_hashes,
                    HexNumber(int(env.number)): client_hash,
                },
            )
            head_hash = client_hash

        if self.post.root:
            got_alloc = t8n.get_post_state_alloc(self.post)
            self.post.verify_post_alloc(got_alloc)

        fixture = BlockchainEngineStatefulFixture(
            fork=self.fork,
            last_block_hash=head_hash,
            config=FixtureConfig(fork=self.fork),
            snapshot_block_number=HexNumber(snapshot_block["number"]),
            snapshot_block_hash=Hash(snapshot_block["hash"]),
            start_block_number=HexNumber(start_block_number),
            start_block_hash=start_block_hash,
            setup_payloads=setup_payloads,
            payloads=execution_payloads,
            benchmark_gas_used=(
                HexNumber(benchmark_gas_used)
                if benchmark_gas_used is not None
                else None
            ),
        )
        metadata: Dict[str, Any] = {}
        if t8n.extract_opcode_count:
            metadata["opcode_counts"] = execution_opcode_counts
        return FillResult(
            fixture=fixture,
            gas_optimization=None,
            benchmark_gas_used=benchmark_gas_used,
            benchmark_block_gas_used=benchmark_block_gas_used,
            benchmark_opcode_count=benchmark_opcode_count,
            metadata=metadata,
            post_verifications=PostVerifications.from_alloc(self.post),
        )

    def generate(
        self,
        t8n: FillerBackend,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the BlockchainTest fixture."""
        if fixture_format == BlockchainEngineStatefulFixture:
            return self.make_stateful_fixture(t8n)
        if fixture_format in [
            BlockchainEngineFixture,
            BlockchainEngineXFixture,
            BlockchainEngineSyncFixture,
        ]:
            return self.make_hive_fixture(t8n, fixture_format)
        elif fixture_format == BlockchainFixture:
            return self.make_fixture(t8n)

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            blocks: List[List[Transaction]] = []
            for block in self.blocks:
                blocks += [block.txs]
            # Pass gas validation params for benchmark tests
            # If not benchmark mode, skip gas used validation
            if self.operation_mode != OpMode.BENCHMARKING:
                self.skip_gas_used_validation = True

            benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
            return TransactionPost(
                blocks=blocks,
                post=self.post,
                benchmark_mode=benchmark_mode,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

include_full_post_state_in_output = True class-attribute instance-attribute

Include the post state in the fixture output. Otherwise, the state verification is only performed based on the state root.

include_tx_receipts_in_output = True class-attribute instance-attribute

Include transaction receipts in the fixture output.

discard_fixture_format_by_marks(fixture_format, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard a fixture format from filling if the appropriate marker is
    used.
    """
    marker_names = [m.name for m in markers]
    if (
        fixture_format != BlockchainFixture
        and "blockchain_test_only" in marker_names
    ):
        return True
    if (
        fixture_format
        not in [BlockchainEngineFixture, BlockchainEngineXFixture]
        and "blockchain_test_engine_only" in marker_names
    ):
        return True
    return False

model_post_init(__context)

Model post-init to assert static (pre-fill/execute) checks.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def model_post_init(self, __context: Any, /) -> None:
    """
    Model post-init to assert static (pre-fill/execute) checks.
    """
    super().model_post_init(__context)
    if self.is_inclusion_test:
        # Verify that the blockchain contains at most one invalid
        # transaction which must be located at the end of the last block.
        for i, block in enumerate(self.blocks):
            if i != (len(self.blocks) - 1):
                valid_tx_range = block.txs[:]
            else:
                valid_tx_range = block.txs[:-1]
            if any(tx.error is not None for tx in valid_tx_range):
                raise Exception(
                    "test correctness: in an inclusion test the only "
                    "transaction allowed to produce an exception is the "
                    "last transaction of the last block, but block "
                    f"{i} contains an invalid transaction elsewhere"
                )
    for i, block in enumerate(self.blocks):
        expectation = block.expected_block_access_list
        if (
            expectation is not None
            and expectation.has_modifier
            and not block.exception
            and block.engine_api_error_code is None
        ):
            raise Exception(
                f"test correctness: block {i} modifies its block access "
                "list or its encoding but declares no `exception` or "
                "`engine_api_error_code`, so the corrupted block access "
                "list would be filled as valid. Declare the exception "
                "the modified block access list must cause, or drop the "
                "modifier."
            )
        if (
            block.engine_new_payload_block_access_list is not None
            and expectation is not None
            and expectation.has_rlp_modifier
        ):
            raise Exception(
                f"test correctness: block {i} sets "
                "`engine_new_payload_block_access_list` and re-encodes "
                "the block access list with `modify_rlp`; the explicit "
                "payload override would discard the re-encoding. Keep "
                "one: `engine_new_payload_block_access_list` delivers "
                "arbitrary bytes, `modify_rlp` re-encodes the list the "
                "transition tool produced."
            )

get_genesis_environment()

Get the genesis environment for pre-allocation groups.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def get_genesis_environment(self) -> Environment:
    """Get the genesis environment for pre-allocation groups."""
    # Checked before the defaults below add the mix hash under
    # `prev_randao`, which every genesis header carries.
    self.genesis_environment.check_fork_fields(
        self.fork.transitions_from()
    )
    modified_values = self.genesis_environment.set_fork_requirements(
        self.fork.transitions_from()
    ).model_dump(exclude_unset=True)
    return Environment(**(GENESIS_ENVIRONMENT_DEFAULTS | modified_values))

make_genesis(*, apply_pre_allocation_blockchain)

Create a genesis block from the blockchain test definition.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def make_genesis(
    self, *, apply_pre_allocation_blockchain: bool
) -> Tuple[Alloc, FixtureBlock]:
    """Create a genesis block from the blockchain test definition."""
    env = self.get_genesis_environment()
    assert env.withdrawals is None or len(env.withdrawals) == 0, (
        "withdrawals must be empty at genesis"
    )
    assert (
        env.parent_beacon_block_root is None
        or env.parent_beacon_block_root == Hash(0)
    ), "parent_beacon_block_root must be empty at genesis"

    pre_alloc = self.pre
    if apply_pre_allocation_blockchain:
        pre_alloc = Alloc.merge(
            Alloc.model_validate(
                self.fork.transitions_to().pre_allocation_blockchain()
            ),
            pre_alloc,
        )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")
    if pre_alloc.state_commitment() is None:
        pre_alloc.migrate_state_commitment(
            self.fork.transitions_from().state_commitment()
        )
    state_root = pre_alloc.state_root()
    genesis = FixtureHeader.genesis(
        self.fork.transitions_from(), env, state_root
    )

    return (
        pre_alloc,
        FixtureBlockBase(
            header=genesis,
            withdrawals=None if env.withdrawals is None else [],
        ).with_rlp(txs=[]),
    )

generate_block_data(t8n, block, previous_env, previous_alloc)

Generate common block data for both make_fixture and make_hive_fixture.

t8n is any backend satisfying FillerBackend. The default compute path passes a concrete TransitionTool; stateful filling will pass an ClientBackend that drives testing_buildBlockV1 against a live client.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def generate_block_data(
    self,
    t8n: FillerBackend,
    block: Block,
    previous_env: Environment,
    previous_alloc: Alloc | LazyAlloc,
) -> BuiltBlock:
    """
    Generate common block data for both make_fixture and make_hive_fixture.

    ``t8n`` is any backend satisfying ``FillerBackend``. The
    default compute path passes a concrete ``TransitionTool``; stateful
    filling will pass an ``ClientBackend`` that drives
    ``testing_buildBlockV1`` against a live client.
    """
    env = block.set_environment(previous_env)
    fork = self.fork.fork_at(
        block_number=env.number, timestamp=env.timestamp
    )
    env = env.set_fork_requirements(fork)
    env.check_fork_fields(fork)
    txs = block.txs[:]
    if any("gas_limit" not in tx.model_fields_set for tx in block.txs):
        max_tx_gas_limit = Transaction.calculate_max_gas_limit(
            txs=txs,
            env_gas_limit=int(env.gas_limit),
            transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
            state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
        )
        if max_tx_gas_limit == 0:
            raise Exception(
                "test correctness: unable to automatically calculate gas "
                "limit for transactions (No remaining gas)."
            )
        txs = [
            tx.with_gas_limit(
                max_gas_limit=max_tx_gas_limit,
                transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
                state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
            )
            for tx in txs
        ]
    txs = [tx.with_signature_and_sender() for tx in txs]

    if (failing_tx_count := len([tx for tx in txs if tx.error])) > 0:
        if failing_tx_count > 1:
            raise Exception(
                "test correctness: only one transaction can produce "
                "an exception in a block"
            )
        if not txs[-1].error:
            raise Exception(
                "test correctness: the transaction that produces an "
                "exception must be the last transaction in the block"
            )

    transition_tool_output = t8n.evaluate(
        transition_tool_data=TransitionTool.TransitionToolData(
            alloc=previous_alloc,
            txs=txs,
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=fork.get_reward(),
            blob_schedule=fork.blob_schedule(),
        ),
        slow_request=self.is_tx_gas_heavy_test,
    )

    # One special case of the invalid transactions is the blob gas used,
    # since this value is not included in the transition tool result, but
    # it is included in the block header, and some clients check it before
    # executing the block by simply counting the type-3 txs, we need to set
    # the correct value by default.
    blob_gas_used: int | None = None
    if fork.supports_blobs():
        if (blob_gas_per_blob := fork.blob_gas_per_blob()) > 0:
            blob_gas_used = blob_gas_per_blob * count_blobs(txs)

    # Prepare slot_number for header initialization
    slot_number_value: ZeroPaddedHexNumber | None = None
    if fork.header_slot_number_required():
        slot_number_value = ZeroPaddedHexNumber(
            int(env.slot_number) if env.slot_number is not None else 0
        )

    header = FixtureHeader(
        **(
            transition_tool_output.result.model_dump(
                exclude_none=True,
                exclude={"blob_gas_used", "transactions_trie"},
            )
            | env.model_dump(
                exclude_none=True,
                exclude={"blob_gas_used", "slot_number"},
            )
        ),
        blob_gas_used=blob_gas_used,
        transactions_trie=Transaction.list_root(txs),
        extra_data=(
            block.extra_data if block.extra_data is not None else b""
        ),
        slot_number=slot_number_value,
        fork=fork,
    )

    if block.header_verify is not None:
        # Verify the header after transition tool processing.
        try:
            block.header_verify.verify(header)
        except Exception as e:
            raise Exception(
                f"Verification of block {int(env.number)} failed"
            ) from e

    if block.expected_gas_used is not None:
        gas_used = int(transition_tool_output.result.gas_used)
        assert gas_used == block.expected_gas_used, (
            f"gas_used ({gas_used}) does not match expected_gas_used "
            f"({block.expected_gas_used})"
            f", difference: {gas_used - block.expected_gas_used}"
        )

    requests_list: List[Bytes] | None = None
    if fork.header_requests_required():
        assert transition_tool_output.result.requests is not None, (
            "Requests are required for this block"
        )
        requests = Requests(
            requests_lists=list(transition_tool_output.result.requests)
        )

        if Hash(requests) != header.requests_hash:
            raise Exception(
                "Requests root in header does not match the requests "
                "root in the transition tool output: "
                f"{header.requests_hash} != {Hash(requests)}"
            )

        requests_list = requests.requests_list

    if block.requests is not None:
        header.requests_hash = Hash(
            Requests(requests_lists=list(block.requests))
        )
        requests_list = block.requests

    # Decode BAL from RLP bytes provided by the transition tool.
    t8n_bal_rlp = transition_tool_output.result.block_access_list
    t8n_bal: BlockAccessList | None = None
    if t8n_bal_rlp is not None:
        t8n_bal = BlockAccessList.from_rlp(t8n_bal_rlp)

    if fork.header_bal_hash_required():
        assert t8n_bal is not None, (
            "Block access list is required for this block but was not "
            "provided by the transition tool"
        )

        if t8n.attests_block_access_list_hash:
            computed_block_access_list_hash = Hash(t8n_bal.rlp.keccak256())
            assert (
                computed_block_access_list_hash
                == header.block_access_list_hash
            ), (
                "Block access list hash in header does not match the "
                f"computed hash from BAL: {header.block_access_list_hash} "
                f"!= {computed_block_access_list_hash}"
            )

    if block.rlp_modifier is not None:
        # Modify any parameter specified in the `rlp_modifier` after
        # transition tool processing.
        header = block.rlp_modifier.apply(header)
        header.fork = fork  # Deleted during `apply` because `exclude=True`

    # Process block access list - apply transformer if present for invalid
    # tests
    bal = t8n_bal
    bal_rlp_override: Bytes | None = None

    # Always validate BAL structural integrity (ordering, duplicates)
    # if present
    if t8n_bal is not None:
        t8n_bal.validate_structure()

    # If expected BAL is defined, verify against it
    if (
        block.expected_block_access_list is not None
        and t8n_bal is not None
    ):
        block.expected_block_access_list.verify_against(t8n_bal)

        bal = block.expected_block_access_list.modify_if_invalid_test(
            t8n_bal
        )
        if bal.rlp != t8n_bal.rlp:
            # Compare bytes, not contents: an encoding override also
            # moves the header commitment.
            header.block_access_list_hash = Hash(bal.rlp.keccak256())
        bal_rlp_override = block.expected_block_access_list.modified_rlp(
            bal
        )
        if (
            block.expected_block_access_list.has_modifier
            and bal.rlp == t8n_bal.rlp
            and (
                bal_rlp_override is None or bal_rlp_override == t8n_bal.rlp
            )
        ):
            raise Exception(
                f"test correctness: block number {int(env.number)}'s "
                "block access list modifier left the list unchanged, so "
                "the block would be labelled invalid for no reason. Make "
                "the modifier change the list, or drop it along with the "
                "exception."
            )
        if bal.has_rlp_override and (
            block.engine_new_payload_block_access_list is not None
            or bal_rlp_override is not None
        ):
            raise Exception(
                f"test correctness: block number {int(env.number)} "
                "re-encodes the block access list with `override_rlp` "
                "and also replaces the payload bytes, so the payload "
                "would not carry what the header commits to. Keep one: "
                "`override_rlp` commits the header to its re-encoding, "
                "`modify_rlp` re-encodes the payload only, "
                "`engine_new_payload_block_access_list` delivers "
                "arbitrary bytes."
            )

    built_block_kwargs: Dict[str, Any] = dict(
        header=header,
        alloc=transition_tool_output.alloc,
        state_root=transition_tool_output.result.state_root,
        env=env,
        txs=txs,
        ommers=[],
        withdrawals=env.withdrawals,
        requests=requests_list,
        result=transition_tool_output.result,
        expected_exception=block.exception,
        engine_api_error_code=block.engine_api_error_code,
        rlp_modifier=block.rlp_modifier,
        fork=fork,
        block_access_list=bal,
        engine_new_payload_block_access_list=(
            block.engine_new_payload_block_access_list
            if block.engine_new_payload_block_access_list is not None
            else bal_rlp_override
        ),
        engine_new_payload_slot_number=(
            block.engine_new_payload_slot_number
        ),
    )
    built_block: BuiltBlock
    if transition_tool_output.engine_payload is not None:
        built_block = TestingBuildBlock(
            **built_block_kwargs,
            engine_payload=transition_tool_output.engine_payload,
        )
    else:
        built_block = BuiltBlock(**built_block_kwargs)

    try:
        rejected_txs = built_block.verify_transactions(
            transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
        )
        if (
            not rejected_txs
            and block.rlp_modifier is None
            and block.requests is None
            and not block.skip_exception_verification
            and block.engine_new_payload_block_access_list is None
            and block.engine_new_payload_slot_number is None
            and not (
                block.expected_block_access_list is not None
                and block.expected_block_access_list.has_modifier
            )
        ):
            # Only verify block level exception if: - No transaction
            # exception was raised, because these are not reported as block
            # exceptions. - No RLP modifier was specified, because the
            # modifier is what normally produces the block exception. - No
            # requests were specified, because modified requests are also
            # what normally produces the block exception. - No engine
            # payload BAL override was specified, because it corrupts only
            # the engine payload after the transition tool has run. - No
            # BAL modifier was specified, because a rewritten BAL, whether
            # in contents or in encoding, is applied after the transition
            # tool has run and is what produces the block exception.
            built_block.verify_block_exception(
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
        verify_result(transition_tool_output.result, env)
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(previous_alloc)
        pprint(transition_tool_output.alloc.materialize())
        raise e

    if len(rejected_txs) > 0 and block.exception is None:
        print_traces(t8n.get_traces())
        raise Exception(
            "one or more transactions in `BlockchainTest` are "
            + "intrinsically invalid, but the block was not expected "
            + "to be invalid. Please verify whether the transaction "
            + "was indeed expected to fail and add the proper "
            + "`block.exception`"
        )

    return built_block

verify_post_state(t8n, t8n_state, expected_state=None)

Verify post alloc after all block/s or payload/s are generated.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def verify_post_state(
    self,
    t8n: FillerBackend,
    t8n_state: Alloc,
    expected_state: Alloc | None = None,
) -> None:
    """Verify post alloc after all block/s or payload/s are generated."""
    try:
        if expected_state:
            expected_state.verify_post_alloc(t8n_state)
        else:
            self.post.verify_post_alloc(t8n_state)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

make_fixture(t8n)

Create a fixture from the blockchain test definition.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def make_fixture(
    self,
    t8n: FillerBackend,
) -> FillResult:
    """Create a fixture from the blockchain test definition."""
    for i, block in enumerate(self.blocks):
        overrides = block.engine_payload_only_overrides()
        if overrides:
            raise Exception(
                f"test correctness: block {i} sets {', '.join(overrides)}"
                ", which only reach the engine payload and cannot be "
                "expressed in an RLP blockchain fixture. Mark the test "
                "`blockchain_test_engine_only`."
            )

    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

    pre, genesis = self.make_genesis(apply_pre_allocation_blockchain=True)

    alloc: Alloc | LazyAlloc = pre
    state_root = genesis.header.state_root
    env = environment_from_parent_header(genesis.header)
    head = genesis.header.block_hash
    invalid_blocks = 0
    benchmark_gas_used: int | None = None
    benchmark_block_gas_used: int | None = None
    benchmark_opcode_count: OpcodeCount | None = None
    for i, block in enumerate(self.blocks):
        # This is the most common case, the RLP needs to be constructed
        # based on the transactions to be included in the block.
        # Set the environment according to the block to execute.
        built_block = self.generate_block_data(
            t8n=t8n,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
        )
        block_number = int(built_block.header.number)
        if (
            built_block.block_access_list is not None
            and built_block.block_access_list.has_rlp_override
        ):
            raise Exception(
                f"test correctness: block {i}'s block access list "
                "modifier re-encodes the list, but block RLP does not "
                "carry the block access list, so an RLP blockchain "
                "fixture cannot deliver the re-encoded bytes. Mark the "
                "test `blockchain_test_engine_only`."
            )
        is_last_block = block is self.blocks[-1]
        if is_last_block and self.operation_mode == OpMode.BENCHMARKING:
            benchmark_gas_used = built_block.cumulative_gas_used()
            benchmark_block_gas_used = built_block.block_gas_used()
            benchmark_opcode_count = built_block.result.opcode_count
        if built_block.result.receipts:
            self.validate_receipt_status(
                receipts=built_block.result.receipts,
                block_number=block_number,
            )
        include_receipts = (
            block.include_receipts_in_output
            if block.include_receipts_in_output is not None
            else self.include_tx_receipts_in_output
        )
        fixture_blocks.append(
            built_block.get_fixture_block(
                include_receipts=include_receipts
            )
        )

        # BAL verification already done in to_fixture_bal() if
        # expected_block_access_list set

        if block.exception is None:
            # Update env, alloc and last block hash for the next block.
            alloc = built_block.alloc
            state_root = built_block.state_root
            env = apply_new_parent(built_block.env, built_block.header)
            head = built_block.header.block_hash
        else:
            invalid_blocks += 1

        if block.expected_post_state:
            self.verify_post_state(
                t8n,
                t8n_state=alloc.materialize()
                if isinstance(alloc, LazyAlloc)
                else alloc,
                expected_state=block.expected_post_state,
            )
    self.check_exception_test(exception=invalid_blocks > 0)
    alloc = alloc.materialize() if isinstance(alloc, LazyAlloc) else alloc
    self.verify_post_state(t8n, t8n_state=alloc)
    fixture = BlockchainFixture(
        fork=self.fork,
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        post_state=alloc
        if self.include_full_post_state_in_output
        else None,
        post_state_hash=state_root
        if not self.include_full_post_state_in_output
        else None,
        config=FixtureConfig(
            fork=self.fork,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                self.fork.transitions_to().blob_schedule()
            ),
            chain_id=self.chain_id,
        ),
    )
    return FillResult(
        fixture=fixture,
        gas_optimization=None,
        benchmark_gas_used=benchmark_gas_used,
        benchmark_block_gas_used=benchmark_block_gas_used,
        benchmark_opcode_count=benchmark_opcode_count,
        post_verifications=PostVerifications.from_alloc(self.post),
    )

make_hive_fixture(t8n, fixture_format=BlockchainEngineFixture)

Create a hive fixture from the blocktest definition.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def make_hive_fixture(
    self,
    t8n: FillerBackend,
    fixture_format: FixtureFormat
    | LabeledFixtureFormat = BlockchainEngineFixture,
) -> FillResult:
    """Create a hive fixture from the blocktest definition."""
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, genesis = self.make_genesis(
        apply_pre_allocation_blockchain=fixture_format
        != BlockchainEngineXFixture,
    )
    alloc: Alloc | LazyAlloc = pre
    state_root = genesis.header.state_root
    env = environment_from_parent_header(genesis.header)
    head_hash = genesis.header.block_hash
    invalid_blocks = 0
    benchmark_gas_used: int | None = None
    benchmark_block_gas_used: int | None = None
    benchmark_opcode_count: OpcodeCount | None = None
    for block in self.blocks:
        built_block = self.generate_block_data(
            t8n=t8n,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
        )
        block_number = int(built_block.header.number)
        is_last_block = block is self.blocks[-1]
        if is_last_block and self.operation_mode == OpMode.BENCHMARKING:
            benchmark_gas_used = built_block.cumulative_gas_used()
            benchmark_block_gas_used = built_block.block_gas_used()
            benchmark_opcode_count = built_block.result.opcode_count
        if built_block.result.receipts:
            self.validate_receipt_status(
                receipts=built_block.result.receipts,
                block_number=block_number,
            )
        fixture_payloads.append(
            built_block.get_fixture_engine_new_payload()
        )
        if block.exception is None:
            alloc = built_block.alloc
            state_root = built_block.state_root
            env = apply_new_parent(built_block.env, built_block.header)
            head_hash = built_block.header.block_hash
        else:
            invalid_blocks += 1

        if block.expected_post_state:
            self.verify_post_state(
                t8n,
                t8n_state=alloc.materialize()
                if isinstance(alloc, LazyAlloc)
                else alloc,
                expected_state=block.expected_post_state,
            )
    self.check_exception_test(exception=invalid_blocks > 0)
    fcu_version = (
        self.fork.transitions_from().engine_forkchoice_updated_version()
    )
    assert fcu_version is not None, (
        "A hive fixture was requested but no forkchoice update is defined."
        " The framework should never try to execute this test case."
    )

    alloc = alloc.materialize() if isinstance(alloc, LazyAlloc) else alloc
    self.verify_post_state(t8n, t8n_state=alloc)

    # Create base fixture data, common to all fixture formats
    fixture_data: Dict[str, Any] = {
        "fork": self.fork,
        "genesis": genesis.header,
        "payloads": fixture_payloads,
        "last_block_hash": head_hash,
        "post_state_hash": state_root
        if not self.include_full_post_state_in_output
        else None,
        "config": FixtureConfig(
            fork=self.fork,
            chain_id=self.chain_id,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                self.fork.transitions_to().blob_schedule()
            ),
        ),
    }

    # Add format-specific fields
    fixture: BaseFixture
    if fixture_format == BlockchainEngineXFixture:
        # For Engine X format, exclude pre (will be provided via shared
        # state) and prepare for state diff optimization
        pre_alloc_group_hash = self.pre.get_alloc_grouping_hash()
        if pre_alloc_group_hash is None:
            raise ValueError(
                "Engine X fixtures require a pre-alloc group; was phase 1 "
                "run?"
            )
        fixture_data["pre_hash"] = pre_alloc_group_hash
        fixture_data["post_state_diff"] = alloc.calculate_diff(self.pre)
    elif fixture_format == BlockchainEngineSyncFixture:
        # Sync fixture format
        assert genesis.header.block_hash != head_hash, (
            "Invalid payload tests negative test via sync is not "
            "supported yet."
        )
        # Most clients require the header to start the sync process, so we
        # create an empty block on top of the last block of the test to
        # send it as new payload and trigger the sync process.
        sync_built_block = self.generate_block_data(
            t8n=t8n,
            block=Block(),
            previous_env=env,
            previous_alloc=alloc,
        )
        fixture_data.update(
            {
                "sync_payload": (
                    sync_built_block.get_fixture_engine_new_payload()
                ),
                "pre": pre,
                "post_state": alloc
                if self.include_full_post_state_in_output
                else None,
            }
        )
    else:
        # Standard engine fixture
        fixture_data.update(
            {
                "pre": pre,
                "post_state": alloc
                if self.include_full_post_state_in_output
                else None,
            }
        )
    fixture = fixture_format.format_class()(**fixture_data)

    return FillResult(
        fixture=fixture,
        gas_optimization=None,
        benchmark_gas_used=benchmark_gas_used,
        benchmark_block_gas_used=benchmark_block_gas_used,
        benchmark_opcode_count=benchmark_opcode_count,
        post_verifications=PostVerifications.from_alloc(self.post),
    )

make_stateful_fixture(t8n)

Create a BlockchainEngineStatefulFixture against a live client.

Differs from make_hive_fixture:

• No genesis building: the client already has warm state from a snapshot. The backend (typically ClientBackend) owns snapshot_block / start_block captured at session start.
• pre.fund_eoa / pre.deploy_contract calls are materialised into a synthetic setup block prepended to self.blocks, instead of being baked into genesis alloc.
• Payloads are partitioned by FixtureEngineNewPayload.phase into setup_payloads (setup-phase txs) and payloads (execution-phase txs).
• post is verified against the live client (the oracle) via get_post_state_alloc; there is no t8n post alloc to diff.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def make_stateful_fixture(
    self,
    t8n: FillerBackend,
) -> FillResult:
    """
    Create a ``BlockchainEngineStatefulFixture`` against a live client.

    Differs from ``make_hive_fixture``:

    - No genesis building: the client already has warm state from a
      snapshot. The backend (typically ``ClientBackend``) owns
      ``snapshot_block`` / ``start_block`` captured at session start.
    - ``pre.fund_eoa`` / ``pre.deploy_contract`` calls are materialised
      into a synthetic setup block prepended to ``self.blocks``,
      instead of being baked into genesis alloc.
    - Payloads are partitioned by ``FixtureEngineNewPayload.phase``
      into ``setup_payloads`` (setup-phase txs) and ``payloads``
      (execution-phase txs).
    - ``post`` is verified against the live client (the oracle) via
      ``get_post_state_alloc``; there is no t8n post alloc to diff.
    """
    if not isinstance(t8n, ClientBackend):
        raise RuntimeError(
            "make_stateful_fixture requires a ClientBackend; got "
            f"{type(t8n).__name__}."
        )
    if t8n.snapshot_block is None or t8n.start_block is None:
        raise RuntimeError(
            "ClientBackend.snapshot_block / .start_block must be "
            "captured by the fill-stateful pre-run before fill."
        )
    snapshot_block = t8n.snapshot_block
    start_block = t8n.start_block

    # Mirror execute.py's pre-send flow so pending-tx funding amounts
    # materialise before we drain the queue: required-balances →
    # resolve deferred deploys/stubs/fund_addresses → run
    # minimum_balance_for_pending_transactions. Tests whose Alloc
    # does not expose ``pending_transactions`` skip this entirely.
    pending_getter = getattr(self.pre, "pending_transactions", None)
    resolve_deferred = getattr(self.pre, "resolve_deferred_checks", None)
    min_balance = getattr(
        self.pre, "minimum_balance_for_pending_transactions", None
    )
    if (
        callable(pending_getter)
        and callable(resolve_deferred)
        and callable(min_balance)
    ):
        execute_plan = self.execute(execute_format=TransactionPost)
        session_fork = self.fork.fork_at(block_number=0, timestamp=0)
        # Session fees pinned on the backend by the fill-stateful plugin.
        gas_price = t8n.gas_price
        max_fee_per_gas = t8n.max_fee_per_gas
        max_priority_fee_per_gas = t8n.max_priority_fee_per_gas
        max_fee_per_blob_gas = t8n.max_fee_per_blob_gas
        if not all(
            [
                gas_price,
                max_fee_per_gas,
                max_priority_fee_per_gas,
                max_fee_per_blob_gas,
            ]
        ):
            raise RuntimeError(
                "make_stateful_fixture requires the backend to carry "
                f"non-zero session fees; got gas_price={gas_price}, "
                f"max_fee_per_gas={max_fee_per_gas}, "
                f"max_priority_fee_per_gas={max_priority_fee_per_gas}, "
                f"max_fee_per_blob_gas={max_fee_per_blob_gas}."
            )
        execute_plan.prepare_transactions(
            env=Environment(gas_limit=HexNumber(start_block["gasLimit"])),
            gas_price=gas_price,
            max_fee_per_gas=max_fee_per_gas,
            max_priority_fee_per_gas=max_priority_fee_per_gas,
            max_fee_per_blob_gas=max_fee_per_blob_gas,
            fork=session_fork,
        )
        required_balances = execute_plan.get_required_sender_balances(
            fork=session_fork,
        )
        resolve_deferred()
        min_balance(
            required_balances,
            gas_price=gas_price,
            max_fee_per_gas=max_fee_per_gas,
            max_priority_fee_per_gas=max_priority_fee_per_gas,
            max_fee_per_blob_gas=max_fee_per_blob_gas,
        )

    self.pre.verify_deployed_accounts(
        int(HexNumber(start_block["number"]))
    )

    # Materialise queued pre-alloc txs into a synthetic setup block.
    blocks_to_process: List[Block] = []
    if callable(pending_getter):
        setup_txs = pending_getter()
        if setup_txs:
            blocks_to_process.append(Block(txs=setup_txs))
    # Each block must be single-phase (Block.phase asserts otherwise);
    # mixed blocks (e.g. EIP-7702 authorization + benchmark exec) are
    # split into contiguous phase runs so benchmark gas isn't
    # swallowed into ``setupEngineNewPayloads``.
    blocks_to_process.extend(_split_blocks_by_phase(self.blocks))

    # Chain off the session start_block. We pull parent_* from a
    # FixtureHeader-validated copy of the client's block dict, but
    # seed block_hashes with the client's hash directly — FixtureHeader
    # recomputes block_hash from RLP and that diverges from the
    # client's authoritative hash unless every header byte is
    # reproduced exactly.
    start_block_number = int(HexNumber(start_block["number"]))
    start_block_hash = Hash(start_block["hash"])
    parent_header = FixtureHeader.model_validate(start_block)
    env = Environment(
        parent_difficulty=parent_header.difficulty,
        parent_timestamp=parent_header.timestamp,
        parent_base_fee_per_gas=parent_header.base_fee_per_gas,
        parent_blob_gas_used=parent_header.blob_gas_used,
        parent_excess_blob_gas=parent_header.excess_blob_gas,
        parent_gas_used=parent_header.gas_used,
        parent_gas_limit=parent_header.gas_limit,
        parent_ommers_hash=parent_header.ommers_hash,
        block_hashes={
            HexNumber(start_block_number): start_block_hash,
        },
    )

    setup_payloads: List[FixtureEngineNewPayload] = []
    execution_payloads: List[FixtureEngineNewPayload] = []
    # Aligned 1:1 with execution_payloads; None when no trace.
    execution_opcode_counts: List[Dict[str, int] | None] = []
    head_hash = start_block_hash
    benchmark_gas_used: int | None = None
    benchmark_block_gas_used: int | None = None
    benchmark_opcode_count: OpcodeCount | None = None
    # Alloc is not authoritative in stateful mode; pass self.pre as a
    # placeholder — ClientBackend ignores it.
    alloc: Alloc | LazyAlloc = self.pre
    for block in blocks_to_process:
        built_block = self.generate_block_data(
            t8n=t8n,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
        )
        assert isinstance(built_block, TestingBuildBlock), (
            "ClientBackend must return TestingBuildBlock; got "
            f"{type(built_block).__name__}"
        )
        payload = payload_metadata_to_fixture(
            built_block.engine_payload, phase=block.phase
        )
        # The client's authoritative block hash (the FixtureHeader RLP
        # hash diverges — the client picks fields like gas_limit).
        client_hash = Hash(
            built_block.engine_payload.payload_response.execution_payload.block_hash
        )
        if payload.phase == TestPhase.SETUP:
            setup_payloads.append(payload)
        else:
            execution_payloads.append(payload)
            block_opcode_count = t8n.extract_block_opcode_count(
                client_hash
            )
            execution_opcode_counts.append(
                block_opcode_count.model_dump()
                if block_opcode_count is not None
                else None
            )
            # Setup blocks (pre-alloc funding/deploys) are exempt:
            # ``expected_receipt_status`` describes the test's own
            # transactions, and setup txs always succeed.
            if built_block.result.receipts:
                self.validate_receipt_status(
                    receipts=built_block.result.receipts,
                    block_number=int(built_block.header.number),
                )
            if self.operation_mode == OpMode.BENCHMARKING:
                benchmark_gas_used = built_block.cumulative_gas_used()
                benchmark_block_gas_used = built_block.block_gas_used()
                # Consumed by BenchmarkTest's opcode-count verification.
                benchmark_opcode_count = block_opcode_count
        # apply_new_parent records the RLP hash; the next block's
        # parent_hash must point at what the client actually built.
        env = apply_new_parent(built_block.env, built_block.header)
        env = env.copy(
            block_hashes={
                **env.block_hashes,
                HexNumber(int(env.number)): client_hash,
            },
        )
        head_hash = client_hash

    if self.post.root:
        got_alloc = t8n.get_post_state_alloc(self.post)
        self.post.verify_post_alloc(got_alloc)

    fixture = BlockchainEngineStatefulFixture(
        fork=self.fork,
        last_block_hash=head_hash,
        config=FixtureConfig(fork=self.fork),
        snapshot_block_number=HexNumber(snapshot_block["number"]),
        snapshot_block_hash=Hash(snapshot_block["hash"]),
        start_block_number=HexNumber(start_block_number),
        start_block_hash=start_block_hash,
        setup_payloads=setup_payloads,
        payloads=execution_payloads,
        benchmark_gas_used=(
            HexNumber(benchmark_gas_used)
            if benchmark_gas_used is not None
            else None
        ),
    )
    metadata: Dict[str, Any] = {}
    if t8n.extract_opcode_count:
        metadata["opcode_counts"] = execution_opcode_counts
    return FillResult(
        fixture=fixture,
        gas_optimization=None,
        benchmark_gas_used=benchmark_gas_used,
        benchmark_block_gas_used=benchmark_block_gas_used,
        benchmark_opcode_count=benchmark_opcode_count,
        metadata=metadata,
        post_verifications=PostVerifications.from_alloc(self.post),
    )

generate(t8n, fixture_format)

Generate the BlockchainTest fixture.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def generate(
    self,
    t8n: FillerBackend,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the BlockchainTest fixture."""
    if fixture_format == BlockchainEngineStatefulFixture:
        return self.make_stateful_fixture(t8n)
    if fixture_format in [
        BlockchainEngineFixture,
        BlockchainEngineXFixture,
        BlockchainEngineSyncFixture,
    ]:
        return self.make_hive_fixture(t8n, fixture_format)
    elif fixture_format == BlockchainFixture:
        return self.make_fixture(t8n)

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, execute_format)

Generate the list of test fixtures.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        blocks: List[List[Transaction]] = []
        for block in self.blocks:
            blocks += [block.txs]
        # Pass gas validation params for benchmark tests
        # If not benchmark mode, skip gas used validation
        if self.operation_mode != OpMode.BENCHMARKING:
            self.skip_gas_used_validation = True

        benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
        return TransactionPost(
            blocks=blocks,
            post=self.post,
            benchmark_mode=benchmark_mode,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

Header

Bases: CamelModel

Header type used to describe block header properties in test specs.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

class Header(CamelModel):
    """Header type used to describe block header properties in test specs."""

    parent_hash: Hash | None = None
    ommers_hash: Hash | None = None
    fee_recipient: Address | None = None
    state_root: Hash | None = None
    transactions_trie: Hash | None = None
    receipts_root: Hash | None = None
    logs_bloom: Bloom | None = None
    difficulty: HexNumber | None = None
    number: HexNumber | None = None
    gas_limit: HexNumber | None = None
    gas_used: HexNumber | None = None
    timestamp: HexNumber | None = None
    extra_data: Bytes | None = None
    prev_randao: Hash | None = None
    nonce: HeaderNonce | None = None
    base_fee_per_gas: Removable | HexNumber | None = None
    withdrawals_root: Removable | Hash | None = None
    blob_gas_used: Removable | HexNumber | None = None
    excess_blob_gas: Removable | HexNumber | None = None
    parent_beacon_block_root: Removable | Hash | None = None
    requests_hash: Removable | Hash | None = None
    block_access_list_hash: Removable | Hash | None = None
    slot_number: Removable | HexNumber | None = None

    REMOVE_FIELD: ClassVar[Removable] = Removable()
    """
    Sentinel object used to specify that a header field should be removed.
    """
    EMPTY_FIELD: ClassVar[Removable] = Removable()
    """
    Sentinel object used to specify that a header field must be empty during
    verification.

    This can be used in a test to explicitly skip a field in a block's RLP
    encoding that would otherwise be included in the (json) output when the
    model is serialized. For example:

    ```
    header_modifier = Header(
        excess_blob_gas=Header.REMOVE_FIELD,
    )
    block = Block(
        timestamp=TIMESTAMP,
        rlp_modifier=header_modifier,
        exception=BlockException.INCORRECT_BLOCK_FORMAT,
        engine_api_error_code=EngineAPIError.InvalidParams,
    )
    ```
    """

    model_config = ConfigDict(
        **CamelModel.model_config,
        arbitrary_types_allowed=True,
    )

    @model_serializer(mode="wrap", when_used="json")
    def _serialize_model(self, serializer: Any, info: Any) -> Dict[str, Any]:
        """Exclude Removable fields from serialization."""
        del info
        data = serializer(self)
        return {k: v for k, v in data.items() if not isinstance(v, Removable)}

    @field_validator("withdrawals_root", mode="before")
    @classmethod
    def validate_withdrawals_root(cls, value: Any) -> Any:
        """Convert a list of withdrawals into the withdrawals root hash."""
        if isinstance(value, list):
            return Withdrawal.list_root(value)
        return value

    def apply(self, target: FixtureHeader) -> FixtureHeader:
        """
        Produce a fixture header copy with the set values from the modifier.
        """
        overrides = {
            k: (v if v is not Header.REMOVE_FIELD else None)
            for k, v in self.model_dump(exclude_none=True).items()
        }
        unknown = overrides.keys() - target.__class__.model_fields.keys()
        if unknown:
            raise ValueError(
                f"Header fields {unknown} do not exist on "
                f"{target.__class__.__name__}. Check for field name "
                f"mismatches between Header and {target.__class__.__name__}."
            )
        return target.copy(**overrides)

    def verify(self, target: FixtureHeader) -> None:
        """Verify that the header fields from self are as expected."""
        for field_name in self.__class__.model_fields:
            baseline_value = getattr(self, field_name)
            if baseline_value is not None:
                assert baseline_value is not Header.REMOVE_FIELD, (
                    "invalid header"
                )
                value = getattr(target, field_name)
                if baseline_value is Header.EMPTY_FIELD:
                    assert value is None, (
                        f"invalid header field {field_name}, "
                        f"got {value}, want None"
                    )
                    continue
                assert value == baseline_value, (
                    f"invalid header field ({field_name}) value, "
                    + f"got {value}, want {baseline_value}"
                )

REMOVE_FIELD = Removable() class-attribute

Sentinel object used to specify that a header field should be removed.

EMPTY_FIELD = Removable() class-attribute

Sentinel object used to specify that a header field must be empty during verification.

This can be used in a test to explicitly skip a field in a block's RLP encoding that would otherwise be included in the (json) output when the model is serialized. For example:

header_modifier = Header(
    excess_blob_gas=Header.REMOVE_FIELD,
)
block = Block(
    timestamp=TIMESTAMP,
    rlp_modifier=header_modifier,
    exception=BlockException.INCORRECT_BLOCK_FORMAT,
    engine_api_error_code=EngineAPIError.InvalidParams,
)

validate_withdrawals_root(value) classmethod

Convert a list of withdrawals into the withdrawals root hash.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

@field_validator("withdrawals_root", mode="before")
@classmethod
def validate_withdrawals_root(cls, value: Any) -> Any:
    """Convert a list of withdrawals into the withdrawals root hash."""
    if isinstance(value, list):
        return Withdrawal.list_root(value)
    return value

apply(target)

Produce a fixture header copy with the set values from the modifier.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def apply(self, target: FixtureHeader) -> FixtureHeader:
    """
    Produce a fixture header copy with the set values from the modifier.
    """
    overrides = {
        k: (v if v is not Header.REMOVE_FIELD else None)
        for k, v in self.model_dump(exclude_none=True).items()
    }
    unknown = overrides.keys() - target.__class__.model_fields.keys()
    if unknown:
        raise ValueError(
            f"Header fields {unknown} do not exist on "
            f"{target.__class__.__name__}. Check for field name "
            f"mismatches between Header and {target.__class__.__name__}."
        )
    return target.copy(**overrides)

verify(target)

Verify that the header fields from self are as expected.

Source code in packages/testing/src/execution_testing/specs/blockchain.py

def verify(self, target: FixtureHeader) -> None:
    """Verify that the header fields from self are as expected."""
    for field_name in self.__class__.model_fields:
        baseline_value = getattr(self, field_name)
        if baseline_value is not None:
            assert baseline_value is not Header.REMOVE_FIELD, (
                "invalid header"
            )
            value = getattr(target, field_name)
            if baseline_value is Header.EMPTY_FIELD:
                assert value is None, (
                    f"invalid header field {field_name}, "
                    f"got {value}, want None"
                )
                continue
            assert value == baseline_value, (
                f"invalid header field ({field_name}) value, "
                + f"got {value}, want {baseline_value}"
            )

StateTest

Bases: BaseTest

Filler type that tests transactions over the period of a single block.

Source code in packages/testing/src/execution_testing/specs/state.py

class StateTest(BaseTest):
    """
    Filler type that tests transactions over the period of a single block.
    """

    env: Environment = Field(default_factory=Environment)
    pre: Alloc
    post: Alloc
    tx: Transaction
    block_exception: (
        List[TransactionException | BlockException]
        | TransactionException
        | BlockException
        | None
    ) = None
    engine_api_error_code: Optional[EngineAPIError] = None
    blockchain_test_header_verify: Optional[Header] = None
    blockchain_test_rlp_modifier: Optional[Header] = None
    expected_block_access_list: Optional[BlockAccessListExpectation] = None
    chain_id: int = 1

    supported_fixture_formats: ClassVar[
        Sequence[FixtureFormat | LabeledFixtureFormat]
    ] = [
        StateFixture,
    ] + [
        fixture_format.with_label_suffix(
            "from_state_test",
            f"A {fixture_format.format_id()} generated from a state_test",
        )
        for fixture_format in BlockchainTest.supported_fixture_formats
        # Exclude sync fixtures from state tests - they don't make sense for
        # state tests
        if "Sync" not in fixture_format.format_class().__name__
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "state_test",
            "An execute test derived from a state test",
        ),
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "state_test_only": "Only generate a state test fixture",
    }

    def verify_modified_gas_limit(
        self,
        *,
        t8n: TransitionTool,
        base_tool_result: Result,
        base_tool_alloc: Alloc,
        fork: Fork,
        current_gas_limit: int,
        pre_alloc: Alloc,
        env: Environment,
        ignore_gas_differences: bool,
    ) -> bool:
        """Verify a new lower gas limit yields the same transaction outcome."""
        base_traces = base_tool_result.traces
        assert base_traces is not None, (
            "Traces not collected for gas optimization"
        )
        new_tx = self.tx.copy(
            gas_limit=current_gas_limit
        ).with_signature_and_sender()
        modified_tool_output = t8n.evaluate(
            transition_tool_data=TransitionTool.TransitionToolData(
                alloc=pre_alloc,
                txs=[new_tx],
                env=env,
                fork=fork,
                chain_id=self.chain_id,
                reward=0,  # Reward on state tests is always zero
                blob_schedule=fork.blob_schedule(),
                state_test=True,
            ),
            slow_request=self.is_tx_gas_heavy_test,
        )
        modified_traces = modified_tool_output.result.traces
        assert modified_traces is not None, (
            "Traces not collected for gas optimization"
        )
        if not base_traces.are_equivalent(
            modified_tool_output.result.traces,
            ignore_gas_differences,
        ):
            logger.debug(
                f"Traces are not equivalent (gas_limit={current_gas_limit})"
            )
            return False
        modified_tool_alloc = modified_tool_output.alloc.materialize()
        try:
            self.post.verify_post_alloc(modified_tool_alloc)
        except Exception as e:
            logger.debug(
                f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
            )
            logger.debug(e)
            return False
        try:
            verify_transactions(
                txs=[new_tx],
                result=modified_tool_output.result,
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
        except Exception as e:
            logger.debug(
                "Transactions are not equivalent "
                f"(gas_limit={current_gas_limit})"
            )
            logger.debug(e)
            return False
        if len(base_tool_alloc.root) != len(modified_tool_alloc.root):
            logger.debug(
                f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
            )
            return False
        if base_tool_alloc.root.keys() != modified_tool_alloc.root.keys():
            logger.debug(
                f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
            )
            return False
        for k in base_tool_alloc.root.keys():
            if k not in modified_tool_alloc:
                logger.debug(
                    "Post alloc is not equivalent "
                    f"(gas_limit={current_gas_limit})"
                )
                return False
            base_account = base_tool_alloc[k]
            modified_account = modified_tool_alloc[k]
            if (modified_account is None) != (base_account is None):
                logger.debug(
                    "Post alloc is not equivalent "
                    f"(gas_limit={current_gas_limit})"
                )
                return False
            if (
                modified_account is not None
                and base_account is not None
                and base_account.nonce != modified_account.nonce
            ):
                logger.debug(
                    "Post alloc is not equivalent "
                    f"(gas_limit={current_gas_limit})"
                )
                return False
        logger.debug(
            f"Gas limit is equivalent (gas_limit={current_gas_limit})"
        )
        return True

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard a fixture format from filling if the appropriate marker is
        used.
        """
        if "state_test_only" in [m.name for m in markers]:
            return fixture_format != StateFixture
        return False

    def _generate_blockchain_genesis_environment(self) -> Environment:
        """
        Generate the genesis environment for the BlockchainTest formatted test.
        """
        assert self.env.number >= 1, (
            "genesis block number cannot be negative, set state test "
            "env.number to at least 1"
        )
        assert self.env.timestamp >= 1, (
            "genesis timestamp cannot be negative, set state test "
            "env.timestamp to at least 1"
        )
        # There's only a handful of values that we need to set in the genesis
        # for the environment values at block 1 to make sense:
        # - Number: Needs to be N minus 1
        # - Timestamp: Needs to be zero, because the subsequent
        #              block can come at any time.
        # - Gas Limit: Changes from parent to child, needs to be set in genesis
        # - Base Fee Per Gas: Block's base fee depends on the parent's value
        # - Excess Blob Gas: Block's excess blob gas value depends on
        #                    the parent's value
        genesis_block_number = self.env.number - 1
        genesis_timestamp = 0
        genesis_fork = self.fork.fork_at(
            block_number=genesis_block_number, timestamp=genesis_timestamp
        )
        kwargs: Dict[str, Any] = {
            "number": genesis_block_number,
            "timestamp": genesis_timestamp,
        }

        if "gas_limit" in self.env.model_fields_set:
            kwargs["gas_limit"] = self.env.gas_limit

        if self.env.base_fee_per_gas:
            # Calculate genesis base fee per gas from state test's block#1 env
            kwargs["base_fee_per_gas"] = HexNumber(
                int(str(self.env.base_fee_per_gas), 0) * 8 // 7
            )

        if self.env.excess_blob_gas:
            # The excess blob gas environment value means the value of the
            # context (block header) where the transaction is executed. In a
            # blockchain test, we need to indirectly set the excess blob gas by
            # setting the excess blob gas of the genesis block to the expected
            # value plus the BLOB_TARGET_GAS_PER_BLOCK, which is the value that
            # will be subtracted from the excess blob gas when the first block
            # is mined.
            kwargs["excess_blob_gas"] = self.env.excess_blob_gas + (
                genesis_fork.target_blobs_per_block()
                * genesis_fork.blob_gas_per_blob()
            )

        return Environment(**kwargs)

    def _generate_blockchain_blocks(self) -> List[Block]:
        """
        Generate the single block that represents this state test in a
        BlockchainTest format.
        """
        number = self.env.number
        timestamp = self.env.timestamp
        fork = self.fork.fork_at(block_number=number, timestamp=timestamp)
        kwargs = {
            "number": self.env.number,
            "timestamp": self.env.timestamp,
            "prev_randao": self.env.prev_randao,
            "fee_recipient": self.env.fee_recipient,
            "gas_limit": self.env.gas_limit,
            "extra_data": self.env.extra_data,
            "withdrawals": self.env.withdrawals,
            "parent_beacon_block_root": self.env.parent_beacon_block_root,
            "slot_number": self.env.slot_number,
            "txs": [self.tx],
            "ommers": [],
            "header_verify": self.blockchain_test_header_verify,
            "rlp_modifier": self.blockchain_test_rlp_modifier,
            "expected_block_access_list": self.expected_block_access_list,
        }
        if not fork.header_prev_randao_required():
            kwargs["difficulty"] = self.env.difficulty
        if "block_exception" in self.model_fields_set:
            kwargs["exception"] = self.block_exception  # type: ignore
        elif "error" in self.tx.model_fields_set:
            kwargs["exception"] = self.tx.error  # type: ignore
        return [Block(**kwargs)]

    def generate_blockchain_test(self) -> BlockchainTest:
        """Generate a BlockchainTest fixture from this StateTest fixture."""
        # Checked before the genesis derivation below, which asks the
        # fork for blob constants a fork without blobs does not have.
        self.env.check_fork_fields(
            self.fork.fork_at(
                block_number=self.env.number, timestamp=self.env.timestamp
            )
        )
        return BlockchainTest.from_test(
            base_test=self,
            genesis_environment=self._generate_blockchain_genesis_environment(),
            pre=self.pre,
            post=self.post,
            blocks=self._generate_blockchain_blocks(),
        )

    def make_state_test_fixture(
        self,
        t8n: TransitionTool,
    ) -> FillResult:
        """Create a fixture from the state test definition."""
        # We can't generate a state test fixture that names a transition fork,
        # so we get the fork at the block number and timestamp of the state
        # test
        fork = self.fork.fork_at(
            block_number=self.env.number, timestamp=self.env.timestamp
        )

        env = self.env.set_fork_requirements(fork)
        env.check_fork_fields(fork)
        tx = self.tx.with_gas_limit(
            max_gas_limit=env.gas_limit,
            transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
            state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
        ).with_signature_and_sender(keep_secret_key=True)
        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation()),
            self.pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")

        transition_tool_output = t8n.evaluate(
            transition_tool_data=TransitionTool.TransitionToolData(
                alloc=pre_alloc,
                txs=[tx],
                env=env,
                fork=fork,
                chain_id=self.chain_id,
                reward=0,  # Reward on state tests is always zero
                blob_schedule=fork.blob_schedule(),
                state_test=True,
            ),
            slow_request=self.is_tx_gas_heavy_test,
        )
        output_alloc = transition_tool_output.alloc.materialize()

        try:
            self.post.verify_post_alloc(output_alloc)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

        try:
            verify_transactions(
                txs=[tx],
                result=transition_tool_output.result,
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(output_alloc)
            raise e

        gas_optimization: int | None = None

        if (
            self.operation_mode == OpMode.OPTIMIZE_GAS
            or self.operation_mode == OpMode.OPTIMIZE_GAS_POST_PROCESSING
        ):
            ignore_gas_differences = (
                self.operation_mode == OpMode.OPTIMIZE_GAS_POST_PROCESSING
            )
            base_tool_output = transition_tool_output
            base_tool_alloc = base_tool_output.alloc.materialize()
            base_tool_result = base_tool_output.result

            assert base_tool_result.traces is not None, "Traces not found."

            # First try reducing the gas limit only by one, if the validation
            # fails, it means that the traces change even with the slightest
            # modification to the gas.
            tx_gas_limit = int(tx.gas_limit)

            if self.verify_modified_gas_limit(
                t8n=t8n,
                base_tool_result=base_tool_result,
                base_tool_alloc=base_tool_alloc,
                fork=fork,
                current_gas_limit=tx_gas_limit - 1,
                pre_alloc=pre_alloc,
                env=env,
                ignore_gas_differences=ignore_gas_differences,
            ):
                minimum_gas_limit = 0
                maximum_gas_limit = tx_gas_limit
                while minimum_gas_limit < maximum_gas_limit:
                    current_gas_limit = (
                        maximum_gas_limit + minimum_gas_limit
                    ) // 2
                    if self.verify_modified_gas_limit(
                        t8n=t8n,
                        base_tool_result=base_tool_result,
                        base_tool_alloc=base_tool_alloc,
                        fork=fork,
                        current_gas_limit=current_gas_limit,
                        pre_alloc=pre_alloc,
                        env=env,
                        ignore_gas_differences=ignore_gas_differences,
                    ):
                        maximum_gas_limit = current_gas_limit
                    else:
                        minimum_gas_limit = current_gas_limit + 1
                        if (
                            self.gas_optimization_max_gas_limit is not None
                            and minimum_gas_limit
                            > self.gas_optimization_max_gas_limit
                        ):
                            raise Exception(
                                "Requires more than the minimum "
                                f"{self.gas_optimization_max_gas_limit} "
                                "wanted."
                            )

                assert self.verify_modified_gas_limit(
                    t8n=t8n,
                    base_tool_result=base_tool_result,
                    base_tool_alloc=base_tool_alloc,
                    fork=fork,
                    current_gas_limit=minimum_gas_limit,
                    pre_alloc=pre_alloc,
                    env=env,
                    ignore_gas_differences=ignore_gas_differences,
                )
                gas_optimization = current_gas_limit
            else:
                raise Exception("Impossible to compare.")

        if len(transition_tool_output.result.receipts) == 1:
            receipt = FixtureTransactionReceipt.from_transaction_receipt(
                transition_tool_output.result.receipts[0], tx
            )
            receipt_root = FixtureTransactionReceipt.list_root([receipt])
            assert (
                transition_tool_output.result.receipts_root == receipt_root
            ), (
                f"Receipts root mismatch: "
                f"{transition_tool_output.result.receipts_root} != "
                f"{receipt_root.hex()}"
                f"Receipt: {receipt.rlp()}"
            )
        else:
            receipt = None

        fixture = StateFixture(
            env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
            pre=pre_alloc,
            post={
                fork: [
                    FixtureForkPost(
                        state_root=transition_tool_output.result.state_root,
                        logs_hash=transition_tool_output.result.logs_hash,
                        receipt=receipt,
                        tx_bytes=tx.rlp(),
                        expect_exception=tx.error,
                        state=output_alloc,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
            config=FixtureConfig(
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                    fork.blob_schedule()
                ),
                chain_id=self.chain_id,
            ),
        )
        return FillResult(
            fixture=fixture,
            gas_optimization=gas_optimization,
            benchmark_gas_used=transition_tool_output.result.gas_used,
            benchmark_opcode_count=transition_tool_output.result.opcode_count,
            post_verifications=PostVerifications.from_alloc(self.post),
        )

    def get_genesis_environment(self) -> Environment:
        """Get the genesis environment for pre-allocation groups."""
        return self.generate_blockchain_test().get_genesis_environment()

    def generate(
        self,
        t8n: TransitionTool,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the BlockchainTest fixture."""
        self.check_exception_test(exception=self.tx.error is not None)
        if fixture_format in BlockchainTest.supported_fixture_formats:
            return self.generate_blockchain_test().generate(
                t8n=t8n, fixture_format=fixture_format
            )
        elif fixture_format == StateFixture:
            return self.make_state_test_fixture(t8n)

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            # Pass gas validation params for benchmark tests
            # If not benchmark mode, skip gas used validation
            if self.operation_mode != OpMode.BENCHMARKING:
                self.skip_gas_used_validation = True
            benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
            return TransactionPost(
                blocks=[[self.tx]],
                post=self.post,
                benchmark_mode=benchmark_mode,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

verify_modified_gas_limit(*, t8n, base_tool_result, base_tool_alloc, fork, current_gas_limit, pre_alloc, env, ignore_gas_differences)

Verify a new lower gas limit yields the same transaction outcome.

Source code in packages/testing/src/execution_testing/specs/state.py

def verify_modified_gas_limit(
    self,
    *,
    t8n: TransitionTool,
    base_tool_result: Result,
    base_tool_alloc: Alloc,
    fork: Fork,
    current_gas_limit: int,
    pre_alloc: Alloc,
    env: Environment,
    ignore_gas_differences: bool,
) -> bool:
    """Verify a new lower gas limit yields the same transaction outcome."""
    base_traces = base_tool_result.traces
    assert base_traces is not None, (
        "Traces not collected for gas optimization"
    )
    new_tx = self.tx.copy(
        gas_limit=current_gas_limit
    ).with_signature_and_sender()
    modified_tool_output = t8n.evaluate(
        transition_tool_data=TransitionTool.TransitionToolData(
            alloc=pre_alloc,
            txs=[new_tx],
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            blob_schedule=fork.blob_schedule(),
            state_test=True,
        ),
        slow_request=self.is_tx_gas_heavy_test,
    )
    modified_traces = modified_tool_output.result.traces
    assert modified_traces is not None, (
        "Traces not collected for gas optimization"
    )
    if not base_traces.are_equivalent(
        modified_tool_output.result.traces,
        ignore_gas_differences,
    ):
        logger.debug(
            f"Traces are not equivalent (gas_limit={current_gas_limit})"
        )
        return False
    modified_tool_alloc = modified_tool_output.alloc.materialize()
    try:
        self.post.verify_post_alloc(modified_tool_alloc)
    except Exception as e:
        logger.debug(
            f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
        )
        logger.debug(e)
        return False
    try:
        verify_transactions(
            txs=[new_tx],
            result=modified_tool_output.result,
            transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
        )
    except Exception as e:
        logger.debug(
            "Transactions are not equivalent "
            f"(gas_limit={current_gas_limit})"
        )
        logger.debug(e)
        return False
    if len(base_tool_alloc.root) != len(modified_tool_alloc.root):
        logger.debug(
            f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
        )
        return False
    if base_tool_alloc.root.keys() != modified_tool_alloc.root.keys():
        logger.debug(
            f"Post alloc is not equivalent (gas_limit={current_gas_limit})"
        )
        return False
    for k in base_tool_alloc.root.keys():
        if k not in modified_tool_alloc:
            logger.debug(
                "Post alloc is not equivalent "
                f"(gas_limit={current_gas_limit})"
            )
            return False
        base_account = base_tool_alloc[k]
        modified_account = modified_tool_alloc[k]
        if (modified_account is None) != (base_account is None):
            logger.debug(
                "Post alloc is not equivalent "
                f"(gas_limit={current_gas_limit})"
            )
            return False
        if (
            modified_account is not None
            and base_account is not None
            and base_account.nonce != modified_account.nonce
        ):
            logger.debug(
                "Post alloc is not equivalent "
                f"(gas_limit={current_gas_limit})"
            )
            return False
    logger.debug(
        f"Gas limit is equivalent (gas_limit={current_gas_limit})"
    )
    return True

discard_fixture_format_by_marks(fixture_format, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in packages/testing/src/execution_testing/specs/state.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard a fixture format from filling if the appropriate marker is
    used.
    """
    if "state_test_only" in [m.name for m in markers]:
        return fixture_format != StateFixture
    return False

generate_blockchain_test()

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in packages/testing/src/execution_testing/specs/state.py

def generate_blockchain_test(self) -> BlockchainTest:
    """Generate a BlockchainTest fixture from this StateTest fixture."""
    # Checked before the genesis derivation below, which asks the
    # fork for blob constants a fork without blobs does not have.
    self.env.check_fork_fields(
        self.fork.fork_at(
            block_number=self.env.number, timestamp=self.env.timestamp
        )
    )
    return BlockchainTest.from_test(
        base_test=self,
        genesis_environment=self._generate_blockchain_genesis_environment(),
        pre=self.pre,
        post=self.post,
        blocks=self._generate_blockchain_blocks(),
    )

make_state_test_fixture(t8n)

Create a fixture from the state test definition.

Source code in packages/testing/src/execution_testing/specs/state.py

def make_state_test_fixture(
    self,
    t8n: TransitionTool,
) -> FillResult:
    """Create a fixture from the state test definition."""
    # We can't generate a state test fixture that names a transition fork,
    # so we get the fork at the block number and timestamp of the state
    # test
    fork = self.fork.fork_at(
        block_number=self.env.number, timestamp=self.env.timestamp
    )

    env = self.env.set_fork_requirements(fork)
    env.check_fork_fields(fork)
    tx = self.tx.with_gas_limit(
        max_gas_limit=env.gas_limit,
        transaction_gas_limit_cap=fork.transaction_gas_limit_cap(),
        state_gas_reservoir_enabled=fork.state_gas_reservoir_enabled(),
    ).with_signature_and_sender(keep_secret_key=True)
    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation()),
        self.pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")

    transition_tool_output = t8n.evaluate(
        transition_tool_data=TransitionTool.TransitionToolData(
            alloc=pre_alloc,
            txs=[tx],
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            blob_schedule=fork.blob_schedule(),
            state_test=True,
        ),
        slow_request=self.is_tx_gas_heavy_test,
    )
    output_alloc = transition_tool_output.alloc.materialize()

    try:
        self.post.verify_post_alloc(output_alloc)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

    try:
        verify_transactions(
            txs=[tx],
            result=transition_tool_output.result,
            transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
        )
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(output_alloc)
        raise e

    gas_optimization: int | None = None

    if (
        self.operation_mode == OpMode.OPTIMIZE_GAS
        or self.operation_mode == OpMode.OPTIMIZE_GAS_POST_PROCESSING
    ):
        ignore_gas_differences = (
            self.operation_mode == OpMode.OPTIMIZE_GAS_POST_PROCESSING
        )
        base_tool_output = transition_tool_output
        base_tool_alloc = base_tool_output.alloc.materialize()
        base_tool_result = base_tool_output.result

        assert base_tool_result.traces is not None, "Traces not found."

        # First try reducing the gas limit only by one, if the validation
        # fails, it means that the traces change even with the slightest
        # modification to the gas.
        tx_gas_limit = int(tx.gas_limit)

        if self.verify_modified_gas_limit(
            t8n=t8n,
            base_tool_result=base_tool_result,
            base_tool_alloc=base_tool_alloc,
            fork=fork,
            current_gas_limit=tx_gas_limit - 1,
            pre_alloc=pre_alloc,
            env=env,
            ignore_gas_differences=ignore_gas_differences,
        ):
            minimum_gas_limit = 0
            maximum_gas_limit = tx_gas_limit
            while minimum_gas_limit < maximum_gas_limit:
                current_gas_limit = (
                    maximum_gas_limit + minimum_gas_limit
                ) // 2
                if self.verify_modified_gas_limit(
                    t8n=t8n,
                    base_tool_result=base_tool_result,
                    base_tool_alloc=base_tool_alloc,
                    fork=fork,
                    current_gas_limit=current_gas_limit,
                    pre_alloc=pre_alloc,
                    env=env,
                    ignore_gas_differences=ignore_gas_differences,
                ):
                    maximum_gas_limit = current_gas_limit
                else:
                    minimum_gas_limit = current_gas_limit + 1
                    if (
                        self.gas_optimization_max_gas_limit is not None
                        and minimum_gas_limit
                        > self.gas_optimization_max_gas_limit
                    ):
                        raise Exception(
                            "Requires more than the minimum "
                            f"{self.gas_optimization_max_gas_limit} "
                            "wanted."
                        )

            assert self.verify_modified_gas_limit(
                t8n=t8n,
                base_tool_result=base_tool_result,
                base_tool_alloc=base_tool_alloc,
                fork=fork,
                current_gas_limit=minimum_gas_limit,
                pre_alloc=pre_alloc,
                env=env,
                ignore_gas_differences=ignore_gas_differences,
            )
            gas_optimization = current_gas_limit
        else:
            raise Exception("Impossible to compare.")

    if len(transition_tool_output.result.receipts) == 1:
        receipt = FixtureTransactionReceipt.from_transaction_receipt(
            transition_tool_output.result.receipts[0], tx
        )
        receipt_root = FixtureTransactionReceipt.list_root([receipt])
        assert (
            transition_tool_output.result.receipts_root == receipt_root
        ), (
            f"Receipts root mismatch: "
            f"{transition_tool_output.result.receipts_root} != "
            f"{receipt_root.hex()}"
            f"Receipt: {receipt.rlp()}"
        )
    else:
        receipt = None

    fixture = StateFixture(
        env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
        pre=pre_alloc,
        post={
            fork: [
                FixtureForkPost(
                    state_root=transition_tool_output.result.state_root,
                    logs_hash=transition_tool_output.result.logs_hash,
                    receipt=receipt,
                    tx_bytes=tx.rlp(),
                    expect_exception=tx.error,
                    state=output_alloc,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
        config=FixtureConfig(
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(
                fork.blob_schedule()
            ),
            chain_id=self.chain_id,
        ),
    )
    return FillResult(
        fixture=fixture,
        gas_optimization=gas_optimization,
        benchmark_gas_used=transition_tool_output.result.gas_used,
        benchmark_opcode_count=transition_tool_output.result.opcode_count,
        post_verifications=PostVerifications.from_alloc(self.post),
    )

get_genesis_environment()

Get the genesis environment for pre-allocation groups.

Source code in packages/testing/src/execution_testing/specs/state.py

def get_genesis_environment(self) -> Environment:
    """Get the genesis environment for pre-allocation groups."""
    return self.generate_blockchain_test().get_genesis_environment()

generate(t8n, fixture_format)

Generate the BlockchainTest fixture.

Source code in packages/testing/src/execution_testing/specs/state.py

def generate(
    self,
    t8n: TransitionTool,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the BlockchainTest fixture."""
    self.check_exception_test(exception=self.tx.error is not None)
    if fixture_format in BlockchainTest.supported_fixture_formats:
        return self.generate_blockchain_test().generate(
            t8n=t8n, fixture_format=fixture_format
        )
    elif fixture_format == StateFixture:
        return self.make_state_test_fixture(t8n)

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, execute_format)

Generate the list of test fixtures.

Source code in packages/testing/src/execution_testing/specs/state.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        # Pass gas validation params for benchmark tests
        # If not benchmark mode, skip gas used validation
        if self.operation_mode != OpMode.BENCHMARKING:
            self.skip_gas_used_validation = True
        benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
        return TransactionPost(
            blocks=[[self.tx]],
            post=self.post,
            benchmark_mode=benchmark_mode,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

TransactionTest

Bases: BaseTest

Filler type that tests the transaction over the period of a single block.

Source code in packages/testing/src/execution_testing/specs/transaction.py

class TransactionTest(BaseTest):
    """
    Filler type that tests the transaction over the period of a single block.
    """

    tx: Transaction
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[
        Sequence[FixtureFormat | LabeledFixtureFormat]
    ] = [
        TransactionFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "transaction_test",
            "An execute test derived from a transaction test",
        ),
    ]

    def make_transaction_test_fixture(
        self,
    ) -> FillResult:
        """Create a fixture from the transaction test definition."""
        fork = self.fork.transitions_from()
        if self.tx.error is not None:
            result = FixtureResult(
                exception=self.tx.error,
                hash=None,
                intrinsic_gas=0,
                sender=None,
            )
        else:
            intrinsic_gas_cost_calculator = (
                fork.transitions_from().transaction_intrinsic_cost_calculator()
            )
            intrinsic_gas = intrinsic_gas_cost_calculator(
                calldata=self.tx.data,
                contract_creation=self.tx.to is None,
                access_list=self.tx.access_list,
                authorization_list_or_count=self.tx.authorization_list,
                sends_value=self.tx.value > 0,
                recipient_type=(
                    RecipientType.SELF
                    if self.tx.to == self.tx.sender
                    else RecipientType.CONTRACT
                ),
            )
            result = FixtureResult(
                exception=None,
                hash=self.tx.hash,
                intrinsic_gas=intrinsic_gas,
                sender=self.tx.sender,
            )

        fixture = TransactionFixture(
            result={
                fork: result,
            },
            transaction=self.tx.with_signature_and_sender().rlp(),
        )
        return FillResult(
            fixture=fixture,
            gas_optimization=None,
            benchmark_gas_used=None,
            benchmark_opcode_count=None,
        )

    def generate(
        self,
        t8n: TransitionTool,
        fixture_format: FixtureFormat | LabeledFixtureFormat,
    ) -> FillResult:
        """Generate the TransactionTest fixture."""
        del t8n

        self.check_exception_test(exception=self.tx.error is not None)
        if fixture_format == TransactionFixture:
            return self.make_transaction_test_fixture()

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        execute_format: ExecuteFormat | LabeledExecuteFormat,
    ) -> BaseExecute:
        """Execute the transaction test by sending it to the live network."""
        if execute_format == TransactionPost:
            benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
            return TransactionPost(
                blocks=[[self.tx]],
                post={},
                benchmark_mode=benchmark_mode,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

make_transaction_test_fixture()

Create a fixture from the transaction test definition.

Source code in packages/testing/src/execution_testing/specs/transaction.py

def make_transaction_test_fixture(
    self,
) -> FillResult:
    """Create a fixture from the transaction test definition."""
    fork = self.fork.transitions_from()
    if self.tx.error is not None:
        result = FixtureResult(
            exception=self.tx.error,
            hash=None,
            intrinsic_gas=0,
            sender=None,
        )
    else:
        intrinsic_gas_cost_calculator = (
            fork.transitions_from().transaction_intrinsic_cost_calculator()
        )
        intrinsic_gas = intrinsic_gas_cost_calculator(
            calldata=self.tx.data,
            contract_creation=self.tx.to is None,
            access_list=self.tx.access_list,
            authorization_list_or_count=self.tx.authorization_list,
            sends_value=self.tx.value > 0,
            recipient_type=(
                RecipientType.SELF
                if self.tx.to == self.tx.sender
                else RecipientType.CONTRACT
            ),
        )
        result = FixtureResult(
            exception=None,
            hash=self.tx.hash,
            intrinsic_gas=intrinsic_gas,
            sender=self.tx.sender,
        )

    fixture = TransactionFixture(
        result={
            fork: result,
        },
        transaction=self.tx.with_signature_and_sender().rlp(),
    )
    return FillResult(
        fixture=fixture,
        gas_optimization=None,
        benchmark_gas_used=None,
        benchmark_opcode_count=None,
    )

generate(t8n, fixture_format)

Generate the TransactionTest fixture.

Source code in packages/testing/src/execution_testing/specs/transaction.py

def generate(
    self,
    t8n: TransitionTool,
    fixture_format: FixtureFormat | LabeledFixtureFormat,
) -> FillResult:
    """Generate the TransactionTest fixture."""
    del t8n

    self.check_exception_test(exception=self.tx.error is not None)
    if fixture_format == TransactionFixture:
        return self.make_transaction_test_fixture()

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, execute_format)

Execute the transaction test by sending it to the live network.

Source code in packages/testing/src/execution_testing/specs/transaction.py

def execute(
    self,
    *,
    execute_format: ExecuteFormat | LabeledExecuteFormat,
) -> BaseExecute:
    """Execute the transaction test by sending it to the live network."""
    if execute_format == TransactionPost:
        benchmark_mode = self.operation_mode == OpMode.BENCHMARKING
        return TransactionPost(
            blocks=[[self.tx]],
            post={},
            benchmark_mode=benchmark_mode,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")