Ethereum Test Fixtures package

Ethereum test fixture format definitions.

BaseFixture

Bases: CamelModel

Represents a base Ethereum test fixture of any type.

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

class BaseFixture(CamelModel):
    """Represents a base Ethereum test fixture of any type."""

    # Base Fixture class properties
    formats: ClassVar[Dict[str, Type["BaseFixture"]]] = {}
    formats_type_adapter: ClassVar[TypeAdapter]

    info: Dict[str, Dict[str, Any] | str] = Field(
        default_factory=dict, alias="_info"
    )
    post_verifications: PostVerifications | None = Field(
        default=None, alias="postVerifications"
    )

    # Fixture format properties
    format_name: ClassVar[str] = ""
    output_file_extension: ClassVar[str] = ".json"
    description: ClassVar[str] = "Unknown fixture format; it has not been set."
    format_phases: ClassVar[Set[FixtureFillingPhase]] = {
        # Normal fixture types can be filled whether the pre-alloc phase was
        # executed or not.
        FixtureFillingPhase.FILL,
        FixtureFillingPhase.FILL_AFTER_PRE_ALLOC_GENERATION,
    }
    transition_tool_cache_key: ClassVar[str] = ""

    @classmethod
    def output_base_dir_name(cls) -> str:
        """
        Return name of the subdirectory where this type of fixture should be
        dumped to.
        """
        return cls.format_name.replace("test", "tests")

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        """
        Register all subclasses of BaseFixture with a fixture format name set
        as possible fixture formats.
        """
        if cls.format_name:
            # Register the new fixture format
            BaseFixture.formats[cls.format_name] = cls
            if len(BaseFixture.formats) > 1:
                BaseFixture.formats_type_adapter = TypeAdapter(
                    Annotated[
                        Union[
                            tuple(
                                Annotated[fixture_format, Tag(format_name)]
                                for (
                                    format_name,
                                    fixture_format,
                                ) in BaseFixture.formats.items()
                            )
                        ],
                        Discriminator(fixture_format_discriminator),
                    ]
                )
            else:
                BaseFixture.formats_type_adapter = TypeAdapter(cls)

    @model_validator(mode="wrap")
    @classmethod
    def _parse_into_subclass(
        cls, v: Any, handler: ValidatorFunctionWrapHandler
    ) -> "BaseFixture":
        """Parse the fixture into the correct subclass."""
        if cls is BaseFixture:
            return BaseFixture.formats_type_adapter.validate_python(v)
        return handler(v)

    @cached_property
    def json_dict(self) -> Dict[str, Any]:
        """Returns the JSON representation of the fixture."""
        return self.model_dump(
            mode="json", by_alias=True, exclude_none=True, exclude={"info"}
        )

    @cached_property
    def hash(self) -> str:
        """Returns the hash of the fixture."""
        json_str = json.dumps(
            self.json_dict, sort_keys=True, separators=(",", ":")
        )
        h = hashlib.sha256(json_str.encode("utf-8")).hexdigest()
        return f"0x{h}"

    def json_dict_with_info(self, hash_only: bool = False) -> Dict[str, Any]:
        """Return JSON representation of the fixture with the info field."""
        dict_with_info = self.json_dict.copy()
        dict_with_info["_info"] = {"hash": self.hash}
        if not hash_only:
            dict_with_info["_info"].update(self.info)
        return dict_with_info

    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)
        self.info["fixture-format"] = self.format_name

    def fill_info(
        self,
        t8n_version: str,
        test_case_description: str,
        fixture_source_url: str,
        ref_spec: ReferenceSpec | None,
        _info_metadata: Dict[str, Any] | None,
        metadata: Dict[str, Any] | None = None,
    ) -> None:
        """Fill the info field for this fixture."""
        if "comment" not in self.info:
            self.info["comment"] = "`execution-specs` generated test"
        self.info["filling-transition-tool"] = t8n_version
        self.info["description"] = test_case_description
        self.info["url"] = fixture_source_url
        if metadata:
            self.info["metadata"] = metadata
        if ref_spec is not None:
            ref_spec.write_info(self.info)
        if _info_metadata:
            self.info.update(_info_metadata)

    def get_fork(self) -> Fork | TransitionFork | None:
        """Return fork of the fixture as a string."""
        raise NotImplementedError

    @classmethod
    def format_class(cls) -> "Type[BaseFixture]":
        """Get the fixture format."""
        return cls

    @classmethod
    def format_id(cls) -> str:
        """Get string used as identifier for this format."""
        return cls.format_name.lower()

    @classmethod
    def is_variant(cls, variant: str) -> bool:
        """
        Return whether this format is the named variant.

        A plain format never is: only a label can name a variant.
        """
        del variant
        return False

    @classmethod
    def with_label_suffix(
        cls,
        suffix: str,
        description: str | None = None,
        *,
        transition_tool_cache_key: str | None = None,
        variant: str | None = None,
    ) -> "LabeledFixtureFormat":
        """
        Return this format labeled `<format_id>_<suffix>`.

        Use this instead of building a `LabeledFixtureFormat` by hand when a
        spec type re-labels the formats of another one: the label is derived
        from `format_id()`, so a format that already carries a label derives a
        distinct label per label instead of collapsing them all onto its
        format name.

        `transition_tool_cache_key` defaults to this format's key, which is
        what a label that only renames the same fixture wants. A suffix that
        asks the transition tool for something different must pass its own
        key, or an empty string to opt out of caching.

        `variant` names what the label asks its spec type to fill differently,
        for that spec type to query with `is_variant()` rather than comparing
        formats.
        """
        return LabeledFixtureFormat(
            cls,
            f"{cls.format_id()}_{suffix}",
            description
            if description is not None
            else f"A {cls.format_id()} {suffix.replace('_', ' ')}",
            transition_tool_cache_key=transition_tool_cache_key,
            variant=variant,
        )

    @classmethod
    def marks(
        cls, *, transition_tool_cache_key: str | None = None
    ) -> List[pytest.MarkDecorator | pytest.Mark]:
        """
        Get list of pytest marks that need to be added to a test produced
        with this fixture format.

        `transition_tool_cache_key` overrides the format's own key.
        """
        cache_key = (
            cls.transition_tool_cache_key
            if transition_tool_cache_key is None
            else transition_tool_cache_key
        )
        marks: List[pytest.MarkDecorator | pytest.Mark] = [
            getattr(
                pytest.mark,
                cls.format_name.lower(),
            ),
        ]
        if cache_key:
            marks.append(pytest.mark.transition_tool_cache_key(cache_key))
        return marks

    @classmethod
    def supports_fork(cls, fork: Fork | TransitionFork) -> bool:
        """
        Return whether the fixture can be generated for the given fork.

        By default, all fixtures support all forks.
        """
        del fork
        return True

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fork: Fork | TransitionFork,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard a fixture format from filling if the appropriate marker is
        used.
        """
        del fork, markers
        return False

output_base_dir_name() classmethod

Return name of the subdirectory where this type of fixture should be dumped to.

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

@classmethod
def output_base_dir_name(cls) -> str:
    """
    Return name of the subdirectory where this type of fixture should be
    dumped to.
    """
    return cls.format_name.replace("test", "tests")

__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/fixtures/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.
    """
    if cls.format_name:
        # Register the new fixture format
        BaseFixture.formats[cls.format_name] = cls
        if len(BaseFixture.formats) > 1:
            BaseFixture.formats_type_adapter = TypeAdapter(
                Annotated[
                    Union[
                        tuple(
                            Annotated[fixture_format, Tag(format_name)]
                            for (
                                format_name,
                                fixture_format,
                            ) in BaseFixture.formats.items()
                        )
                    ],
                    Discriminator(fixture_format_discriminator),
                ]
            )
        else:
            BaseFixture.formats_type_adapter = TypeAdapter(cls)

json_dict cached property

Returns the JSON representation of the fixture.

hash cached property

Returns the hash of the fixture.

json_dict_with_info(hash_only=False)

Return JSON representation of the fixture with the info field.

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

def json_dict_with_info(self, hash_only: bool = False) -> Dict[str, Any]:
    """Return JSON representation of the fixture with the info field."""
    dict_with_info = self.json_dict.copy()
    dict_with_info["_info"] = {"hash": self.hash}
    if not hash_only:
        dict_with_info["_info"].update(self.info)
    return dict_with_info

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/fixtures/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)
    self.info["fixture-format"] = self.format_name

fill_info(t8n_version, test_case_description, fixture_source_url, ref_spec, _info_metadata, metadata=None)

Fill the info field for this fixture.

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

def fill_info(
    self,
    t8n_version: str,
    test_case_description: str,
    fixture_source_url: str,
    ref_spec: ReferenceSpec | None,
    _info_metadata: Dict[str, Any] | None,
    metadata: Dict[str, Any] | None = None,
) -> None:
    """Fill the info field for this fixture."""
    if "comment" not in self.info:
        self.info["comment"] = "`execution-specs` generated test"
    self.info["filling-transition-tool"] = t8n_version
    self.info["description"] = test_case_description
    self.info["url"] = fixture_source_url
    if metadata:
        self.info["metadata"] = metadata
    if ref_spec is not None:
        ref_spec.write_info(self.info)
    if _info_metadata:
        self.info.update(_info_metadata)

get_fork()

Return fork of the fixture as a string.

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

def get_fork(self) -> Fork | TransitionFork | None:
    """Return fork of the fixture as a string."""
    raise NotImplementedError

format_class() classmethod

Get the fixture format.

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

@classmethod
def format_class(cls) -> "Type[BaseFixture]":
    """Get the fixture format."""
    return cls

format_id() classmethod

Get string used as identifier for this format.

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

@classmethod
def format_id(cls) -> str:
    """Get string used as identifier for this format."""
    return cls.format_name.lower()

is_variant(variant) classmethod

Return whether this format is the named variant.

A plain format never is: only a label can name a variant.

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

@classmethod
def is_variant(cls, variant: str) -> bool:
    """
    Return whether this format is the named variant.

    A plain format never is: only a label can name a variant.
    """
    del variant
    return False

with_label_suffix(suffix, description=None, *, transition_tool_cache_key=None, variant=None) classmethod

Return this format labeled <format_id>_<suffix>.

Use this instead of building a LabeledFixtureFormat by hand when a spec type re-labels the formats of another one: the label is derived from format_id(), so a format that already carries a label derives a distinct label per label instead of collapsing them all onto its format name.

transition_tool_cache_key defaults to this format's key, which is what a label that only renames the same fixture wants. A suffix that asks the transition tool for something different must pass its own key, or an empty string to opt out of caching.

variant names what the label asks its spec type to fill differently, for that spec type to query with is_variant() rather than comparing formats.

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

@classmethod
def with_label_suffix(
    cls,
    suffix: str,
    description: str | None = None,
    *,
    transition_tool_cache_key: str | None = None,
    variant: str | None = None,
) -> "LabeledFixtureFormat":
    """
    Return this format labeled `<format_id>_<suffix>`.

    Use this instead of building a `LabeledFixtureFormat` by hand when a
    spec type re-labels the formats of another one: the label is derived
    from `format_id()`, so a format that already carries a label derives a
    distinct label per label instead of collapsing them all onto its
    format name.

    `transition_tool_cache_key` defaults to this format's key, which is
    what a label that only renames the same fixture wants. A suffix that
    asks the transition tool for something different must pass its own
    key, or an empty string to opt out of caching.

    `variant` names what the label asks its spec type to fill differently,
    for that spec type to query with `is_variant()` rather than comparing
    formats.
    """
    return LabeledFixtureFormat(
        cls,
        f"{cls.format_id()}_{suffix}",
        description
        if description is not None
        else f"A {cls.format_id()} {suffix.replace('_', ' ')}",
        transition_tool_cache_key=transition_tool_cache_key,
        variant=variant,
    )

