Ethereum Test Tools Package

Module containing tools for generating cross-client Ethereum execution layer tests.

AccessList

Bases: CamelModel

Access List for transactions.

Source code in src/ethereum_test_base_types/composite_types.py

class AccessList(CamelModel):
    """Access List for transactions."""

    address: Address
    storage_keys: List[Hash]

    def to_list(self) -> List[Address | List[Hash]]:
        """Return access list as a list of serializable elements."""
        return [self.address, self.storage_keys]

to_list()

Return access list as a list of serializable elements.

Source code in src/ethereum_test_base_types/composite_types.py

def to_list(self) -> List[Address | List[Hash]]:
    """Return access list as a list of serializable elements."""
    return [self.address, self.storage_keys]

Account

Bases: CamelModel

State associated with an address.

Source code in src/ethereum_test_base_types/composite_types.py

class Account(CamelModel):
    """State associated with an address."""

    nonce: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0)
    """
    The scalar value equal to a) the number of transactions sent by
    an Externally Owned Account, b) the amount of contracts created by a
    contract.
    """
    balance: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0)
    """
    The amount of Wei (10<sup>-18</sup> Eth) the account has.
    """
    code: Bytes = Bytes(b"")
    """
    Bytecode contained by the account.
    """
    storage: Storage = Field(default_factory=Storage)
    """
    Storage within a contract.
    """

    NONEXISTENT: ClassVar[None] = None
    """
    Sentinel object used to specify when an account should not exist in the
    state.
    """

    @dataclass(kw_only=True)
    class NonceMismatchError(Exception):
        """
        Test expected a certain nonce value for an account but a different
        value was found.
        """

        address: Address
        want: int | None
        got: int | None

        def __init__(self, address: Address, want: int | None, got: int | None, *args):
            """Initialize the exception with the address, wanted and got values."""
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string."""
            label_str = ""
            if self.address.label is not None:
                label_str = f" ({self.address.label})"
            return (
                f"unexpected nonce for account {self.address}{label_str}: "
                + f"want {self.want}, got {self.got}"
            )

    @dataclass(kw_only=True)
    class BalanceMismatchError(Exception):
        """
        Test expected a certain balance for an account but a different
        value was found.
        """

        address: Address
        want: int | None
        got: int | None

        def __init__(self, address: Address, want: int | None, got: int | None, *args):
            """Initialize the exception with the address, wanted and got values."""
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string."""
            label_str = ""
            if self.address.label is not None:
                label_str = f" ({self.address.label})"
            return (
                f"unexpected balance for account {self.address}{label_str}: "
                + f"want {self.want}, got {self.got}"
            )

    @dataclass(kw_only=True)
    class CodeMismatchError(Exception):
        """
        Test expected a certain bytecode for an account but a different
        one was found.
        """

        address: Address
        want: bytes | None
        got: bytes | None

        def __init__(self, address: Address, want: bytes | None, got: bytes | None, *args):
            """Initialize the exception with the address, wanted and got values."""
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string."""
            label_str = ""
            if self.address.label is not None:
                label_str = f" ({self.address.label})"
            return (
                f"unexpected code for account {self.address}{label_str}: "
                + f"want {self.want}, got {self.got}"
            )

    def check_alloc(self: "Account", address: Address, account: "Account"):
        """
        Check the returned alloc against an expected account in post state.
        Raises exception on failure.
        """
        if "nonce" in self.model_fields_set:
            if self.nonce != account.nonce:
                raise Account.NonceMismatchError(
                    address=address,
                    want=self.nonce,
                    got=account.nonce,
                )

        if "balance" in self.model_fields_set:
            if self.balance != account.balance:
                raise Account.BalanceMismatchError(
                    address=address,
                    want=self.balance,
                    got=account.balance,
                )

        if "code" in self.model_fields_set:
            if self.code != account.code:
                raise Account.CodeMismatchError(
                    address=address,
                    want=self.code,
                    got=account.code,
                )

        if "storage" in self.model_fields_set:
            self.storage.must_be_equal(address=address, other=account.storage)

    def __bool__(self: "Account") -> bool:
        """Return True on a non-empty account."""
        return any((self.nonce, self.balance, self.code, self.storage))

    @classmethod
    def with_code(cls: Type, code: BytesConvertible) -> "Account":
        """Create account with provided `code` and nonce of `1`."""
        return Account(nonce=HexNumber(1), code=Bytes(code))

    @classmethod
    def merge(
        cls: Type, account_1: "Dict | Account | None", account_2: "Dict | Account | None"
    ) -> "Account":
        """Create a merged account from two sources."""

        def to_kwargs_dict(account: "Dict | Account | None") -> Dict:
            if account is None:
                return {}
            if isinstance(account, dict):
                return account
            elif isinstance(account, cls):
                return account.model_dump(exclude_unset=True)
            raise TypeError(f"Unexpected type for account merge: {type(account)}")

        kwargs = to_kwargs_dict(account_1)
        kwargs.update(to_kwargs_dict(account_2))

        return cls(**kwargs)

nonce: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0) class-attribute instance-attribute

The scalar value equal to a) the number of transactions sent by an Externally Owned Account, b) the amount of contracts created by a contract.

balance: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0) class-attribute instance-attribute

The amount of Wei (10^-18 Eth) the account has.

code: Bytes = Bytes(b'') class-attribute instance-attribute

Bytecode contained by the account.

storage: Storage = Field(default_factory=Storage) class-attribute instance-attribute

Storage within a contract.

NONEXISTENT: None = None class-attribute

Sentinel object used to specify when an account should not exist in the state.

NonceMismatchError dataclass

Bases: Exception

Test expected a certain nonce value for an account but a different value was found.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class NonceMismatchError(Exception):
    """
    Test expected a certain nonce value for an account but a different
    value was found.
    """

    address: Address
    want: int | None
    got: int | None

    def __init__(self, address: Address, want: int | None, got: int | None, *args):
        """Initialize the exception with the address, wanted and got values."""
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string."""
        label_str = ""
        if self.address.label is not None:
            label_str = f" ({self.address.label})"
        return (
            f"unexpected nonce for account {self.address}{label_str}: "
            + f"want {self.want}, got {self.got}"
        )

__init__(address, want, got, *args)

Initialize the exception with the address, wanted and got values.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, address: Address, want: int | None, got: int | None, *args):
    """Initialize the exception with the address, wanted and got values."""
    super().__init__(args)
    self.address = address
    self.want = want
    self.got = got

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    label_str = ""
    if self.address.label is not None:
        label_str = f" ({self.address.label})"
    return (
        f"unexpected nonce for account {self.address}{label_str}: "
        + f"want {self.want}, got {self.got}"
    )

BalanceMismatchError dataclass

Bases: Exception

Test expected a certain balance for an account but a different value was found.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class BalanceMismatchError(Exception):
    """
    Test expected a certain balance for an account but a different
    value was found.
    """

    address: Address
    want: int | None
    got: int | None

    def __init__(self, address: Address, want: int | None, got: int | None, *args):
        """Initialize the exception with the address, wanted and got values."""
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string."""
        label_str = ""
        if self.address.label is not None:
            label_str = f" ({self.address.label})"
        return (
            f"unexpected balance for account {self.address}{label_str}: "
            + f"want {self.want}, got {self.got}"
        )

__init__(address, want, got, *args)

Initialize the exception with the address, wanted and got values.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, address: Address, want: int | None, got: int | None, *args):
    """Initialize the exception with the address, wanted and got values."""
    super().__init__(args)
    self.address = address
    self.want = want
    self.got = got

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    label_str = ""
    if self.address.label is not None:
        label_str = f" ({self.address.label})"
    return (
        f"unexpected balance for account {self.address}{label_str}: "
        + f"want {self.want}, got {self.got}"
    )

CodeMismatchError dataclass

Bases: Exception

Test expected a certain bytecode for an account but a different one was found.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class CodeMismatchError(Exception):
    """
    Test expected a certain bytecode for an account but a different
    one was found.
    """

    address: Address
    want: bytes | None
    got: bytes | None

    def __init__(self, address: Address, want: bytes | None, got: bytes | None, *args):
        """Initialize the exception with the address, wanted and got values."""
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string."""
        label_str = ""
        if self.address.label is not None:
            label_str = f" ({self.address.label})"
        return (
            f"unexpected code for account {self.address}{label_str}: "
            + f"want {self.want}, got {self.got}"
        )

__init__(address, want, got, *args)

Initialize the exception with the address, wanted and got values.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, address: Address, want: bytes | None, got: bytes | None, *args):
    """Initialize the exception with the address, wanted and got values."""
    super().__init__(args)
    self.address = address
    self.want = want
    self.got = got

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    label_str = ""
    if self.address.label is not None:
        label_str = f" ({self.address.label})"
    return (
        f"unexpected code for account {self.address}{label_str}: "
        + f"want {self.want}, got {self.got}"
    )

check_alloc(address, account)

Check the returned alloc against an expected account in post state. Raises exception on failure.

Source code in src/ethereum_test_base_types/composite_types.py

def check_alloc(self: "Account", address: Address, account: "Account"):
    """
    Check the returned alloc against an expected account in post state.
    Raises exception on failure.
    """
    if "nonce" in self.model_fields_set:
        if self.nonce != account.nonce:
            raise Account.NonceMismatchError(
                address=address,
                want=self.nonce,
                got=account.nonce,
            )

    if "balance" in self.model_fields_set:
        if self.balance != account.balance:
            raise Account.BalanceMismatchError(
                address=address,
                want=self.balance,
                got=account.balance,
            )

    if "code" in self.model_fields_set:
        if self.code != account.code:
            raise Account.CodeMismatchError(
                address=address,
                want=self.code,
                got=account.code,
            )

    if "storage" in self.model_fields_set:
        self.storage.must_be_equal(address=address, other=account.storage)

__bool__()

Return True on a non-empty account.

Source code in src/ethereum_test_base_types/composite_types.py

def __bool__(self: "Account") -> bool:
    """Return True on a non-empty account."""
    return any((self.nonce, self.balance, self.code, self.storage))

with_code(code) classmethod

Create account with provided code and nonce of 1.

Source code in src/ethereum_test_base_types/composite_types.py

@classmethod
def with_code(cls: Type, code: BytesConvertible) -> "Account":
    """Create account with provided `code` and nonce of `1`."""
    return Account(nonce=HexNumber(1), code=Bytes(code))

merge(account_1, account_2) classmethod

Create a merged account from two sources.

Source code in src/ethereum_test_base_types/composite_types.py

@classmethod
def merge(
    cls: Type, account_1: "Dict | Account | None", account_2: "Dict | Account | None"
) -> "Account":
    """Create a merged account from two sources."""

    def to_kwargs_dict(account: "Dict | Account | None") -> Dict:
        if account is None:
            return {}
        if isinstance(account, dict):
            return account
        elif isinstance(account, cls):
            return account.model_dump(exclude_unset=True)
        raise TypeError(f"Unexpected type for account merge: {type(account)}")

    kwargs = to_kwargs_dict(account_1)
    kwargs.update(to_kwargs_dict(account_2))

    return cls(**kwargs)

Address

Bases: FixedSizeBytes[20]

Class that helps represent Ethereum addresses in tests.

Source code in src/ethereum_test_base_types/base_types.py

class Address(FixedSizeBytes[20]):  # type: ignore
    """Class that helps represent Ethereum addresses in tests."""

    label: str | None = None

    def __new__(
        cls,
        input_bytes: "FixedSizeBytesConvertible | Address",
        *args,
        label: str | None = None,
        **kwargs,
    ):
        """Create a new Address object with an optional label."""
        instance = super(Address, cls).__new__(cls, input_bytes, *args, **kwargs)
        if isinstance(input_bytes, Address) and label is None:
            instance.label = input_bytes.label
        else:
            instance.label = label
        return instance

__new__(input_bytes, *args, label=None, **kwargs)

Create a new Address object with an optional label.

Source code in src/ethereum_test_base_types/base_types.py

def __new__(
    cls,
    input_bytes: "FixedSizeBytesConvertible | Address",
    *args,
    label: str | None = None,
    **kwargs,
):
    """Create a new Address object with an optional label."""
    instance = super(Address, cls).__new__(cls, input_bytes, *args, **kwargs)
    if isinstance(input_bytes, Address) and label is None:
        instance.label = input_bytes.label
    else:
        instance.label = label
    return instance

Bytes

Bases: bytes, ToStringSchema

Class that helps represent bytes of variable length in tests.

Source code in src/ethereum_test_base_types/base_types.py

class Bytes(bytes, ToStringSchema):
    """Class that helps represent bytes of variable length in tests."""

    def __new__(cls, input_bytes: BytesConvertible = b""):
        """Create a new Bytes object."""
        if type(input_bytes) is cls:
            return input_bytes
        return super(Bytes, cls).__new__(cls, to_bytes(input_bytes))

    def __hash__(self) -> int:
        """Return the hash of the bytes."""
        return super(Bytes, self).__hash__()

    def __str__(self) -> str:
        """Return the hexadecimal representation of the bytes."""
        return self.hex()

    def hex(self, *args, **kwargs) -> str:
        """Return the hexadecimal representation of the bytes."""
        return "0x" + super().hex(*args, **kwargs)

    @classmethod
    def or_none(cls, input_bytes: "Bytes | BytesConvertible | None") -> "Bytes | None":
        """Convert the input to a Bytes while accepting None."""
        if input_bytes is None:
            return input_bytes
        return cls(input_bytes)

    def keccak256(self) -> "Hash":
        """Return the keccak256 hash of the opcode byte representation."""
        k = keccak.new(digest_bits=256)
        return Hash(k.update(bytes(self)).digest())

    def sha256(self) -> "Hash":
        """Return the sha256 hash of the opcode byte representation."""
        return Hash(sha256(self).digest())

__new__(input_bytes=b'')

Create a new Bytes object.

Source code in src/ethereum_test_base_types/base_types.py

def __new__(cls, input_bytes: BytesConvertible = b""):
    """Create a new Bytes object."""
    if type(input_bytes) is cls:
        return input_bytes
    return super(Bytes, cls).__new__(cls, to_bytes(input_bytes))

__hash__()

Return the hash of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

def __hash__(self) -> int:
    """Return the hash of the bytes."""
    return super(Bytes, self).__hash__()

__str__()

Return the hexadecimal representation of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

def __str__(self) -> str:
    """Return the hexadecimal representation of the bytes."""
    return self.hex()

hex(*args, **kwargs)

Return the hexadecimal representation of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

def hex(self, *args, **kwargs) -> str:
    """Return the hexadecimal representation of the bytes."""
    return "0x" + super().hex(*args, **kwargs)

or_none(input_bytes) classmethod

Convert the input to a Bytes while accepting None.

Source code in src/ethereum_test_base_types/base_types.py

@classmethod
def or_none(cls, input_bytes: "Bytes | BytesConvertible | None") -> "Bytes | None":
    """Convert the input to a Bytes while accepting None."""
    if input_bytes is None:
        return input_bytes
    return cls(input_bytes)

keccak256()

Return the keccak256 hash of the opcode byte representation.

Source code in src/ethereum_test_base_types/base_types.py

def keccak256(self) -> "Hash":
    """Return the keccak256 hash of the opcode byte representation."""
    k = keccak.new(digest_bits=256)
    return Hash(k.update(bytes(self)).digest())

sha256()

Return the sha256 hash of the opcode byte representation.

Source code in src/ethereum_test_base_types/base_types.py

def sha256(self) -> "Hash":
    """Return the sha256 hash of the opcode byte representation."""
    return Hash(sha256(self).digest())

Hash

Bases: FixedSizeBytes[32]

Class that helps represent hashes in tests.

Source code in src/ethereum_test_base_types/base_types.py

class Hash(FixedSizeBytes[32]):  # type: ignore
    """Class that helps represent hashes in tests."""

    pass

ReferenceSpec

Reference Specification Description Abstract Class.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

class ReferenceSpec:
    """Reference Specification Description Abstract Class."""

    @abstractmethod
    def name(self) -> str:
        """Return the name of the spec."""
        pass

    @abstractmethod
    def has_known_version(self) -> bool:
        """Return true if the reference spec object is hard-coded with a latest known version."""
        pass

    @abstractmethod
    def known_version(self) -> str:
        """Return the latest known version in the reference."""
        pass

    @abstractmethod
    def api_url(self) -> str:
        """Return the URL required to poll the version from an API, if needed."""
        pass

    @abstractmethod
    def latest_version(self) -> str:
        """Return a digest that points to the latest version of the spec."""
        pass

    @abstractmethod
    def is_outdated(self) -> bool:
        """
        Check whether the reference specification has been updated since the
        test was last updated.
        """
        pass

    @abstractmethod
    def write_info(self, info: Dict[str, Dict[str, Any] | str]):
        """Write info about the reference specification used into the output fixture."""
        pass

    @staticmethod
    @abstractmethod
    def parseable_from_module(module_dict: Dict[str, Any]) -> bool:
        """Check whether the module's dict contains required reference spec information."""
        pass

    @staticmethod
    @abstractmethod
    def parse_from_module(module_dict: Dict[str, Any]) -> "ReferenceSpec":
        """Parse the module's dict into a reference spec."""
        pass

name() abstractmethod

Return the name of the spec.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def name(self) -> str:
    """Return the name of the spec."""
    pass

has_known_version() abstractmethod

Return true if the reference spec object is hard-coded with a latest known version.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def has_known_version(self) -> bool:
    """Return true if the reference spec object is hard-coded with a latest known version."""
    pass

known_version() abstractmethod

Return the latest known version in the reference.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def known_version(self) -> str:
    """Return the latest known version in the reference."""
    pass

api_url() abstractmethod

Return the URL required to poll the version from an API, if needed.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def api_url(self) -> str:
    """Return the URL required to poll the version from an API, if needed."""
    pass

latest_version() abstractmethod

Return a digest that points to the latest version of the spec.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def latest_version(self) -> str:
    """Return a digest that points to the latest version of the spec."""
    pass

is_outdated() abstractmethod

Check whether the reference specification has been updated since the test was last updated.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def is_outdated(self) -> bool:
    """
    Check whether the reference specification has been updated since the
    test was last updated.
    """
    pass

write_info(info) abstractmethod

Write info about the reference specification used into the output fixture.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@abstractmethod
def write_info(self, info: Dict[str, Dict[str, Any] | str]):
    """Write info about the reference specification used into the output fixture."""
    pass

parseable_from_module(module_dict) abstractmethod staticmethod

Check whether the module's dict contains required reference spec information.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@staticmethod
@abstractmethod
def parseable_from_module(module_dict: Dict[str, Any]) -> bool:
    """Check whether the module's dict contains required reference spec information."""
    pass

parse_from_module(module_dict) abstractmethod staticmethod

Parse the module's dict into a reference spec.

Source code in src/ethereum_test_base_types/reference_spec/reference_spec.py

@staticmethod
@abstractmethod
def parse_from_module(module_dict: Dict[str, Any]) -> "ReferenceSpec":
    """Parse the module's dict into a reference spec."""
    pass

BlockException

Bases: ExceptionBase

Exception raised when a block is invalid, but not due to a transaction.

E.g. all transactions in the block are valid, and can be applied to the state, but the block header contains an invalid field.

Source code in src/ethereum_test_exceptions/exceptions.py

