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]]:
        """
        Returns the access list as a list of serializable elements.
        """
        return [self.address, self.storage_keys]

to_list()

Returns the 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]]:
    """
    Returns the 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 NonceMismatch(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):
            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 BalanceMismatch(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):
            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 CodeMismatch(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):
            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"):
        """
        Checks 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.NonceMismatch(
                    address=address,
                    want=self.nonce,
                    got=account.nonce,
                )

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

        if "code" in self.model_fields_set:
            if self.code != account.code:
                raise Account.CodeMismatch(
                    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:
        """
        Returns 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.

NonceMismatch 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 NonceMismatch(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):
        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}"
        )

__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}"
    )

BalanceMismatch 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 BalanceMismatch(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):
        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}"
        )

__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}"
    )

CodeMismatch 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 CodeMismatch(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):
        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}"
        )

__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)

Checks 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"):
    """
    Checks 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.NonceMismatch(
                address=address,
                want=self.nonce,
                got=account.nonce,
            )

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

    if "code" in self.model_fields_set:
        if self.code != account.code:
            raise Account.CodeMismatch(
                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__()

Returns True on a non-empty account.

Source code in src/ethereum_test_base_types/composite_types.py

def __bool__(self: "Account") -> bool:
    """
    Returns 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: "FixedSizeBytesConvertible | Address", *, label: str | None = None):
        """
        Creates a new Address object with an optional label.
        """
        instance = super(Address, cls).__new__(cls, input)
        if isinstance(input, Address) and label is None:
            instance.label = input.label
        else:
            instance.label = label
        return instance

__new__(input, *, label=None)

Creates a new Address object with an optional label.

Source code in src/ethereum_test_base_types/base_types.py

def __new__(cls, input: "FixedSizeBytesConvertible | Address", *, label: str | None = None):
    """
    Creates a new Address object with an optional label.
    """
    instance = super(Address, cls).__new__(cls, input)
    if isinstance(input, Address) and label is None:
        instance.label = input.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: BytesConvertible = b""):
        """
        Creates a new Bytes object.
        """
        if type(input) is cls:
            return input
        return super(Bytes, cls).__new__(cls, to_bytes(input))

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

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

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

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

    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=b'')

Creates a new Bytes object.

Source code in src/ethereum_test_base_types/base_types.py

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

__hash__()

Returns the hash of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

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

__str__()

Returns the hexadecimal representation of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

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

hex(*args, **kwargs)

Returns the hexadecimal representation of the bytes.

Source code in src/ethereum_test_base_types/base_types.py

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

or_none(input) classmethod

Converts 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 | BytesConvertible | None") -> "Bytes | None":
    """
    Converts the input to a Bytes while accepting None.
    """
    if input is None:
        return input
    return cls(input)

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:
        """
        Returns the name of the spec.
        """
        pass

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

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

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

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

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

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

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

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

name() abstractmethod

Returns the name of the spec.

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

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

has_known_version() abstractmethod

Returns 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:
    """
    Returns true if the reference spec object is hard-coded with a latest
    known version.
    """
    pass

known_version() abstractmethod

Returns 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:
    """
    Returns the latest known version in the reference.
    """
    pass

api_url() abstractmethod

Returns 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:
    """
    Returns the URL required to poll the version from an API, if needed.
    """
    pass

latest_version() abstractmethod

Returns 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:
    """
    Returns a digest that points to the latest version of the spec.
    """
    pass

is_outdated() abstractmethod

Checks 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:
    """
    Checks whether the reference specification has been updated since the
    test was last updated.
    """
    pass

write_info(info) abstractmethod

Writes 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, str]):
    """
    Writes info about the reference specification used into the output
    fixture.
    """
    pass

parseable_from_module(module_dict) abstractmethod staticmethod

Checks 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:
    """
    Checks whether the module's dict contains required reference spec
    information.
    """
    pass

parse_from_module(module_dict) abstractmethod staticmethod

Parses 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":
    """
    Parses 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_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
    """

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_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

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."""

    info: Dict[str, str] = Field(default_factory=dict, alias="_info")

    # Fixture format properties
    fixture_format_name: ClassVar[str] = "unset"
    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:
        """
        Returns the name of the subdirectory where this type of fixture should be dumped to.
        """
        return cls.fixture_format_name.replace("test", "tests")

    @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]:
        """
        Returns the 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,
    ):
        """
        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.fixture_format_name
        if ref_spec is not None:
            ref_spec.write_info(self.info)

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

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

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

output_base_dir_name() classmethod

Returns the 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:
    """
    Returns the name of the subdirectory where this type of fixture should be dumped to.
    """
    return cls.fixture_format_name.replace("test", "tests")

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)