marks(*, transition_tool_cache_key=None) classmethod

Get list of pytest marks that need to be added to a test produced with this fixture format.

transition_tool_cache_key overrides the format's own key.

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

@classmethod
def marks(
    cls, *, transition_tool_cache_key: str | None = None
) -> List[pytest.MarkDecorator | pytest.Mark]:
    """
    Get list of pytest marks that need to be added to a test produced
    with this fixture format.

    `transition_tool_cache_key` overrides the format's own key.
    """
    cache_key = (
        cls.transition_tool_cache_key
        if transition_tool_cache_key is None
        else transition_tool_cache_key
    )
    marks: List[pytest.MarkDecorator | pytest.Mark] = [
        getattr(
            pytest.mark,
            cls.format_name.lower(),
        ),
    ]
    if cache_key:
        marks.append(pytest.mark.transition_tool_cache_key(cache_key))
    return marks

supports_fork(fork) classmethod

Return whether the fixture can be generated for the given fork.

By default, all fixtures support all forks.

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

@classmethod
def supports_fork(cls, fork: Fork | TransitionFork) -> bool:
    """
    Return whether the fixture can be generated for the given fork.

    By default, all fixtures support all forks.
    """
    del fork
    return True

discard_fixture_format_by_marks(fork, markers) classmethod

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

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

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fork: Fork | TransitionFork,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard a fixture format from filling if the appropriate marker is
    used.
    """
    del fork, markers
    return False

FixtureFillingPhase

Bases: Enum

Execution phase for fixture generation.

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

class FixtureFillingPhase(Enum):
    """Execution phase for fixture generation."""

    FILL = auto()
    PRE_ALLOC_GENERATION = auto()
    FILL_AFTER_PRE_ALLOC_GENERATION = auto()
    FILL_STATEFUL = auto()

LabeledFixtureFormat

Represents a fixture format with a custom label.

This label will be used in the test id and also will be added as a marker to the generated test case when filling the test.

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

class LabeledFixtureFormat:
    """
    Represents a fixture format with a custom label.

    This label will be used in the test id and also will be added as a marker
    to the generated test case when filling the test.
    """

    format: Type[BaseFixture]
    label: str
    description: str
    base: "LabeledFixtureFormat | None"
    _transition_tool_cache_key: str | None
    _variant: str | None

    registered_labels: ClassVar[Dict[str, "LabeledFixtureFormat"]] = {}

    def __init__(
        self,
        fixture_format: "Type[BaseFixture] | LabeledFixtureFormat",
        label: str,
        description: str,
        *,
        transition_tool_cache_key: str | None = None,
        variant: str | None = None,
    ):
        """
        Initialize the fixture format with a custom label.

        `transition_tool_cache_key` defaults to the wrapped format's key. A
        label that asks the transition tool for something different must set
        its own key, or an empty string to opt out of caching, since labels
        sharing a key also share cached output.

        `variant` names what this label asks its spec type to fill
        differently, and defaults to the wrapped label's variant.

        Wrapping a label rather than a plain format keeps what that inner
        label decided: its variant, its cache key and its fork/marker vetoes
        still apply, so re-labeling never silently reverts them to the plain
        format's.
        """
        self.format = fixture_format.format_class()
        self.base = (
            fixture_format
            if isinstance(fixture_format, LabeledFixtureFormat)
            else None
        )
        self.label = label
        self.description = description
        self._transition_tool_cache_key = transition_tool_cache_key
        self._variant = variant
        if label not in LabeledFixtureFormat.registered_labels:
            LabeledFixtureFormat.registered_labels[label] = self

    @property
    def format_name(self) -> str:
        """Get the filling format name."""
        return self.format.format_name

    @property
    def format_phases(self) -> Set[FixtureFillingPhase]:
        """Get the filling format phases where it should be included."""
        return self.format.format_phases

    def format_class(self) -> Type[BaseFixture]:
        """Get the format without label."""
        return self.format

    def supports_fork(self, fork: Fork | TransitionFork) -> bool:
        """
        Return whether this label can be filled for the given fork.

        Defers to the label this one was derived from, or to the wrapped
        format. A label whose fixture only makes sense for some forks
        overrides this.
        """
        if self.base is not None:
            return self.base.supports_fork(fork)
        return self.format.supports_fork(fork)

    def discard_fixture_format_by_marks(
        self,
        fork: Fork | TransitionFork,
        markers: List[pytest.Mark],
    ) -> bool:
        """
        Discard this label from filling if the appropriate marker is used.

        Defers to the label this one was derived from, or to the wrapped
        format, so a label can veto itself without affecting the other labels
        of the same format.
        """
        if self.base is not None:
            return self.base.discard_fixture_format_by_marks(fork, markers)
        return self.format.discard_fixture_format_by_marks(fork, markers)

    def format_id(self) -> str:
        """Get string used as identifier for this format."""
        return self.label

    @property
    def variant(self) -> str | None:
        """
        Get what this label asks its spec type to fill differently.

        Falls back to the variant of the label this one was derived from, so a
        variant survives being re-labeled.
        """
        if self._variant is not None:
            return self._variant
        if self.base is not None:
            return self.base.variant
        return None

    def is_variant(self, variant: str) -> bool:
        """
        Return whether this label is the named variant.

        A spec type asks this instead of comparing the format it was handed
        against the label it declared: comparing cannot tell a variant from
        the plain format it wraps, and it stops matching as soon as another
        spec type re-labels the variant.
        """
        return self.variant == variant

    def labels(self) -> List[str]:
        """
        Get this label and every label it was derived from, outermost last.
        """
        labels = self.base.labels() if self.base is not None else []
        labels.append(self.label)
        return labels

    def with_label_suffix(
        self,
        suffix: str,
        description: str | None = None,
        *,
        transition_tool_cache_key: str | None = None,
        variant: str | None = None,
    ) -> "LabeledFixtureFormat":
        """
        Return this label re-labeled as `<label>_<suffix>`.

        The derived label keeps this label's fork/marker vetoes, so every label
        of one format derives its own distinct label rather than all of them
        collapsing onto the format name.

        `transition_tool_cache_key` defaults to this label's key, so a label
        that opted out of caching stays opted out. A suffix that asks the
        transition tool for something different must pass its own key.

        `variant` defaults to this label's variant, so re-labeling a variant
        keeps filling that variant.
        """
        return LabeledFixtureFormat(
            self,
            f"{self.format_id()}_{suffix}",
            description
            if description is not None
            else f"A {self.format_id()} {suffix.replace('_', ' ')}",
            transition_tool_cache_key=transition_tool_cache_key,
            variant=variant,
        )

    def marks(self) -> List[pytest.MarkDecorator | pytest.Mark]:
        """
        Get list of pytest marks that need to be added to a test produced
        with this fixture format.

        Every label this one was derived from is marked too, so selecting a
        label also selects the labels another spec type derived from it.
        """
        marks: List[pytest.MarkDecorator | pytest.Mark] = self.format.marks(
            transition_tool_cache_key=self.transition_tool_cache_key
        )
        for label in self.labels():
            if label.lower() != self.format.format_name.lower():
                marks.append(
                    getattr(
                        pytest.mark,
                        label.lower(),
                    ),
                )
        return marks

    @property
    def transition_tool_cache_key(self) -> str:
        """
        Get the transition tool cache key.

        Falls back to the key of the label this one was derived from, and then
        to the wrapped format's, so a label that opted out of caching does not
        opt back in when it is re-labeled.
        """
        if self._transition_tool_cache_key is not None:
            return self._transition_tool_cache_key
        if self.base is not None:
            return self.base.transition_tool_cache_key
        return self.format.transition_tool_cache_key

    def __eq__(self, other: Any) -> bool:
        """
        Check if two labeled fixture formats are equal.

        Two labeled formats are equal only when they share both format and
        label, so one format can carry more than one label.

        If the other object is a FixtureFormat type, the format of the labeled
        fixture format will be compared with the format of the other object.
        """
        if isinstance(other, LabeledFixtureFormat):
            return self.format == other.format and self.label == other.label
        if isinstance(other, type) and issubclass(other, BaseFixture):
            return self.format == other
        return False

    def __hash__(self) -> int:
        """
        Return the hash of the wrapped format.

        A labeled format compares equal to the plain format it wraps, so both
        must hash alike. Two labels of one format collide, which is allowed
        since they no longer compare equal.
        """
        return hash(self.format)

__init__(fixture_format, label, description, *, transition_tool_cache_key=None, variant=None)

Initialize the fixture format with a custom label.

transition_tool_cache_key defaults to the wrapped format's key. A label that asks the transition tool for something different must set its own key, or an empty string to opt out of caching, since labels sharing a key also share cached output.

variant names what this label asks its spec type to fill differently, and defaults to the wrapped label's variant.

Wrapping a label rather than a plain format keeps what that inner label decided: its variant, its cache key and its fork/marker vetoes still apply, so re-labeling never silently reverts them to the plain format's.

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

def __init__(
    self,
    fixture_format: "Type[BaseFixture] | LabeledFixtureFormat",
    label: str,
    description: str,
    *,
    transition_tool_cache_key: str | None = None,
    variant: str | None = None,
):
    """
    Initialize the fixture format with a custom label.

    `transition_tool_cache_key` defaults to the wrapped format's key. A
    label that asks the transition tool for something different must set
    its own key, or an empty string to opt out of caching, since labels
    sharing a key also share cached output.

    `variant` names what this label asks its spec type to fill
    differently, and defaults to the wrapped label's variant.

    Wrapping a label rather than a plain format keeps what that inner
    label decided: its variant, its cache key and its fork/marker vetoes
    still apply, so re-labeling never silently reverts them to the plain
    format's.
    """
    self.format = fixture_format.format_class()
    self.base = (
        fixture_format
        if isinstance(fixture_format, LabeledFixtureFormat)
        else None
    )
    self.label = label
    self.description = description
    self._transition_tool_cache_key = transition_tool_cache_key
    self._variant = variant
    if label not in LabeledFixtureFormat.registered_labels:
        LabeledFixtureFormat.registered_labels[label] = self

format_name property

Get the filling format name.

format_phases property

Get the filling format phases where it should be included.

format_class()

Get the format without label.

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

def format_class(self) -> Type[BaseFixture]:
    """Get the format without label."""
    return self.format

supports_fork(fork)

Return whether this label can be filled for the given fork.

Defers to the label this one was derived from, or to the wrapped format. A label whose fixture only makes sense for some forks overrides this.

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

def supports_fork(self, fork: Fork | TransitionFork) -> bool:
    """
    Return whether this label can be filled for the given fork.

    Defers to the label this one was derived from, or to the wrapped
    format. A label whose fixture only makes sense for some forks
    overrides this.
    """
    if self.base is not None:
        return self.base.supports_fork(fork)
    return self.format.supports_fork(fork)

discard_fixture_format_by_marks(fork, markers)

Discard this label from filling if the appropriate marker is used.

Defers to the label this one was derived from, or to the wrapped format, so a label can veto itself without affecting the other labels of the same format.

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

def discard_fixture_format_by_marks(
    self,
    fork: Fork | TransitionFork,
    markers: List[pytest.Mark],
) -> bool:
    """
    Discard this label from filling if the appropriate marker is used.

    Defers to the label this one was derived from, or to the wrapped
    format, so a label can veto itself without affecting the other labels
    of the same format.
    """
    if self.base is not None:
        return self.base.discard_fixture_format_by_marks(fork, markers)
    return self.format.discard_fixture_format_by_marks(fork, markers)

format_id()

Get string used as identifier for this format.

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

def format_id(self) -> str:
    """Get string used as identifier for this format."""
    return self.label

variant property

Get what this label asks its spec type to fill differently.

Falls back to the variant of the label this one was derived from, so a variant survives being re-labeled.

is_variant(variant)

Return whether this label is the named variant.

A spec type asks this instead of comparing the format it was handed against the label it declared: comparing cannot tell a variant from the plain format it wraps, and it stops matching as soon as another spec type re-labels the variant.

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

def is_variant(self, variant: str) -> bool:
    """
    Return whether this label is the named variant.

    A spec type asks this instead of comparing the format it was handed
    against the label it declared: comparing cannot tell a variant from
    the plain format it wraps, and it stops matching as soon as another
    spec type re-labels the variant.
    """
    return self.variant == variant

labels()

Get this label and every label it was derived from, outermost last.

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

def labels(self) -> List[str]:
    """
    Get this label and every label it was derived from, outermost last.
    """
    labels = self.base.labels() if self.base is not None else []
    labels.append(self.label)
    return labels

with_label_suffix(suffix, description=None, *, transition_tool_cache_key=None, variant=None)

Return this label re-labeled as <label>_<suffix>.

The derived label keeps this label's fork/marker vetoes, so every label of one format derives its own distinct label rather than all of them collapsing onto the format name.

transition_tool_cache_key defaults to this label's key, so a label that opted out of caching stays opted out. A suffix that asks the transition tool for something different must pass its own key.

variant defaults to this label's variant, so re-labeling a variant keeps filling that variant.

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

def with_label_suffix(
    self,
    suffix: str,
    description: str | None = None,
    *,
    transition_tool_cache_key: str | None = None,
    variant: str | None = None,
) -> "LabeledFixtureFormat":
    """
    Return this label re-labeled as `<label>_<suffix>`.

    The derived label keeps this label's fork/marker vetoes, so every label
    of one format derives its own distinct label rather than all of them
    collapsing onto the format name.

    `transition_tool_cache_key` defaults to this label's key, so a label
    that opted out of caching stays opted out. A suffix that asks the
    transition tool for something different must pass its own key.

    `variant` defaults to this label's variant, so re-labeling a variant
    keeps filling that variant.
    """
    return LabeledFixtureFormat(
        self,
        f"{self.format_id()}_{suffix}",
        description
        if description is not None
        else f"A {self.format_id()} {suffix.replace('_', ' ')}",
        transition_tool_cache_key=transition_tool_cache_key,
        variant=variant,
    )

marks()

Get list of pytest marks that need to be added to a test produced with this fixture format.

Every label this one was derived from is marked too, so selecting a label also selects the labels another spec type derived from it.

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

def marks(self) -> List[pytest.MarkDecorator | pytest.Mark]:
    """
    Get list of pytest marks that need to be added to a test produced
    with this fixture format.

    Every label this one was derived from is marked too, so selecting a
    label also selects the labels another spec type derived from it.
    """
    marks: List[pytest.MarkDecorator | pytest.Mark] = self.format.marks(
        transition_tool_cache_key=self.transition_tool_cache_key
    )
    for label in self.labels():
        if label.lower() != self.format.format_name.lower():
            marks.append(
                getattr(
                    pytest.mark,
                    label.lower(),
                ),
            )
    return marks

transition_tool_cache_key property

Get the transition tool cache key.

Falls back to the key of the label this one was derived from, and then to the wrapped format's, so a label that opted out of caching does not opt back in when it is re-labeled.

__eq__(other)

Check if two labeled fixture formats are equal.

Two labeled formats are equal only when they share both format and label, so one format can carry more than one label.

If the other object is a FixtureFormat type, the format of the labeled fixture format will be compared with the format of the other object.

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

def __eq__(self, other: Any) -> bool:
    """
    Check if two labeled fixture formats are equal.

    Two labeled formats are equal only when they share both format and
    label, so one format can carry more than one label.

    If the other object is a FixtureFormat type, the format of the labeled
    fixture format will be compared with the format of the other object.
    """
    if isinstance(other, LabeledFixtureFormat):
        return self.format == other.format and self.label == other.label
    if isinstance(other, type) and issubclass(other, BaseFixture):
        return self.format == other
    return False

__hash__()

Return the hash of the wrapped format.

A labeled format compares equal to the plain format it wraps, so both must hash alike. Two labels of one format collide, which is allowed since they no longer compare equal.

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

def __hash__(self) -> int:
    """
    Return the hash of the wrapped format.

    A labeled format compares equal to the plain format it wraps, so both
    must hash alike. Two labels of one format collide, which is allowed
    since they no longer compare equal.
    """
    return hash(self.format)

strip_fixture_format_from_node(item)

Remove fixture format suffix from a test nodeid.

Used for cache keys and xdist grouping to ensure related fixture formats (e.g., blockchain_test and blockchain_test_engine) share the same key.

Example

'test.py::test[fork_Osaka-state_test]' -> 'test.py::test[fork_Osaka]'

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

def strip_fixture_format_from_node(
    item: PytestItemProtocol,
) -> str:
    """
    Remove fixture format suffix from a test nodeid.

    Used for cache keys and xdist grouping to ensure related fixture formats
    (e.g., blockchain_test and blockchain_test_engine) share the same key.

    Example:
        'test.py::test[fork_Osaka-state_test]' -> 'test.py::test[fork_Osaka]'

    """
    fixture_format_id_marker = item.get_closest_marker("fixture_format_id")
    nodeid = item.nodeid
    if fixture_format_id_marker is None:
        return nodeid
    assert len(fixture_format_id_marker.args) == 1
    fixture_id = fixture_format_id_marker.args[0]
    if fixture_id not in nodeid:
        return nodeid
    return nodeid.replace(fixture_id, "")

BlockchainEngineFixture

Bases: BlockchainEngineFixtureCommon

Engine specific test fixture information.

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

class BlockchainEngineFixture(BlockchainEngineFixtureCommon):
    """Engine specific test fixture information."""

    format_name: ClassVar[str] = "blockchain_test_engine"
    description: ClassVar[str] = (
        "Tests that generate a blockchain test fixture in Engine API format."
    )
    pre: Alloc
    genesis: FixtureHeader = Field(..., alias="genesisBlockHeader")
    post_state: Alloc | None = Field(None)
    payloads: List[FixtureEngineNewPayload] = Field(
        ..., alias="engineNewPayloads"
    )
    transition_tool_cache_key: ClassVar[str] = "blockchain_test"

BlockchainEngineFixtureCommon

Bases: BaseFixture

Base blockchain test fixture model for Engine API based execution.

Similar to BlockchainFixtureCommon but excludes the 'pre' field to avoid duplicating large pre-allocations.

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

@post_state_validator()
class BlockchainEngineFixtureCommon(BaseFixture):
    """
    Base blockchain test fixture model for Engine API based execution.

    Similar to BlockchainFixtureCommon but excludes the 'pre' field to avoid
    duplicating large pre-allocations.
    """

    fork: Fork | TransitionFork = Field(..., alias="network")
    post_state_hash: Hash | None = Field(None)
    # FIXME: lastBlockHash
    last_block_hash: Hash = Field(..., alias="lastblockhash")
    config: FixtureConfig

    def get_fork(self) -> Fork | TransitionFork | None:
        """Return fixture's `Fork`."""
        return self.fork

    @classmethod
    def supports_fork(cls, fork: Fork | TransitionFork) -> bool:
        """
        Return whether the fixture can be generated for the given fork.

        The Engine API is available only on Paris and afterwards.
        """
        return fork.fork_at(block_number=0, timestamp=0) >= Paris