@unique
class BlockException(ExceptionBase):
    """
    Exception raised when a block is invalid, but not due to a transaction.

    E.g. all transactions in the block are valid, and can be applied to the state, but the
    block header contains an invalid field.
    """

    TOO_MANY_UNCLES = auto()
    """
    Block declares too many uncles over the allowed limit.
    """
    UNCLE_IN_CHAIN = auto()
    """
    Block declares uncle header that is already imported into chain.
    """
    UNCLE_IS_ANCESTOR = auto()
    """
    Block declares uncle header that is directly a parent of this block.
    """
    UNCLE_IS_BROTHER = auto()
    """
    Block declares two similar uncle headers.
    """
    UNCLE_PARENT_INCORRECT = auto()
    """
    Block declares uncle header that is an outdated block to be an uncle.
    """
    EXTRA_DATA_TOO_BIG = auto()
    """
    Block header's extra data >32 bytes.
    """
    EXTRA_DATA_INVALID_DAO = auto()
    """
    Block header's extra data after dao fork must be a fixed pre defined hash.
    """
    UNKNOWN_PARENT = auto()
    """
    Block header's parent hash does not correspond to any of existing blocks on chain.
    """
    UNCLE_UNKNOWN_PARENT = auto()
    """
    Uncle header's parent hash does not correspond to any of existing blocks on chain.
    """
    UNKNOWN_PARENT_ZERO = auto()
    """
    Block header's parent hash is zero hash.
    """
    GASLIMIT_TOO_BIG = auto()
    """
    Block header's gas limit > 0x7fffffffffffffff.
    """
    INVALID_BLOCK_NUMBER = auto()
    """
    Block header's number != parent header's number + 1.
    """
    INVALID_BLOCK_TIMESTAMP_OLDER_THAN_PARENT = auto()
    """
    Block header's timestamp <= parent header's timestamp.
    """
    INVALID_DIFFICULTY = auto()
    """
    Block header's difficulty does not match the difficulty formula calculated from previous block.
    """
    INVALID_LOG_BLOOM = auto()
    """
    Block header's logs bloom hash does not match the actually computed log bloom.
    """
    INVALID_STATE_ROOT = auto()
    """
    Block header's state root hash does not match the actually computed hash of the state.
    """
    INVALID_RECEIPTS_ROOT = auto()
    """
    Block header's receipts root hash does not match the actually computed hash of receipts.
    """
    INVALID_TRANSACTIONS_ROOT = auto()
    """
    Block header's transactions root hash does not match the actually computed hash of tx tree.
    """
    INVALID_UNCLES_HASH = auto()
    """
    Block header's uncle hash does not match the actually computed hash of block's uncles.
    """
    GAS_USED_OVERFLOW = auto()
    """
    Block transactions consume more gas than block header allow.
    """
    INVALID_GASLIMIT = auto()
    """
    Block header's gas limit does not match the gas limit formula calculated from previous block.
    """
    INVALID_BASEFEE_PER_GAS = auto()
    """
    Block header's base_fee_per_gas field is calculated incorrect.
    """
    INVALID_GAS_USED = auto()
    """
    Block header's actual gas used does not match the provided header's value
    """
    INVALID_WITHDRAWALS_ROOT = auto()
    """
    Block header's withdrawals root does not match calculated withdrawals root.
    """
    INCORRECT_BLOCK_FORMAT = auto()
    """
    Block's format is incorrect, contains invalid fields, is missing fields, or contains fields of
    a fork that is not active yet.
    """
    BLOB_GAS_USED_ABOVE_LIMIT = auto()
    """
    Block's blob gas used in header is above the limit.
    """
    INCORRECT_BLOB_GAS_USED = auto()
    """
    Block's blob gas used in header is incorrect.
    """
    INCORRECT_EXCESS_BLOB_GAS = auto()
    """
    Block's excess blob gas in header is incorrect.
    """
    RLP_STRUCTURES_ENCODING = auto()
    """
    Block's rlp encoding is valid but ethereum structures in it are invalid.
    """
    RLP_WITHDRAWALS_NOT_READ = auto()
    """
    Block's rlp encoding is missing withdrawals.
    """
    RLP_INVALID_FIELD_OVERFLOW_64 = auto()
    """
    One of block's fields rlp is overflow 2**64 value.
    """
    RLP_INVALID_ADDRESS = auto()
    """
    Block withdrawals address is rlp of invalid address != 20 bytes.
    """
    INVALID_REQUESTS = auto()
    """
    Block's requests are invalid.
    """
    IMPORT_IMPOSSIBLE_LEGACY = auto()
    """
    Legacy block import is impossible in this chain configuration.
    """
    IMPORT_IMPOSSIBLE_LEGACY_WRONG_PARENT = auto()
    """
    Legacy block import is impossible, trying to import on top of a block that is not legacy.
    """
    IMPORT_IMPOSSIBLE_LONDON_WRONG_PARENT = auto()
    """
    Trying to import london (basefee) block on top of block that is not 1559.
    """
    IMPORT_IMPOSSIBLE_PARIS_WRONG_POW = auto()
    """
    Trying to import paris(merge) block with PoW enabled.
    """
    IMPORT_IMPOSSIBLE_PARIS_WRONG_POS = auto()
    """
    Trying to import paris(merge) block with PoS enabled before TTD is reached.
    """
    IMPORT_IMPOSSIBLE_LONDON_OVER_PARIS = auto()
    """
    Trying to import london looking block over paris network (POS).
    """
    IMPORT_IMPOSSIBLE_PARIS_OVER_SHANGHAI = auto()
    """
    Trying to import paris block on top of shanghai block.
    """
    IMPORT_IMPOSSIBLE_SHANGHAI = auto()
    """
    Shanghai block import is impossible in this chain configuration.
    """
    IMPORT_IMPOSSIBLE_UNCLES_OVER_PARIS = auto()
    """
    Trying to import a block after paris fork that has not empty uncles hash.
    """
    IMPORT_IMPOSSIBLE_DIFFICULTY_OVER_PARIS = auto()
    """
    Trying to import a block after paris fork that has difficulty != 0.
    """

TOO_MANY_UNCLES = auto() class-attribute instance-attribute

Block declares too many uncles over the allowed limit.

UNCLE_IN_CHAIN = auto() class-attribute instance-attribute

Block declares uncle header that is already imported into chain.

UNCLE_IS_ANCESTOR = auto() class-attribute instance-attribute

Block declares uncle header that is directly a parent of this block.

UNCLE_IS_BROTHER = auto() class-attribute instance-attribute

Block declares two similar uncle headers.

UNCLE_PARENT_INCORRECT = auto() class-attribute instance-attribute

Block declares uncle header that is an outdated block to be an uncle.

EXTRA_DATA_TOO_BIG = auto() class-attribute instance-attribute

Block header's extra data >32 bytes.

EXTRA_DATA_INVALID_DAO = auto() class-attribute instance-attribute

Block header's extra data after dao fork must be a fixed pre defined hash.

UNKNOWN_PARENT = auto() class-attribute instance-attribute

Block header's parent hash does not correspond to any of existing blocks on chain.

UNCLE_UNKNOWN_PARENT = auto() class-attribute instance-attribute

Uncle header's parent hash does not correspond to any of existing blocks on chain.

UNKNOWN_PARENT_ZERO = auto() class-attribute instance-attribute

Block header's parent hash is zero hash.

GASLIMIT_TOO_BIG = auto() class-attribute instance-attribute

Block header's gas limit > 0x7fffffffffffffff.

INVALID_BLOCK_NUMBER = auto() class-attribute instance-attribute

Block header's number != parent header's number + 1.

INVALID_BLOCK_TIMESTAMP_OLDER_THAN_PARENT = auto() class-attribute instance-attribute

Block header's timestamp <= parent header's timestamp.

INVALID_DIFFICULTY = auto() class-attribute instance-attribute

Block header's difficulty does not match the difficulty formula calculated from previous block.

INVALID_LOG_BLOOM = auto() class-attribute instance-attribute

Block header's logs bloom hash does not match the actually computed log bloom.

INVALID_STATE_ROOT = auto() class-attribute instance-attribute

Block header's state root hash does not match the actually computed hash of the state.

INVALID_RECEIPTS_ROOT = auto() class-attribute instance-attribute

Block header's receipts root hash does not match the actually computed hash of receipts.

INVALID_TRANSACTIONS_ROOT = auto() class-attribute instance-attribute

Block header's transactions root hash does not match the actually computed hash of tx tree.

INVALID_UNCLES_HASH = auto() class-attribute instance-attribute

Block header's uncle hash does not match the actually computed hash of block's uncles.

GAS_USED_OVERFLOW = auto() class-attribute instance-attribute

Block transactions consume more gas than block header allow.

INVALID_GASLIMIT = auto() class-attribute instance-attribute

Block header's gas limit does not match the gas limit formula calculated from previous block.

INVALID_BASEFEE_PER_GAS = auto() class-attribute instance-attribute

Block header's base_fee_per_gas field is calculated incorrect.

INVALID_GAS_USED = auto() class-attribute instance-attribute

Block header's actual gas used does not match the provided header's value

INVALID_WITHDRAWALS_ROOT = auto() class-attribute instance-attribute

Block header's withdrawals root does not match calculated withdrawals root.

INCORRECT_BLOCK_FORMAT = auto() class-attribute instance-attribute

Block's format is incorrect, contains invalid fields, is missing fields, or contains fields of a fork that is not active yet.

BLOB_GAS_USED_ABOVE_LIMIT = auto() class-attribute instance-attribute

Block's blob gas used in header is above the limit.

INCORRECT_BLOB_GAS_USED = auto() class-attribute instance-attribute

Block's blob gas used in header is incorrect.

INCORRECT_EXCESS_BLOB_GAS = auto() class-attribute instance-attribute

Block's excess blob gas in header is incorrect.

RLP_STRUCTURES_ENCODING = auto() class-attribute instance-attribute

Block's rlp encoding is valid but ethereum structures in it are invalid.

RLP_WITHDRAWALS_NOT_READ = auto() class-attribute instance-attribute

Block's rlp encoding is missing withdrawals.

RLP_INVALID_FIELD_OVERFLOW_64 = auto() class-attribute instance-attribute

One of block's fields rlp is overflow 2**64 value.

RLP_INVALID_ADDRESS = auto() class-attribute instance-attribute

Block withdrawals address is rlp of invalid address != 20 bytes.

INVALID_REQUESTS = auto() class-attribute instance-attribute

Block's requests are invalid.

IMPORT_IMPOSSIBLE_LEGACY = auto() class-attribute instance-attribute

Legacy block import is impossible in this chain configuration.

IMPORT_IMPOSSIBLE_LEGACY_WRONG_PARENT = auto() class-attribute instance-attribute

Legacy block import is impossible, trying to import on top of a block that is not legacy.

IMPORT_IMPOSSIBLE_LONDON_WRONG_PARENT = auto() class-attribute instance-attribute

Trying to import london (basefee) block on top of block that is not 1559.

IMPORT_IMPOSSIBLE_PARIS_WRONG_POW = auto() class-attribute instance-attribute

Trying to import paris(merge) block with PoW enabled.

IMPORT_IMPOSSIBLE_PARIS_WRONG_POS = auto() class-attribute instance-attribute

Trying to import paris(merge) block with PoS enabled before TTD is reached.

IMPORT_IMPOSSIBLE_LONDON_OVER_PARIS = auto() class-attribute instance-attribute

Trying to import london looking block over paris network (POS).

IMPORT_IMPOSSIBLE_PARIS_OVER_SHANGHAI = auto() class-attribute instance-attribute

Trying to import paris block on top of shanghai block.

IMPORT_IMPOSSIBLE_SHANGHAI = auto() class-attribute instance-attribute

Shanghai block import is impossible in this chain configuration.

IMPORT_IMPOSSIBLE_UNCLES_OVER_PARIS = auto() class-attribute instance-attribute

Trying to import a block after paris fork that has not empty uncles hash.

IMPORT_IMPOSSIBLE_DIFFICULTY_OVER_PARIS = auto() class-attribute instance-attribute

Trying to import a block after paris fork that has difficulty != 0.

EngineAPIError

Bases: IntEnum

List of Engine API errors.

Source code in src/ethereum_test_exceptions/engine_api.py

class EngineAPIError(IntEnum):
    """List of Engine API errors."""

    ParseError = -32700
    InvalidRequest = -32600
    MethodNotFound = -32601
    InvalidParams = -32602
    InternalError = -32603
    ServerError = -32000
    UnknownPayload = -38001
    InvalidForkchoiceState = -38002
    InvalidPayloadAttributes = -38003
    TooLargeRequest = -38004
    UnsupportedFork = -38005

EOFException

Bases: ExceptionBase

Exception raised when an EOF container is invalid.

Source code in src/ethereum_test_exceptions/exceptions.py

@unique
class EOFException(ExceptionBase):
    """Exception raised when an EOF container is invalid."""

    DEFAULT_EXCEPTION = auto()
    """
    Expect some exception, not yet known.
    """

    UNDEFINED_EXCEPTION = auto()
    """
    Indicates that exception string is not mapped to an exception enum.
    """

    UNDEFINED_INSTRUCTION = auto()
    """
    EOF container has undefined instruction in it's body code.
    """

    UNKNOWN_VERSION = auto()
    """
    EOF container has an unknown version.
    """
    INCOMPLETE_MAGIC = auto()
    """
    EOF container has not enough bytes to read magic.
    """
    INVALID_MAGIC = auto()
    """
    EOF container has not allowed magic version byte.
    """
    INVALID_VERSION = auto()
    """
    EOF container version bytes mismatch.
    """
    INVALID_NON_RETURNING_FLAG = auto()
    """
    EOF container's section has non-returning flag set incorrectly.
    """
    INVALID_RJUMP_DESTINATION = auto()
    """
    Code has RJUMP instruction with invalid parameters.
    """
    MISSING_TYPE_HEADER = auto()
    """
    EOF container missing types section.
    """
    INVALID_TYPE_SECTION_SIZE = auto()
    """
    EOF container types section has wrong size.
    """
    INVALID_TYPE_BODY = auto()
    """
    EOF container types body section bytes are wrong.
    """
    MISSING_CODE_HEADER = auto()
    """
    EOF container missing code section.
    """
    INVALID_CODE_SECTION = auto()
    """
    EOF container code section bytes are incorrect.
    """
    INCOMPLETE_CODE_HEADER = auto()
    """
    EOF container code header missing bytes.
    """
    INCOMPLETE_DATA_HEADER = auto()
    """
    EOF container data header missing bytes.
    """
    ZERO_SECTION_SIZE = auto()
    """
    EOF container data header construction is wrong.
    """
    MISSING_DATA_SECTION = auto()
    """
    EOF container missing data section
    """
    INCOMPLETE_CONTAINER = auto()
    """
    EOF container bytes are incomplete.
    """
    INVALID_SECTION_BODIES_SIZE = auto()
    """
    Sections bodies does not match sections headers.
    """
    TRAILING_BYTES = auto()
    """
    EOF container has bytes beyond data section.
    """
    MISSING_TERMINATOR = auto()
    """
    EOF container missing terminator bytes between header and body.
    """
    MISSING_HEADERS_TERMINATOR = auto()
    """
    Some type of another exception about missing headers terminator.
    """
    INVALID_FIRST_SECTION_TYPE = auto()
    """
    EOF container header does not have types section first.
    """
    INCOMPLETE_SECTION_NUMBER = auto()
    """
    EOF container header has section that is missing declaration bytes.
    """
    INCOMPLETE_SECTION_SIZE = auto()
    """
    EOF container header has section that is defined incorrectly.
    """
    TOO_MANY_CODE_SECTIONS = auto()
    """
    EOF container header has too many code sections.
    """
    MISSING_STOP_OPCODE = auto()
    """
    EOF container's code missing STOP bytecode at it's end.
    """
    INPUTS_OUTPUTS_NUM_ABOVE_LIMIT = auto()
    """
    EOF container code section inputs/outputs number is above the limit
    """
    UNREACHABLE_INSTRUCTIONS = auto()
    """
    EOF container's code have instructions that are unreachable.
    """
    UNREACHABLE_CODE_SECTIONS = auto()
    """
    EOF container's body have code sections that are unreachable.
    """
    STACK_UNDERFLOW = auto()
    """
    EOF container's code produces an stack underflow.
    """
    STACK_OVERFLOW = auto()
    """
    EOF container's code produces an stack overflow.
    """
    STACK_HEIGHT_MISMATCH = auto()
    """
    EOF container section stack height mismatch.
    """
    MAX_STACK_HEIGHT_ABOVE_LIMIT = auto()
    """
    EOF container's specified max stack height is above the limit.
    """
    STACK_HIGHER_THAN_OUTPUTS = auto()
    """
    EOF container section stack height is higher than the outputs.
    when returning
    """
    JUMPF_DESTINATION_INCOMPATIBLE_OUTPUTS = auto()
    """
    EOF container section JUMPF's to a destination section with incompatible outputs.
    """
    INVALID_MAX_STACK_HEIGHT = auto()
    """
    EOF container section's specified max stack height does not match the actual stack height.
    """
    INVALID_DATALOADN_INDEX = auto()
    """
    A DATALOADN instruction has out-of-bounds index for the data section.
    """
    TRUNCATED_INSTRUCTION = auto()
    """
    EOF container's code section has truncated instruction.
    """
    TOPLEVEL_CONTAINER_TRUNCATED = auto()
    """
    Top-level EOF container has data section truncated
    """
    ORPHAN_SUBCONTAINER = auto()
    """
    EOF container has an unreferenced subcontainer.
    '"""
    CONTAINER_SIZE_ABOVE_LIMIT = auto()
    """
    EOF container is above size limit
    """
    INVALID_CONTAINER_SECTION_INDEX = auto()
    """
    Instruction references container section that does not exist.
    """
    INCOMPATIBLE_CONTAINER_KIND = auto()
    """
    Incompatible instruction found in a container of a specific kind.
    """
    TOO_MANY_CONTAINERS = auto()
    """
    EOF container header has too many sub-containers.
    """
    INVALID_CODE_SECTION_INDEX = auto()
    """
    CALLF Operation referes to a non-existent code section
    """
    UNEXPECTED_HEADER_KIND = auto()
    """
    Header parsing encounterd a section kind it wasn't expecting
    """
    CALLF_TO_NON_RETURNING = auto()
    """
    CALLF instruction targeting a non-returning code section
    """
    EOFCREATE_WITH_TRUNCATED_CONTAINER = auto()
    """
    EOFCREATE with truncated container
    """

DEFAULT_EXCEPTION = auto() class-attribute instance-attribute

Expect some exception, not yet known.

UNDEFINED_EXCEPTION = auto() class-attribute instance-attribute

Indicates that exception string is not mapped to an exception enum.

UNDEFINED_INSTRUCTION = auto() class-attribute instance-attribute

EOF container has undefined instruction in it's body code.

UNKNOWN_VERSION = auto() class-attribute instance-attribute

EOF container has an unknown version.

INCOMPLETE_MAGIC = auto() class-attribute instance-attribute

EOF container has not enough bytes to read magic.

INVALID_MAGIC = auto() class-attribute instance-attribute

EOF container has not allowed magic version byte.

INVALID_VERSION = auto() class-attribute instance-attribute

EOF container version bytes mismatch.

INVALID_NON_RETURNING_FLAG = auto() class-attribute instance-attribute

EOF container's section has non-returning flag set incorrectly.

INVALID_RJUMP_DESTINATION = auto() class-attribute instance-attribute

Code has RJUMP instruction with invalid parameters.

MISSING_TYPE_HEADER = auto() class-attribute instance-attribute

EOF container missing types section.

INVALID_TYPE_SECTION_SIZE = auto() class-attribute instance-attribute

EOF container types section has wrong size.

INVALID_TYPE_BODY = auto() class-attribute instance-attribute

EOF container types body section bytes are wrong.

MISSING_CODE_HEADER = auto() class-attribute instance-attribute

EOF container missing code section.

INVALID_CODE_SECTION = auto() class-attribute instance-attribute

EOF container code section bytes are incorrect.

INCOMPLETE_CODE_HEADER = auto() class-attribute instance-attribute

EOF container code header missing bytes.

INCOMPLETE_DATA_HEADER = auto() class-attribute instance-attribute

EOF container data header missing bytes.

ZERO_SECTION_SIZE = auto() class-attribute instance-attribute

EOF container data header construction is wrong.

MISSING_DATA_SECTION = auto() class-attribute instance-attribute

EOF container missing data section

INCOMPLETE_CONTAINER = auto() class-attribute instance-attribute

EOF container bytes are incomplete.

INVALID_SECTION_BODIES_SIZE = auto() class-attribute instance-attribute

Sections bodies does not match sections headers.

TRAILING_BYTES = auto() class-attribute instance-attribute

EOF container has bytes beyond data section.

MISSING_TERMINATOR = auto() class-attribute instance-attribute