Returns the 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]:
    """
    Returns the 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)

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,
):
    """
    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.fixture_format_name
    if ref_spec is not None:
        ref_spec.write_info(self.info)

get_fork()

Returns the fork of the fixture as a string.

Source code in src/ethereum_test_fixtures/base.py

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

supports_fork(fork) classmethod

Returns 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:
    """
    Returns 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:
        """
        Returns the 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:
        """
        Adds a 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:
        """
        Dumps 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: FixtureVerifier) -> None:
        """
        Runs `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.is_verifiable(fixture.__class__):
                    info = self.json_path_to_test_item[fixture_path]
                    verify_fixtures_dump_dir = self._get_verify_fixtures_dump_dir(info)
                    evm_fixture_verification.verify_fixture(
                        fixture.__class__,
                        fixture_path,
                        fixture_name=None,
                        debug_output_path=verify_fixtures_dump_dir,
                    )

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

Returns the 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:
    """
    Returns the 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)

Adds a 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:
    """
    Adds a 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()

Dumps all collected fixtures to their respective files.

Source code in src/ethereum_test_fixtures/collector.py

def dump_fixtures(self) -> None:
    """
    Dumps 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)

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

Source code in src/ethereum_test_fixtures/collector.py

def verify_fixture_files(self, evm_fixture_verification: FixtureVerifier) -> None:
    """
    Runs `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.is_verifiable(fixture.__class__):
                info = self.json_path_to_test_item[fixture_path]
                verify_fixtures_dump_dir = self._get_verify_fixtures_dump_dir(info)
                evm_fixture_verification.verify_fixture(
                    fixture.__class__,
                    fixture_path,
                    fixture_name=None,
                    debug_output_path=verify_fixtures_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]:
        """
        Converts a 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:
        """
        Converts a 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]:
        """
        The 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()

Converts a 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]:
    """
    Converts a 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()

Converts a test name to a single test name.

Source code in src/ethereum_test_fixtures/collector.py

def get_single_test_name(self) -> str:
    """
    Converts a 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')

The 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]:
    """
    The 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[List[FixtureFormat]] = []
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = []

    @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:
        """
        Returns the 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)),
        )

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()

Returns the 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:
    """
    Returns the 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[List[FixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
    ]
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = [
        TransactionPost,
    ]

    def make_genesis(
        self,
        fork: Fork,
    ) -> Tuple[Alloc, FixtureBlock]:
        """
        Create a genesis block from the blockchain test definition.
        """
        env = self.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()),
            self.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,
            target_blobs_per_block=(
                fork.target_blobs_per_block(0, 0)
                if fork.header_target_blobs_per_block_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),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            slow_request=slow,
        )

        try:
            rejected_txs = verify_transactions(
                t8n.exception_mapper, txs, 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,
        )

    def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
        """
        Returns 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, alloc: Alloc):
        """
        Verifies the post alloc after all block/s or payload/s are generated.
        """
        try:
            self.post.verify_post_alloc(alloc)
        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,
    ) -> Fixture:
        """
        Create a fixture from the blockchain test definition.
        """
        fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

        pre, genesis = self.make_genesis(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,
                    ),
                )

        self.verify_post_state(t8n, alloc)
        return Fixture(
            fork=self.network_info(fork, eips),
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            post_state=alloc,
        )

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

        pre, genesis = self.make_genesis(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
        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, 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,
            )

        return EngineFixture(
            fork=self.network_info(fork, eips),
            genesis=genesis.header,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre=pre,
            post_state=alloc,
            sync_payload=sync_payload,
            last_block_hash=head_hash,
        )

    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}")

make_genesis(fork)

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

def make_genesis(
    self,
    fork: Fork,
) -> Tuple[Alloc, FixtureBlock]:
    """
    Create a genesis block from the blockchain test definition.
    """
    env = self.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()),
        self.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,
        target_blobs_per_block=(
            fork.target_blobs_per_block(0, 0)
            if fork.header_target_blobs_per_block_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),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        slow_request=slow,
    )

    try:
        rejected_txs = verify_transactions(
            t8n.exception_mapper, txs, 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)

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

Source code in src/ethereum_test_specs/blockchain.py