get_fork()

Return fixture's Fork.

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

def get_fork(self) -> Fork | TransitionFork | None:
    """Return fixture's `Fork`."""
    return self.fork

supports_fork(fork) classmethod

Return whether the fixture can be generated for the given fork.

The Engine API is available only on Paris and afterwards.

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

@classmethod
def supports_fork(cls, fork: Fork | TransitionFork) -> bool:
    """
    Return whether the fixture can be generated for the given fork.

    The Engine API is available only on Paris and afterwards.
    """
    return fork.fork_at(block_number=0, timestamp=0) >= Paris

BlockchainEngineStatefulFixture

Bases: BlockchainEngineFixtureCommon

Engine fixture for snapshot-based stateful testing.

Instead of embedding pre-allocation or referencing a computed group, this fixture references an external snapshot that the consumer must pre-load. Setup payloads deploy contracts and seed accounts on top of the snapshot before the actual test payloads execute.

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

class BlockchainEngineStatefulFixture(BlockchainEngineFixtureCommon):
    """
    Engine fixture for snapshot-based stateful testing.

    Instead of embedding pre-allocation or referencing a computed group,
    this fixture references an external snapshot that the consumer must
    pre-load. Setup payloads deploy contracts and seed accounts on top
    of the snapshot before the actual test payloads execute.
    """

    model_config = CamelModel.model_config | {"extra": "ignore"}

    format_name: ClassVar[str] = "blockchain_test_stateful_engine"
    description: ClassVar[str] = (
        "Tests that generate a Blockchain Test fixture for "
        "snapshot-based stateful Engine API testing."
    )
    format_phases: ClassVar[Set[FixtureFillingPhase]] = {
        FixtureFillingPhase.FILL_STATEFUL,
    }

    snapshot_block_number: HexNumber
    snapshot_block_hash: Hash
    start_block_number: HexNumber
    start_block_hash: Hash

    setup_payloads: List[FixtureEngineNewPayload] = Field(
        default_factory=list,
        alias="setupEngineNewPayloads",
        description=(
            "Per-test setup-phase payloads applied on top of "
            "``start_block_hash`` before the test's execution payloads."
        ),
    )
    payloads: List[FixtureEngineNewPayload] = Field(
        ..., alias="engineNewPayloads"
    )
    benchmark_gas_used: HexNumber | None = Field(
        None,
        alias="benchmarkGasUsed",
        description=(
            "Total gas consumed by the execution payloads when this fixture "
            "was filled, as reported by the live client. Populated for "
            "``BENCHMARKING`` operation mode; ``None`` otherwise. Lets "
            "consumers compare expected vs. observed gas without replay."
        ),
    )

BlockchainEngineSyncFixture

Bases: BlockchainEngineFixture

Engine Sync specific test fixture information.

This fixture format is specifically designed for sync testing where: - The client under test receives all payloads - A sync client attempts to sync from the client under test - Both client types are parametrized from hive client config

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

class BlockchainEngineSyncFixture(BlockchainEngineFixture):
    """
    Engine Sync specific test fixture information.

    This fixture format is specifically designed for sync testing where:
    - The client under test receives all payloads
    - A sync client attempts to sync from the client under test
    - Both client types are parametrized from hive client config
    """

    format_name: ClassVar[str] = "blockchain_test_sync"
    description: ClassVar[str] = (
        "Tests that generate a blockchain test fixture for Engine API "
        "testing with client sync."
    )
    transition_tool_cache_key: ClassVar[str] = ""
    sync_payload: FixtureEngineNewPayload | None = None

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fork: Fork | TransitionFork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard the fixture format based on the provided markers."""
        del fork
        marker_names = [m.name for m in markers]
        return "verify_sync" not in marker_names

discard_fixture_format_by_marks(fork, markers) classmethod

Discard the fixture format based on the provided markers.

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

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fork: Fork | TransitionFork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard the fixture format based on the provided markers."""
    del fork
    marker_names = [m.name for m in markers]
    return "verify_sync" not in marker_names