EOF container missing terminator bytes between header and body.

MISSING_HEADERS_TERMINATOR = auto() class-attribute instance-attribute

Some type of another exception about missing headers terminator.

INVALID_FIRST_SECTION_TYPE = auto() class-attribute instance-attribute

EOF container header does not have types section first.

INCOMPLETE_SECTION_NUMBER = auto() class-attribute instance-attribute

EOF container header has section that is missing declaration bytes.

INCOMPLETE_SECTION_SIZE = auto() class-attribute instance-attribute

EOF container header has section that is defined incorrectly.

TOO_MANY_CODE_SECTIONS = auto() class-attribute instance-attribute

EOF container header has too many code sections.

MISSING_STOP_OPCODE = auto() class-attribute instance-attribute

EOF container's code missing STOP bytecode at it's end.

INPUTS_OUTPUTS_NUM_ABOVE_LIMIT = auto() class-attribute instance-attribute

EOF container code section inputs/outputs number is above the limit

UNREACHABLE_INSTRUCTIONS = auto() class-attribute instance-attribute

EOF container's code have instructions that are unreachable.

UNREACHABLE_CODE_SECTIONS = auto() class-attribute instance-attribute

EOF container's body have code sections that are unreachable.

STACK_UNDERFLOW = auto() class-attribute instance-attribute

EOF container's code produces an stack underflow.

STACK_OVERFLOW = auto() class-attribute instance-attribute

EOF container's code produces an stack overflow.

STACK_HEIGHT_MISMATCH = auto() class-attribute instance-attribute

EOF container section stack height mismatch.

MAX_STACK_HEIGHT_ABOVE_LIMIT = auto() class-attribute instance-attribute

EOF container's specified max stack height is above the limit.

STACK_HIGHER_THAN_OUTPUTS = auto() class-attribute instance-attribute

EOF container section stack height is higher than the outputs. when returning

JUMPF_DESTINATION_INCOMPATIBLE_OUTPUTS = auto() class-attribute instance-attribute

EOF container section JUMPF's to a destination section with incompatible outputs.

INVALID_MAX_STACK_HEIGHT = auto() class-attribute instance-attribute

EOF container section's specified max stack height does not match the actual stack height.

INVALID_DATALOADN_INDEX = auto() class-attribute instance-attribute

A DATALOADN instruction has out-of-bounds index for the data section.

TRUNCATED_INSTRUCTION = auto() class-attribute instance-attribute

EOF container's code section has truncated instruction.

TOPLEVEL_CONTAINER_TRUNCATED = auto() class-attribute instance-attribute

Top-level EOF container has data section truncated

ORPHAN_SUBCONTAINER = auto() class-attribute instance-attribute

EOF container has an unreferenced subcontainer. '

CONTAINER_SIZE_ABOVE_LIMIT = auto() class-attribute instance-attribute

EOF container is above size limit

INVALID_CONTAINER_SECTION_INDEX = auto() class-attribute instance-attribute

Instruction references container section that does not exist.

INCOMPATIBLE_CONTAINER_KIND = auto() class-attribute instance-attribute

Incompatible instruction found in a container of a specific kind.

TOO_MANY_CONTAINERS = auto() class-attribute instance-attribute

EOF container header has too many sub-containers.

INVALID_CODE_SECTION_INDEX = auto() class-attribute instance-attribute

CALLF Operation referes to a non-existent code section

UNEXPECTED_HEADER_KIND = auto() class-attribute instance-attribute

Header parsing encounterd a section kind it wasn't expecting

CALLF_TO_NON_RETURNING = auto() class-attribute instance-attribute

CALLF instruction targeting a non-returning code section

EOFCREATE_WITH_TRUNCATED_CONTAINER = auto() class-attribute instance-attribute

EOFCREATE with truncated container

TransactionException

Bases: ExceptionBase

Exception raised when a transaction is invalid, and thus cannot be executed.

If a transaction with any of these exceptions is included in a block, the block is invalid.

Source code in src/ethereum_test_exceptions/exceptions.py

@unique
class TransactionException(ExceptionBase):
    """
    Exception raised when a transaction is invalid, and thus cannot be executed.

    If a transaction with any of these exceptions is included in a block, the block is invalid.
    """

    TYPE_NOT_SUPPORTED = auto()
    """
    Transaction type is not supported on this chain configuration.
    """
    SENDER_NOT_EOA = auto()
    """
    Transaction is coming from address that is not exist anymore.
    """
    ADDRESS_TOO_SHORT = auto()
    """
    Transaction `to` is not allowed to be less than 20 bytes.
    """
    ADDRESS_TOO_LONG = auto()
    """
    Transaction `to` is not allowed to be more than 20 bytes.
    """
    NONCE_MISMATCH_TOO_HIGH = auto()
    """
    Transaction nonce > sender.nonce.
    """
    NONCE_MISMATCH_TOO_LOW = auto()
    """
    Transaction nonce < sender.nonce.
    """
    NONCE_TOO_BIG = auto()
    """
    Transaction `nonce` is not allowed to be max_uint64 - 1 (this is probably TransactionTest).
    """
    NONCE_IS_MAX = auto()
    """
    Transaction `nonce` is not allowed to be max_uint64 - 1 (this is StateTests).
    """
    NONCE_OVERFLOW = auto()
    """
    Transaction `nonce` is not allowed to be more than uint64.
    """
    GASLIMIT_OVERFLOW = auto()
    """
    Transaction gaslimit exceeds 2^64-1 maximum value.
    """
    VALUE_OVERFLOW = auto()
    """
    Transaction value exceeds 2^256-1 maximum value.
    """
    GASPRICE_OVERFLOW = auto()
    """
    Transaction gasPrice exceeds 2^256-1 maximum value.
    """
    GASLIMIT_PRICE_PRODUCT_OVERFLOW = auto()
    """
    Transaction gasPrice * gasLimit exceeds 2^256-1 maximum value.
    """
    INVALID_SIGNATURE_VRS = auto()
    """
    Invalid transaction v, r, s values.
    """
    RLP_INVALID_SIGNATURE_R = auto()
    """
    Error reading transaction signature R value.
    """
    RLP_INVALID_SIGNATURE_S = auto()
    """
    Error reading transaction signature S value.
    """
    RLP_LEADING_ZEROS_GASLIMIT = auto()
    """
    Error reading transaction gaslimit field RLP.
    """
    RLP_LEADING_ZEROS_GASPRICE = auto()
    """
    Error reading transaction gasprice field RLP.
    """
    RLP_LEADING_ZEROS_VALUE = auto()
    """
    Error reading transaction value field RLP.
    """
    RLP_LEADING_ZEROS_NONCE = auto()
    """
    Error reading transaction nonce field RLP.
    """
    RLP_LEADING_ZEROS_R = auto()
    """
    Error reading transaction signature R field RLP.
    """
    RLP_LEADING_ZEROS_S = auto()
    """
    Error reading transaction signature S field RLP.
    """
    RLP_LEADING_ZEROS_V = auto()
    """
    Error reading transaction signature V field RLP.
    """
    RLP_LEADING_ZEROS_BASEFEE = auto()
    """
    Error reading transaction basefee field RLP.
    """
    RLP_LEADING_ZEROS_PRIORITY_FEE = auto()
    """
    Error reading transaction priority fee field RLP.
    """
    RLP_LEADING_ZEROS_DATA_SIZE = auto()
    """
    Error reading transaction data field RLP, (rlp field length has leading zeros).
    """
    RLP_LEADING_ZEROS_NONCE_SIZE = auto()
    """
    Error reading transaction nonce field RLP, (rlp field length has leading zeros).
    """
    RLP_TOO_FEW_ELEMENTS = auto()
    """
    Error reading transaction RLP, structure has too few elements than expected.
    """
    RLP_TOO_MANY_ELEMENTS = auto()
    """
    Error reading transaction RLP, structure has too many elements than expected.
    """
    RLP_ERROR_EOF = auto()
    """
    Error reading transaction RLP, rlp stream unexpectedly finished.
    """
    RLP_ERROR_SIZE = auto()
    """
    Error reading transaction RLP, rlp size is invalid.
    """
    RLP_ERROR_SIZE_LEADING_ZEROS = auto()
    """
    Error reading transaction RLP, field size has leading zeros.
    """
    INVALID_CHAINID = auto()
    """
    Transaction chain id encoding is incorrect.
    """
    RLP_INVALID_DATA = auto()
    """
    Transaction data field is invalid rlp.
    """
    RLP_INVALID_GASLIMIT = auto()
    """
    Transaction gaslimit field is invalid rlp.
    """
    RLP_INVALID_NONCE = auto()
    """
    Transaction nonce field is invalid rlp.
    """
    RLP_INVALID_TO = auto()
    """
    Transaction to field is invalid rlp.
    """
    RLP_INVALID_ACCESS_LIST_ADDRESS_TOO_LONG = auto()
    """
    Transaction access list address is > 20 bytes.
    """
    RLP_INVALID_ACCESS_LIST_ADDRESS_TOO_SHORT = auto()
    """
    Transaction access list address is < 20 bytes.
    """
    RLP_INVALID_ACCESS_LIST_STORAGE_TOO_LONG = auto()
    """
    Transaction access list storage hash > 32 bytes.
    """
    RLP_INVALID_ACCESS_LIST_STORAGE_TOO_SHORT = auto()
    """
    Transaction access list storage hash < 32 bytes.
    """
    RLP_INVALID_HEADER = auto()
    """
    Transaction failed to read from RLP as rlp header is invalid.
    """
    RLP_INVALID_VALUE = auto()
    """
    Transaction value field is invalid rlp/structure.
    """
    EC_RECOVERY_FAIL = auto()
    """
    Transaction has correct signature, but ec recovery failed.
    """
    INSUFFICIENT_ACCOUNT_FUNDS = auto()
    """
    Transaction's sender does not have enough funds to pay for the transaction.
    """
    INSUFFICIENT_MAX_FEE_PER_GAS = auto()
    """
    Transaction's max-fee-per-gas is lower than the block base-fee.
    """
    PRIORITY_OVERFLOW = auto()
    """
    Transaction's max-priority-fee-per-gas is exceeds 2^256-1 maximum value.
    """
    PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS = auto()
    """
    Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas.
    """
    PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS_2 = auto()
    """
    Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas (TransactionTests).
    """
    INSUFFICIENT_MAX_FEE_PER_BLOB_GAS = auto()
    """
    Transaction's max-fee-per-blob-gas is lower than the block's blob-gas price.
    """
    INTRINSIC_GAS_TOO_LOW = auto()
    """
    Transaction's gas limit is too low.
    """
    INITCODE_SIZE_EXCEEDED = auto()
    """
    Transaction's initcode for a contract-creating transaction is too large.
    """
    TYPE_3_TX_PRE_FORK = auto()
    """
    Transaction type 3 included before activation fork.
    """
    TYPE_3_TX_ZERO_BLOBS_PRE_FORK = auto()
    """
    Transaction type 3, with zero blobs, included before activation fork.
    """
    TYPE_3_TX_INVALID_BLOB_VERSIONED_HASH = auto()
    """
    Transaction contains a blob versioned hash with an invalid version.
    """
    TYPE_3_TX_WITH_FULL_BLOBS = auto()
    """
    Transaction contains full blobs (network-version of the transaction).
    """
    TYPE_3_TX_BLOB_COUNT_EXCEEDED = auto()
    """
    Transaction contains too many blob versioned hashes.
    """
    TYPE_3_TX_CONTRACT_CREATION = auto()
    """
    Transaction is a type 3 transaction and has an empty `to`.
    """
    TYPE_3_TX_MAX_BLOB_GAS_ALLOWANCE_EXCEEDED = auto()
    """
    Transaction causes block to go over blob gas limit.
    """
    GAS_ALLOWANCE_EXCEEDED = auto()
    """
    Transaction causes block to go over blob gas limit.
    """
    TYPE_3_TX_ZERO_BLOBS = auto()
    """
    Transaction is type 3, but has no blobs.
    """
    TYPE_4_EMPTY_AUTHORIZATION_LIST = auto()
    """
    Transaction is type 4, but has an empty authorization list.
    """
    TYPE_4_INVALID_AUTHORITY_SIGNATURE = auto()
    """
    Transaction authority signature is invalid
    """
    TYPE_4_INVALID_AUTHORITY_SIGNATURE_S_TOO_HIGH = auto()
    """
    Transaction authority signature is invalid
    """
    TYPE_4_TX_CONTRACT_CREATION = auto()
    """
    Transaction is a type 4 transaction and has an empty `to`.
    """
    TYPE_4_INVALID_AUTHORIZATION_FORMAT = auto()
    """
    Transaction is type 4, but contains an authorization that has an invalid format.
    """

TYPE_NOT_SUPPORTED = auto() class-attribute instance-attribute

Transaction type is not supported on this chain configuration.

SENDER_NOT_EOA = auto() class-attribute instance-attribute

Transaction is coming from address that is not exist anymore.

ADDRESS_TOO_SHORT = auto() class-attribute instance-attribute

Transaction to is not allowed to be less than 20 bytes.

ADDRESS_TOO_LONG = auto() class-attribute instance-attribute

Transaction to is not allowed to be more than 20 bytes.

NONCE_MISMATCH_TOO_HIGH = auto() class-attribute instance-attribute

Transaction nonce > sender.nonce.

NONCE_MISMATCH_TOO_LOW = auto() class-attribute instance-attribute

Transaction nonce < sender.nonce.

NONCE_TOO_BIG = auto() class-attribute instance-attribute

Transaction nonce is not allowed to be max_uint64 - 1 (this is probably TransactionTest).

NONCE_IS_MAX = auto() class-attribute instance-attribute

Transaction nonce is not allowed to be max_uint64 - 1 (this is StateTests).

NONCE_OVERFLOW = auto() class-attribute instance-attribute

Transaction nonce is not allowed to be more than uint64.

GASLIMIT_OVERFLOW = auto() class-attribute instance-attribute

Transaction gaslimit exceeds 2^64-1 maximum value.

VALUE_OVERFLOW = auto() class-attribute instance-attribute

Transaction value exceeds 2^256-1 maximum value.

GASPRICE_OVERFLOW = auto() class-attribute instance-attribute

Transaction gasPrice exceeds 2^256-1 maximum value.

GASLIMIT_PRICE_PRODUCT_OVERFLOW = auto() class-attribute instance-attribute

Transaction gasPrice * gasLimit exceeds 2^256-1 maximum value.

INVALID_SIGNATURE_VRS = auto() class-attribute instance-attribute

Invalid transaction v, r, s values.

RLP_INVALID_SIGNATURE_R = auto() class-attribute instance-attribute

Error reading transaction signature R value.

RLP_INVALID_SIGNATURE_S = auto() class-attribute instance-attribute

Error reading transaction signature S value.

RLP_LEADING_ZEROS_GASLIMIT = auto() class-attribute instance-attribute

Error reading transaction gaslimit field RLP.

RLP_LEADING_ZEROS_GASPRICE = auto() class-attribute instance-attribute

Error reading transaction gasprice field RLP.

RLP_LEADING_ZEROS_VALUE = auto() class-attribute instance-attribute

Error reading transaction value field RLP.

RLP_LEADING_ZEROS_NONCE = auto() class-attribute instance-attribute

Error reading transaction nonce field RLP.

RLP_LEADING_ZEROS_R = auto() class-attribute instance-attribute

Error reading transaction signature R field RLP.

RLP_LEADING_ZEROS_S = auto() class-attribute instance-attribute

Error reading transaction signature S field RLP.

RLP_LEADING_ZEROS_V = auto() class-attribute instance-attribute

Error reading transaction signature V field RLP.

RLP_LEADING_ZEROS_BASEFEE = auto() class-attribute instance-attribute

Error reading transaction basefee field RLP.

RLP_LEADING_ZEROS_PRIORITY_FEE = auto() class-attribute instance-attribute

Error reading transaction priority fee field RLP.

RLP_LEADING_ZEROS_DATA_SIZE = auto() class-attribute instance-attribute

Error reading transaction data field RLP, (rlp field length has leading zeros).

RLP_LEADING_ZEROS_NONCE_SIZE = auto() class-attribute instance-attribute

Error reading transaction nonce field RLP, (rlp field length has leading zeros).

RLP_TOO_FEW_ELEMENTS = auto() class-attribute instance-attribute

Error reading transaction RLP, structure has too few elements than expected.

RLP_TOO_MANY_ELEMENTS = auto() class-attribute instance-attribute

Error reading transaction RLP, structure has too many elements than expected.

RLP_ERROR_EOF = auto() class-attribute instance-attribute

Error reading transaction RLP, rlp stream unexpectedly finished.

RLP_ERROR_SIZE = auto() class-attribute instance-attribute

Error reading transaction RLP, rlp size is invalid.

RLP_ERROR_SIZE_LEADING_ZEROS = auto() class-attribute instance-attribute

Error reading transaction RLP, field size has leading zeros.

INVALID_CHAINID = auto() class-attribute instance-attribute

Transaction chain id encoding is incorrect.

RLP_INVALID_DATA = auto() class-attribute instance-attribute

Transaction data field is invalid rlp.

RLP_INVALID_GASLIMIT = auto() class-attribute instance-attribute

Transaction gaslimit field is invalid rlp.

RLP_INVALID_NONCE = auto() class-attribute instance-attribute

Transaction nonce field is invalid rlp.

RLP_INVALID_TO = auto() class-attribute instance-attribute

Transaction to field is invalid rlp.

RLP_INVALID_ACCESS_LIST_ADDRESS_TOO_LONG = auto() class-attribute instance-attribute

Transaction access list address is > 20 bytes.

RLP_INVALID_ACCESS_LIST_ADDRESS_TOO_SHORT = auto() class-attribute instance-attribute

Transaction access list address is < 20 bytes.

RLP_INVALID_ACCESS_LIST_STORAGE_TOO_LONG = auto() class-attribute instance-attribute

Transaction access list storage hash > 32 bytes.

RLP_INVALID_ACCESS_LIST_STORAGE_TOO_SHORT = auto() class-attribute instance-attribute

Transaction access list storage hash < 32 bytes.

RLP_INVALID_HEADER = auto() class-attribute instance-attribute

Transaction failed to read from RLP as rlp header is invalid.

RLP_INVALID_VALUE = auto() class-attribute instance-attribute

Transaction value field is invalid rlp/structure.

EC_RECOVERY_FAIL = auto() class-attribute instance-attribute

Transaction has correct signature, but ec recovery failed.

INSUFFICIENT_ACCOUNT_FUNDS = auto() class-attribute instance-attribute

Transaction's sender does not have enough funds to pay for the transaction.

INSUFFICIENT_MAX_FEE_PER_GAS = auto() class-attribute instance-attribute

Transaction's max-fee-per-gas is lower than the block base-fee.

PRIORITY_OVERFLOW = auto() class-attribute instance-attribute

Transaction's max-priority-fee-per-gas is exceeds 2^256-1 maximum value.

PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS = auto() class-attribute instance-attribute

Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas.

PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS_2 = auto() class-attribute instance-attribute

Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas (TransactionTests).

INSUFFICIENT_MAX_FEE_PER_BLOB_GAS = auto() class-attribute instance-attribute

Transaction's max-fee-per-blob-gas is lower than the block's blob-gas price.

INTRINSIC_GAS_TOO_LOW = auto() class-attribute instance-attribute

Transaction's gas limit is too low.

INITCODE_SIZE_EXCEEDED = auto() class-attribute instance-attribute

Transaction's initcode for a contract-creating transaction is too large.

TYPE_3_TX_PRE_FORK = auto() class-attribute instance-attribute

Transaction type 3 included before activation fork.

TYPE_3_TX_ZERO_BLOBS_PRE_FORK = auto() class-attribute instance-attribute

Transaction type 3, with zero blobs, included before activation fork.

TYPE_3_TX_INVALID_BLOB_VERSIONED_HASH = auto() class-attribute instance-attribute

Transaction contains a blob versioned hash with an invalid version.

TYPE_3_TX_WITH_FULL_BLOBS = auto() class-attribute instance-attribute