def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
    """
    Returns 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, alloc)

Verifies the 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, alloc: Alloc):
    """
    Verifies the post alloc after all block/s or payload/s are generated.
    """
    try:
        self.post.verify_post_alloc(alloc)
    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,
) -> Fixture:
    """
    Create a fixture from the blockchain test definition.
    """
    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

    pre, genesis = self.make_genesis(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,
                ),
            )

    self.verify_post_state(t8n, alloc)
    return Fixture(
        fork=self.network_info(fork, eips),
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        post_state=alloc,
    )

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,
) -> EngineFixture:
    """
    Create a hive fixture from the blocktest definition.
    """
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, genesis = self.make_genesis(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
    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, 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,
        )

    return EngineFixture(
        fork=self.network_info(fork, eips),
        genesis=genesis.header,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre=pre,
        post_state=alloc,
        sync_payload=sync_payload,
        last_block_hash=head_hash,
    )

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}")

BlockchainTestEngine

Bases: BlockchainTest

Filler type that tests multiple blocks (valid or invalid) in a chain, only for the Engine API.

Source code in src/ethereum_test_specs/blockchain.py

class BlockchainTestEngine(BlockchainTest):
    """
    Filler type that tests multiple blocks (valid or invalid) in a chain, only for the Engine API.
    """

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        BlockchainEngineFixture,
    ]

EOFStateTest

Bases: EOFTest

Filler type that tests EOF containers and also generates a state/blockchain test.

Source code in src/ethereum_test_specs/eof.py

class EOFStateTest(EOFTest):
    """
    Filler type that tests EOF containers and also generates a state/blockchain test.
    """

    deploy_tx: bool = False
    tx_gas_limit: int = 10_000_000
    tx_data: Bytes = Bytes(b"")
    tx_sender_funding_amount: int = 1_000_000_000_000_000_000_000
    env: Environment = Field(default_factory=Environment)
    container_post: Account = Field(default_factory=Account)
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        EOFFixture,
        StateFixture,
        BlockchainFixture,
        BlockchainEngineFixture,
    ]

    @model_validator(mode="before")
    @classmethod
    def check_container_type(cls, data: Any) -> Any:
        """
        Check if the container exception matches the expected exception.
        """
        if isinstance(data, dict):
            container = data.get("data")
            deploy_tx = data.get("deploy_tx")
            container_kind = data.get("container_kind")
            if deploy_tx is None:
                if (
                    container is not None
                    and isinstance(container, Container)
                    and "kind" in container.model_fields_set
                    and container.kind == ContainerKind.INITCODE
                ) or (container_kind is not None and container_kind == ContainerKind.INITCODE):
                    data["deploy_tx"] = True
        return data

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

    def generate_state_test(self) -> StateTest:
        """
        Generate the StateTest filler.
        """
        assert self.pre is not None, "pre must be set to generate a StateTest."
        tx = Transaction(
            sender=self.pre.fund_eoa(amount=self.tx_sender_funding_amount),
            gas_limit=self.tx_gas_limit,
        )
        post = Alloc()
        if self.expect_exception is not None:  # Invalid EOF
            tx.to = None  # Make EIP-7698 create transaction
            tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
            post[tx.created_contract] = None  # Expect failure.
        elif self.deploy_tx:
            tx.to = None  # Make EIP-7698 create transaction
            tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
            post[tx.created_contract] = self.container_post  # Successful.
        else:
            tx.to = self.pre.deploy_contract(code=self.data)
            tx.data = self.tx_data
            post[tx.to] = self.container_post
        return StateTest(
            pre=self.pre,
            tx=tx,
            env=self.env,
            post=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 self.data 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 (
            StateFixture,
            BlockchainFixture,
            BlockchainEngineFixture,
        ):
            return self.generate_state_test().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().execute(
                fork=fork, execute_format=execute_format, eips=eips
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

check_container_type(data) classmethod

Check if the container exception matches the expected exception.

Source code in src/ethereum_test_specs/eof.py

@model_validator(mode="before")
@classmethod
def check_container_type(cls, data: Any) -> Any:
    """
    Check if the container exception matches the expected exception.
    """
    if isinstance(data, dict):
        container = data.get("data")
        deploy_tx = data.get("deploy_tx")
        container_kind = data.get("container_kind")
        if deploy_tx is None:
            if (
                container is not None
                and isinstance(container, Container)
                and "kind" in container.model_fields_set
                and container.kind == ContainerKind.INITCODE
            ) or (container_kind is not None and container_kind == ContainerKind.INITCODE):
                data["deploy_tx"] = True
    return data

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"

generate_state_test()

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self) -> StateTest:
    """
    Generate the StateTest filler.
    """
    assert self.pre is not None, "pre must be set to generate a StateTest."
    tx = Transaction(
        sender=self.pre.fund_eoa(amount=self.tx_sender_funding_amount),
        gas_limit=self.tx_gas_limit,
    )
    post = Alloc()
    if self.expect_exception is not None:  # Invalid EOF
        tx.to = None  # Make EIP-7698 create transaction
        tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
        post[tx.created_contract] = None  # Expect failure.
    elif self.deploy_tx:
        tx.to = None  # Make EIP-7698 create transaction
        tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
        post[tx.created_contract] = self.container_post  # Successful.
    else:
        tx.to = self.pre.deploy_contract(code=self.data)
        tx.data = self.tx_data
        post[tx.to] = self.container_post
    return StateTest(
        pre=self.pre,
        tx=tx,
        env=self.env,
        post=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 self.data 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 (
        StateFixture,
        BlockchainFixture,
        BlockchainEngineFixture,
    ):
        return self.generate_state_test().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().execute(
            fork=fork, execute_format=execute_format, eips=eips
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

EOFTest

Bases: BaseTest

Filler type that tests EOF containers.

Source code in src/ethereum_test_specs/eof.py

class EOFTest(BaseTest):
    """
    Filler type that tests EOF containers.
    """

    data: Bytes
    expect_exception: EOFExceptionInstanceOrList | None = None
    container_kind: ContainerKind | None = None

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        EOFFixture,
    ]

    @model_validator(mode="before")
    @classmethod
    def check_container_exception(cls, data: Any) -> Any:
        """
        Check if the container exception matches the expected exception.
        """
        if isinstance(data, dict):
            container = data.get("data")
            expect_exception = data.get("expect_exception")
            container_kind = data.get("container_kind")
            if container is not None and isinstance(container, Container):
                if (
                    "validity_error" in container.model_fields_set
                    and container.validity_error is not None
                ):
                    if expect_exception is not None:
                        assert container.validity_error == expect_exception, (
                            f"Container validity error {container.validity_error} "
                            f"does not match expected exception {expect_exception}."
                        )
                    if expect_exception is None:
                        data["expect_exception"] = container.validity_error
                if "kind" in container.model_fields_set:
                    if container_kind is not None:
                        assert container.kind == container_kind, (
                            f"Container kind type {str(container.kind)} "
                            f"does not match test {container_kind}."
                        )
                    if container.kind != ContainerKind.RUNTIME:
                        data["container_kind"] = container.kind
        return data

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

    def make_eof_test_fixture(
        self,
        *,
        request: pytest.FixtureRequest,
        fork: Fork,
        eips: Optional[List[int]],
    ) -> EOFFixture:
        """
        Generate the EOF test fixture.
        """
        if self.data in existing_tests:
            pytest.fail(
                f"Duplicate EOF test: {self.data}, existing test: {existing_tests[self.data]}"
            )
        existing_tests[self.data] = request.node.nodeid
        vectors = [
            Vector(
                code=self.data,
                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!")
            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=str(vector.code))
            self.verify_result(result, expected_result, vector.code)

        return fixture

    def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
        """
        Checks 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 UnexpectedEOFException(
                    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 ExpectedEOFException(
                    code=code,
                    expected=f"{expected_string}",
                )
            elif actual_exception in expected_result.exception:
                return
            else:
                raise EOFExceptionMismatch(
                    code=code,
                    expected=f"{expected_string}",
                    got=f"{actual_exception} ({actual_message})",
                )

    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)

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