BlockchainEngineXFixture

Bases: BlockchainEngineFixtureCommon

Engine X specific test fixture information.

Uses pre-allocation groups (and a single client instance) for efficient test execution without client restarts.

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

@post_state_validator(alternate_field="post_state_diff")
class BlockchainEngineXFixture(BlockchainEngineFixtureCommon):
    """
    Engine X specific test fixture information.

    Uses pre-allocation groups (and a single client instance) for efficient
    test execution without client restarts.
    """

    # Allow extra fields: BlockchainEngineXFixture is constructed from shared
    # fixture_data that has fields for other fixture formats (e.g. genesis).
    model_config = CamelModel.model_config | {"extra": "ignore"}

    format_name: ClassVar[str] = "blockchain_test_engine_x"
    description: ClassVar[str] = (
        "Tests that generate a Blockchain Test Engine X fixture."
    )
    format_phases: ClassVar[Set[FixtureFillingPhase]] = {
        FixtureFillingPhase.FILL_AFTER_PRE_ALLOC_GENERATION,
        FixtureFillingPhase.PRE_ALLOC_GENERATION,
    }
    transition_tool_cache_key: ClassVar[str] = ""

    pre_hash: AllocGroupHash
    """Hash of the pre-allocation group this test belongs to."""

    post_state_diff: Alloc
    """
    State difference from genesis after test execution (efficiency
    optimization).
    """

    payloads: List[FixtureEngineNewPayload] = Field(
        ..., alias="engineNewPayloads"
    )
    """Engine API payloads for blockchain execution."""

pre_hash instance-attribute

Hash of the pre-allocation group this test belongs to.

post_state_diff instance-attribute

State difference from genesis after test execution (efficiency optimization).

payloads = Field(..., alias='engineNewPayloads') class-attribute instance-attribute

Engine API payloads for blockchain execution.

BlockchainFixture

Bases: BlockchainFixtureCommon

Cross-client specific blockchain test model use in JSON fixtures.

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

class BlockchainFixture(BlockchainFixtureCommon):
    """Cross-client specific blockchain test model use in JSON fixtures."""

    format_name: ClassVar[str] = "blockchain_test"
    description: ClassVar[str] = (
        "Tests that generate a blockchain test fixture."
    )

    genesis_rlp: Bytes = Field(..., alias="genesisRLP")
    blocks: List[FixtureBlock | InvalidFixtureBlock]
    seal_engine: Literal["NoProof"] = Field("NoProof")
    transition_tool_cache_key: ClassVar[str] = "blockchain_test"

BlockchainFixtureCommon

Bases: BaseFixture

Base blockchain test fixture model.

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

@post_state_validator()
class BlockchainFixtureCommon(BaseFixture):
    """Base blockchain test fixture model."""

    fork: Fork | TransitionFork = Field(..., alias="network")
    genesis: FixtureHeader = Field(..., alias="genesisBlockHeader")
    pre: Alloc
    post_state: Alloc | None = Field(None)
    post_state_hash: Hash | None = Field(None)
    # FIXME: lastBlockHash
    last_block_hash: Hash = Field(..., alias="lastblockhash")
    config: FixtureConfig

    @model_validator(mode="before")
    @classmethod
    def config_defaults_for_backwards_compatibility(cls, data: Any) -> Any:
        """
        Check if the config field is populated, otherwise use the root-level
        field values for backwards compatibility.
        """
        if isinstance(data, dict):
            if "config" not in data:
                data["config"] = {}
            if isinstance(data["config"], dict):
                if "network" not in data["config"]:
                    data["config"]["network"] = data["network"]
                if "chainid" not in data["config"]:
                    data["config"]["chainid"] = "0x01"
        return data

    def get_fork(self) -> Fork | TransitionFork | None:
        """Return fork of the fixture as a string."""
        return self.fork

config_defaults_for_backwards_compatibility(data) classmethod

Check if the config field is populated, otherwise use the root-level field values for backwards compatibility.

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

@model_validator(mode="before")
@classmethod
def config_defaults_for_backwards_compatibility(cls, data: Any) -> Any:
    """
    Check if the config field is populated, otherwise use the root-level
    field values for backwards compatibility.
    """
    if isinstance(data, dict):
        if "config" not in data:
            data["config"] = {}
        if isinstance(data["config"], dict):
            if "network" not in data["config"]:
                data["config"]["network"] = data["network"]
            if "chainid" not in data["config"]:
                data["config"]["chainid"] = "0x01"
    return data

get_fork()

Return fork of the fixture as a string.

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

def get_fork(self) -> Fork | TransitionFork | None:
    """Return fork of the fixture as a string."""
    return self.fork

FixtureCollector dataclass

Collects all fixtures generated by the test cases.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

@dataclass(kw_only=True)
class FixtureCollector:
    """Collects all fixtures generated by the test cases."""

    output_dir: Path
    single_fixture_per_file: bool
    filler_path: Path
    base_dump_dir: Optional[Path] = None
    generate_index: bool = True
    # Worker ID for partial files. None = read from env var.
    worker_id: Optional[str] = None

    # Internal state (only used for stdout mode)
    all_fixtures: Dict[Path, Fixtures] = field(default_factory=dict)

    # Streaming file handles - kept open for module duration
    _partial_fixture_files: Dict[Path, IO[str]] = field(default_factory=dict)
    _partial_index_file: Optional[IO[str]] = field(default=None)
    _worker_id_cached: bool = field(default=False, init=False)

    # Lightweight tracking for verification (path, format class, debug_path)
    # Only stores metadata, not fixture data - memory efficient
    _fixtures_to_verify: List[Tuple[Path, type, Optional[Path]]] = field(
        default_factory=list
    )

    def get_fixture_basename(self, info: TestInfo) -> Path:
        """Return basename of the fixture file for a given test case."""
        module_relative_output_dir = info.get_module_relative_output_dir(
            self.filler_path
        )

        if self.single_fixture_per_file:
            return module_relative_output_dir / info.get_single_test_name(
                mode="test"
            )
        return module_relative_output_dir / info.get_single_test_name(
            mode="module"
        )

    def _get_worker_id(self) -> str | None:
        """Get the worker ID (from constructor or environment)."""
        if self.worker_id is not None:
            return self.worker_id
        if not self._worker_id_cached:
            # Cache the env var lookup
            env_worker_id = os.environ.get("PYTEST_XDIST_WORKER")
            if env_worker_id:
                self.worker_id = env_worker_id
            self._worker_id_cached = True
        return self.worker_id

    def add_fixture(
        self,
        info: TestInfo,
        fixture: BaseFixture,
        output_subdir: Path | None = None,
    ) -> Path:
        """Add fixture and immediately stream to partial JSONL file."""
        fixture_basename = self.get_fixture_basename(info)
        if (
            output_subdir is not None
            and SUBFOLDER_LEVEL_SEPARATOR in output_subdir.name
        ):
            parts = fixture_basename.parts
            if parts and parts[0] == "benchmark":
                # Strip the "benchmark/" prefix from the fixture path so
                # files land directly under the gas-limit subdirectory.
                fixture_basename = Path(*parts[1:])

        format_output_dir = self.output_dir / fixture.output_base_dir_name()
        if output_subdir is not None and self.output_dir.name != "stdout":
            format_output_dir = format_output_dir / output_subdir

        fixture_path = format_output_dir / fixture_basename.with_suffix(
            fixture.output_file_extension
        )

        # Stream fixture directly to partial JSONL (no memory accumulation)
        if self.output_dir.name != "stdout":
            self._stream_fixture_to_partial(
                fixture_path, info.get_id(), fixture
            )
            # Track for verification (lightweight - only path and format class)
            debug_path = self._get_consume_direct_dump_dir(info)
            self._fixtures_to_verify.append(
                (fixture_path, fixture.__class__, debug_path)
            )
        else:
            # stdout mode: accumulate for final JSON dump
            if fixture_path not in self.all_fixtures:
                self.all_fixtures[fixture_path] = Fixtures(root={})
            self.all_fixtures[fixture_path][info.get_id()] = fixture

        # Stream index entry directly to partial JSONL
        if self.generate_index and self.output_dir.name != "stdout":
            relative_path = fixture_path.relative_to(self.output_dir)
            fixture_fork = fixture.get_fork()
            index_entry = TestCaseIndexFile(
                id=info.get_id(),
                json_path=relative_path,
                fixture_hash=fixture.hash,
                fork=fixture_fork,
                format=fixture.format_class(),
                pre_hash=getattr(fixture, "pre_hash", None),
            )
            self._stream_index_entry_to_partial(index_entry)

        return fixture_path

    def _get_partial_fixture_file(self, fixture_path: Path) -> "IO[str]":
        """Get or create a file handle for streaming fixtures."""
        worker_id = self._get_worker_id()
        suffix = f".{worker_id}" if worker_id else ".main"
        partial_path = fixture_path.with_suffix(f".partial{suffix}.jsonl")

        if partial_path not in self._partial_fixture_files:
            partial_path.parent.mkdir(parents=True, exist_ok=True)
            self._partial_fixture_files[partial_path] = open(partial_path, "a")

        return self._partial_fixture_files[partial_path]

    def _stream_fixture_to_partial(
        self,
        fixture_path: Path,
        fixture_id: str,
        fixture: BaseFixture,
    ) -> None:
        """Stream a single fixture to its partial JSONL file."""
        value = json.dumps(fixture.json_dict_with_info(), indent=4)
        line = json.dumps({"k": fixture_id, "v": value}) + "\n"

        f = self._get_partial_fixture_file(fixture_path)
        f.write(line)
        f.flush()  # Ensure data is written immediately

    def _get_partial_index_file(self) -> "IO[str]":
        """Get or create the file handle for streaming index entries."""
        if self._partial_index_file is None:
            worker_id = self._get_worker_id()
            suffix = f".{worker_id}" if worker_id else ".main"
            partial_index_path = (
                self.output_dir / ".meta" / f"partial_index{suffix}.jsonl"
            )
            partial_index_path.parent.mkdir(parents=True, exist_ok=True)
            self._partial_index_file = open(partial_index_path, "a")

        return self._partial_index_file

    def _stream_index_entry_to_partial(self, entry: TestCaseIndexFile) -> None:
        """Stream a single index entry to partial JSONL file."""
        f = self._get_partial_index_file()
        f.write(entry.model_dump_json(exclude_none=True) + "\n")
        f.flush()  # Ensure data is written immediately

    def close_streaming_files(self) -> None:
        """Close all open streaming file handles."""
        for f in self._partial_fixture_files.values():
            f.close()
        self._partial_fixture_files.clear()

        if self._partial_index_file is not None:
            self._partial_index_file.close()
            self._partial_index_file = None

    def dump_fixtures(self) -> None:
        """Dump collected fixtures (only used for stdout mode)."""
        if self.output_dir.name == "stdout":
            combined_fixtures = {
                k: to_json(v)
                for fixture in self.all_fixtures.values()
                for k, v in fixture.items()
            }
            json.dump(combined_fixtures, sys.stdout, indent=4)
            self.all_fixtures.clear()
        # For file output, fixtures are already streamed in add_fixture()

    def _get_consume_direct_dump_dir(self, info: TestInfo) -> Path | None:
        """
        Directory to dump the current test function's fixture.json and fixture
        verification debug output.
        """
        if not self.base_dump_dir:
            return None
        if self.single_fixture_per_file:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_parameter"
            )
        else:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_function"
            )

    def verify_fixture_files(
        self, evm_fixture_verification: FixtureConsumer
    ) -> None:
        """
        Run `evm [state|block]test` on each fixture.

        For streaming mode, uses lightweight tracking of fixture paths/formats
        rather than keeping full fixtures in memory.
        """
        if self.output_dir.name == "stdout":
            # stdout mode: fixtures are in memory
            for fixture_path, name_fixture_dict in self.all_fixtures.items():
                for _fixture_name, fixture in name_fixture_dict.items():
                    if evm_fixture_verification.can_consume(fixture.__class__):
                        evm_fixture_verification.consume_fixture(
                            fixture.__class__,
                            fixture_path,
                            fixture_name=None,
                            debug_output_path=None,
                        )
        else:
            # Streaming mode: use tracked fixture metadata
            for entry in self._fixtures_to_verify:
                fixture_path, fixture_format, debug_path = entry
                if evm_fixture_verification.can_consume(fixture_format):
                    evm_fixture_verification.consume_fixture(
                        fixture_format,
                        fixture_path,
                        fixture_name=None,
                        debug_output_path=debug_path,
                    )
            # Clear tracking after verification
            self._fixtures_to_verify.clear()