Transaction contains full blobs (network-version of the transaction).

TYPE_3_TX_BLOB_COUNT_EXCEEDED = auto() class-attribute instance-attribute

Transaction contains too many blob versioned hashes.

TYPE_3_TX_CONTRACT_CREATION = auto() class-attribute instance-attribute

Transaction is a type 3 transaction and has an empty to.

TYPE_3_TX_MAX_BLOB_GAS_ALLOWANCE_EXCEEDED = auto() class-attribute instance-attribute

Transaction causes block to go over blob gas limit.

GAS_ALLOWANCE_EXCEEDED = auto() class-attribute instance-attribute

Transaction causes block to go over blob gas limit.

TYPE_3_TX_ZERO_BLOBS = auto() class-attribute instance-attribute

Transaction is type 3, but has no blobs.

TYPE_4_EMPTY_AUTHORIZATION_LIST = auto() class-attribute instance-attribute

Transaction is type 4, but has an empty authorization list.

TYPE_4_INVALID_AUTHORITY_SIGNATURE = auto() class-attribute instance-attribute

Transaction authority signature is invalid

TYPE_4_INVALID_AUTHORITY_SIGNATURE_S_TOO_HIGH = auto() class-attribute instance-attribute

Transaction authority signature is invalid

TYPE_4_TX_CONTRACT_CREATION = auto() class-attribute instance-attribute

Transaction is a type 4 transaction and has an empty to.

TYPE_4_INVALID_AUTHORIZATION_FORMAT = auto() class-attribute instance-attribute

Transaction is type 4, but contains an authorization that has an invalid format.

BaseFixture

Bases: CamelModel

Represents a base Ethereum test fixture of any type.

Source code in src/ethereum_test_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")

    # Fixture format properties
    format_name: ClassVar[str] = ""
    output_file_extension: ClassVar[str] = ".json"
    description: ClassVar[str] = "Unknown fixture format; it has not been set."

    @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):
        """
        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 fill_info(
        self,
        t8n_version: str,
        test_case_description: str,
        fixture_source_url: str,
        ref_spec: ReferenceSpec | None,
        _info_metadata: Dict[str, Any],
    ):
        """Fill the info field for this fixture."""
        if "comment" not in self.info:
            self.info["comment"] = "`execution-spec-tests` generated test"
        self.info["filling-transition-tool"] = t8n_version
        self.info["description"] = test_case_description
        self.info["url"] = fixture_source_url
        self.info["fixture_format"] = self.format_name
        if ref_spec is not None:
            ref_spec.write_info(self.info)
        if _info_metadata:
            self.info.update(_info_metadata)

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

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

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

output_base_dir_name() classmethod

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

Source code in src/ethereum_test_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 src/ethereum_test_fixtures/base.py

@classmethod
def __pydantic_init_subclass__(cls, **kwargs):
    """
    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: Dict[str, Any] cached property

Returns the JSON representation of the fixture.

hash: str 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 src/ethereum_test_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

fill_info(t8n_version, test_case_description, fixture_source_url, ref_spec, _info_metadata)

Fill the info field for this fixture.

Source code in src/ethereum_test_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],
):
    """Fill the info field for this fixture."""
    if "comment" not in self.info:
        self.info["comment"] = "`execution-spec-tests` generated test"
    self.info["filling-transition-tool"] = t8n_version
    self.info["description"] = test_case_description
    self.info["url"] = fixture_source_url
    self.info["fixture_format"] = self.format_name
    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 src/ethereum_test_fixtures/base.py

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

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 src/ethereum_test_fixtures/base.py

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

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

FixtureCollector dataclass

Collects all fixtures generated by the test cases.

Source code in src/ethereum_test_fixtures/collector.py

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

    output_dir: Path
    flat_output: bool
    single_fixture_per_file: bool
    filler_path: Path
    base_dump_dir: Optional[Path] = None

    # Internal state
    all_fixtures: Dict[Path, Fixtures] = field(default_factory=dict)
    json_path_to_test_item: Dict[Path, TestInfo] = field(default_factory=dict)

    def get_fixture_basename(self, info: TestInfo) -> Path:
        """Return basename of the fixture file for a given test case."""
        if self.flat_output:
            if self.single_fixture_per_file:
                return Path(strip_test_prefix(info.get_single_test_name()))
            return Path(strip_test_prefix(info.original_name))
        else:
            relative_fixture_output_dir = Path(info.path).parent / strip_test_prefix(
                Path(info.path).stem
            )
            module_relative_output_dir = get_module_relative_output_dir(
                relative_fixture_output_dir, self.filler_path
            )

            if self.single_fixture_per_file:
                return module_relative_output_dir / strip_test_prefix(info.get_single_test_name())
            return module_relative_output_dir / strip_test_prefix(info.original_name)

    def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> Path:
        """Add fixture to the list of fixtures of a given test case."""
        fixture_basename = self.get_fixture_basename(info)

        fixture_path = (
            self.output_dir
            / fixture.output_base_dir_name()
            / fixture_basename.with_suffix(fixture.output_file_extension)
        )
        if fixture_path not in self.all_fixtures.keys():  # relevant when we group by test function
            self.all_fixtures[fixture_path] = Fixtures(root={})
            self.json_path_to_test_item[fixture_path] = info

        self.all_fixtures[fixture_path][info.id] = fixture

        return fixture_path

    def dump_fixtures(self) -> None:
        """Dump all collected fixtures to their respective files."""
        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)
            return
        os.makedirs(self.output_dir, exist_ok=True)
        for fixture_path, fixtures in self.all_fixtures.items():
            os.makedirs(fixture_path.parent, exist_ok=True)
            if len({fixture.__class__ for fixture in fixtures.values()}) != 1:
                raise TypeError("All fixtures in a single file must have the same format.")
            fixtures.collect_into_file(fixture_path)

    def verify_fixture_files(self, evm_fixture_verification: FixtureConsumer) -> None:
        """Run `evm [state|block]test` on each fixture."""
        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__):
                    info = self.json_path_to_test_item[fixture_path]
                    consume_direct_dump_dir = self._get_consume_direct_dump_dir(info)
                    evm_fixture_verification.consume_fixture(
                        fixture.__class__,
                        fixture_path,
                        fixture_name=None,
                        debug_output_path=consume_direct_dump_dir,
                    )

    def _get_consume_direct_dump_dir(
        self,
        info: TestInfo,
    ):
        """
        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"
            )

get_fixture_basename(info)

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

Source code in src/ethereum_test_fixtures/collector.py

def get_fixture_basename(self, info: TestInfo) -> Path:
    """Return basename of the fixture file for a given test case."""
    if self.flat_output:
        if self.single_fixture_per_file:
            return Path(strip_test_prefix(info.get_single_test_name()))
        return Path(strip_test_prefix(info.original_name))
    else:
        relative_fixture_output_dir = Path(info.path).parent / strip_test_prefix(
            Path(info.path).stem
        )
        module_relative_output_dir = get_module_relative_output_dir(
            relative_fixture_output_dir, self.filler_path
        )

        if self.single_fixture_per_file:
            return module_relative_output_dir / strip_test_prefix(info.get_single_test_name())
        return module_relative_output_dir / strip_test_prefix(info.original_name)

add_fixture(info, fixture)

Add fixture to the list of fixtures of a given test case.

Source code in src/ethereum_test_fixtures/collector.py

def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> Path:
    """Add fixture to the list of fixtures of a given test case."""
    fixture_basename = self.get_fixture_basename(info)

    fixture_path = (
        self.output_dir
        / fixture.output_base_dir_name()
        / fixture_basename.with_suffix(fixture.output_file_extension)
    )
    if fixture_path not in self.all_fixtures.keys():  # relevant when we group by test function
        self.all_fixtures[fixture_path] = Fixtures(root={})
        self.json_path_to_test_item[fixture_path] = info

    self.all_fixtures[fixture_path][info.id] = fixture

    return fixture_path

dump_fixtures()

Dump all collected fixtures to their respective files.

Source code in src/ethereum_test_fixtures/collector.py

def dump_fixtures(self) -> None:
    """Dump all collected fixtures to their respective files."""
    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)
        return
    os.makedirs(self.output_dir, exist_ok=True)
    for fixture_path, fixtures in self.all_fixtures.items():
        os.makedirs(fixture_path.parent, exist_ok=True)
        if len({fixture.__class__ for fixture in fixtures.values()}) != 1:
            raise TypeError("All fixtures in a single file must have the same format.")
        fixtures.collect_into_file(fixture_path)

verify_fixture_files(evm_fixture_verification)

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

Source code in src/ethereum_test_fixtures/collector.py

def verify_fixture_files(self, evm_fixture_verification: FixtureConsumer) -> None:
    """Run `evm [state|block]test` on each fixture."""
    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__):
                info = self.json_path_to_test_item[fixture_path]
                consume_direct_dump_dir = self._get_consume_direct_dump_dir(info)
                evm_fixture_verification.consume_fixture(
                    fixture.__class__,
                    fixture_path,
                    fixture_name=None,
                    debug_output_path=consume_direct_dump_dir,
                )

TestInfo dataclass

Contains test information from the current node.

Source code in src/ethereum_test_fixtures/collector.py

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

    name: str  # pytest: Item.name
    id: str  # pytest: Item.nodeid
    original_name: str  # pytest: Item.originalname
    path: Path  # pytest: Item.path

    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) -> str:
        """Convert test name to a single test name."""
        test_name, test_parameters = self.get_name_and_parameters()
        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 = get_module_relative_output_dir(self.path, 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()
        flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{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_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 src/ethereum_test_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()

Convert test name to a single test name.

Source code in src/ethereum_test_fixtures/collector.py

def get_single_test_name(self) -> str:
    """Convert test name to a single test name."""
    test_name, test_parameters = self.get_name_and_parameters()
    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 src/ethereum_test_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 = get_module_relative_output_dir(self.path, 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()
    flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{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.")

BaseTest

Bases: BaseModel

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

Source code in src/ethereum_test_specs/base.py

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

    tag: str = ""

    # Transition tool specific fields
    t8n_dump_dir: Path | None = Field(None, exclude=True)
    _t8n_call_counter: Iterator[int] = count(0)

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = []
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = []

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

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

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

    @abstractmethod
    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the list of test fixtures."""
        pass

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        raise Exception(f"Unsupported execute format: {execute_format}")

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

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

    def get_next_transition_tool_output_path(self) -> str:
        """Return path to the next transition tool output file."""
        if not self.t8n_dump_dir:
            return ""
        return path.join(
            self.t8n_dump_dir,
            str(next(self._t8n_call_counter)),
        )

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

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

Source code in src/ethereum_test_specs/base.py

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

discard_execute_format_by_marks(execute_format, fork, markers) classmethod

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

Source code in src/ethereum_test_specs/base.py

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

generate(*, request, t8n, fork, fixture_format, eips=None) abstractmethod

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py

@abstractmethod
def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the list of test fixtures."""
    pass

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    raise Exception(f"Unsupported execute format: {execute_format}")

pytest_parameter_name() classmethod

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

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

Source code in src/ethereum_test_specs/base.py

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

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

get_next_transition_tool_output_path()

Return path to the next transition tool output file.

Source code in src/ethereum_test_specs/base.py

def get_next_transition_tool_output_path(self) -> str:
    """Return path to the next transition tool output file."""
    if not self.t8n_dump_dir:
        return ""
    return path.join(
        self.t8n_dump_dir,
        str(next(self._t8n_call_counter)),
    )

BlockchainTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/blockchain.py

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

    pre: Alloc
    post: Alloc
    blocks: List[Block]
    genesis_environment: Environment = Field(default_factory=Environment)
    verify_sync: bool = False
    chain_id: int = 1

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

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

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

    @staticmethod
    def make_genesis(
        genesis_environment: Environment,
        pre: Alloc,
        fork: Fork,
    ) -> Tuple[Alloc, FixtureBlock]:
        """Create a genesis block from the blockchain test definition."""
        env = genesis_environment.set_fork_requirements(fork)
        assert env.withdrawals is None or len(env.withdrawals) == 0, (
            "withdrawals must be empty at genesis"
        )
        assert env.parent_beacon_block_root is None or env.parent_beacon_block_root == Hash(0), (
            "parent_beacon_block_root must be empty at genesis"
        )

        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation_blockchain()),
            pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")
        state_root = pre_alloc.state_root()
        genesis = FixtureHeader(
            parent_hash=0,
            ommers_hash=EmptyOmmersRoot,
            fee_recipient=0,
            state_root=state_root,
            transactions_trie=EmptyTrieRoot,
            receipts_root=EmptyTrieRoot,
            logs_bloom=0,
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=0,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data=b"\x00",
            prev_randao=0,
            nonce=0,
            base_fee_per_gas=env.base_fee_per_gas,
            blob_gas_used=env.blob_gas_used,
            excess_blob_gas=env.excess_blob_gas,
            withdrawals_root=(
                Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
            ),
            parent_beacon_block_root=env.parent_beacon_block_root,
            requests_hash=Requests() if fork.header_requests_required(0, 0) else None,
            fork=fork,
        )

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

    def generate_block_data(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Alloc,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | None, Alloc, Environment]:
        """Generate common block data for both make_fixture and make_hive_fixture."""
        if block.rlp and block.exception is not None:
            raise Exception(
                "test correctness: post-state cannot be verified if the "
                + "block's rlp is supplied and the block is not supposed "
                + "to produce an exception"
            )

        env = block.set_environment(previous_env)
        env = env.set_fork_requirements(fork)

        txs = [tx.with_signature_and_sender() for tx in block.txs]

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

        transition_tool_output = t8n.evaluate(
            alloc=previous_alloc,
            txs=txs,
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=fork.get_reward(env.number, env.timestamp),
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            slow_request=slow,
        )

        try:
            rejected_txs = verify_transactions(
                txs=txs,
                exception_mapper=t8n.exception_mapper,
                result=transition_tool_output.result,
            )
            verify_result(transition_tool_output.result, env)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(previous_alloc)
            pprint(transition_tool_output.alloc)
            raise e

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

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

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

        if block.header_verify is not None:
            # Verify the header after transition tool processing.
            block.header_verify.verify(header)

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

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

            requests_list = requests.requests_list

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

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

        return (
            header,
            txs,
            requests_list,
            transition_tool_output.alloc,
            env,
        )

    @staticmethod
    def network_info(fork: Fork, eips: Optional[List[int]] = None):
        """Return fixture network information for the fork & EIP/s."""
        return (
            "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
            if eips
            else fork.blockchain_test_network_name()
        )

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

    def make_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> BlockchainFixture:
        """Create a fixture from the blockchain test definition."""
        fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)

        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head = genesis.header.block_hash

        for block in self.blocks:
            if block.rlp is None:
                # This is the most common case, the RLP needs to be constructed
                # based on the transactions to be included in the block.
                # Set the environment according to the block to execute.
                header, txs, _, new_alloc, new_env = self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    eips=eips,
                    slow=slow,
                )
                fixture_block = FixtureBlockBase(
                    header=header,
                    txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                    ommers=[],
                    withdrawals=(
                        [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                        if new_env.withdrawals is not None
                        else None
                    ),
                    fork=fork,
                ).with_rlp(txs=txs)
                if block.exception is None:
                    fixture_blocks.append(fixture_block)
                    # Update env, alloc and last block hash for the next block.
                    alloc = new_alloc
                    env = apply_new_parent(new_env, header)
                    head = header.block_hash
                else:
                    fixture_blocks.append(
                        InvalidFixtureBlock(
                            rlp=fixture_block.rlp,
                            expect_exception=block.exception,
                            rlp_decoded=(
                                None
                                if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                                else fixture_block.without_rlp()
                            ),
                        ),
                    )
            else:
                assert block.exception is not None, (
                    "test correctness: if the block's rlp is hard-coded, "
                    + "the block is expected to produce an exception"
                )
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=block.rlp,
                        expect_exception=block.exception,
                    ),
                )

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )

        self.verify_post_state(t8n, t8n_state=alloc)
        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainFixture(
            fork=network_info,
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            post_state=alloc,
            config=FixtureConfig(
                fork=network_info,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

    def make_hive_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> BlockchainEngineFixture:
        """Create a hive fixture from the blocktest definition."""
        fixture_payloads: List[FixtureEngineNewPayload] = []

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)
        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head_hash = genesis.header.block_hash

        for block in self.blocks:
            header, txs, requests, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
                slow=slow,
            )
            if block.rlp is None:
                fixture_payloads.append(
                    FixtureEngineNewPayload.from_fixture_header(
                        fork=fork,
                        header=header,
                        transactions=txs,
                        withdrawals=new_env.withdrawals,
                        requests=requests,
                        validation_error=block.exception,
                        error_code=block.engine_api_error_code,
                    )
                )
                if block.exception is None:
                    alloc = new_alloc
                    env = apply_new_parent(env, header)
                    head_hash = header.block_hash

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )

        fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
        assert fcu_version is not None, (
            "A hive fixture was requested but no forkchoice update is defined."
            " The framework should never try to execute this test case."
        )

        self.verify_post_state(t8n, t8n_state=alloc)

        sync_payload: Optional[FixtureEngineNewPayload] = None
        if self.verify_sync:
            # Test is marked for syncing verification.
            assert genesis.header.block_hash != head_hash, (
                "Invalid payload tests negative test via sync is not supported yet."
            )

            # Most clients require the header to start the sync process, so we create an empty
            # block on top of the last block of the test to send it as new payload and trigger the
            # sync process.
            sync_header, _, requests, _, _ = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=Block(),
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
            )
            sync_payload = FixtureEngineNewPayload.from_fixture_header(
                fork=fork,
                header=sync_header,
                transactions=[],
                withdrawals=[],
                requests=requests,
                validation_error=None,
                error_code=None,
            )

        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainEngineFixture(
            fork=network_info,
            genesis=genesis.header,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre=pre,
            post_state=alloc,
            sync_payload=sync_payload,
            last_block_hash=head_hash,
            config=FixtureConfig(
                fork=network_info,
                chain_id=self.chain_id,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            ),
        )

    def generate(
        self,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        t8n.reset_traces()
        if fixture_format == BlockchainEngineFixture:
            return self.make_hive_fixture(t8n, fork, eips, slow=is_slow_test(request))
        elif fixture_format == BlockchainFixture:
            return self.make_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            txs: List[Transaction] = []
            for block in self.blocks:
                txs += block.txs
            return TransactionPost(
                transactions=txs,
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

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

Source code in src/ethereum_test_specs/blockchain.py

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

make_genesis(genesis_environment, pre, fork) staticmethod

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

@staticmethod
def make_genesis(
    genesis_environment: Environment,
    pre: Alloc,
    fork: Fork,
) -> Tuple[Alloc, FixtureBlock]:
    """Create a genesis block from the blockchain test definition."""
    env = genesis_environment.set_fork_requirements(fork)
    assert env.withdrawals is None or len(env.withdrawals) == 0, (
        "withdrawals must be empty at genesis"
    )
    assert env.parent_beacon_block_root is None or env.parent_beacon_block_root == Hash(0), (
        "parent_beacon_block_root must be empty at genesis"
    )

    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation_blockchain()),
        pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")
    state_root = pre_alloc.state_root()
    genesis = FixtureHeader(
        parent_hash=0,
        ommers_hash=EmptyOmmersRoot,
        fee_recipient=0,
        state_root=state_root,
        transactions_trie=EmptyTrieRoot,
        receipts_root=EmptyTrieRoot,
        logs_bloom=0,
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=0,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data=b"\x00",
        prev_randao=0,
        nonce=0,
        base_fee_per_gas=env.base_fee_per_gas,
        blob_gas_used=env.blob_gas_used,
        excess_blob_gas=env.excess_blob_gas,
        withdrawals_root=(
            Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
        ),
        parent_beacon_block_root=env.parent_beacon_block_root,
        requests_hash=Requests() if fork.header_requests_required(0, 0) else None,
        fork=fork,
    )

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

generate_block_data(t8n, fork, block, previous_env, previous_alloc, eips=None, slow=False)

Generate common block data for both make_fixture and make_hive_fixture.

Source code in src/ethereum_test_specs/blockchain.py

def generate_block_data(
    self,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Alloc,
    eips: Optional[List[int]] = None,
    slow: bool = False,
) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | None, Alloc, Environment]:
    """Generate common block data for both make_fixture and make_hive_fixture."""
    if block.rlp and block.exception is not None:
        raise Exception(
            "test correctness: post-state cannot be verified if the "
            + "block's rlp is supplied and the block is not supposed "
            + "to produce an exception"
        )

    env = block.set_environment(previous_env)
    env = env.set_fork_requirements(fork)

    txs = [tx.with_signature_and_sender() for tx in block.txs]

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

    transition_tool_output = t8n.evaluate(
        alloc=previous_alloc,
        txs=txs,
        env=env,
        fork=fork,
        chain_id=self.chain_id,
        reward=fork.get_reward(env.number, env.timestamp),
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        slow_request=slow,
    )

    try:
        rejected_txs = verify_transactions(
            txs=txs,
            exception_mapper=t8n.exception_mapper,
            result=transition_tool_output.result,
        )
        verify_result(transition_tool_output.result, env)
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(previous_alloc)
        pprint(transition_tool_output.alloc)
        raise e

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

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

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

    if block.header_verify is not None:
        # Verify the header after transition tool processing.
        block.header_verify.verify(header)

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

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

        requests_list = requests.requests_list

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

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

    return (
        header,
        txs,
        requests_list,
        transition_tool_output.alloc,
        env,
    )

network_info(fork, eips=None) staticmethod

Return fixture network information for the fork & EIP/s.

Source code in src/ethereum_test_specs/blockchain.py

@staticmethod
def network_info(fork: Fork, eips: Optional[List[int]] = None):
    """Return fixture network information for the fork & EIP/s."""
    return (
        "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
        if eips
        else fork.blockchain_test_network_name()
    )

verify_post_state(t8n, t8n_state, expected_state=None)

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

Source code in src/ethereum_test_specs/blockchain.py

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

make_fixture(t8n, fork, eips=None, slow=False)

Create a fixture from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

def make_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    slow: bool = False,
) -> BlockchainFixture:
    """Create a fixture from the blockchain test definition."""
    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)

    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head = genesis.header.block_hash

    for block in self.blocks:
        if block.rlp is None:
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            header, txs, _, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
                slow=slow,
            )
            fixture_block = FixtureBlockBase(
                header=header,
                txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                ommers=[],
                withdrawals=(
                    [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                    if new_env.withdrawals is not None
                    else None
                ),
                fork=fork,
            ).with_rlp(txs=txs)
            if block.exception is None:
                fixture_blocks.append(fixture_block)
                # Update env, alloc and last block hash for the next block.
                alloc = new_alloc
                env = apply_new_parent(new_env, header)
                head = header.block_hash
            else:
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=fixture_block.rlp,
                        expect_exception=block.exception,
                        rlp_decoded=(
                            None
                            if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                            else fixture_block.without_rlp()
                        ),
                    ),
                )
        else:
            assert block.exception is not None, (
                "test correctness: if the block's rlp is hard-coded, "
                + "the block is expected to produce an exception"
            )
            fixture_blocks.append(
                InvalidFixtureBlock(
                    rlp=block.rlp,
                    expect_exception=block.exception,
                ),
            )

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )

    self.verify_post_state(t8n, t8n_state=alloc)
    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainFixture(
        fork=network_info,
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        post_state=alloc,
        config=FixtureConfig(
            fork=network_info,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

make_hive_fixture(t8n, fork, eips=None, slow=False)

Create a hive fixture from the blocktest definition.

Source code in src/ethereum_test_specs/blockchain.py

def make_hive_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    slow: bool = False,
) -> BlockchainEngineFixture:
    """Create a hive fixture from the blocktest definition."""
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)
    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head_hash = genesis.header.block_hash

    for block in self.blocks:
        header, txs, requests, new_alloc, new_env = self.generate_block_data(
            t8n=t8n,
            fork=fork,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
            eips=eips,
            slow=slow,
        )
        if block.rlp is None:
            fixture_payloads.append(
                FixtureEngineNewPayload.from_fixture_header(
                    fork=fork,
                    header=header,
                    transactions=txs,
                    withdrawals=new_env.withdrawals,
                    requests=requests,
                    validation_error=block.exception,
                    error_code=block.engine_api_error_code,
                )
            )
            if block.exception is None:
                alloc = new_alloc
                env = apply_new_parent(env, header)
                head_hash = header.block_hash

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )

    fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
    assert fcu_version is not None, (
        "A hive fixture was requested but no forkchoice update is defined."
        " The framework should never try to execute this test case."
    )

    self.verify_post_state(t8n, t8n_state=alloc)

    sync_payload: Optional[FixtureEngineNewPayload] = None
    if self.verify_sync:
        # Test is marked for syncing verification.
        assert genesis.header.block_hash != head_hash, (
            "Invalid payload tests negative test via sync is not supported yet."
        )

        # Most clients require the header to start the sync process, so we create an empty
        # block on top of the last block of the test to send it as new payload and trigger the
        # sync process.
        sync_header, _, requests, _, _ = self.generate_block_data(
            t8n=t8n,
            fork=fork,
            block=Block(),
            previous_env=env,
            previous_alloc=alloc,
            eips=eips,
        )
        sync_payload = FixtureEngineNewPayload.from_fixture_header(
            fork=fork,
            header=sync_header,
            transactions=[],
            withdrawals=[],
            requests=requests,
            validation_error=None,
            error_code=None,
        )

    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainEngineFixture(
        fork=network_info,
        genesis=genesis.header,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre=pre,
        post_state=alloc,
        sync_payload=sync_payload,
        last_block_hash=head_hash,
        config=FixtureConfig(
            fork=network_info,
            chain_id=self.chain_id,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
        ),
    )

generate(request, t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/blockchain.py

def generate(
    self,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    t8n.reset_traces()
    if fixture_format == BlockchainEngineFixture:
        return self.make_hive_fixture(t8n, fork, eips, slow=is_slow_test(request))
    elif fixture_format == BlockchainFixture:
        return self.make_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/blockchain.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        txs: List[Transaction] = []
        for block in self.blocks:
            txs += block.txs
        return TransactionPost(
            transactions=txs,
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

EOFStateTest

Bases: EOFTest, Transaction

Filler type that generates an EOF test for container validation, and also tests the container during runtime using a state test (and blockchain test).

In the state or blockchain test, the container is first deployed to the pre-allocation and then a transaction is sent to the deployed container.

Container deployment/validation is not tested like in the EOFTest unless the container under test is an initcode container.

All fields from ethereum_test_types.Transaction are available for use in the test.

Source code in src/ethereum_test_specs/eof.py

class EOFStateTest(EOFTest, Transaction):
    """
    Filler type that generates an EOF test for container validation, and also tests the container
    during runtime using a state test (and blockchain test).

    In the state or blockchain test, the container is first deployed to the pre-allocation and
    then a transaction is sent to the deployed container.

    Container deployment/validation is **not** tested like in the `EOFTest` unless the container
    under test is an initcode container.

    All fields from `ethereum_test_types.Transaction` are available for use in the test.
    """

    gas_limit: HexNumber = Field(HexNumber(10_000_000), serialization_alias="gas")
    """
    Gas limit for the transaction that deploys the container.
    """
    tx_sender_funding_amount: int = 1_000_000_000_000_000_000_000
    """
    Amount of funds to send to the sender EOA before the transaction.
    """
    env: Environment = Field(default_factory=Environment)
    """
    Environment object that is used during State Test generation.
    """
    container_post: Account = Field(default_factory=Account)
    """
    Account object used to verify the container post state.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"eof_{fixture_format.format_name}",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"eof_{execute_format.format_name}",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """Workaround for pytest parameter name."""
        return "eof_state_test"

    def model_post_init(self, __context):
        """Prepare the transaction parameters required to fill the test."""
        assert self.pre is not None, "pre must be set to generate a StateTest."

        EOFTest.model_post_init(self, __context)

        self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
        if self.post is None:
            self.post = Alloc()

        if self.expect_exception is not None:  # Invalid EOF
            self.to = None  # Make EIP-7698 create transaction
            self.data = Bytes(
                bytes(self.container) + self.data
            )  # by concatenating container and tx data.

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.created_contract] = None  # Expect failure.
        elif self.container_kind == ContainerKind.INITCODE:
            self.to = None  # Make EIP-7698 create transaction
            self.data = Bytes(
                bytes(self.container) + self.data
            )  # by concatenating container and tx data.

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.created_contract] = self.container_post  # Successful.
        else:
            self.to = self.pre.deploy_contract(code=self.container)

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.to] = self.container_post

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        assert self.pre is not None, "pre must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."

        return StateTest(
            pre=self.pre,
            tx=self,
            env=self.env,
            post=self.post,
            t8n_dump_dir=self.t8n_dump_dir,
        )

    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        if fixture_format == EOFFixture:
            if Bytes(self.container) in existing_tests:
                # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
                # state fixtures with the same data.
                pytest.skip(f"Duplicate EOF container on EOFStateTest: {request.node.nodeid}")
            return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
        elif fixture_format in StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).generate(
                request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )

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