check_container_exception(data) classmethod

Check if the container exception matches the expected exception.

Source code in src/ethereum_test_specs/eof.py

@model_validator(mode="before")
@classmethod
def check_container_exception(cls, data: Any) -> Any:
    """
    Check if the container exception matches the expected exception.
    """
    if isinstance(data, dict):
        container = data.get("data")
        expect_exception = data.get("expect_exception")
        container_kind = data.get("container_kind")
        if container is not None and isinstance(container, Container):
            if (
                "validity_error" in container.model_fields_set
                and container.validity_error is not None
            ):
                if expect_exception is not None:
                    assert container.validity_error == expect_exception, (
                        f"Container validity error {container.validity_error} "
                        f"does not match expected exception {expect_exception}."
                    )
                if expect_exception is None:
                    data["expect_exception"] = container.validity_error
            if "kind" in container.model_fields_set:
                if container_kind is not None:
                    assert container.kind == container_kind, (
                        f"Container kind type {str(container.kind)} "
                        f"does not match test {container_kind}."
                    )
                if container.kind != ContainerKind.RUNTIME:
                    data["container_kind"] = container.kind
    return data

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"

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.
    """
    if self.data in existing_tests:
        pytest.fail(
            f"Duplicate EOF test: {self.data}, existing test: {existing_tests[self.data]}"
        )
    existing_tests[self.data] = request.node.nodeid
    vectors = [
        Vector(
            code=self.data,
            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!")
        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=str(vector.code))
        self.verify_result(result, expected_result, vector.code)

    return fixture

verify_result(result, expected_result, code)

Checks 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):
    """
    Checks 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 UnexpectedEOFException(
                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 ExpectedEOFException(
                code=code,
                expected=f"{expected_string}",
            )
        elif actual_exception in expected_result.exception:
            return
        else:
            raise EOFExceptionMismatch(
                code=code,
                expected=f"{expected_string}",
                got=f"{actual_exception} ({actual_message})",
            )

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)

    raise Exception(f"Unknown fixture format: {fixture_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[List[FixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
        StateFixture,
    ]
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = [
        TransactionPost,
    ]

    def _generate_blockchain_genesis_environment(self) -> Environment:
        """
        Generate the genesis environment for the BlockchainTest formatted test.
        """
        assert (
            self.env.number >= 1
        ), "genesis block number cannot be negative, set state test env.number to 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 + TARGET_BLOB_GAS_PER_BLOCK
            )

        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) -> BlockchainTest:
        """
        Generate a BlockchainTest fixture from this StateTest fixture.
        """
        return BlockchainTest(
            genesis_environment=self._generate_blockchain_genesis_environment(),
            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,
    ) -> Fixture:
        """
        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
            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(t8n.exception_mapper, [tx], 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 Fixture(
            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,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
        )

    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().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}")

generate_blockchain_test()

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in src/ethereum_test_specs/state.py

def generate_blockchain_test(self) -> BlockchainTest:
    """
    Generate a BlockchainTest fixture from this StateTest fixture.
    """
    return BlockchainTest(
        genesis_environment=self._generate_blockchain_genesis_environment(),
        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,
) -> Fixture:
    """
    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
        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(t8n.exception_mapper, [tx], 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 Fixture(
        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,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
    )

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().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[List[FixtureFormat]] = [
        TransactionFixture,
    ]
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = [
        TransactionPost,
    ]

    def make_transaction_test_fixture(
        self,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> Fixture:
        """
        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 Fixture(
            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,
) -> Fixture:
    """
    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 Fixture(
        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.
    """

    def set_environment(self, env: Environment) -> Environment:
        """
        Creates a 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
        if not isinstance(self.target_blobs_per_block, Removable):
            new_env_values["target_blobs_per_block"] = self.target_blobs_per_block
        """
        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.

set_environment(env)

Creates a 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:
    """
    Creates a 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
    if not isinstance(self.target_blobs_per_block, Removable):
        new_env_values["target_blobs_per_block"] = self.target_blobs_per_block
    """
    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
    target_blobs_per_block: Removable | HexNumber | None = None

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

    This can be used in a test to explicitly skip a field in a block's RLP encoding.
    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):
        """
        Helper validator to 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:
        """
        Produces 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):
        """
        Verifies 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

Helper validator to 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):
    """
    Helper validator to convert a list of withdrawals into the withdrawals root hash.
    """
    if isinstance(value, list):
        return Withdrawal.list_root(value)
    return value

apply(target)

Produces 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:
    """
    Produces 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)

Verifies that the header fields from self are as expected.

Source code in src/ethereum_test_specs/blockchain.py

def verify(self, target: FixtureHeader):
    """
    Verifies 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 = Bytecode(),
        if_false: Bytecode | Op = Bytecode(),
        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 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=Bytecode(), if_false=Bytecode(), 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 = Bytecode(),
    if_false: Bytecode | Op = Bytecode(),
    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 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 = Bytecode(),
        initcode_length: int | None = None,
        initcode_prefix: Bytecode = Bytecode(),
        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.
        """
        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=Bytecode(), initcode_length=None, initcode_prefix=Bytecode(), 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 = Bytecode(),
    initcode_length: int | None = None,
    initcode_prefix: Bytecode = Bytecode(),
    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.
    """
    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=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=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

extend_with_defaults(defaults, cases, **parametrize_kwargs)

Extends 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]:
    """
    Extends 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}