get_fixture_basename(info)

Return basename of the fixture file for a given test case.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_fixture_basename(self, info: TestInfo) -> Path:
    """Return basename of the fixture file for a given test case."""
    module_relative_output_dir = info.get_module_relative_output_dir(
        self.filler_path
    )

    if self.single_fixture_per_file:
        return module_relative_output_dir / info.get_single_test_name(
            mode="test"
        )
    return module_relative_output_dir / info.get_single_test_name(
        mode="module"
    )

add_fixture(info, fixture, output_subdir=None)

Add fixture and immediately stream to partial JSONL file.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def add_fixture(
    self,
    info: TestInfo,
    fixture: BaseFixture,
    output_subdir: Path | None = None,
) -> Path:
    """Add fixture and immediately stream to partial JSONL file."""
    fixture_basename = self.get_fixture_basename(info)
    if (
        output_subdir is not None
        and SUBFOLDER_LEVEL_SEPARATOR in output_subdir.name
    ):
        parts = fixture_basename.parts
        if parts and parts[0] == "benchmark":
            # Strip the "benchmark/" prefix from the fixture path so
            # files land directly under the gas-limit subdirectory.
            fixture_basename = Path(*parts[1:])

    format_output_dir = self.output_dir / fixture.output_base_dir_name()
    if output_subdir is not None and self.output_dir.name != "stdout":
        format_output_dir = format_output_dir / output_subdir

    fixture_path = format_output_dir / fixture_basename.with_suffix(
        fixture.output_file_extension
    )

    # Stream fixture directly to partial JSONL (no memory accumulation)
    if self.output_dir.name != "stdout":
        self._stream_fixture_to_partial(
            fixture_path, info.get_id(), fixture
        )
        # Track for verification (lightweight - only path and format class)
        debug_path = self._get_consume_direct_dump_dir(info)
        self._fixtures_to_verify.append(
            (fixture_path, fixture.__class__, debug_path)
        )
    else:
        # stdout mode: accumulate for final JSON dump
        if fixture_path not in self.all_fixtures:
            self.all_fixtures[fixture_path] = Fixtures(root={})
        self.all_fixtures[fixture_path][info.get_id()] = fixture

    # Stream index entry directly to partial JSONL
    if self.generate_index and self.output_dir.name != "stdout":
        relative_path = fixture_path.relative_to(self.output_dir)
        fixture_fork = fixture.get_fork()
        index_entry = TestCaseIndexFile(
            id=info.get_id(),
            json_path=relative_path,
            fixture_hash=fixture.hash,
            fork=fixture_fork,
            format=fixture.format_class(),
            pre_hash=getattr(fixture, "pre_hash", None),
        )
        self._stream_index_entry_to_partial(index_entry)

    return fixture_path

close_streaming_files()

Close all open streaming file handles.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def close_streaming_files(self) -> None:
    """Close all open streaming file handles."""
    for f in self._partial_fixture_files.values():
        f.close()
    self._partial_fixture_files.clear()

    if self._partial_index_file is not None:
        self._partial_index_file.close()
        self._partial_index_file = None

dump_fixtures()

Dump collected fixtures (only used for stdout mode).

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def dump_fixtures(self) -> None:
    """Dump collected fixtures (only used for stdout mode)."""
    if self.output_dir.name == "stdout":
        combined_fixtures = {
            k: to_json(v)
            for fixture in self.all_fixtures.values()
            for k, v in fixture.items()
        }
        json.dump(combined_fixtures, sys.stdout, indent=4)
        self.all_fixtures.clear()

verify_fixture_files(evm_fixture_verification)

Run evm [state|block]test on each fixture.

For streaming mode, uses lightweight tracking of fixture paths/formats rather than keeping full fixtures in memory.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def verify_fixture_files(
    self, evm_fixture_verification: FixtureConsumer
) -> None:
    """
    Run `evm [state|block]test` on each fixture.

    For streaming mode, uses lightweight tracking of fixture paths/formats
    rather than keeping full fixtures in memory.
    """
    if self.output_dir.name == "stdout":
        # stdout mode: fixtures are in memory
        for fixture_path, name_fixture_dict in self.all_fixtures.items():
            for _fixture_name, fixture in name_fixture_dict.items():
                if evm_fixture_verification.can_consume(fixture.__class__):
                    evm_fixture_verification.consume_fixture(
                        fixture.__class__,
                        fixture_path,
                        fixture_name=None,
                        debug_output_path=None,
                    )
    else:
        # Streaming mode: use tracked fixture metadata
        for entry in self._fixtures_to_verify:
            fixture_path, fixture_format, debug_path = entry
            if evm_fixture_verification.can_consume(fixture_format):
                evm_fixture_verification.consume_fixture(
                    fixture_format,
                    fixture_path,
                    fixture_name=None,
                    debug_output_path=debug_path,
                )
        # Clear tracking after verification
        self._fixtures_to_verify.clear()

TestInfo dataclass

Contains test information from the current node.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

@dataclass(kw_only=True, slots=True)
class TestInfo:
    """Contains test information from the current node."""

    __test__ = False  # stop pytest from collecting this class as a test

    name: str  # pytest: Item.name, e.g. test_paris_one[fork_Paris-state_test]
    id: str  # pytest: Item.nodeid, e.g.
    # tests/paris/test_module_paris.py::test_paris_one[...]
    original_name: str  # pytest: Item.originalname, e.g. test_paris_one
    module_path: Path  # pytest: Item.path, e.g.
    # .../tests/paris/test_module_paris.py

    test_prefix: ClassVar[str] = "test_"  # Python test prefix
    filler_suffix: ClassVar[str] = "Filler"  # Static test suffix

    @classmethod
    def strip_test_name(cls, name: str) -> str:
        """Remove test prefix from a python test case name."""
        if name.startswith(cls.test_prefix):
            return name.removeprefix(cls.test_prefix)
        if name.endswith(cls.filler_suffix):
            return name.removesuffix(cls.filler_suffix)
        return name

    def get_name_and_parameters(self) -> Tuple[str, str]:
        """
        Convert test name to a tuple containing the test name and test
        parameters.

        Example: test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore,
        fork_Shanghai
        """
        test_name, parameters = self.name.split("[")
        return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

    def get_single_test_name(
        self, mode: Literal["module", "test"] = "module"
    ) -> str:
        """Convert test name to a single test name."""
        if mode == "module":
            # Use the module name as the test name
            return self.strip_test_name(self.original_name)
        elif mode == "test":
            # Mix the module name and the test name/arguments
            test_name, test_parameters = self.get_name_and_parameters()
            test_name = self.strip_test_name(test_name)
            return f"{test_name}__{test_parameters}"

    def get_dump_dir_path(
        self,
        base_dump_dir: Optional[Path],
        filler_path: Path,
        level: Literal[
            "test_module", "test_function", "test_parameter"
        ] = "test_parameter",
    ) -> Optional[Path]:
        """Path to dump the debug output as defined by the level to dump at."""
        if not base_dump_dir:
            return None
        test_module_relative_dir = self.get_module_relative_output_dir(
            filler_path
        )
        if level == "test_module":
            return Path(base_dump_dir) / Path(
                str(test_module_relative_dir).replace(os.sep, "__")
            )
        test_name, test_parameter_string = self.get_name_and_parameters()
        dir_str = str(test_module_relative_dir).replace(os.sep, "__")
        flat_path = f"{dir_str}__{test_name}"
        if level == "test_function":
            return Path(base_dump_dir) / flat_path
        elif level == "test_parameter":
            return Path(base_dump_dir) / flat_path / test_parameter_string
        raise Exception("Unexpected level.")

    def get_id(self) -> str:
        """Return the test id."""
        return self.id

    def get_module_relative_output_dir(self, filler_path: Path) -> Path:
        """
        Return a directory name for the provided test_module (relative to the
        base ./tests directory) that can be used for output (within the
        configured fixtures output path or the base_dump_dir directory).

        Example: tests/shanghai/eip3855_push0/test_push0.py ->
        shanghai/eip3855_push0/test_push0
        """
        basename = self.module_path.with_suffix("").absolute()
        basename_relative = basename.relative_to(
            os.path.commonpath([filler_path.absolute(), basename])
        )
        module_path = basename_relative.parent / self.strip_test_name(
            basename_relative.stem
        )
        return module_path

strip_test_name(name) classmethod

Remove test prefix from a python test case name.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

@classmethod
def strip_test_name(cls, name: str) -> str:
    """Remove test prefix from a python test case name."""
    if name.startswith(cls.test_prefix):
        return name.removeprefix(cls.test_prefix)
    if name.endswith(cls.filler_suffix):
        return name.removesuffix(cls.filler_suffix)
    return name

get_name_and_parameters()

Convert test name to a tuple containing the test name and test parameters.

Example: test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_name_and_parameters(self) -> Tuple[str, str]:
    """
    Convert test name to a tuple containing the test name and test
    parameters.

    Example: test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore,
    fork_Shanghai
    """
    test_name, parameters = self.name.split("[")
    return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

get_single_test_name(mode='module')

Convert test name to a single test name.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_single_test_name(
    self, mode: Literal["module", "test"] = "module"
) -> str:
    """Convert test name to a single test name."""
    if mode == "module":
        # Use the module name as the test name
        return self.strip_test_name(self.original_name)
    elif mode == "test":
        # Mix the module name and the test name/arguments
        test_name, test_parameters = self.get_name_and_parameters()
        test_name = self.strip_test_name(test_name)
        return f"{test_name}__{test_parameters}"

get_dump_dir_path(base_dump_dir, filler_path, level='test_parameter')