gas_limit: HexNumber = Field(HexNumber(10000000), serialization_alias='gas') class-attribute instance-attribute

Gas limit for the transaction that deploys the container.

tx_sender_funding_amount: int = 1000000000000000000000 class-attribute instance-attribute

Amount of funds to send to the sender EOA before the transaction.

env: Environment = Field(default_factory=Environment) class-attribute instance-attribute

Environment object that is used during State Test generation.

container_post: Account = Field(default_factory=Account) class-attribute instance-attribute

Account object used to verify the container post state.

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """Workaround for pytest parameter name."""
    return "eof_state_test"

model_post_init(__context)

Prepare the transaction parameters required to fill the test.

Source code in src/ethereum_test_specs/eof.py

def model_post_init(self, __context):
    """Prepare the transaction parameters required to fill the test."""
    assert self.pre is not None, "pre must be set to generate a StateTest."

    EOFTest.model_post_init(self, __context)

    self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
    if self.post is None:
        self.post = Alloc()

    if self.expect_exception is not None:  # Invalid EOF
        self.to = None  # Make EIP-7698 create transaction
        self.data = Bytes(
            bytes(self.container) + self.data
        )  # by concatenating container and tx data.

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.created_contract] = None  # Expect failure.
    elif self.container_kind == ContainerKind.INITCODE:
        self.to = None  # Make EIP-7698 create transaction
        self.data = Bytes(
            bytes(self.container) + self.data
        )  # by concatenating container and tx data.

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.created_contract] = self.container_post  # Successful.
    else:
        self.to = self.pre.deploy_contract(code=self.container)

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.to] = self.container_post

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    assert self.pre is not None, "pre must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."

    return StateTest(
        pre=self.pre,
        tx=self,
        env=self.env,
        post=self.post,
        t8n_dump_dir=self.t8n_dump_dir,
    )

generate(*, request, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py

def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    if fixture_format == EOFFixture:
        if Bytes(self.container) in existing_tests:
            # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
            # state fixtures with the same data.
            pytest.skip(f"Duplicate EOF container on EOFStateTest: {request.node.nodeid}")
        return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
    elif fixture_format in StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).generate(
            request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )

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

EOFTest

Bases: BaseTest

Filler type that generates a test for EOF container validation.

A state test is also automatically generated where the container is wrapped in a contract-creating transaction to test deployment/validation on the instantiated blockchain.

Source code in src/ethereum_test_specs/eof.py