Path to dump the debug output as defined by the level to dump at.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_dump_dir_path(
    self,
    base_dump_dir: Optional[Path],
    filler_path: Path,
    level: Literal[
        "test_module", "test_function", "test_parameter"
    ] = "test_parameter",
) -> Optional[Path]:
    """Path to dump the debug output as defined by the level to dump at."""
    if not base_dump_dir:
        return None
    test_module_relative_dir = self.get_module_relative_output_dir(
        filler_path
    )
    if level == "test_module":
        return Path(base_dump_dir) / Path(
            str(test_module_relative_dir).replace(os.sep, "__")
        )
    test_name, test_parameter_string = self.get_name_and_parameters()
    dir_str = str(test_module_relative_dir).replace(os.sep, "__")
    flat_path = f"{dir_str}__{test_name}"
    if level == "test_function":
        return Path(base_dump_dir) / flat_path
    elif level == "test_parameter":
        return Path(base_dump_dir) / flat_path / test_parameter_string
    raise Exception("Unexpected level.")

get_id()

Return the test id.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_id(self) -> str:
    """Return the test id."""
    return self.id

get_module_relative_output_dir(filler_path)

Return a directory name for the provided test_module (relative to the base ./tests directory) that can be used for output (within the configured fixtures output path or the base_dump_dir directory).

Example: tests/shanghai/eip3855_push0/test_push0.py -> shanghai/eip3855_push0/test_push0

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def get_module_relative_output_dir(self, filler_path: Path) -> Path:
    """
    Return a directory name for the provided test_module (relative to the
    base ./tests directory) that can be used for output (within the
    configured fixtures output path or the base_dump_dir directory).

    Example: tests/shanghai/eip3855_push0/test_push0.py ->
    shanghai/eip3855_push0/test_push0
    """
    basename = self.module_path.with_suffix("").absolute()
    basename_relative = basename.relative_to(
        os.path.commonpath([filler_path.absolute(), basename])
    )
    module_path = basename_relative.parent / self.strip_test_name(
        basename_relative.stem
    )
    return module_path

merge_partial_fixture_files(output_dir)

Merge all partial fixture JSONL files into final JSON fixture files.

Called at session end after all workers have written their partials. Each partial file contains JSONL lines: {"k": fixture_id, "v": json_str}

Processes one target file at a time, reading its partials sequentially into a dict. Memory = O(entries per target), freed before next target.

Source code in packages/testing/src/execution_testing/fixtures/collector.py

def merge_partial_fixture_files(output_dir: Path) -> None:
    """
    Merge all partial fixture JSONL files into final JSON fixture files.

    Called at session end after all workers have written their partials.
    Each partial file contains JSONL lines: {"k": fixture_id, "v": json_str}

    Processes one target file at a time, reading its partials sequentially
    into a dict. Memory = O(entries per target), freed before next target.
    """
    # Find all partial files
    partial_files = list(output_dir.rglob("*.partial.*.jsonl"))
    if not partial_files:
        return

    # Group partial files by their target fixture file
    # e.g., "test.partial.gw0.jsonl" -> "test.json"
    partials_by_target: Dict[Path, List[Path]] = {}
    for partial in partial_files:
        # Remove .partial.{worker_id}.jsonl suffix to get target
        name = partial.name
        # Find ".partial." and remove everything after
        idx = name.find(".partial.")
        if idx == -1:
            continue
        target_name = name[:idx] + ".json"
        target_path = partial.parent / target_name
        if target_path not in partials_by_target:
            partials_by_target[target_path] = []
        partials_by_target[target_path].append(partial)

    # Merge each group into its target file
    for target_path, partials in partials_by_target.items():
        # Seed from existing target file (if any) so that
        # repeated single-test sessions accumulate into one file.
        entries: Dict[str, str] = {}
        if target_path.exists():
            with open(target_path) as existing:
                try:
                    existing_data = json.load(existing)
                    for k, v in existing_data.items():
                        entries[k] = json.dumps(v, indent=4)
                except json.JSONDecodeError:
                    pass

        for partial in partials:
            with open(partial) as f:
                for line in f:
                    line = line.strip()
                    if line:
                        entry = json.loads(line)
                        entries[entry["k"]] = entry["v"]

        # Write sorted entries to output file
        with open(target_path, "w") as out_f:
            out_f.write("{\n")
            sorted_keys = sorted(entries.keys())
            last_idx = len(sorted_keys) - 1
            for i, key in enumerate(sorted_keys):
                key_json = json.dumps(key)
                value_indented = entries[key].replace("\n", "\n    ")
                out_f.write(f"    {key_json}: {value_indented}")
                out_f.write(",\n" if i < last_idx else "\n")
            out_f.write("}")

        # Free memory before processing next target
        entries.clear()

        # Clean up partial files
        for partial in partials:
            partial.unlink()
            # Also remove lock files
            lock_file = partial.with_suffix(".lock")
            if lock_file.exists():
                lock_file.unlink()

FixtureConsumer

Bases: ABC

Abstract class for verifying Ethereum test fixtures.

Source code in packages/testing/src/execution_testing/fixtures/consume.py

class FixtureConsumer(ABC):
    """Abstract class for verifying Ethereum test fixtures."""

    fixture_formats: List[FixtureFormat]

    def can_consume(
        self,
        fixture_format: FixtureFormat,
    ) -> bool:
        """Return whether the fixture format is consumable by this consumer."""
        return fixture_format in self.fixture_formats

    @abstractmethod
    def consume_fixture(
        self,
        fixture_format: FixtureFormat,
        fixture_path: Path,
        fixture_name: str | None = None,
        debug_output_path: Path | None = None,
    ) -> None:
        """
        Test the client with the specified fixture using its direct consumer
        interface.
        """
        raise NotImplementedError(
            "The `consume_fixture()` function is not supported by this tool."
        )

can_consume(fixture_format)

Return whether the fixture format is consumable by this consumer.

Source code in packages/testing/src/execution_testing/fixtures/consume.py

def can_consume(
    self,
    fixture_format: FixtureFormat,
) -> bool:
    """Return whether the fixture format is consumable by this consumer."""
    return fixture_format in self.fixture_formats

consume_fixture(fixture_format, fixture_path, fixture_name=None, debug_output_path=None) abstractmethod

Test the client with the specified fixture using its direct consumer interface.

Source code in packages/testing/src/execution_testing/fixtures/consume.py

@abstractmethod
def consume_fixture(
    self,
    fixture_format: FixtureFormat,
    fixture_path: Path,
    fixture_name: str | None = None,
    debug_output_path: Path | None = None,
) -> None:
    """
    Test the client with the specified fixture using its direct consumer
    interface.
    """
    raise NotImplementedError(
        "The `consume_fixture()` function is not supported by this tool."
    )

AccountCheck

Bases: CamelModel

Capture which fields are verified for a single account.

A None value means the field is not checked (it was not explicitly set by the test author). A present value records the expected value that check_alloc would assert against.

Source code in packages/testing/src/execution_testing/fixtures/post_verifications.py

class AccountCheck(CamelModel):
    """
    Capture which fields are verified for a single account.

    A ``None`` value means the field is not checked (it was not
    explicitly set by the test author).  A present value records the
    expected value that ``check_alloc`` would assert against.
    """

    nonce: ZeroPaddedHexNumber | None = None
    balance: ZeroPaddedHexNumber | None = None
    code: Bytes | None = None
    storage: Mapping[ZeroPaddedHexNumber, ZeroPaddedHexNumber] | None = None

PostVerifications

Bases: CamelModel

Record every post-state check performed during a fill session.

Accounts mapped to None represent should-not-exist checks.

Source code in packages/testing/src/execution_testing/fixtures/post_verifications.py

class PostVerifications(CamelModel):
    """
    Record every post-state check performed during a fill session.

    Accounts mapped to ``None`` represent *should-not-exist* checks.
    """

    accounts: Dict[Address, AccountCheck | None]

    @classmethod
    def from_alloc(cls, alloc: Alloc) -> PostVerifications:
        """
        Derive verification checks from an expected post ``Alloc``.

        Walk each address/account pair and inspect
        ``model_fields_set`` to determine which fields will actually
        be compared by ``Account.check_alloc``.
        """
        accounts: Dict[Address, AccountCheck | None] = {}
        for address, account in alloc.root.items():
            if account is None:
                accounts[address] = None
                continue
            accounts[address] = AccountCheck(
                nonce=(
                    account.nonce
                    if "nonce" in account.model_fields_set
                    else None
                ),
                balance=(
                    account.balance
                    if "balance" in account.model_fields_set
                    else None
                ),
                code=(
                    account.code
                    if "code" in account.model_fields_set
                    else None
                ),
                storage=(
                    dict(account.storage.root)
                    if "storage" in account.model_fields_set
                    else None
                ),
            )
        return cls(accounts=accounts)

from_alloc(alloc) classmethod

Derive verification checks from an expected post Alloc.

Walk each address/account pair and inspect model_fields_set to determine which fields will actually be compared by Account.check_alloc.

Source code in packages/testing/src/execution_testing/fixtures/post_verifications.py

@classmethod
def from_alloc(cls, alloc: Alloc) -> PostVerifications:
    """
    Derive verification checks from an expected post ``Alloc``.

    Walk each address/account pair and inspect
    ``model_fields_set`` to determine which fields will actually
    be compared by ``Account.check_alloc``.
    """
    accounts: Dict[Address, AccountCheck | None] = {}
    for address, account in alloc.root.items():
        if account is None:
            accounts[address] = None
            continue
        accounts[address] = AccountCheck(
            nonce=(
                account.nonce
                if "nonce" in account.model_fields_set
                else None
            ),
            balance=(
                account.balance
                if "balance" in account.model_fields_set
                else None
            ),
            code=(
                account.code
                if "code" in account.model_fields_set
                else None
            ),
            storage=(
                dict(account.storage.root)
                if "storage" in account.model_fields_set
                else None
            ),
        )
    return cls(accounts=accounts)

PreAllocGroup

Bases: PreAllocGroupCommon

Pre-allocation group for tests with identical Environment and fork values.

Grouping is still keyed on the tests' fixture Environment and fork, but that Environment lives in phase 1 (PreAllocGroupBuilder) only: what reaches the final group is the genesis header derived from it.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

class PreAllocGroup(PreAllocGroupCommon):
    """
    Pre-allocation group for tests with identical Environment and fork values.

    Grouping is still keyed on the tests' fixture Environment and fork, but
    that Environment lives in phase 1 (`PreAllocGroupBuilder`) only: what
    reaches the final group is the `genesis` header derived from it.
    """

    pre: GroupPreAlloc
    genesis: FixtureHeader
    pre_account_count: int
    test_count: int

    def model_post_init(self, __context: Any) -> None:
        """
        Model post init method to cache the state root in GroupPreAlloc.
        """
        super().model_post_init(__context)
        self.pre._cached_state_root = self.genesis.state_root
        self.pre._pre_alloc_group_hash = self.group_hash

    def hash(self) -> Hash:
        """
        Return a Hash based on the canonical JSON of the model.
        """
        canonical_json = json.dumps(
            self.model_dump(mode="json", by_alias=True, exclude_none=True),
            sort_keys=True,
        )
        return Hash(sha256(canonical_json.encode("utf-8")).digest())

model_post_init(__context)

Model post init method to cache the state root in GroupPreAlloc.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def model_post_init(self, __context: Any) -> None:
    """
    Model post init method to cache the state root in GroupPreAlloc.
    """
    super().model_post_init(__context)
    self.pre._cached_state_root = self.genesis.state_root
    self.pre._pre_alloc_group_hash = self.group_hash

hash()

Return a Hash based on the canonical JSON of the model.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def hash(self) -> Hash:
    """
    Return a Hash based on the canonical JSON of the model.
    """
    canonical_json = json.dumps(
        self.model_dump(mode="json", by_alias=True, exclude_none=True),
        sort_keys=True,
    )
    return Hash(sha256(canonical_json.encode("utf-8")).digest())

PreAllocGroupBuilder

Bases: PreAllocGroupCommon

Pre-allocation group temporary builder.

This file must NOT be saved as the final output of the filling process.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

class PreAllocGroupBuilder(PreAllocGroupCommon):
    """
    Pre-allocation group temporary builder.

    This file must _NOT_ be saved as the final output of the filling process.
    """

    environment: Environment = Field(
        ..., description="Grouping environment for this test group"
    )
    pre: Alloc

    def model_post_init(self, __context: Any) -> None:
        """
        Seed the pre-alloc's commitment scheme from its genesis fork.
        """
        super().model_post_init(__context)
        self.pre.migrate_state_commitment(
            self.fork.transitions_from().state_commitment()
        )

    def get_pre_account_count(self) -> int:
        """Return the amount of accounts the pre-allocation group holds."""
        return len(self.pre.root)

    def get_test_count(self) -> int:
        """Return the amount of tests that use this pre-allocation group."""
        return len(self.test_ids)

    def calculate_genesis(self) -> FixtureHeader:
        """Get the genesis header for this group."""
        return FixtureHeader.genesis(
            self.fork.transitions_from(),
            self.environment,
            self.pre.state_root(),
        )

    def add_test_alloc(self, test_id: str, new_pre: Alloc) -> None:
        """Adds a pre to this builder's pre."""
        assert self.pre.state_commitment() == new_pre.state_commitment()
        self.pre = Alloc.merge(
            self.pre,
            new_pre,
            key_collision_mode=Alloc.KeyCollisionMode.ALLOW_IDENTICAL_ACCOUNTS,
        )
        self.test_ids.append(test_id)

    def build(self) -> "PreAllocGroup":
        """Build the pre-alloc group."""
        return PreAllocGroup(
            test_ids=self.test_ids,
            fork=self.fork,
            chain_id=self.chain_id,
            group_salt=self.group_salt,
            group_hash=self.group_hash,
            pre=self.pre.model_dump(),
            pre_account_count=self.get_pre_account_count(),
            test_count=self.get_test_count(),
            genesis=self.calculate_genesis(),
        )

    def to_partial_file(
        self, file: Path, worker_id: Optional[str] = None
    ) -> None:
        """
        Save PreAllocGroupBuilder to a partial file (no locking).

        Each worker writes its own partial file, which are merged at session
        end by merge_partial_group_files(). This eliminates lock contention
        that caused workers to take 30-180+ seconds each.

        Saves the builder format (without genesis/state_root) to avoid
        expensive state root computation during Phase 1. State root is
        computed once when the `build` method is used to construct the final
        `PreAllocGroup`.
        """
        suffix = f".{worker_id}" if worker_id else ".main"
        partial_path = file.with_suffix(f".partial{suffix}.json")
        partial_path.write_text(
            self.model_dump_json(by_alias=True, exclude_none=True, indent=2)
        )

model_post_init(__context)

Seed the pre-alloc's commitment scheme from its genesis fork.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def model_post_init(self, __context: Any) -> None:
    """
    Seed the pre-alloc's commitment scheme from its genesis fork.
    """
    super().model_post_init(__context)
    self.pre.migrate_state_commitment(
        self.fork.transitions_from().state_commitment()
    )

get_pre_account_count()

Return the amount of accounts the pre-allocation group holds.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def get_pre_account_count(self) -> int:
    """Return the amount of accounts the pre-allocation group holds."""
    return len(self.pre.root)

get_test_count()

Return the amount of tests that use this pre-allocation group.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def get_test_count(self) -> int:
    """Return the amount of tests that use this pre-allocation group."""
    return len(self.test_ids)

calculate_genesis()

Get the genesis header for this group.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def calculate_genesis(self) -> FixtureHeader:
    """Get the genesis header for this group."""
    return FixtureHeader.genesis(
        self.fork.transitions_from(),
        self.environment,
        self.pre.state_root(),
    )

add_test_alloc(test_id, new_pre)

Adds a pre to this builder's pre.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def add_test_alloc(self, test_id: str, new_pre: Alloc) -> None:
    """Adds a pre to this builder's pre."""
    assert self.pre.state_commitment() == new_pre.state_commitment()
    self.pre = Alloc.merge(
        self.pre,
        new_pre,
        key_collision_mode=Alloc.KeyCollisionMode.ALLOW_IDENTICAL_ACCOUNTS,
    )
    self.test_ids.append(test_id)

build()

Build the pre-alloc group.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def build(self) -> "PreAllocGroup":
    """Build the pre-alloc group."""
    return PreAllocGroup(
        test_ids=self.test_ids,
        fork=self.fork,
        chain_id=self.chain_id,
        group_salt=self.group_salt,
        group_hash=self.group_hash,
        pre=self.pre.model_dump(),
        pre_account_count=self.get_pre_account_count(),
        test_count=self.get_test_count(),
        genesis=self.calculate_genesis(),
    )

to_partial_file(file, worker_id=None)

Save PreAllocGroupBuilder to a partial file (no locking).

Each worker writes its own partial file, which are merged at session end by merge_partial_group_files(). This eliminates lock contention that caused workers to take 30-180+ seconds each.

Saves the builder format (without genesis/state_root) to avoid expensive state root computation during Phase 1. State root is computed once when the build method is used to construct the final PreAllocGroup.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def to_partial_file(
    self, file: Path, worker_id: Optional[str] = None
) -> None:
    """
    Save PreAllocGroupBuilder to a partial file (no locking).

    Each worker writes its own partial file, which are merged at session
    end by merge_partial_group_files(). This eliminates lock contention
    that caused workers to take 30-180+ seconds each.

    Saves the builder format (without genesis/state_root) to avoid
    expensive state root computation during Phase 1. State root is
    computed once when the `build` method is used to construct the final
    `PreAllocGroup`.
    """
    suffix = f".{worker_id}" if worker_id else ".main"
    partial_path = file.with_suffix(f".partial{suffix}.json")
    partial_path.write_text(
        self.model_dump_json(by_alias=True, exclude_none=True, indent=2)
    )

PreAllocGroupBuilders

Bases: EthereumTestRootModel

Root model mapping pre-allocation group builders to group hashes.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

class PreAllocGroupBuilders(EthereumTestRootModel):
    """Root model mapping pre-allocation group builders to group hashes."""

    root: Dict[AllocGroupHash, PreAllocGroupBuilder] = Field(
        default_factory=dict
    )

    def to_folder(self, folder: Path, worker_id: Optional[str] = None) -> None:
        """
        Save PreAllocGroups to a folder as partial files.

        Each worker writes its own partial files (no lock contention).
        Call merge_partial_group_files() on master after all workers finish.
        """
        for key, value in self.root.items():
            assert value is not None, f"Value for key {key} is None"
            value.to_partial_file(folder / f"{key}.json", worker_id=worker_id)

    def add_test_pre(
        self,
        *,
        pre_alloc_hash: AllocGroupHash,
        test_id: str,
        fork: Fork | TransitionFork,
        chain_id: int,
        environment: Environment,
        pre: Alloc,
        group_salt: str | None = None,
    ) -> None:
        """Adds a single test to the appropriate group based on the hash."""
        if pre_alloc_hash in self.root:
            # Update existing group - just merge pre-allocations
            group = self.root[pre_alloc_hash]
            assert group.fork == fork, (
                f"Incompatible fork: {group.fork}!={fork}"
            )
            assert group.chain_id == chain_id, (
                f"Incompatible chain id: {group.chain_id}!={chain_id}"
            )
            assert group.group_salt == group_salt, (
                f"Incompatible group salt: {group.group_salt}!={group_salt}"
            )
            group.add_test_alloc(test_id, pre)
        else:
            # Create new group - use Environment instead of expensive genesis
            # generation
            group = PreAllocGroupBuilder(
                test_ids=[test_id],
                fork=fork,
                chain_id=chain_id,
                environment=environment,
                group_salt=group_salt,
                group_hash=pre_alloc_hash,
                pre=Alloc.merge(
                    Alloc.model_validate(
                        fork.transitions_to().pre_allocation_blockchain()
                    ),
                    pre,
                ),
            )
            self.root[pre_alloc_hash] = group

to_folder(folder, worker_id=None)

Save PreAllocGroups to a folder as partial files.

Each worker writes its own partial files (no lock contention). Call merge_partial_group_files() on master after all workers finish.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def to_folder(self, folder: Path, worker_id: Optional[str] = None) -> None:
    """
    Save PreAllocGroups to a folder as partial files.

    Each worker writes its own partial files (no lock contention).
    Call merge_partial_group_files() on master after all workers finish.
    """
    for key, value in self.root.items():
        assert value is not None, f"Value for key {key} is None"
        value.to_partial_file(folder / f"{key}.json", worker_id=worker_id)

add_test_pre(*, pre_alloc_hash, test_id, fork, chain_id, environment, pre, group_salt=None)

Adds a single test to the appropriate group based on the hash.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def add_test_pre(
    self,
    *,
    pre_alloc_hash: AllocGroupHash,
    test_id: str,
    fork: Fork | TransitionFork,
    chain_id: int,
    environment: Environment,
    pre: Alloc,
    group_salt: str | None = None,
) -> None:
    """Adds a single test to the appropriate group based on the hash."""
    if pre_alloc_hash in self.root:
        # Update existing group - just merge pre-allocations
        group = self.root[pre_alloc_hash]
        assert group.fork == fork, (
            f"Incompatible fork: {group.fork}!={fork}"
        )
        assert group.chain_id == chain_id, (
            f"Incompatible chain id: {group.chain_id}!={chain_id}"
        )
        assert group.group_salt == group_salt, (
            f"Incompatible group salt: {group.group_salt}!={group_salt}"
        )
        group.add_test_alloc(test_id, pre)
    else:
        # Create new group - use Environment instead of expensive genesis
        # generation
        group = PreAllocGroupBuilder(
            test_ids=[test_id],
            fork=fork,
            chain_id=chain_id,
            environment=environment,
            group_salt=group_salt,
            group_hash=pre_alloc_hash,
            pre=Alloc.merge(
                Alloc.model_validate(
                    fork.transitions_to().pre_allocation_blockchain()
                ),
                pre,
            ),
        )
        self.root[pre_alloc_hash] = group

PreAllocGroups

Bases: EthereumTestRootModel

Root model mapping pre-allocation group hashes to test groups.

If lazy_load is True, the groups are not loaded from the folder until they are accessed.

Iterating will fail if lazy_load is True.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