class EOFTest(BaseTest):
    """
    Filler type that generates a test for EOF container validation.

    A state test is also automatically generated where the container is wrapped in a
    contract-creating transaction to test deployment/validation on the instantiated blockchain.
    """

    container: Container
    """
    EOF container that will be tested for validity.

    The only supported type at the moment is `ethereum_test_types.eof.v1.Container`.

    If an invalid container needs to be tested, and it cannot be generated using the
    Container class features, the `raw_bytes` field can be used to provide the raw
    container bytes.
    """
    expect_exception: EOFExceptionInstanceOrList | None = None
    """
    Expected exception that the container should raise when parsed by an EOF parser.

    Can be a single exception or a list of exceptions that the container is expected to raise,
    in which case the test will pass if any of the exceptions are raised.

    The list of supported exceptions can be found in the `ethereum_test_exceptions.EOFException`
    class.
    """
    container_kind: ContainerKind = ContainerKind.RUNTIME
    """
    Container kind type that the container should be treated as.

    The container kind can be one of the following:
    - `ContainerKind.INITCODE`: The container is an initcode container.
    - `ContainerKind.RUNTIME`: The container is a runtime container.

    The default value is `ContainerKind.RUNTIME`.
    """
    deployed_container: Container | None = None
    """
    To be used when the container is an initcode container and the expected deployed container is
    known.

    The value is only used when a State Test is generated from this EOF test to set the expected
    deployed container that should be found in the post state.

    If this field is not set, and the container is valid:
      - If the container kind is `ContainerKind.RUNTIME`, the deployed container is assumed to be
        the container itself, and an initcode container that wraps the container is generated
        automatically.
      - If the container kind is `ContainerKind.INITCODE`, `model_post_init` will attempt to infer
        the deployed container from the sections of the init-container, and the first
        container-type section will be used. An error will be raised if the deployed container
        cannot be inferred.

    If the value is set to `None`, it is assumed that the container is invalid and the test will
    expect that no contract is created.

    It is considered an error if:
      - The `deployed_container` field is set and the `container_kind` field is not set to
        `ContainerKind.INITCODE`.
      - The `deployed_container` field is set and the `expect_exception` is not `None`.

    The deployed container is **not** executed at any point during the EOF validation test nor
    the generated State Test. For container runtime testing use the `EOFStateTest` class.
    """
    pre: Alloc | None = None
    """
    Pre alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and should otherwise be left unset.
    """
    post: Alloc | None = None
    """
    Post alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and is normally not set by the user.
    """
    sender: EOA | None = None
    """
    Sender EOA object that is used during State Test generation.

    This field is automatically set by the `model_post_init` method and should otherwise be left
    unset.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_eof_test",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"{execute_format.format_name}_from_eof_test",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "eof_test_only": "Only generate an EOF test fixture",
    }

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

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """Workaround for pytest parameter name."""
        return "eof_test"

    def model_post_init(self, __context):
        """Prepare the test exception based on the container."""
        if self.container.validity_error is not None:
            if self.expect_exception is not None:
                assert self.expect_exception == self.container.validity_error, (
                    f"Container validity error {self.container.validity_error} "
                    f"does not match expected exception {self.expect_exception}."
                )
            self.expect_exception = self.container.validity_error
            assert self.deployed_container is None, (
                "deployed_container must be None for invalid containers."
            )
        if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
            if (
                "kind" in self.container.model_fields_set
                and "container_kind" in self.model_fields_set
            ):
                assert self.container.kind == self.container_kind, (
                    f"Container kind type {str(self.container.kind)} "
                    f"does not match test {self.container_kind}."
                )
            elif "kind" in self.container.model_fields_set:
                self.container_kind = self.container.kind
            elif "container_kind" in self.model_fields_set:
                self.container.kind = self.container_kind

        assert self.pre is not None, "pre must be set to generate a StateTest."
        self.sender = self.pre.fund_eoa()
        if self.post is None:
            self.post = Alloc()

    def make_eof_test_fixture(
        self,
        *,
        request: pytest.FixtureRequest,
        fork: Fork,
        eips: Optional[List[int]],
    ) -> EOFFixture:
        """Generate the EOF test fixture."""
        container_bytes = Bytes(self.container)
        if container_bytes in existing_tests:
            pytest.fail(
                f"Duplicate EOF test: {container_bytes}, "
                f"existing test: {existing_tests[container_bytes]}"
            )
        existing_tests[container_bytes] = request.node.nodeid
        vectors = [
            Vector(
                code=container_bytes,
                container_kind=self.container_kind,
                results={
                    fork.blockchain_test_network_name(): Result(
                        exception=self.expect_exception,
                        valid=self.expect_exception is None,
                    ),
                },
            )
        ]
        fixture = EOFFixture(vectors=dict(enumerate(vectors)))
        try:
            eof_parse = EOFParse()
        except FileNotFoundError as e:
            warnings.warn(
                f"{e} Skipping EOF fixture verification. Fixtures may be invalid!", stacklevel=2
            )
            return fixture

        for _, vector in fixture.vectors.items():
            expected_result = vector.results.get(fork.blockchain_test_network_name())
            if expected_result is None:
                raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
            args = []
            if vector.container_kind == ContainerKind.INITCODE:
                args.append("--initcode")
            result = eof_parse.run(*args, input_value=str(vector.code))
            self.verify_result(result, expected_result, vector.code)

        return fixture

    def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
        """Check that the reported exception string matches the expected error."""
        parser = EvmoneExceptionMapper()
        actual_message = result.stdout.strip()
        actual_exception = parser.message_to_exception(actual_message)

        if expected_result.exception is None:
            if "OK" in actual_message:
                return
            else:
                raise UnexpectedEOFExceptionError(
                    code=code, got=f"{actual_exception} ({actual_message})"
                )
        else:
            expected_string = to_pipe_str(expected_result.exception)
            print(expected_string)
            print(actual_exception)
            if "OK" in actual_message:
                raise ExpectedEOFExceptionError(
                    code=code,
                    expected=f"{expected_string}",
                )
            elif actual_exception in expected_result.exception:
                return
            else:
                raise EOFExceptionMismatchError(
                    code=code,
                    expected=f"{expected_string}",
                    got=f"{actual_exception} ({actual_message})",
                )

    def generate_eof_contract_create_transaction(self) -> Transaction:
        """Generate a transaction that creates a contract."""
        assert self.sender is not None, "sender must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."

        initcode: Container
        deployed_container: Container | Bytes | None = None
        if self.container_kind == ContainerKind.INITCODE:
            initcode = self.container
            if "deployed_container" in self.model_fields_set:
                # In the case of an initcontainer where we know the deployed container,
                # we can use the initcontainer as-is.
                deployed_container = self.deployed_container
            elif self.expect_exception is None:
                # We have a valid init-container, but we don't know the deployed container.
                # Try to infer the deployed container from the sections of the init-container.
                assert self.container.raw_bytes is None, (
                    "deployed_container must be set for initcode containers with raw_bytes."
                )
                for section in self.container.sections:
                    if section.kind == SectionKind.CONTAINER:
                        deployed_container = section.data
                        break

                assert deployed_container is not None, (
                    "Unable to infer deployed container for init-container. "
                    "Use field `deployed_container` to set the expected deployed container."
                )
        else:
            assert self.deployed_container is None, (
                "deployed_container must be None for runtime containers."
            )
            initcode = Container(
                sections=[
                    Section.Code(Op.RETURNCONTRACT[0](0, 0)),
                    Section.Container(self.container),
                ]
            )
            deployed_container = self.container

        tx = Transaction(
            sender=self.sender,
            to=None,
            gas_limit=10_000_000,
            data=initcode,
        )

        if self.expect_exception is not None or deployed_container is None:
            self.post[tx.created_contract] = None
        else:
            self.post[tx.created_contract] = Account(
                code=deployed_container,
            )
        return tx

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        return StateTest(
            pre=self.pre,
            tx=self.generate_eof_contract_create_transaction(),
            env=Environment(),
            post=self.post,
            t8n_dump_dir=self.t8n_dump_dir,
        )

    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        if fixture_format == EOFFixture:
            return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
        elif fixture_format in StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).generate(
                request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )
        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            return self.generate_state_test(fork).execute(
                fork=fork, execute_format=execute_format, eips=eips
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

container: Container instance-attribute

EOF container that will be tested for validity.

The only supported type at the moment is ethereum_test_types.eof.v1.Container.

If an invalid container needs to be tested, and it cannot be generated using the Container class features, the raw_bytes field can be used to provide the raw container bytes.

expect_exception: EOFExceptionInstanceOrList | None = None class-attribute instance-attribute

Expected exception that the container should raise when parsed by an EOF parser.

Can be a single exception or a list of exceptions that the container is expected to raise, in which case the test will pass if any of the exceptions are raised.

The list of supported exceptions can be found in the ethereum_test_exceptions.EOFException class.

container_kind: ContainerKind = ContainerKind.RUNTIME class-attribute instance-attribute

Container kind type that the container should be treated as.

The container kind can be one of the following: - ContainerKind.INITCODE: The container is an initcode container. - ContainerKind.RUNTIME: The container is a runtime container.

The default value is ContainerKind.RUNTIME.

deployed_container: Container | None = None class-attribute instance-attribute

To be used when the container is an initcode container and the expected deployed container is known.

The value is only used when a State Test is generated from this EOF test to set the expected deployed container that should be found in the post state.

If this field is not set, and the container is valid: - If the container kind is ContainerKind.RUNTIME, the deployed container is assumed to be the container itself, and an initcode container that wraps the container is generated automatically. - If the container kind is ContainerKind.INITCODE, model_post_init will attempt to infer the deployed container from the sections of the init-container, and the first container-type section will be used. An error will be raised if the deployed container cannot be inferred.

If the value is set to None, it is assumed that the container is invalid and the test will expect that no contract is created.

It is considered an error if

• The deployed_container field is set and the container_kind field is not set to ContainerKind.INITCODE.
• The deployed_container field is set and the expect_exception is not None.

The deployed container is not executed at any point during the EOF validation test nor the generated State Test. For container runtime testing use the EOFStateTest class.

pre: Alloc | None = None class-attribute instance-attribute

Pre alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and should otherwise be left unset.

post: Alloc | None = None class-attribute instance-attribute

Post alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and is normally not set by the user.

sender: EOA | None = None class-attribute instance-attribute

Sender EOA object that is used during State Test generation.

This field is automatically set by the model_post_init method and should otherwise be left unset.

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

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

Source code in src/ethereum_test_specs/eof.py

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

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """Workaround for pytest parameter name."""
    return "eof_test"

model_post_init(__context)

Prepare the test exception based on the container.

Source code in src/ethereum_test_specs/eof.py

def model_post_init(self, __context):
    """Prepare the test exception based on the container."""
    if self.container.validity_error is not None:
        if self.expect_exception is not None:
            assert self.expect_exception == self.container.validity_error, (
                f"Container validity error {self.container.validity_error} "
                f"does not match expected exception {self.expect_exception}."
            )
        self.expect_exception = self.container.validity_error
        assert self.deployed_container is None, (
            "deployed_container must be None for invalid containers."
        )
    if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
        if (
            "kind" in self.container.model_fields_set
            and "container_kind" in self.model_fields_set
        ):
            assert self.container.kind == self.container_kind, (
                f"Container kind type {str(self.container.kind)} "
                f"does not match test {self.container_kind}."
            )
        elif "kind" in self.container.model_fields_set:
            self.container_kind = self.container.kind
        elif "container_kind" in self.model_fields_set:
            self.container.kind = self.container_kind

    assert self.pre is not None, "pre must be set to generate a StateTest."
    self.sender = self.pre.fund_eoa()
    if self.post is None:
        self.post = Alloc()

make_eof_test_fixture(*, request, fork, eips)

Generate the EOF test fixture.

Source code in src/ethereum_test_specs/eof.py

def make_eof_test_fixture(
    self,
    *,
    request: pytest.FixtureRequest,
    fork: Fork,
    eips: Optional[List[int]],
) -> EOFFixture:
    """Generate the EOF test fixture."""
    container_bytes = Bytes(self.container)
    if container_bytes in existing_tests:
        pytest.fail(
            f"Duplicate EOF test: {container_bytes}, "
            f"existing test: {existing_tests[container_bytes]}"
        )
    existing_tests[container_bytes] = request.node.nodeid
    vectors = [
        Vector(
            code=container_bytes,
            container_kind=self.container_kind,
            results={
                fork.blockchain_test_network_name(): Result(
                    exception=self.expect_exception,
                    valid=self.expect_exception is None,
                ),
            },
        )
    ]
    fixture = EOFFixture(vectors=dict(enumerate(vectors)))
    try:
        eof_parse = EOFParse()
    except FileNotFoundError as e:
        warnings.warn(
            f"{e} Skipping EOF fixture verification. Fixtures may be invalid!", stacklevel=2
        )
        return fixture

    for _, vector in fixture.vectors.items():
        expected_result = vector.results.get(fork.blockchain_test_network_name())
        if expected_result is None:
            raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
        args = []
        if vector.container_kind == ContainerKind.INITCODE:
            args.append("--initcode")
        result = eof_parse.run(*args, input_value=str(vector.code))
        self.verify_result(result, expected_result, vector.code)

    return fixture

verify_result(result, expected_result, code)

Check that the reported exception string matches the expected error.

Source code in src/ethereum_test_specs/eof.py

def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
    """Check that the reported exception string matches the expected error."""
    parser = EvmoneExceptionMapper()
    actual_message = result.stdout.strip()
    actual_exception = parser.message_to_exception(actual_message)

    if expected_result.exception is None:
        if "OK" in actual_message:
            return
        else:
            raise UnexpectedEOFExceptionError(
                code=code, got=f"{actual_exception} ({actual_message})"
            )
    else:
        expected_string = to_pipe_str(expected_result.exception)
        print(expected_string)
        print(actual_exception)
        if "OK" in actual_message:
            raise ExpectedEOFExceptionError(
                code=code,
                expected=f"{expected_string}",
            )
        elif actual_exception in expected_result.exception:
            return
        else:
            raise EOFExceptionMismatchError(
                code=code,
                expected=f"{expected_string}",
                got=f"{actual_exception} ({actual_message})",
            )

generate_eof_contract_create_transaction()

Generate a transaction that creates a contract.

Source code in src/ethereum_test_specs/eof.py

def generate_eof_contract_create_transaction(self) -> Transaction:
    """Generate a transaction that creates a contract."""
    assert self.sender is not None, "sender must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."

    initcode: Container
    deployed_container: Container | Bytes | None = None
    if self.container_kind == ContainerKind.INITCODE:
        initcode = self.container
        if "deployed_container" in self.model_fields_set:
            # In the case of an initcontainer where we know the deployed container,
            # we can use the initcontainer as-is.
            deployed_container = self.deployed_container
        elif self.expect_exception is None:
            # We have a valid init-container, but we don't know the deployed container.
            # Try to infer the deployed container from the sections of the init-container.
            assert self.container.raw_bytes is None, (
                "deployed_container must be set for initcode containers with raw_bytes."
            )
            for section in self.container.sections:
                if section.kind == SectionKind.CONTAINER:
                    deployed_container = section.data
                    break

            assert deployed_container is not None, (
                "Unable to infer deployed container for init-container. "
                "Use field `deployed_container` to set the expected deployed container."
            )
    else:
        assert self.deployed_container is None, (
            "deployed_container must be None for runtime containers."
        )
        initcode = Container(
            sections=[
                Section.Code(Op.RETURNCONTRACT[0](0, 0)),
                Section.Container(self.container),
            ]
        )
        deployed_container = self.container

    tx = Transaction(
        sender=self.sender,
        to=None,
        gas_limit=10_000_000,
        data=initcode,
    )

    if self.expect_exception is not None or deployed_container is None:
        self.post[tx.created_contract] = None
    else:
        self.post[tx.created_contract] = Account(
            code=deployed_container,
        )
    return tx

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    return StateTest(
        pre=self.pre,
        tx=self.generate_eof_contract_create_transaction(),
        env=Environment(),
        post=self.post,
        t8n_dump_dir=self.t8n_dump_dir,
    )

generate(*, request, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py

def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    if fixture_format == EOFFixture:
        return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
    elif fixture_format in StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).generate(
            request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )
    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/eof.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        return self.generate_state_test(fork).execute(
            fork=fork, execute_format=execute_format, eips=eips
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

StateTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/state.py

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

    env: Environment = Field(default_factory=Environment)
    pre: Alloc
    post: Alloc
    tx: Transaction
    engine_api_error_code: Optional[EngineAPIError] = None
    blockchain_test_header_verify: Optional[Header] = None
    blockchain_test_rlp_modifier: Optional[Header] = None
    chain_id: int = 1

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        StateFixture,
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_state_test",
        )
        for fixture_format in BlockchainTest.supported_fixture_formats
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

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

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

    def _generate_blockchain_genesis_environment(self, *, fork: Fork) -> Environment:
        """Generate the genesis environment for the BlockchainTest formatted test."""
        assert self.env.number >= 1, (
            "genesis block number cannot be negative, set state test env.number to 1"
        )

        # Modify values to the proper values for the genesis block
        # TODO: All of this can be moved to a new method in `Fork`
        updated_values: Dict[str, Any] = {
            "withdrawals": None,
            "parent_beacon_block_root": None,
            "number": self.env.number - 1,
        }
        if self.env.excess_blob_gas:
            # The excess blob gas environment value means the value of the context (block header)
            # where the transaction is executed. In a blockchain test, we need to indirectly
            # set the excess blob gas by setting the excess blob gas of the genesis block
            # to the expected value plus the TARGET_BLOB_GAS_PER_BLOCK, which is the value
            # that will be subtracted from the excess blob gas when the first block is mined.
            updated_values["excess_blob_gas"] = self.env.excess_blob_gas + (
                fork.target_blobs_per_block() * fork.blob_gas_per_blob()
            )

        return self.env.copy(**updated_values)

    def _generate_blockchain_blocks(self) -> List[Block]:
        """Generate the single block that represents this state test in a BlockchainTest format."""
        return [
            Block(
                number=self.env.number,
                timestamp=self.env.timestamp,
                fee_recipient=self.env.fee_recipient,
                difficulty=self.env.difficulty,
                gas_limit=self.env.gas_limit,
                extra_data=self.env.extra_data,
                withdrawals=self.env.withdrawals,
                parent_beacon_block_root=self.env.parent_beacon_block_root,
                txs=[self.tx],
                ommers=[],
                exception=self.tx.error,
                header_verify=self.blockchain_test_header_verify,
                rlp_modifier=self.blockchain_test_rlp_modifier,
            )
        ]

    def generate_blockchain_test(self, *, fork: Fork) -> BlockchainTest:
        """Generate a BlockchainTest fixture from this StateTest fixture."""
        return BlockchainTest(
            genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
            pre=self.pre,
            post=self.post,
            blocks=self._generate_blockchain_blocks(),
            t8n_dump_dir=self.t8n_dump_dir,
        )

    def make_state_test_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> StateFixture:
        """Create a fixture from the state test definition."""
        # We can't generate a state test fixture that names a transition fork,
        # so we get the fork at the block number and timestamp of the state test
        fork = fork.fork_at(self.env.number, self.env.timestamp)

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

        transition_tool_output = t8n.evaluate(
            alloc=pre_alloc,
            txs=[tx],
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            state_test=True,
            slow_request=slow,
        )

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

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

        return StateFixture(
            env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
            pre=pre_alloc,
            post={
                fork.blockchain_test_network_name(): [
                    FixtureForkPost(
                        state_root=transition_tool_output.result.state_root,
                        logs_hash=transition_tool_output.result.logs_hash,
                        tx_bytes=tx.rlp,
                        expect_exception=tx.error,
                        state=transition_tool_output.alloc,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
            config=FixtureConfig(
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

    def generate(
        self,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        if fixture_format in BlockchainTest.supported_fixture_formats:
            return self.generate_blockchain_test(fork=fork).generate(
                request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )
        elif fixture_format == StateFixture:
            return self.make_state_test_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            return TransactionPost(
                transactions=[self.tx],
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

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

Source code in src/ethereum_test_specs/state.py

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

generate_blockchain_test(*, fork)

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in src/ethereum_test_specs/state.py

def generate_blockchain_test(self, *, fork: Fork) -> BlockchainTest:
    """Generate a BlockchainTest fixture from this StateTest fixture."""
    return BlockchainTest(
        genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
        pre=self.pre,
        post=self.post,
        blocks=self._generate_blockchain_blocks(),
        t8n_dump_dir=self.t8n_dump_dir,
    )

make_state_test_fixture(t8n, fork, eips=None, slow=False)

Create a fixture from the state test definition.

Source code in src/ethereum_test_specs/state.py

def make_state_test_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    slow: bool = False,
) -> StateFixture:
    """Create a fixture from the state test definition."""
    # We can't generate a state test fixture that names a transition fork,
    # so we get the fork at the block number and timestamp of the state test
    fork = fork.fork_at(self.env.number, self.env.timestamp)

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

    transition_tool_output = t8n.evaluate(
        alloc=pre_alloc,
        txs=[tx],
        env=env,
        fork=fork,
        chain_id=self.chain_id,
        reward=0,  # Reward on state tests is always zero
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        state_test=True,
        slow_request=slow,
    )

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

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

    return StateFixture(
        env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
        pre=pre_alloc,
        post={
            fork.blockchain_test_network_name(): [
                FixtureForkPost(
                    state_root=transition_tool_output.result.state_root,
                    logs_hash=transition_tool_output.result.logs_hash,
                    tx_bytes=tx.rlp,
                    expect_exception=tx.error,
                    state=transition_tool_output.alloc,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
        config=FixtureConfig(
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

generate(request, t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/state.py

def generate(
    self,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    if fixture_format in BlockchainTest.supported_fixture_formats:
        return self.generate_blockchain_test(fork=fork).generate(
            request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )
    elif fixture_format == StateFixture:
        return self.make_state_test_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/state.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        return TransactionPost(
            transactions=[self.tx],
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

TransactionTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/transaction.py

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

    tx: Transaction
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        TransactionFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

    def make_transaction_test_fixture(
        self,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> TransactionFixture:
        """Create a fixture from the transaction test definition."""
        if self.tx.error is not None:
            result = FixtureResult(
                exception=self.tx.error,
                hash=None,
                intrinsic_gas=0,
                sender=None,
            )
        else:
            intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
            intrinsic_gas = intrinsic_gas_cost_calculator(
                calldata=self.tx.data,
                contract_creation=self.tx.to is None,
                access_list=self.tx.access_list,
                authorization_list_or_count=self.tx.authorization_list,
            )
            result = FixtureResult(
                exception=None,
                hash=self.tx.hash,
                intrinsic_gas=intrinsic_gas,
                sender=self.tx.sender,
            )

        return TransactionFixture(
            result={
                fork.blockchain_test_network_name(): result,
            },
            transaction=self.tx.with_signature_and_sender().rlp,
        )

    def generate(
        self,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the TransactionTest fixture."""
        if fixture_format == TransactionFixture:
            return self.make_transaction_test_fixture(fork, eips)

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Execute the transaction test by sending it to the live network."""
        if execute_format == TransactionPost:
            return TransactionPost(
                transactions=[self.tx],
                post={},
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

make_transaction_test_fixture(fork, eips=None)

Create a fixture from the transaction test definition.

Source code in src/ethereum_test_specs/transaction.py

def make_transaction_test_fixture(
    self,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> TransactionFixture:
    """Create a fixture from the transaction test definition."""
    if self.tx.error is not None:
        result = FixtureResult(
            exception=self.tx.error,
            hash=None,
            intrinsic_gas=0,
            sender=None,
        )
    else:
        intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
        intrinsic_gas = intrinsic_gas_cost_calculator(
            calldata=self.tx.data,
            contract_creation=self.tx.to is None,
            access_list=self.tx.access_list,
            authorization_list_or_count=self.tx.authorization_list,
        )
        result = FixtureResult(
            exception=None,
            hash=self.tx.hash,
            intrinsic_gas=intrinsic_gas,
            sender=self.tx.sender,
        )

    return TransactionFixture(
        result={
            fork.blockchain_test_network_name(): result,
        },
        transaction=self.tx.with_signature_and_sender().rlp,
    )

generate(request, t8n, fork, fixture_format, eips=None)

Generate the TransactionTest fixture.

Source code in src/ethereum_test_specs/transaction.py

def generate(
    self,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the TransactionTest fixture."""
    if fixture_format == TransactionFixture:
        return self.make_transaction_test_fixture(fork, eips)

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

execute(*, fork, execute_format, eips=None)

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