class PreAllocGroups(EthereumTestRootModel):
    """
    Root model mapping pre-allocation group hashes to test groups.

    If lazy_load is True, the groups are not loaded from the folder until they
    are accessed.

    Iterating will fail if lazy_load is True.
    """

    root: Dict[AllocGroupHash, PreAllocGroup | None] = Field(
        default_factory=dict
    )

    _folder_source: Path | None = PrivateAttr(None)

    def __setitem__(self, key: AllocGroupHash, value: Any) -> None:
        """Set item in root dict."""
        assert self._folder_source is None, (
            "Cannot set item in root dict after folder source is set"
        )
        self.root[key] = value

    @classmethod
    def from_folder(cls, folder: Path, *, lazy_load: bool = False) -> Self:
        """Create PreAllocGroups from a folder of pre-allocation files."""
        # First check for collision failures
        for fail_file in folder.glob("*.fail"):
            with open(fail_file) as f:
                raise Alloc.CollisionError.from_json(json.loads(f.read()))

        data: Dict[AllocGroupHash, PreAllocGroup | None] = {}
        for file in folder.glob("*.json"):
            if lazy_load:
                data[AllocGroupHash(file.stem)] = None
            else:
                data[AllocGroupHash(file.stem)] = PreAllocGroup.from_file(file)
        instance = cls(root=data)
        if lazy_load:
            instance._folder_source = folder
        return instance

    def __getitem__(self, item: AllocGroupHash) -> PreAllocGroup:
        """Get item from root dict."""
        if self._folder_source is None:
            value = self.root[item]
            assert value is not None, f"Item {item} is None"
            return value
        else:
            if self.root[item] is None:
                self.root[item] = PreAllocGroup.from_file(
                    self._folder_source / f"{item}.json"
                )
            result = self.root[item]
            assert result is not None
            return result

    def __iter__(self) -> Iterator[AllocGroupHash]:  # type: ignore [override]
        """Iterate over root dict."""
        return iter(self.root)

    def __contains__(self, item: AllocGroupHash) -> bool:
        """Check if item in root dict."""
        return item in self.root

    def __len__(self) -> int:
        """Get length of root dict."""
        return len(self.root)

    def keys(self) -> KeysView[AllocGroupHash]:
        """Get keys from root dict."""
        return self.root.keys()

    def values(self) -> Generator[PreAllocGroup, None, None]:
        """Get values from root dict."""
        for value in self.root.values():
            assert value is not None, "Value is None"
            yield value

    def items(
        self,
    ) -> Generator[Tuple[AllocGroupHash, PreAllocGroup], None, None]:
        """Get items from root dict."""
        for key, value in self.root.items():
            assert value is not None, f"Value for key {key} is None"
            yield key, value

__setitem__(key, value)

Set item in root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def __setitem__(self, key: AllocGroupHash, value: Any) -> None:
    """Set item in root dict."""
    assert self._folder_source is None, (
        "Cannot set item in root dict after folder source is set"
    )
    self.root[key] = value

from_folder(folder, *, lazy_load=False) classmethod

Create PreAllocGroups from a folder of pre-allocation files.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

@classmethod
def from_folder(cls, folder: Path, *, lazy_load: bool = False) -> Self:
    """Create PreAllocGroups from a folder of pre-allocation files."""
    # First check for collision failures
    for fail_file in folder.glob("*.fail"):
        with open(fail_file) as f:
            raise Alloc.CollisionError.from_json(json.loads(f.read()))

    data: Dict[AllocGroupHash, PreAllocGroup | None] = {}
    for file in folder.glob("*.json"):
        if lazy_load:
            data[AllocGroupHash(file.stem)] = None
        else:
            data[AllocGroupHash(file.stem)] = PreAllocGroup.from_file(file)
    instance = cls(root=data)
    if lazy_load:
        instance._folder_source = folder
    return instance

__getitem__(item)

Get item from root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def __getitem__(self, item: AllocGroupHash) -> PreAllocGroup:
    """Get item from root dict."""
    if self._folder_source is None:
        value = self.root[item]
        assert value is not None, f"Item {item} is None"
        return value
    else:
        if self.root[item] is None:
            self.root[item] = PreAllocGroup.from_file(
                self._folder_source / f"{item}.json"
            )
        result = self.root[item]
        assert result is not None
        return result

__iter__()

Iterate over root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def __iter__(self) -> Iterator[AllocGroupHash]:  # type: ignore [override]
    """Iterate over root dict."""
    return iter(self.root)

__contains__(item)

Check if item in root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def __contains__(self, item: AllocGroupHash) -> bool:
    """Check if item in root dict."""
    return item in self.root

__len__()

Get length of root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def __len__(self) -> int:
    """Get length of root dict."""
    return len(self.root)

keys()

Get keys from root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def keys(self) -> KeysView[AllocGroupHash]:
    """Get keys from root dict."""
    return self.root.keys()

values()

Get values from root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def values(self) -> Generator[PreAllocGroup, None, None]:
    """Get values from root dict."""
    for value in self.root.values():
        assert value is not None, "Value is None"
        yield value

items()

Get items from root dict.

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def items(
    self,
) -> Generator[Tuple[AllocGroupHash, PreAllocGroup], None, None]:
    """Get items from root dict."""
    for key, value in self.root.items():
        assert value is not None, f"Value for key {key} is None"
        yield key, value

pack_pre_alloc_groups(folder)

Merge fine-grained pre-allocation groups into fewer, larger ones.

Phase 1 keys every test's group on the exact content of any hard-coded accounts it sets (modified_accounts_salt), so a test that pins accounts to fixed addresses lands in its own group even when it could safely share a genesis with others. This is conservative: it splits far more than the genuine address conflicts require. groupstats shows this dominates the group count, with most groups a single test.

This pass reclaims that while preserving each test's isolation. Groups are bucketed by everything a shared genesis requires (fork, chain id, and environment), by the explicit pre_alloc_group marker salt (so a test that demands its own genesis keeps it), and then by their reserved-address footprint (see _reserved_addresses), so two tests only share a genesis when they agree on every precompile and shared address. Within a bucket the reserved accounts are identical and the remaining (test-private) addresses are unique to one group, so the union is always conflict-free and the whole bucket collapses to a single group.

The packing is deterministic: buckets are processed in sorted order and each group's id is derived from its sorted test ids, so a re-fill of the same tests reproduces the same groups.

Called on the master process after merge_partial_group_files, replacing the fine-grained files in folder with the packed ones. Also writes a test id -> group index file (see read_test_group_index), so phase 2 workers can find a test's group without scanning every group file; each entry records the test's fine-grained phase 1 hash as a fingerprint so a stale folder is detected (see packed_group_hash_for_test).

Source code in packages/testing/src/execution_testing/fixtures/pre_alloc_groups.py

def pack_pre_alloc_groups(folder: Path) -> None:
    """
    Merge fine-grained pre-allocation groups into fewer, larger ones.

    Phase 1 keys every test's group on the exact content of any hard-coded
    accounts it sets (`modified_accounts_salt`), so a test that pins accounts
    to fixed addresses lands in its own group even when it could safely share a
    genesis with others. This is conservative: it splits far more than the
    genuine address conflicts require. `groupstats` shows this dominates the
    group count, with most groups a single test.

    This pass reclaims that while preserving each test's isolation. Groups
    are bucketed by everything a shared genesis requires (fork, chain id, and
    environment), by the explicit `pre_alloc_group` marker salt (so a test
    that demands its own genesis keeps it), and then by their
    reserved-address footprint (see `_reserved_addresses`), so two tests only
    share a genesis when they agree on every precompile and shared address.
    Within a bucket the reserved accounts are identical and the remaining
    (test-private) addresses are unique to one group, so the union is always
    conflict-free and the whole bucket collapses to a single group.

    The packing is deterministic: buckets are processed in sorted order and
    each group's id is derived from its sorted test ids, so a re-fill of the
    same tests reproduces the same groups.

    Called on the master process after `merge_partial_group_files`, replacing
    the fine-grained files in `folder` with the packed ones. Also writes a
    test id -> group index file (see `read_test_group_index`), so phase 2
    workers can find a test's group without scanning every group file; each
    entry records the test's fine-grained phase 1 hash as a fingerprint so a
    stale folder is detected (see `packed_group_hash_for_test`).
    """
    files = sorted(folder.glob("*.json"))
    if not files:
        return

    builders = []
    phase1_hash_by_test: Dict[str, AllocGroupHash] = {}
    for file in files:
        builder = PreAllocGroupBuilder.from_file(file)
        for test_id in builder.test_ids:
            phase1_hash_by_test[test_id] = AllocGroupHash(file.stem)
        builders.append(builder)

    genesis_buckets: Dict[
        Tuple[Fork | TransitionFork, int, str, str], List[PreAllocGroupBuilder]
    ] = defaultdict(list)
    for builder in builders:
        key = (
            builder.fork,
            builder.chain_id,
            builder.group_salt or "",
            builder.environment.canonical_json(),
        )
        genesis_buckets[key].append(builder)

    # Drop the fine-grained files up front; the packed files written below are
    # named by content hash and never clash with the (now stale) originals.
    for file in files:
        file.unlink()

    test_group_index = GroupIndexEntries()
    for genesis_key in sorted(genesis_buckets):
        bucket = genesis_buckets[genesis_key]
        reserved = _reserved_addresses(bucket)

        packed: Dict[Tuple[Tuple[str, str], ...], PreAllocGroupBuilder] = {}
        for builder in bucket:
            signature = _reserved_signature(builder, reserved)
            if signature in packed:
                merged = packed[signature]
                merged.pre.root.update(builder.pre.root)
                merged.test_ids.extend(builder.test_ids)
            else:
                packed[signature] = builder

        for merged in packed.values():
            merged.test_ids.sort()
            packed_hash = _packed_group_hash(merged.test_ids)
            merged.group_hash = packed_hash
            (folder / f"{packed_hash}.json").write_text(
                merged.build().model_dump_json(
                    by_alias=True, exclude_none=True, indent=2
                )
            )
            for test_id in merged.test_ids:
                test_group_index[test_id] = GroupIndexEntry(
                    group_hash=packed_hash,
                    phase1_hash=phase1_hash_by_test[test_id],
                )
    test_group_index.to_file(folder / TEST_GROUP_INDEX_FILE)

StateFixture

Bases: BaseFixture

Fixture for a single StateTest.

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

class StateFixture(BaseFixture):
    """Fixture for a single StateTest."""

    format_name: ClassVar[str] = "state_test"
    description: ClassVar[str] = "Tests that generate a state test fixture."

    env: FixtureEnvironment
    pre: Alloc
    transaction: FixtureTransaction
    post: Mapping[Fork | TransitionFork, List[FixtureForkPost]]
    config: FixtureConfig

    def get_fork(self) -> Fork | TransitionFork | None:
        """Return fork of the fixture as a string."""
        forks = list(self.post.keys())
        assert len(forks) == 1, "Expected state test fixture with single fork"
        return forks[0]

get_fork()

Return fork of the fixture as a string.

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

def get_fork(self) -> Fork | TransitionFork | None:
    """Return fork of the fixture as a string."""
    forks = list(self.post.keys())
    assert len(forks) == 1, "Expected state test fixture with single fork"
    return forks[0]

TransactionFixture

Bases: BaseFixture

Fixture for a single TransactionTest.

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

class TransactionFixture(BaseFixture):
    """Fixture for a single TransactionTest."""

    format_name: ClassVar[str] = "transaction_test"
    description: ClassVar[str] = (
        "Tests that generate a transaction test fixture."
    )

    result: Mapping[Fork | TransitionFork, FixtureResult]
    transaction: Bytes = Field(..., alias="txbytes")

    def get_fork(self) -> Fork | TransitionFork | None:
        """Return the fork of the fixture as a string."""
        forks = list(self.result.keys())
        assert len(forks) == 1, (
            "Expected transaction test fixture with single fork"
        )
        return forks[0]

get_fork()

Return the fork of the fixture as a string.

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

def get_fork(self) -> Fork | TransitionFork | None:
    """Return the fork of the fixture as a string."""
    forks = list(self.result.keys())
    assert len(forks) == 1, (
        "Expected transaction test fixture with single fork"
    )
    return forks[0]