Source code in src/ethereum_test_specs/transaction.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Execute the transaction test by sending it to the live network."""
    if execute_format == TransactionPost:
        return TransactionPost(
            transactions=[self.tx],
            post={},
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

Block

Bases: Header

Block type used to describe block properties in test specs.

Source code in src/ethereum_test_specs/blockchain.py

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

    rlp: Bytes | None = None
    """
    If set, blockchain test will skip generating the block and will pass this value directly to
    the Fixture.

    Only meant to be used to simulate blocks with bad formats, and therefore
    requires the block to produce an exception.
    """
    header_verify: Header | None = None
    """
    If set, the block header will be verified against the specified values.
    """
    rlp_modifier: Header | None = None
    """
    An RLP modifying header which values would be used to override the ones
    returned by the `ethereum_clis.TransitionTool`.
    """
    exception: (
        List[TransactionException | BlockException] | TransactionException | BlockException | None
    ) = None
    """
    If set, the block is expected to be rejected by the client.
    """
    engine_api_error_code: EngineAPIError | None = None
    """
    If set, the block is expected to produce an error response from the Engine API.
    """
    txs: List[Transaction] = Field(default_factory=list)
    """
    List of transactions included in the block.
    """
    ommers: List[Header] | None = None
    """
    List of ommer headers included in the block.
    """
    withdrawals: List[Withdrawal] | None = None
    """
    List of withdrawals to perform for this block.
    """
    requests: List[Bytes] | None = None
    """
    Custom list of requests to embed in this block.
    """
    expected_post_state: Alloc | None = None
    """
    Post state for verification after block execution in BlockchainTest
    """

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

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

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

        return env.copy(**new_env_values)

rlp: Bytes | None = None class-attribute instance-attribute

If set, blockchain test will skip generating the block and will pass this value directly to the Fixture.

Only meant to be used to simulate blocks with bad formats, and therefore requires the block to produce an exception.

header_verify: Header | None = None class-attribute instance-attribute

If set, the block header will be verified against the specified values.

rlp_modifier: Header | None = None class-attribute instance-attribute

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

exception: List[TransactionException | BlockException] | TransactionException | BlockException | None = None class-attribute instance-attribute

If set, the block is expected to be rejected by the client.

engine_api_error_code: EngineAPIError | None = None class-attribute instance-attribute

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

txs: List[Transaction] = Field(default_factory=list) class-attribute instance-attribute

List of transactions included in the block.

ommers: List[Header] | None = None class-attribute instance-attribute

List of ommer headers included in the block.

withdrawals: List[Withdrawal] | None = None class-attribute instance-attribute

List of withdrawals to perform for this block.

requests: List[Bytes] | None = None class-attribute instance-attribute

Custom list of requests to embed in this block.

expected_post_state: Alloc | None = None class-attribute instance-attribute

Post state for verification after block execution in BlockchainTest

set_environment(env)

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

Source code in src/ethereum_test_specs/blockchain.py

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

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

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

    return env.copy(**new_env_values)

Header

Bases: CamelModel

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

Source code in src/ethereum_test_specs/blockchain.py

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

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

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

    This can be used in a test to explicitly skip a field in a block's RLP encoding.
    included in the (json) output when the model is serialized. For example:
    ```
    header_modifier = Header(
        excess_blob_gas=Header.REMOVE_FIELD,
    )
    block = Block(
        timestamp=TIMESTAMP,
        rlp_modifier=header_modifier,
        exception=BlockException.INCORRECT_BLOCK_FORMAT,
        engine_api_error_code=EngineAPIError.InvalidParams,
    )
    ```
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        # explicitly set Removable items to None so they are not included in the serialization
        # (in combination with exclude_None=True in model.dump()).
        json_encoders={
            Removable: lambda x: None,
        },
    )

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

    def apply(self, target: FixtureHeader) -> FixtureHeader:
        """Produce a fixture header copy with the set values from the modifier."""
        return target.copy(
            **{
                k: (v if v is not Header.REMOVE_FIELD else None)
                for k, v in self.model_dump(exclude_none=True).items()
            }
        )

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

REMOVE_FIELD: Removable = Removable() class-attribute

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

EMPTY_FIELD: Removable = Removable() class-attribute

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

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

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

validate_withdrawals_root(value) classmethod

Convert a list of withdrawals into the withdrawals root hash.

Source code in src/ethereum_test_specs/blockchain.py

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

apply(target)

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

Source code in src/ethereum_test_specs/blockchain.py

def apply(self, target: FixtureHeader) -> FixtureHeader:
    """Produce a fixture header copy with the set values from the modifier."""
    return target.copy(
        **{
            k: (v if v is not Header.REMOVE_FIELD else None)
            for k, v in self.model_dump(exclude_none=True).items()
        }
    )

verify(target)

Verify that the header fields from self are as expected.

Source code in src/ethereum_test_specs/blockchain.py

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

CalldataCase

Bases: Case

Small helper class to represent a single case whose condition depends on the value of the contract's calldata in a Switch case statement.

By default the calldata is read from position zero, but this can be overridden using position.

The condition is generated automatically based on the value (and optionally position) and may not be set directly.

Source code in src/ethereum_test_tools/code/generators.py

class CalldataCase(Case):
    """
    Small helper class to represent a single case whose condition depends
    on the value of the contract's calldata in a Switch case statement.

    By default the calldata is read from position zero, but this can be
    overridden using `position`.

    The `condition` is generated automatically based on the `value` (and
    optionally `position`) and may not be set directly.
    """

    def __init__(self, value: int | str | Bytecode, position: int = 0, **kwargs):
        """Generate the condition base on `value` and `position`."""
        condition = Op.EQ(Op.CALLDATALOAD(position), value)
        super().__init__(condition=condition, **kwargs)

__init__(value, position=0, **kwargs)

Generate the condition base on value and position.

Source code in src/ethereum_test_tools/code/generators.py

def __init__(self, value: int | str | Bytecode, position: int = 0, **kwargs):
    """Generate the condition base on `value` and `position`."""
    condition = Op.EQ(Op.CALLDATALOAD(position), value)
    super().__init__(condition=condition, **kwargs)

Case dataclass

Small helper class to represent a single, generic case in a Switch cases list.

Source code in src/ethereum_test_tools/code/generators.py

@dataclass(kw_only=True)
class Case:
    """
    Small helper class to represent a single, generic case in a `Switch` cases
    list.
    """

    condition: Bytecode | Op
    action: Bytecode | Op
    terminating: bool | None = None

    @property
    def is_terminating(self) -> bool:
        """Returns whether the case is terminating."""
        return self.terminating if self.terminating is not None else self.action.terminating

is_terminating: bool property

Returns whether the case is terminating.

CodeGasMeasure

Bases: Bytecode

Helper class used to generate bytecode that measures gas usage of a bytecode, taking into account and subtracting any extra overhead gas costs required to execute. By default, the result gas calculation is saved to storage key 0.

Source code in src/ethereum_test_tools/code/generators.py

class CodeGasMeasure(Bytecode):
    """
    Helper class used to generate bytecode that measures gas usage of a
    bytecode, taking into account and subtracting any extra overhead gas costs
    required to execute.
    By default, the result gas calculation is saved to storage key 0.
    """

    code: Bytecode
    """
    Bytecode to be executed to measure the gas usage.
    """
    overhead_cost: int
    """
    Extra gas cost to be subtracted from extra operations.
    """
    extra_stack_items: int
    """
    Extra stack items that remain at the end of the execution.
    To be considered when subtracting the value of the previous GAS operation,
    and to be popped at the end of the execution.
    """
    sstore_key: int
    """
    Storage key to save the gas used.
    """

    def __new__(
        cls,
        *,
        code: Bytecode,
        overhead_cost: int = 0,
        extra_stack_items: int = 0,
        sstore_key: int = 0,
        stop: bool = True,
    ):
        """Assemble the bytecode that measures gas usage."""
        res = Op.GAS + code + Op.GAS
        # We need to swap and pop for each extra stack item that remained from
        # the execution of the code
        res += (Op.SWAP1 + Op.POP) * extra_stack_items
        res += (
            Op.SWAP1
            + Op.SUB
            + Op.PUSH1(overhead_cost + 2)
            + Op.SWAP1
            + Op.SUB
            + Op.PUSH1(sstore_key)
            + Op.SSTORE
        )
        if stop:
            res += Op.STOP

        instance = super().__new__(cls, res)
        instance.code = code
        instance.overhead_cost = overhead_cost
        instance.extra_stack_items = extra_stack_items
        instance.sstore_key = sstore_key
        return instance

code: Bytecode instance-attribute

Bytecode to be executed to measure the gas usage.

overhead_cost: int instance-attribute

Extra gas cost to be subtracted from extra operations.

extra_stack_items: int instance-attribute

Extra stack items that remain at the end of the execution. To be considered when subtracting the value of the previous GAS operation, and to be popped at the end of the execution.

sstore_key: int instance-attribute

Storage key to save the gas used.

__new__(*, code, overhead_cost=0, extra_stack_items=0, sstore_key=0, stop=True)

Assemble the bytecode that measures gas usage.

Source code in src/ethereum_test_tools/code/generators.py

def __new__(
    cls,
    *,
    code: Bytecode,
    overhead_cost: int = 0,
    extra_stack_items: int = 0,
    sstore_key: int = 0,
    stop: bool = True,
):
    """Assemble the bytecode that measures gas usage."""
    res = Op.GAS + code + Op.GAS
    # We need to swap and pop for each extra stack item that remained from
    # the execution of the code
    res += (Op.SWAP1 + Op.POP) * extra_stack_items
    res += (
        Op.SWAP1
        + Op.SUB
        + Op.PUSH1(overhead_cost + 2)
        + Op.SWAP1
        + Op.SUB
        + Op.PUSH1(sstore_key)
        + Op.SSTORE
    )
    if stop:
        res += Op.STOP

    instance = super().__new__(cls, res)
    instance.code = code
    instance.overhead_cost = overhead_cost
    instance.extra_stack_items = extra_stack_items
    instance.sstore_key = sstore_key
    return instance

Conditional

Bases: Bytecode

Helper class used to generate conditional bytecode.

Source code in src/ethereum_test_tools/code/generators.py

class Conditional(Bytecode):
    """Helper class used to generate conditional bytecode."""

    def __new__(
        cls,
        *,
        condition: Bytecode | Op,
        if_true: Bytecode | Op | None = None,
        if_false: Bytecode | Op | None = None,
        evm_code_type: EVMCodeType = EVMCodeType.LEGACY,
    ):
        """
        Assemble the conditional bytecode by generating the necessary jump and
        jumpdest opcodes surrounding the condition and the two possible execution
        paths.

        In the future, PC usage should be replaced by using RJUMP and RJUMPI
        """
        if if_true is None:
            if_true = Bytecode()
        if if_false is None:
            if_false = Bytecode()

        if evm_code_type == EVMCodeType.LEGACY:
            # First we append a jumpdest to the start of the true branch
            if_true = Op.JUMPDEST + if_true

            # Then we append the unconditional jump to the end of the false branch, used to skip
            # the true branch
            if_false += Op.JUMP(Op.ADD(Op.PC, len(if_true) + 3))

            # Then we need to do the conditional jump by skipping the false branch
            condition = Op.JUMPI(Op.ADD(Op.PC, len(if_false) + 3), condition)

            # Finally we append the condition, false and true branches, plus the jumpdest at the
            # very end
            bytecode = condition + if_false + if_true + Op.JUMPDEST

        elif evm_code_type == EVMCodeType.EOF_V1:
            if not if_false.terminating:
                if_false += Op.RJUMP[len(if_true)]
            condition = Op.RJUMPI[len(if_false)](condition)

            # Finally we append the condition, false and true branches
            bytecode = condition + if_false + if_true

        return super().__new__(cls, bytecode)

__new__(*, condition, if_true=None, if_false=None, evm_code_type=EVMCodeType.LEGACY)

Assemble the conditional bytecode by generating the necessary jump and jumpdest opcodes surrounding the condition and the two possible execution paths.

In the future, PC usage should be replaced by using RJUMP and RJUMPI

Source code in src/ethereum_test_tools/code/generators.py

def __new__(
    cls,
    *,
    condition: Bytecode | Op,
    if_true: Bytecode | Op | None = None,
    if_false: Bytecode | Op | None = None,
    evm_code_type: EVMCodeType = EVMCodeType.LEGACY,
):
    """
    Assemble the conditional bytecode by generating the necessary jump and
    jumpdest opcodes surrounding the condition and the two possible execution
    paths.

    In the future, PC usage should be replaced by using RJUMP and RJUMPI
    """
    if if_true is None:
        if_true = Bytecode()
    if if_false is None:
        if_false = Bytecode()

    if evm_code_type == EVMCodeType.LEGACY:
        # First we append a jumpdest to the start of the true branch
        if_true = Op.JUMPDEST + if_true

        # Then we append the unconditional jump to the end of the false branch, used to skip
        # the true branch
        if_false += Op.JUMP(Op.ADD(Op.PC, len(if_true) + 3))

        # Then we need to do the conditional jump by skipping the false branch
        condition = Op.JUMPI(Op.ADD(Op.PC, len(if_false) + 3), condition)

        # Finally we append the condition, false and true branches, plus the jumpdest at the
        # very end
        bytecode = condition + if_false + if_true + Op.JUMPDEST

    elif evm_code_type == EVMCodeType.EOF_V1:
        if not if_false.terminating:
            if_false += Op.RJUMP[len(if_true)]
        condition = Op.RJUMPI[len(if_false)](condition)

        # Finally we append the condition, false and true branches
        bytecode = condition + if_false + if_true

    return super().__new__(cls, bytecode)

Initcode

Bases: Bytecode

Helper class used to generate initcode for the specified deployment code.

The execution gas cost of the initcode is calculated, and also the deployment gas costs for the deployed code.

The initcode can be padded to a certain length if necessary, which does not affect the deployed code.

Other costs such as the CREATE2 hashing costs or the initcode_word_cost of EIP-3860 are not taken into account by any of these calculated costs.

Source code in src/ethereum_test_tools/code/generators.py

class Initcode(Bytecode):
    """
    Helper class used to generate initcode for the specified deployment code.

    The execution gas cost of the initcode is calculated, and also the
    deployment gas costs for the deployed code.

    The initcode can be padded to a certain length if necessary, which
    does not affect the deployed code.

    Other costs such as the CREATE2 hashing costs or the initcode_word_cost
    of EIP-3860 are *not* taken into account by any of these calculated
    costs.
    """

    deploy_code: SupportsBytes | Bytes
    """
    Bytecode to be deployed by the initcode.
    """
    execution_gas: int
    """
    Gas cost of executing the initcode, without considering deployment gas
    costs.
    """
    deployment_gas: int
    """
    Gas cost of deploying the cost, subtracted after initcode execution,
    """

    def __new__(
        cls,
        *,
        deploy_code: SupportsBytes | Bytes | None = None,
        initcode_length: int | None = None,
        initcode_prefix: Bytecode | None = None,
        initcode_prefix_execution_gas: int = 0,
        padding_byte: int = 0x00,
        name: str = "",
    ):
        """
        Generate legacy initcode that inits a contract with the specified code.
        The initcode can be padded to a specified length for testing purposes.
        """
        if deploy_code is None:
            deploy_code = Bytecode()
        if initcode_prefix is None:
            initcode_prefix = Bytecode()

        initcode = initcode_prefix
        code_length = len(bytes(deploy_code))
        execution_gas = initcode_prefix_execution_gas

        # PUSH2: length=<bytecode length>
        initcode += Op.PUSH2(code_length)
        execution_gas = 3

        # PUSH1: offset=0
        initcode += Op.PUSH1(0)
        execution_gas += 3

        # DUP2
        initcode += Op.DUP2
        execution_gas += 3

        # PUSH1: initcode_length=11 + len(initcode_prefix_bytes) (constant)
        no_prefix_length = 0x0B
        assert no_prefix_length + len(initcode_prefix) <= 0xFF, "initcode prefix too long"
        initcode += Op.PUSH1(no_prefix_length + len(initcode_prefix))
        execution_gas += 3

        # DUP3
        initcode += Op.DUP3
        execution_gas += 3

        # CODECOPY: destinationOffset=0, offset=0, length
        initcode += Op.CODECOPY
        execution_gas += (
            3
            + (3 * ceiling_division(code_length, 32))
            + (3 * code_length)
            + ((code_length * code_length) // 512)
        )

        # RETURN: offset=0, length
        initcode += Op.RETURN
        execution_gas += 0

        initcode_plus_deploy_code = bytes(initcode) + bytes(deploy_code)
        padding_bytes = bytes()

        if initcode_length is not None:
            assert initcode_length >= len(initcode_plus_deploy_code), (
                "specified invalid length for initcode"
            )

            padding_bytes = bytes(
                [padding_byte] * (initcode_length - len(initcode_plus_deploy_code))
            )

        initcode_bytes = initcode_plus_deploy_code + padding_bytes
        instance = super().__new__(
            cls,
            initcode_bytes,
            popped_stack_items=initcode.popped_stack_items,
            pushed_stack_items=initcode.pushed_stack_items,
            max_stack_height=initcode.max_stack_height,
            min_stack_height=initcode.min_stack_height,
        )
        instance._name_ = name
        instance.deploy_code = deploy_code
        instance.execution_gas = execution_gas
        instance.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(bytes(instance.deploy_code))

        return instance

deploy_code: SupportsBytes | Bytes instance-attribute

Bytecode to be deployed by the initcode.

execution_gas: int instance-attribute

Gas cost of executing the initcode, without considering deployment gas costs.

deployment_gas: int instance-attribute

Gas cost of deploying the cost, subtracted after initcode execution,

__new__(*, deploy_code=None, initcode_length=None, initcode_prefix=None, initcode_prefix_execution_gas=0, padding_byte=0, name='')

Generate legacy initcode that inits a contract with the specified code. The initcode can be padded to a specified length for testing purposes.

Source code in src/ethereum_test_tools/code/generators.py

def __new__(
    cls,
    *,
    deploy_code: SupportsBytes | Bytes | None = None,
    initcode_length: int | None = None,
    initcode_prefix: Bytecode | None = None,
    initcode_prefix_execution_gas: int = 0,
    padding_byte: int = 0x00,
    name: str = "",
):
    """
    Generate legacy initcode that inits a contract with the specified code.
    The initcode can be padded to a specified length for testing purposes.
    """
    if deploy_code is None:
        deploy_code = Bytecode()
    if initcode_prefix is None:
        initcode_prefix = Bytecode()

    initcode = initcode_prefix
    code_length = len(bytes(deploy_code))
    execution_gas = initcode_prefix_execution_gas

    # PUSH2: length=<bytecode length>
    initcode += Op.PUSH2(code_length)
    execution_gas = 3

    # PUSH1: offset=0
    initcode += Op.PUSH1(0)
    execution_gas += 3

    # DUP2
    initcode += Op.DUP2
    execution_gas += 3

    # PUSH1: initcode_length=11 + len(initcode_prefix_bytes) (constant)
    no_prefix_length = 0x0B
    assert no_prefix_length + len(initcode_prefix) <= 0xFF, "initcode prefix too long"
    initcode += Op.PUSH1(no_prefix_length + len(initcode_prefix))
    execution_gas += 3

    # DUP3
    initcode += Op.DUP3
    execution_gas += 3

    # CODECOPY: destinationOffset=0, offset=0, length
    initcode += Op.CODECOPY
    execution_gas += (
        3
        + (3 * ceiling_division(code_length, 32))
        + (3 * code_length)
        + ((code_length * code_length) // 512)
    )

    # RETURN: offset=0, length
    initcode += Op.RETURN
    execution_gas += 0

    initcode_plus_deploy_code = bytes(initcode) + bytes(deploy_code)
    padding_bytes = bytes()

    if initcode_length is not None:
        assert initcode_length >= len(initcode_plus_deploy_code), (
            "specified invalid length for initcode"
        )

        padding_bytes = bytes(
            [padding_byte] * (initcode_length - len(initcode_plus_deploy_code))
        )

    initcode_bytes = initcode_plus_deploy_code + padding_bytes
    instance = super().__new__(
        cls,
        initcode_bytes,
        popped_stack_items=initcode.popped_stack_items,
        pushed_stack_items=initcode.pushed_stack_items,
        max_stack_height=initcode.max_stack_height,
        min_stack_height=initcode.min_stack_height,
    )
    instance._name_ = name
    instance.deploy_code = deploy_code
    instance.execution_gas = execution_gas
    instance.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(bytes(instance.deploy_code))

    return instance

Switch

Bases: Bytecode

Helper class used to generate switch-case expressions in EVM bytecode.

Switch-case behavior

• If no condition is met in the list of BytecodeCases conditions, the default_action bytecode is executed.
• If multiple conditions are met, the action from the first valid condition is the only one executed.
• There is no fall through; it is not possible to execute multiple actions.

Source code in src/ethereum_test_tools/code/generators.py

class Switch(Bytecode):
    """
    Helper class used to generate switch-case expressions in EVM bytecode.

    Switch-case behavior:
        - If no condition is met in the list of BytecodeCases conditions,
            the `default_action` bytecode is executed.
        - If multiple conditions are met, the action from the first valid
            condition is the only one executed.
        - There is no fall through; it is not possible to execute multiple
            actions.
    """

    default_action: Bytecode | Op | None
    """
    The default bytecode to execute; if no condition is met, this bytecode is
    executed.
    """

    cases: List[Case]
    """
    A list of Cases: The first element with a condition that
    evaluates to a non-zero value is the one that is executed.
    """

    evm_code_type: EVMCodeType
    """
    The EVM code type to use for the switch-case bytecode.
    """

    def __new__(
        cls,
        *,
        default_action: Bytecode | Op | None = None,
        cases: List[Case],
        evm_code_type: EVMCodeType = EVMCodeType.LEGACY,
    ):
        """
        Assemble the bytecode by looping over the list of cases and adding
        the necessary [R]JUMPI and JUMPDEST opcodes in order to replicate
        switch-case behavior.
        """
        # The length required to jump over subsequent actions to the final JUMPDEST at the end
        # of the switch-case block:
        # - add 6 per case for the length of the JUMPDEST and JUMP(ADD(PC, action_jump_length))
        #   bytecode
        # - add 3 to the total to account for this action's JUMP; the PC within the call
        #   requires a "correction" of 3.

        bytecode = Bytecode()

        # All conditions get pre-pended to this bytecode; if none are met, we reach the default
        if evm_code_type == EVMCodeType.LEGACY:
            action_jump_length = sum(len(case.action) + 6 for case in cases) + 3
            bytecode = default_action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
            # The length required to jump over the default action and its JUMP bytecode
            condition_jump_length = len(bytecode) + 3
        elif evm_code_type == EVMCodeType.EOF_V1:
            action_jump_length = sum(
                len(case.action) + (len(Op.RJUMP[0]) if not case.is_terminating else 0)
                for case in cases
                # On not terminating cases, we need to add 3 bytes for the RJUMP
            )
            bytecode = default_action + Op.RJUMP[action_jump_length]
            # The length required to jump over the default action and its JUMP bytecode
            condition_jump_length = len(bytecode)

        # Reversed: first case in the list has priority; it will become the outer-most onion layer.
        # We build up layers around the default_action, after 1 iteration of the loop, a simplified
        # representation of the bytecode is:
        #
        #  JUMPI(case[n-1].condition)
        #  + default_action + JUMP()
        #  + JUMPDEST + case[n-1].action + JUMP()
        #
        # and after n=len(cases) iterations:
        #
        #  JUMPI(case[0].condition)
        #  + JUMPI(case[1].condition)
        #    ...
        #  + JUMPI(case[n-1].condition)
        #  + default_action + JUMP()
        #  + JUMPDEST + case[n-1].action + JUMP()
        #  + ...
        #  + JUMPDEST + case[1].action + JUMP()
        #  + JUMPDEST + case[0].action + JUMP()
        #
        for case in reversed(cases):
            action = case.action
            if evm_code_type == EVMCodeType.LEGACY:
                action_jump_length -= len(action) + 6
                action = Op.JUMPDEST + action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
                condition = Op.JUMPI(Op.ADD(Op.PC, condition_jump_length), case.condition)
            elif evm_code_type == EVMCodeType.EOF_V1:
                action_jump_length -= len(action) + (
                    len(Op.RJUMP[0]) if not case.is_terminating else 0
                )
                if not case.is_terminating:
                    action += Op.RJUMP[action_jump_length]
                condition = Op.RJUMPI[condition_jump_length](case.condition)
            # wrap the current case around the onion as its next layer
            bytecode = condition + bytecode + action
            condition_jump_length += len(condition) + len(action)

        bytecode += Op.JUMPDEST

        instance = super().__new__(cls, bytecode)
        instance.default_action = default_action
        instance.cases = cases
        return instance

default_action: Bytecode | Op | None instance-attribute

The default bytecode to execute; if no condition is met, this bytecode is executed.

cases: List[Case] instance-attribute

A list of Cases: The first element with a condition that evaluates to a non-zero value is the one that is executed.

evm_code_type: EVMCodeType instance-attribute

The EVM code type to use for the switch-case bytecode.

__new__(*, default_action=None, cases, evm_code_type=EVMCodeType.LEGACY)

Assemble the bytecode by looping over the list of cases and adding the necessary [R]JUMPI and JUMPDEST opcodes in order to replicate switch-case behavior.

Source code in src/ethereum_test_tools/code/generators.py

def __new__(
    cls,
    *,
    default_action: Bytecode | Op | None = None,
    cases: List[Case],
    evm_code_type: EVMCodeType = EVMCodeType.LEGACY,
):
    """
    Assemble the bytecode by looping over the list of cases and adding
    the necessary [R]JUMPI and JUMPDEST opcodes in order to replicate
    switch-case behavior.
    """
    # The length required to jump over subsequent actions to the final JUMPDEST at the end
    # of the switch-case block:
    # - add 6 per case for the length of the JUMPDEST and JUMP(ADD(PC, action_jump_length))
    #   bytecode
    # - add 3 to the total to account for this action's JUMP; the PC within the call
    #   requires a "correction" of 3.

    bytecode = Bytecode()

    # All conditions get pre-pended to this bytecode; if none are met, we reach the default
    if evm_code_type == EVMCodeType.LEGACY:
        action_jump_length = sum(len(case.action) + 6 for case in cases) + 3
        bytecode = default_action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
        # The length required to jump over the default action and its JUMP bytecode
        condition_jump_length = len(bytecode) + 3
    elif evm_code_type == EVMCodeType.EOF_V1:
        action_jump_length = sum(
            len(case.action) + (len(Op.RJUMP[0]) if not case.is_terminating else 0)
            for case in cases
            # On not terminating cases, we need to add 3 bytes for the RJUMP
        )
        bytecode = default_action + Op.RJUMP[action_jump_length]
        # The length required to jump over the default action and its JUMP bytecode
        condition_jump_length = len(bytecode)

    # Reversed: first case in the list has priority; it will become the outer-most onion layer.
    # We build up layers around the default_action, after 1 iteration of the loop, a simplified
    # representation of the bytecode is:
    #
    #  JUMPI(case[n-1].condition)
    #  + default_action + JUMP()
    #  + JUMPDEST + case[n-1].action + JUMP()
    #
    # and after n=len(cases) iterations:
    #
    #  JUMPI(case[0].condition)
    #  + JUMPI(case[1].condition)
    #    ...
    #  + JUMPI(case[n-1].condition)
    #  + default_action + JUMP()
    #  + JUMPDEST + case[n-1].action + JUMP()
    #  + ...
    #  + JUMPDEST + case[1].action + JUMP()
    #  + JUMPDEST + case[0].action + JUMP()
    #
    for case in reversed(cases):
        action = case.action
        if evm_code_type == EVMCodeType.LEGACY:
            action_jump_length -= len(action) + 6
            action = Op.JUMPDEST + action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
            condition = Op.JUMPI(Op.ADD(Op.PC, condition_jump_length), case.condition)
        elif evm_code_type == EVMCodeType.EOF_V1:
            action_jump_length -= len(action) + (
                len(Op.RJUMP[0]) if not case.is_terminating else 0
            )
            if not case.is_terminating:
                action += Op.RJUMP[action_jump_length]
            condition = Op.RJUMPI[condition_jump_length](case.condition)
        # wrap the current case around the onion as its next layer
        bytecode = condition + bytecode + action
        condition_jump_length += len(condition) + len(action)

    bytecode += Op.JUMPDEST

    instance = super().__new__(cls, bytecode)
    instance.default_action = default_action
    instance.cases = cases
    return instance

Yul

Bases: Bytecode

Yul compiler. Compiles Yul source code into bytecode.

Source code in src/ethereum_test_tools/code/yul.py

class Yul(Bytecode):
    """
    Yul compiler.
    Compiles Yul source code into bytecode.
    """

    source: str
    evm_version: str | None

    def __new__(
        cls,
        source: str,
        fork: Optional[Fork] = None,
        binary: Optional[Path | str] = None,
    ):
        """Compile Yul source code into bytecode."""
        solc = Solc(binary)
        evm_version = fork.solc_name() if fork else None

        solc_args = ("--evm-version", evm_version) if evm_version else ()

        result = solc.run(*solc_args, *DEFAULT_SOLC_ARGS, input_value=source)

        if result.returncode:
            stderr_lines = result.stderr.splitlines()
            stderr_message = "\n".join(line.strip() for line in stderr_lines)
            raise Exception(f"failed to compile yul source:\n{stderr_message[7:]}")

        lines = result.stdout.splitlines()

        hex_str = lines[lines.index("Binary representation:") + 1]

        bytecode = bytes.fromhex(hex_str)
        instance = super().__new__(
            cls,
            bytecode,
            popped_stack_items=0,
            pushed_stack_items=0,
        )
        instance.source = source
        instance.evm_version = evm_version
        return instance

__new__(source, fork=None, binary=None)

Compile Yul source code into bytecode.

Source code in src/ethereum_test_tools/code/yul.py

def __new__(
    cls,
    source: str,
    fork: Optional[Fork] = None,
    binary: Optional[Path | str] = None,
):
    """Compile Yul source code into bytecode."""
    solc = Solc(binary)
    evm_version = fork.solc_name() if fork else None

    solc_args = ("--evm-version", evm_version) if evm_version else ()

    result = solc.run(*solc_args, *DEFAULT_SOLC_ARGS, input_value=source)

    if result.returncode:
        stderr_lines = result.stderr.splitlines()
        stderr_message = "\n".join(line.strip() for line in stderr_lines)
        raise Exception(f"failed to compile yul source:\n{stderr_message[7:]}")

    lines = result.stdout.splitlines()

    hex_str = lines[lines.index("Binary representation:") + 1]

    bytecode = bytes.fromhex(hex_str)
    instance = super().__new__(
        cls,
        bytecode,
        popped_stack_items=0,
        pushed_stack_items=0,
    )
    instance.source = source
    instance.evm_version = evm_version
    return instance

DeploymentTestType

Bases: Enum

Represents the type of deployment test.

Source code in src/ethereum_test_tools/utility/generators.py

class DeploymentTestType(Enum):
    """Represents the type of deployment test."""

    DEPLOY_BEFORE_FORK = "deploy_before_fork"
    DEPLOY_AFTER_FORK = "deploy_after_fork"

generate_system_contract_deploy_test(*, fork, tx_json_path, expected_deploy_address, expected_system_contract_storage=None)

Generate a test that verifies the correct deployment of a system contract.

Generates four test cases:

                                | before/after fork | has balance |

------------------------------------|-------------------|-------------| deploy_before_fork-nonzero_balance| before | True | deploy_before_fork-zero_balance | before | False | deploy_after_fork-nonzero_balance | after | True | deploy_after_fork-zero_balance | after | False |

where has balance refers to whether the contract address has a non-zero balance before deployment, or not.

Parameters:

Name	Type	Description	Default
fork	Fork	The fork to test.	required
tx_json_path	Path	Path to the JSON file with the transaction to deploy the system contract. Providing a JSON file is useful to copy-paste the transaction from the EIP.	required
expected_deploy_address	Address	The expected address of the deployed contract.	required
expected_system_contract_storage	Dict | None	The expected storage of the system contract.	None

Source code in src/ethereum_test_tools/utility/generators.py

def generate_system_contract_deploy_test(
    *,
    fork: Fork,
    tx_json_path: Path,
    expected_deploy_address: Address,
    expected_system_contract_storage: Dict | None = None,
):
    """
    Generate a test that verifies the correct deployment of a system contract.

    Generates four test cases:

                                        | before/after fork | has balance |
    ------------------------------------|-------------------|-------------|
    `deploy_before_fork-nonzero_balance`| before            | True        |
    `deploy_before_fork-zero_balance`   | before            | False       |
    `deploy_after_fork-nonzero_balance` | after             | True        |
    `deploy_after_fork-zero_balance`    | after             | False       |

    where `has balance` refers to whether the contract address has a non-zero balance before
    deployment, or not.

    Args:
        fork (Fork): The fork to test.
        tx_json_path (Path): Path to the JSON file with the transaction to deploy the system
            contract.
            Providing a JSON file is useful to copy-paste the transaction from the EIP.
        expected_deploy_address (Address): The expected address of the deployed contract.
        expected_system_contract_storage (Dict | None): The expected storage of the system
            contract.

    """
    with open(tx_json_path, mode="r") as f:
        tx_json = json.loads(f.read())
    if "gasLimit" not in tx_json and "gas" in tx_json:
        tx_json["gasLimit"] = tx_json["gas"]
        del tx_json["gas"]
    if "protected" not in tx_json:
        tx_json["protected"] = False
    deploy_tx = Transaction.model_validate(tx_json).with_signature_and_sender()  # type: ignore
    gas_price = deploy_tx.gas_price
    assert gas_price is not None
    deployer_required_balance = deploy_tx.gas_limit * gas_price
    deployer_address = deploy_tx.sender
    if "hash" in tx_json:
        assert deploy_tx.hash == Hash(tx_json["hash"])
    if "sender" in tx_json:
        assert deploy_tx.sender == Address(tx_json["sender"])

    def decorator(func: SystemContractDeployTestFunction):
        @pytest.mark.parametrize(
            "has_balance",
            [
                pytest.param(ContractAddressHasBalance.NONZERO_BALANCE),
                pytest.param(ContractAddressHasBalance.ZERO_BALANCE),
            ],
            ids=lambda x: x.name.lower(),
        )
        @pytest.mark.parametrize(
            "test_type",
            [
                pytest.param(DeploymentTestType.DEPLOY_BEFORE_FORK),
                pytest.param(DeploymentTestType.DEPLOY_AFTER_FORK),
            ],
            ids=lambda x: x.name.lower(),
        )
        @pytest.mark.execute(pytest.mark.skip(reason="modifies pre-alloc"))
        @pytest.mark.valid_at_transition_to(fork.name())
        def wrapper(
            blockchain_test: BlockchainTestFiller,
            has_balance: ContractAddressHasBalance,
            pre: Alloc,
            test_type: DeploymentTestType,
            fork: Fork,
        ):
            assert deployer_address is not None
            assert deploy_tx.created_contract == expected_deploy_address
            blocks: List[Block] = []

            if test_type == DeploymentTestType.DEPLOY_BEFORE_FORK:
                blocks = [
                    Block(  # Deployment block
                        txs=[deploy_tx],
                        timestamp=14_999,
                    ),
                    Block(  # Empty block on fork
                        txs=[],
                        timestamp=15_000,
                    ),
                ]
            elif test_type == DeploymentTestType.DEPLOY_AFTER_FORK:
                blocks = [
                    Block(  # Empty block on fork
                        txs=[],
                        timestamp=15_000,
                    ),
                    Block(  # Deployment block
                        txs=[deploy_tx],
                        timestamp=15_001,
                    ),
                ]
            balance = 1 if has_balance == ContractAddressHasBalance.NONZERO_BALANCE else 0
            pre[expected_deploy_address] = Account(
                code=b"",  # Remove the code that is automatically allocated on the fork
                nonce=0,
                balance=balance,
            )
            pre[deployer_address] = Account(
                balance=deployer_required_balance,
            )

            expected_deploy_address_int = int.from_bytes(expected_deploy_address, "big")

            post = Alloc()
            fork_pre_allocation = fork.pre_allocation_blockchain()
            assert expected_deploy_address_int in fork_pre_allocation
            expected_code = fork_pre_allocation[expected_deploy_address_int]["code"]
            # Note: balance check is omitted; it may be modified by the underlying, decorated test
            if expected_system_contract_storage is None:
                post[expected_deploy_address] = Account(
                    code=expected_code,
                    nonce=1,
                )
            else:
                post[expected_deploy_address] = Account(
                    storage=expected_system_contract_storage,
                    code=expected_code,
                    nonce=1,
                )
            post[deployer_address] = Account(
                nonce=1,
            )

            # Extra blocks (if any) returned by the decorated function to add after the
            # contract is deployed.
            blocks += list(func(fork=fork, pre=pre, post=post, test_type=test_type))

            blockchain_test(
                pre=pre,
                blocks=blocks,
                post=post,
            )

        wrapper.__name__ = func.__name__  # type: ignore
        wrapper.__doc__ = func.__doc__  # type: ignore

        return wrapper

    return decorator

extend_with_defaults(defaults, cases, **parametrize_kwargs)

Extend test cases with default parameter values.

This utility function extends test case parameters by adding default values from the defaults dictionary to each case in the cases list. If a case already specifies a value for a parameter, its default is ignored.

This function is particularly useful in scenarios where you want to define a common set of default values but allow individual test cases to override them as needed.

The function returns a dictionary that can be directly unpacked and passed to the @pytest.mark.parametrize decorator.

Parameters:

Name	Type	Description	Default
defaults	Dict[str, Any]	A dictionary of default parameter names and their values. These values will be added to each case unless the case already defines a value for each parameter.	required
cases	List[ParameterSet]	A list of pytest.param objects representing different test cases. Its first argument must be a dictionary defining parameter names and values.	required
parametrize_kwargs	Any	Additional keyword arguments to be passed to @pytest.mark.parametrize. These arguments are not modified by this function and are passed through unchanged.	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: A dictionary with the following structure: argnames: A list of parameter names. argvalues: A list of test cases with modified parameter values. parametrize_kwargs: Additional keyword arguments passed through unchanged.

Example

@pytest.mark.parametrize(**extend_with_defaults(
    defaults=dict(
        min_value=0,  # default minimum value is 0
        max_value=100,  # default maximum value is 100
        average=50,  # default average value is 50
    ),
    cases=[
        pytest.param(
            dict(),  # use default values
            id='default_case',
        ),
        pytest.param(
            dict(min_value=10),  # override with min_value=10
            id='min_value_10',
        ),
        pytest.param(
            dict(max_value=200),  # override with max_value=200
            id='max_value_200',
        ),
        pytest.param(
            dict(min_value=-10, max_value=50),  # override both min_value
            # and max_value
            id='min_-10_max_50',
        ),
        pytest.param(
            dict(min_value=20, max_value=80, average=50),  # all defaults
            # are overridden
            id="min_20_max_80_avg_50",
        ),
        pytest.param(
            dict(min_value=100, max_value=0),  # invalid range
            id='invalid_range',
            marks=pytest.mark.xfail(reason='invalid range'),
        )
    ],
))
def test_range(min_value, max_value, average):
    assert min_value <= max_value
    assert min_value <= average <= max_value

The above test will execute with the following sets of parameters:

"default_case": {"min_value": 0, "max_value": 100, "average": 50}
"min_value_10": {"min_value": 10, "max_value": 100, "average": 50}
"max_value_200": {"min_value": 0, "max_value": 200, "average": 50}
"min_-10_max_50": {"min_value": -10, "max_value": 50, "average": 50}
"min_20_max_80_avg_50": {"min_value": 20, "max_value": 80, "average": 50}
"invalid_range": {"min_value": 100, "max_value": 0, "average": 50}  # expected to fail

Notes

• Each case in cases must contain exactly one value, which is a dictionary of parameter values.
• The function performs an in-place update of the cases list, so the original cases list is modified.

Source code in src/ethereum_test_tools/utility/pytest.py

def extend_with_defaults(
    defaults: Dict[str, Any], cases: List[ParameterSet], **parametrize_kwargs: Any
) -> Dict[str, Any]:
    """
    Extend test cases with default parameter values.

    This utility function extends test case parameters by adding default values
    from the `defaults` dictionary to each case in the `cases` list. If a case
    already specifies a value for a parameter, its default is ignored.

    This function is particularly useful in scenarios where you want to define
    a common set of default values but allow individual test cases to override
    them as needed.

    The function returns a dictionary that can be directly unpacked and passed
    to the `@pytest.mark.parametrize` decorator.

    Args:
        defaults (Dict[str, Any]): A dictionary of default parameter names and
            their values. These values will be added to each case unless the case
            already defines a value for each parameter.
        cases (List[ParameterSet]): A list of `pytest.param` objects representing
            different test cases. Its first argument must be a dictionary defining
            parameter names and values.
        parametrize_kwargs (Any): Additional keyword arguments to be passed to
            `@pytest.mark.parametrize`. These arguments are not modified by this
            function and are passed through unchanged.

    Returns:
        Dict[str, Any]: A dictionary with the following structure:
            `argnames`: A list of parameter names.
            `argvalues`: A list of test cases with modified parameter values.
            `parametrize_kwargs`: Additional keyword arguments passed through unchanged.


    Example:
        ```python
        @pytest.mark.parametrize(**extend_with_defaults(
            defaults=dict(
                min_value=0,  # default minimum value is 0
                max_value=100,  # default maximum value is 100
                average=50,  # default average value is 50
            ),
            cases=[
                pytest.param(
                    dict(),  # use default values
                    id='default_case',
                ),
                pytest.param(
                    dict(min_value=10),  # override with min_value=10
                    id='min_value_10',
                ),
                pytest.param(
                    dict(max_value=200),  # override with max_value=200
                    id='max_value_200',
                ),
                pytest.param(
                    dict(min_value=-10, max_value=50),  # override both min_value
                    # and max_value
                    id='min_-10_max_50',
                ),
                pytest.param(
                    dict(min_value=20, max_value=80, average=50),  # all defaults
                    # are overridden
                    id="min_20_max_80_avg_50",
                ),
                pytest.param(
                    dict(min_value=100, max_value=0),  # invalid range
                    id='invalid_range',
                    marks=pytest.mark.xfail(reason='invalid range'),
                )
            ],
        ))
        def test_range(min_value, max_value, average):
            assert min_value <= max_value
            assert min_value <= average <= max_value
        ```

    The above test will execute with the following sets of parameters:

    ```python
    "default_case": {"min_value": 0, "max_value": 100, "average": 50}
    "min_value_10": {"min_value": 10, "max_value": 100, "average": 50}
    "max_value_200": {"min_value": 0, "max_value": 200, "average": 50}
    "min_-10_max_50": {"min_value": -10, "max_value": 50, "average": 50}
    "min_20_max_80_avg_50": {"min_value": 20, "max_value": 80, "average": 50}
    "invalid_range": {"min_value": 100, "max_value": 0, "average": 50}  # expected to fail
    ```

    Notes:
        - Each case in `cases` must contain exactly one value, which is a dictionary
          of parameter values.
        - The function performs an in-place update of the `cases` list, so the
          original `cases` list is modified.

    """
    for i, case in enumerate(cases):
        if not (len(case.values) == 1 and isinstance(case.values[0], dict)):
            raise ValueError(
                "each case must contain exactly one value; a dict of parameter values"
            )
        if set(case.values[0].keys()) - set(defaults.keys()):
            raise UnknownParameterInCasesError()
        # Overwrite values in defaults if the parameter is present in the test case values
        merged_params = {**defaults, **case.values[0]}  # type: ignore
        cases[i] = pytest.param(*merged_params.values(), id=case.id, marks=case.marks)

    return {"argnames": list(defaults), "argvalues": cases, **parametrize_kwargs}