Ethereum Test Types package

Common definitions and types.

TestParameterGroup dataclass

Base class for grouping test parameters in a dataclass. Provides a generic repr method to generate clean test ids, including only non-default optional fields.

Source code in src/ethereum_test_types/helpers.py

@dataclass(kw_only=True, frozen=True, repr=False)
class TestParameterGroup:
    """
    Base class for grouping test parameters in a dataclass. Provides a generic
    __repr__ method to generate clean test ids, including only non-default
    optional fields.
    """

    __test__ = False  # explicitly prevent pytest collecting this class

    def __repr__(self):
        """
        Generate repr string, intended to be used as a test id, based on the class
        name and the values of the non-default optional fields.
        """
        class_name = self.__class__.__name__
        field_strings = []

        for field in fields(self):
            value = getattr(self, field.name)
            # Include the field only if it is not optional or not set to its default value
            if field.default is MISSING or field.default != value:
                field_strings.append(f"{field.name}_{value}")

        return f"{class_name}_{'-'.join(field_strings)}"

__repr__()

Generate repr string, intended to be used as a test id, based on the class name and the values of the non-default optional fields.

Source code in src/ethereum_test_types/helpers.py

def __repr__(self):
    """
    Generate repr string, intended to be used as a test id, based on the class
    name and the values of the non-default optional fields.
    """
    class_name = self.__class__.__name__
    field_strings = []

    for field in fields(self):
        value = getattr(self, field.name)
        # Include the field only if it is not optional or not set to its default value
        if field.default is MISSING or field.default != value:
            field_strings.append(f"{field.name}_{value}")

    return f"{class_name}_{'-'.join(field_strings)}"

add_kzg_version(b_hashes, kzg_version)

Add Kzg Version to each blob hash.

Source code in src/ethereum_test_types/helpers.py

def add_kzg_version(
    b_hashes: List[bytes | SupportsBytes | int | str], kzg_version: int
) -> List[Hash]:
    """Add  Kzg Version to each blob hash."""
    kzg_version_hex = bytes([kzg_version])
    kzg_versioned_hashes = []

    for b_hash in b_hashes:
        b_hash = bytes(Hash(b_hash))
        if isinstance(b_hash, int) or isinstance(b_hash, str):
            kzg_versioned_hashes.append(Hash(kzg_version_hex + b_hash[1:]))
        elif isinstance(b_hash, bytes) or isinstance(b_hash, SupportsBytes):
            if isinstance(b_hash, SupportsBytes):
                b_hash = bytes(b_hash)
            kzg_versioned_hashes.append(Hash(kzg_version_hex + b_hash[1:]))
        else:
            raise TypeError("Blob hash must be either an integer, string or bytes")
    return kzg_versioned_hashes

ceiling_division(a, b)

Calculate ceil without using floating point. Used by many of the EVM's formulas.

Source code in src/ethereum_test_types/helpers.py

def ceiling_division(a: int, b: int) -> int:
    """
    Calculate ceil without using floating point.
    Used by many of the EVM's formulas.
    """
    return -(a // -b)

compute_create2_address(address, salt, initcode)

Compute address of the resulting contract created using the CREATE2 opcode.

Source code in src/ethereum_test_types/helpers.py

def compute_create2_address(
    address: FixedSizeBytesConvertible, salt: FixedSizeBytesConvertible, initcode: BytesConvertible
) -> Address:
    """
    Compute address of the resulting contract created using the `CREATE2`
    opcode.
    """
    hash_bytes = Bytes(
        b"\xff" + Address(address) + Hash(salt) + Bytes(initcode).keccak256()
    ).keccak256()
    return Address(hash_bytes[-20:])

compute_create_address(*, address, nonce=None, salt=0, initcode=b'', opcode=Op.CREATE)

Compute address of the resulting contract created using a transaction or the CREATE opcode.

Source code in src/ethereum_test_types/helpers.py

def compute_create_address(
    *,
    address: FixedSizeBytesConvertible | EOA,
    nonce: int | None = None,
    salt: int = 0,
    initcode: BytesConvertible = b"",
    opcode: Op = Op.CREATE,
) -> Address:
    """
    Compute address of the resulting contract created using a transaction
    or the `CREATE` opcode.
    """
    if opcode == Op.CREATE:
        if isinstance(address, EOA):
            if nonce is None:
                nonce = address.nonce
        else:
            address = Address(address)
        if nonce is None:
            nonce = 0
        hash_bytes = Bytes(eth_rlp.encode([address, int_to_bytes(nonce)])).keccak256()
        return Address(hash_bytes[-20:])
    if opcode == Op.CREATE2:
        return compute_create2_address(address, salt, initcode)
    raise ValueError("Unsupported opcode")

compute_eofcreate_address(address, salt)

Compute address of the resulting contract created using the EOFCREATE opcode.

Source code in src/ethereum_test_types/helpers.py

def compute_eofcreate_address(
    address: FixedSizeBytesConvertible, salt: FixedSizeBytesConvertible
) -> Address:
    """Compute address of the resulting contract created using the `EOFCREATE` opcode."""
    hash_bytes = Bytes(b"\xff" + b"\x00" * 12 + Address(address) + Hash(salt)).keccak256()
    return Address(hash_bytes[-20:])

EOA

Bases: Address

An Externally Owned Account (EOA) is an account controlled by a private key.

The EOA is defined by its address and (optionally) by its corresponding private key.

Source code in src/ethereum_test_types/types.py

class EOA(Address):
    """
    An Externally Owned Account (EOA) is an account controlled by a private key.

    The EOA is defined by its address and (optionally) by its corresponding private key.
    """

    key: Hash | None
    nonce: Number

    def __new__(
        cls,
        address: "FixedSizeBytesConvertible | Address | EOA | None" = None,
        *,
        key: FixedSizeBytesConvertible | None = None,
        nonce: NumberConvertible = 0,
    ):
        """Init the EOA."""
        if address is None:
            if key is None:
                raise ValueError("impossible to initialize EOA without address")
            private_key = PrivateKey(Hash(key))
            public_key = private_key.public_key
            address = Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
        elif isinstance(address, EOA):
            return address
        instance = super(EOA, cls).__new__(cls, address)
        instance.key = Hash(key) if key is not None else None
        instance.nonce = Number(nonce)
        return instance

    def get_nonce(self) -> Number:
        """Return current nonce of the EOA and increments it by one."""
        nonce = self.nonce
        self.nonce = Number(nonce + 1)
        return nonce

    def copy(self) -> "EOA":
        """Return copy of the EOA."""
        return EOA(Address(self), key=self.key, nonce=self.nonce)

__new__(address=None, *, key=None, nonce=0)

Init the EOA.

Source code in src/ethereum_test_types/types.py

def __new__(
    cls,
    address: "FixedSizeBytesConvertible | Address | EOA | None" = None,
    *,
    key: FixedSizeBytesConvertible | None = None,
    nonce: NumberConvertible = 0,
):
    """Init the EOA."""
    if address is None:
        if key is None:
            raise ValueError("impossible to initialize EOA without address")
        private_key = PrivateKey(Hash(key))
        public_key = private_key.public_key
        address = Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
    elif isinstance(address, EOA):
        return address
    instance = super(EOA, cls).__new__(cls, address)
    instance.key = Hash(key) if key is not None else None
    instance.nonce = Number(nonce)
    return instance

get_nonce()

Return current nonce of the EOA and increments it by one.

Source code in src/ethereum_test_types/types.py

def get_nonce(self) -> Number:
    """Return current nonce of the EOA and increments it by one."""
    nonce = self.nonce
    self.nonce = Number(nonce + 1)
    return nonce

copy()

Return copy of the EOA.

Source code in src/ethereum_test_types/types.py

def copy(self) -> "EOA":
    """Return copy of the EOA."""
    return EOA(Address(self), key=self.key, nonce=self.nonce)

Account

Bases: CamelModel

State associated with an address.

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        return cls(**kwargs)

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

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

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

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

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

Bytecode contained by the account.

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

Storage within a contract.

NONEXISTENT: None = None class-attribute

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

NonceMismatchError dataclass

Bases: Exception

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

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

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

__init__(address, want, got, *args)

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

Source code in src/ethereum_test_base_types/composite_types.py

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

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

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

BalanceMismatchError dataclass

Bases: Exception

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

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

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

__init__(address, want, got, *args)

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

Source code in src/ethereum_test_base_types/composite_types.py

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

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

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

CodeMismatchError dataclass

Bases: Exception

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

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

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

__init__(address, want, got, *args)

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

Source code in src/ethereum_test_base_types/composite_types.py

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

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

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

check_alloc(address, account)

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

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

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

__bool__()

Return True on a non-empty account.

Source code in src/ethereum_test_base_types/composite_types.py

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

with_code(code) classmethod

Create account with provided code and nonce of 1.

Source code in src/ethereum_test_base_types/composite_types.py

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

merge(account_1, account_2) classmethod

Create a merged account from two sources.

Source code in src/ethereum_test_base_types/composite_types.py

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

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

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

    return cls(**kwargs)

Alloc

Bases: Alloc

Allocation of accounts in the state, pre and post test execution.

Source code in src/ethereum_test_types/types.py

class Alloc(BaseAlloc):
    """Allocation of accounts in the state, pre and post test execution."""

    _eoa_fund_amount_default: int = PrivateAttr(10**21)

    @dataclass(kw_only=True)
    class UnexpectedAccountError(Exception):
        """Unexpected account found in the allocation."""

        address: Address
        account: Account | None

        def __init__(self, address: Address, account: Account | None, *args):
            """Initialize the exception."""
            super().__init__(args)
            self.address = address
            self.account = account

        def __str__(self):
            """Print exception string."""
            return f"unexpected account in allocation {self.address}: {self.account}"

    @dataclass(kw_only=True)
    class MissingAccountError(Exception):
        """Expected account not found in the allocation."""

        address: Address

        def __init__(self, address: Address, *args):
            """Initialize the exception."""
            super().__init__(args)
            self.address = address

        def __str__(self):
            """Print exception string."""
            return f"Account missing from allocation {self.address}"

    @classmethod
    def merge(cls, alloc_1: "Alloc", alloc_2: "Alloc") -> "Alloc":
        """Return merged allocation of two sources."""
        merged = alloc_1.model_dump()

        for address, other_account in alloc_2.root.items():
            merged_account = Account.merge(merged.get(address, None), other_account)
            if merged_account:
                merged[address] = merged_account
            elif address in merged:
                merged.pop(address, None)

        return Alloc(merged)

    def __iter__(self):
        """Return iterator over the allocation."""
        return iter(self.root)

    def items(self):
        """Return iterator over the allocation items."""
        return self.root.items()

    def __getitem__(self, address: Address | FixedSizeBytesConvertible) -> Account | None:
        """Return account associated with an address."""
        if not isinstance(address, Address):
            address = Address(address)
        return self.root[address]

    def __setitem__(self, address: Address | FixedSizeBytesConvertible, account: Account | None):
        """Set account associated with an address."""
        if not isinstance(address, Address):
            address = Address(address)
        self.root[address] = account

    def __delitem__(self, address: Address | FixedSizeBytesConvertible):
        """Delete account associated with an address."""
        if not isinstance(address, Address):
            address = Address(address)
        self.root.pop(address, None)

    def __eq__(self, other) -> bool:
        """Return True if both allocations are equal."""
        if not isinstance(other, Alloc):
            return False
        return self.root == other.root

    def __contains__(self, address: Address | FixedSizeBytesConvertible) -> bool:
        """Check if an account is in the allocation."""
        if not isinstance(address, Address):
            address = Address(address)
        return address in self.root

    def empty_accounts(self) -> List[Address]:
        """Return list of addresses of empty accounts."""
        return [address for address, account in self.root.items() if not account]

    def state_root(self) -> bytes:
        """Return state root of the allocation."""
        state = State()
        for address, account in self.root.items():
            if account is None:
                continue
            set_account(
                state=state,
                address=FrontierAddress(address),
                account=FrontierAccount(
                    nonce=Uint(account.nonce) if account.nonce is not None else Uint(0),
                    balance=(U256(account.balance) if account.balance is not None else U256(0)),
                    code=account.code if account.code is not None else b"",
                ),
            )
            if account.storage is not None:
                for key, value in account.storage.root.items():
                    set_storage(
                        state=state,
                        address=FrontierAddress(address),
                        key=Bytes32(Hash(key)),
                        value=U256(value),
                    )
        return state_root(state)

    def verify_post_alloc(self, got_alloc: "Alloc"):
        """
        Verify that the allocation matches the expected post in the test.
        Raises exception on unexpected values.
        """
        assert isinstance(got_alloc, Alloc), f"got_alloc is not an Alloc: {got_alloc}"
        for address, account in self.root.items():
            if account is None:
                # Account must not exist
                if address in got_alloc.root and got_alloc.root[address] is not None:
                    raise Alloc.UnexpectedAccountError(address, got_alloc.root[address])
            else:
                if address in got_alloc.root:
                    got_account = got_alloc.root[address]
                    assert isinstance(got_account, Account)
                    assert isinstance(account, Account)
                    account.check_alloc(address, got_account)
                else:
                    raise Alloc.MissingAccountError(address)

    def deploy_contract(
        self,
        code: BytesConvertible,
        *,
        storage: Storage | StorageRootType | None = None,
        balance: NumberConvertible = 0,
        nonce: NumberConvertible = 1,
        address: Address | None = None,
        evm_code_type: EVMCodeType | None = None,
        label: str | None = None,
    ) -> Address:
        """Deploy a contract to the allocation."""
        raise NotImplementedError("deploy_contract is not implemented in the base class")

    def fund_eoa(
        self,
        amount: NumberConvertible | None = None,
        label: str | None = None,
        storage: Storage | None = None,
        delegation: Address | Literal["Self"] | None = None,
        nonce: NumberConvertible | None = None,
    ) -> EOA:
        """Add a previously unused EOA to the pre-alloc with the balance specified by `amount`."""
        raise NotImplementedError("fund_eoa is not implemented in the base class")

    def fund_address(self, address: Address, amount: NumberConvertible):
        """
        Fund an address with a given amount.

        If the address is already present in the pre-alloc the amount will be
        added to its existing balance.
        """
        raise NotImplementedError("fund_address is not implemented in the base class")

UnexpectedAccountError dataclass

Bases: Exception

Unexpected account found in the allocation.

Source code in src/ethereum_test_types/types.py

@dataclass(kw_only=True)
class UnexpectedAccountError(Exception):
    """Unexpected account found in the allocation."""

    address: Address
    account: Account | None

    def __init__(self, address: Address, account: Account | None, *args):
        """Initialize the exception."""
        super().__init__(args)
        self.address = address
        self.account = account

    def __str__(self):
        """Print exception string."""
        return f"unexpected account in allocation {self.address}: {self.account}"

__init__(address, account, *args)

Initialize the exception.

Source code in src/ethereum_test_types/types.py

def __init__(self, address: Address, account: Account | None, *args):
    """Initialize the exception."""
    super().__init__(args)
    self.address = address
    self.account = account

__str__()

Print exception string.

Source code in src/ethereum_test_types/types.py

def __str__(self):
    """Print exception string."""
    return f"unexpected account in allocation {self.address}: {self.account}"

MissingAccountError dataclass

Bases: Exception

Expected account not found in the allocation.

Source code in src/ethereum_test_types/types.py

@dataclass(kw_only=True)
class MissingAccountError(Exception):
    """Expected account not found in the allocation."""

    address: Address

    def __init__(self, address: Address, *args):
        """Initialize the exception."""
        super().__init__(args)
        self.address = address

    def __str__(self):
        """Print exception string."""
        return f"Account missing from allocation {self.address}"

__init__(address, *args)

Initialize the exception.

Source code in src/ethereum_test_types/types.py

def __init__(self, address: Address, *args):
    """Initialize the exception."""
    super().__init__(args)
    self.address = address

__str__()

Print exception string.

Source code in src/ethereum_test_types/types.py

def __str__(self):
    """Print exception string."""
    return f"Account missing from allocation {self.address}"

merge(alloc_1, alloc_2) classmethod

Return merged allocation of two sources.

Source code in src/ethereum_test_types/types.py

@classmethod
def merge(cls, alloc_1: "Alloc", alloc_2: "Alloc") -> "Alloc":
    """Return merged allocation of two sources."""
    merged = alloc_1.model_dump()

    for address, other_account in alloc_2.root.items():
        merged_account = Account.merge(merged.get(address, None), other_account)
        if merged_account:
            merged[address] = merged_account
        elif address in merged:
            merged.pop(address, None)

    return Alloc(merged)

__iter__()

Return iterator over the allocation.

Source code in src/ethereum_test_types/types.py

def __iter__(self):
    """Return iterator over the allocation."""
    return iter(self.root)

items()

Return iterator over the allocation items.

Source code in src/ethereum_test_types/types.py

def items(self):
    """Return iterator over the allocation items."""
    return self.root.items()

__getitem__(address)

Return account associated with an address.

Source code in src/ethereum_test_types/types.py

def __getitem__(self, address: Address | FixedSizeBytesConvertible) -> Account | None:
    """Return account associated with an address."""
    if not isinstance(address, Address):
        address = Address(address)
    return self.root[address]

__setitem__(address, account)

Set account associated with an address.

Source code in src/ethereum_test_types/types.py

def __setitem__(self, address: Address | FixedSizeBytesConvertible, account: Account | None):
    """Set account associated with an address."""
    if not isinstance(address, Address):
        address = Address(address)
    self.root[address] = account

__delitem__(address)

Delete account associated with an address.

Source code in src/ethereum_test_types/types.py

def __delitem__(self, address: Address | FixedSizeBytesConvertible):
    """Delete account associated with an address."""
    if not isinstance(address, Address):
        address = Address(address)
    self.root.pop(address, None)

__eq__(other)

Return True if both allocations are equal.

Source code in src/ethereum_test_types/types.py

def __eq__(self, other) -> bool:
    """Return True if both allocations are equal."""
    if not isinstance(other, Alloc):
        return False
    return self.root == other.root

__contains__(address)

Check if an account is in the allocation.

Source code in src/ethereum_test_types/types.py

def __contains__(self, address: Address | FixedSizeBytesConvertible) -> bool:
    """Check if an account is in the allocation."""
    if not isinstance(address, Address):
        address = Address(address)
    return address in self.root

empty_accounts()

Return list of addresses of empty accounts.

Source code in src/ethereum_test_types/types.py

def empty_accounts(self) -> List[Address]:
    """Return list of addresses of empty accounts."""
    return [address for address, account in self.root.items() if not account]

state_root()

Return state root of the allocation.

Source code in src/ethereum_test_types/types.py

def state_root(self) -> bytes:
    """Return state root of the allocation."""
    state = State()
    for address, account in self.root.items():
        if account is None:
            continue
        set_account(
            state=state,
            address=FrontierAddress(address),
            account=FrontierAccount(
                nonce=Uint(account.nonce) if account.nonce is not None else Uint(0),
                balance=(U256(account.balance) if account.balance is not None else U256(0)),
                code=account.code if account.code is not None else b"",
            ),
        )
        if account.storage is not None:
            for key, value in account.storage.root.items():
                set_storage(
                    state=state,
                    address=FrontierAddress(address),
                    key=Bytes32(Hash(key)),
                    value=U256(value),
                )
    return state_root(state)

verify_post_alloc(got_alloc)

Verify that the allocation matches the expected post in the test. Raises exception on unexpected values.

Source code in src/ethereum_test_types/types.py

def verify_post_alloc(self, got_alloc: "Alloc"):
    """
    Verify that the allocation matches the expected post in the test.
    Raises exception on unexpected values.
    """
    assert isinstance(got_alloc, Alloc), f"got_alloc is not an Alloc: {got_alloc}"
    for address, account in self.root.items():
        if account is None:
            # Account must not exist
            if address in got_alloc.root and got_alloc.root[address] is not None:
                raise Alloc.UnexpectedAccountError(address, got_alloc.root[address])
        else:
            if address in got_alloc.root:
                got_account = got_alloc.root[address]
                assert isinstance(got_account, Account)
                assert isinstance(account, Account)
                account.check_alloc(address, got_account)
            else:
                raise Alloc.MissingAccountError(address)

deploy_contract(code, *, storage=None, balance=0, nonce=1, address=None, evm_code_type=None, label=None)

Deploy a contract to the allocation.

Source code in src/ethereum_test_types/types.py

def deploy_contract(
    self,
    code: BytesConvertible,
    *,
    storage: Storage | StorageRootType | None = None,
    balance: NumberConvertible = 0,
    nonce: NumberConvertible = 1,
    address: Address | None = None,
    evm_code_type: EVMCodeType | None = None,
    label: str | None = None,
) -> Address:
    """Deploy a contract to the allocation."""
    raise NotImplementedError("deploy_contract is not implemented in the base class")

fund_eoa(amount=None, label=None, storage=None, delegation=None, nonce=None)

Add a previously unused EOA to the pre-alloc with the balance specified by amount.

Source code in src/ethereum_test_types/types.py

def fund_eoa(
    self,
    amount: NumberConvertible | None = None,
    label: str | None = None,
    storage: Storage | None = None,
    delegation: Address | Literal["Self"] | None = None,
    nonce: NumberConvertible | None = None,
) -> EOA:
    """Add a previously unused EOA to the pre-alloc with the balance specified by `amount`."""
    raise NotImplementedError("fund_eoa is not implemented in the base class")

fund_address(address, amount)

Fund an address with a given amount.

If the address is already present in the pre-alloc the amount will be added to its existing balance.

Source code in src/ethereum_test_types/types.py

def fund_address(self, address: Address, amount: NumberConvertible):
    """
    Fund an address with a given amount.

    If the address is already present in the pre-alloc the amount will be
    added to its existing balance.
    """
    raise NotImplementedError("fund_address is not implemented in the base class")

AuthorizationTuple

Bases: AuthorizationTupleGeneric[HexNumber]

Authorization tuple for transactions.

Source code in src/ethereum_test_types/types.py

class AuthorizationTuple(AuthorizationTupleGeneric[HexNumber]):
    """Authorization tuple for transactions."""

    signer: EOA | None = None
    secret_key: Hash | None = None

    def model_post_init(self, __context: Any) -> None:
        """Automatically signs the authorization tuple if a secret key or sender are provided."""
        super().model_post_init(__context)
        self.sign()

    def sign(self: "AuthorizationTuple"):
        """Signs the authorization tuple with a private key."""
        signature_bytes: bytes | None = None
        rlp_signing_bytes = self.rlp_signing_bytes()
        if (
            "v" not in self.model_fields_set
            and "r" not in self.model_fields_set
            and "s" not in self.model_fields_set
        ):
            signing_key: Hash | None = None
            if self.secret_key is not None:
                signing_key = self.secret_key
            elif self.signer is not None:
                eoa = self.signer
                assert eoa is not None, "signer must be set"
                signing_key = eoa.key
            assert signing_key is not None, "secret_key or signer must be set"

            signature_bytes = PrivateKey(secret=signing_key).sign_recoverable(
                rlp_signing_bytes, hasher=keccak256
            )
            self.v, self.r, self.s = (
                HexNumber(signature_bytes[64]),
                HexNumber(int.from_bytes(signature_bytes[0:32], byteorder="big")),
                HexNumber(int.from_bytes(signature_bytes[32:64], byteorder="big")),
            )
            self.model_fields_set.add("v")
            self.model_fields_set.add("r")
            self.model_fields_set.add("s")

        if self.signer is None:
            try:
                if not signature_bytes:
                    signature_bytes = (
                        int(self.r).to_bytes(32, byteorder="big")
                        + int(self.s).to_bytes(32, byteorder="big")
                        + bytes([self.v])
                    )
                public_key = PublicKey.from_signature_and_message(
                    signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
                )
                self.signer = EOA(
                    address=Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
                )
            except Exception:
                # Signer remains `None` in this case
                pass

model_post_init(__context)

Automatically signs the authorization tuple if a secret key or sender are provided.

Source code in src/ethereum_test_types/types.py

def model_post_init(self, __context: Any) -> None:
    """Automatically signs the authorization tuple if a secret key or sender are provided."""
    super().model_post_init(__context)
    self.sign()

sign()

Signs the authorization tuple with a private key.

Source code in src/ethereum_test_types/types.py

def sign(self: "AuthorizationTuple"):
    """Signs the authorization tuple with a private key."""
    signature_bytes: bytes | None = None
    rlp_signing_bytes = self.rlp_signing_bytes()
    if (
        "v" not in self.model_fields_set
        and "r" not in self.model_fields_set
        and "s" not in self.model_fields_set
    ):
        signing_key: Hash | None = None
        if self.secret_key is not None:
            signing_key = self.secret_key
        elif self.signer is not None:
            eoa = self.signer
            assert eoa is not None, "signer must be set"
            signing_key = eoa.key
        assert signing_key is not None, "secret_key or signer must be set"

        signature_bytes = PrivateKey(secret=signing_key).sign_recoverable(
            rlp_signing_bytes, hasher=keccak256
        )
        self.v, self.r, self.s = (
            HexNumber(signature_bytes[64]),
            HexNumber(int.from_bytes(signature_bytes[0:32], byteorder="big")),
            HexNumber(int.from_bytes(signature_bytes[32:64], byteorder="big")),
        )
        self.model_fields_set.add("v")
        self.model_fields_set.add("r")
        self.model_fields_set.add("s")

    if self.signer is None:
        try:
            if not signature_bytes:
                signature_bytes = (
                    int(self.r).to_bytes(32, byteorder="big")
                    + int(self.s).to_bytes(32, byteorder="big")
                    + bytes([self.v])
                )
            public_key = PublicKey.from_signature_and_message(
                signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
            )
            self.signer = EOA(
                address=Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
            )
        except Exception:
            # Signer remains `None` in this case
            pass

CamelModel

Bases: CopyValidateModel

A base model that converts field names to camel case when serializing.

For example, the field name current_timestamp in a Python model will be represented as currentTimestamp when it is serialized to json.

Source code in src/ethereum_test_base_types/pydantic.py

class CamelModel(CopyValidateModel):
    """
    A base model that converts field names to camel case when serializing.

    For example, the field name `current_timestamp` in a Python model will be represented
    as `currentTimestamp` when it is serialized to json.
    """

    model_config = ConfigDict(
        alias_generator=to_camel,
        populate_by_name=True,
        validate_default=True,
    )

ConsolidationRequest

Bases: RequestBase, CamelModel

Consolidation Request type.

Source code in src/ethereum_test_types/types.py

class ConsolidationRequest(RequestBase, CamelModel):
    """Consolidation Request type."""

    source_address: Address = Address(0)
    """
    The address of the execution layer account that made the consolidation request.
    """
    source_pubkey: BLSPublicKey
    """
    The public key of the source validator as it currently is in the beacon state.
    """
    target_pubkey: BLSPublicKey
    """
    The public key of the target validator as it currently is in the beacon state.
    """

    type: ClassVar[int] = 2

    def __bytes__(self) -> bytes:
        """Return consolidation's attributes as bytes."""
        return bytes(self.source_address) + bytes(self.source_pubkey) + bytes(self.target_pubkey)

source_address: Address = Address(0) class-attribute instance-attribute

The address of the execution layer account that made the consolidation request.

source_pubkey: BLSPublicKey instance-attribute

The public key of the source validator as it currently is in the beacon state.

target_pubkey: BLSPublicKey instance-attribute

The public key of the target validator as it currently is in the beacon state.

__bytes__()

Return consolidation's attributes as bytes.

Source code in src/ethereum_test_types/types.py

def __bytes__(self) -> bytes:
    """Return consolidation's attributes as bytes."""
    return bytes(self.source_address) + bytes(self.source_pubkey) + bytes(self.target_pubkey)

DepositRequest

Bases: RequestBase, CamelModel

Deposit Request type.

Source code in src/ethereum_test_types/types.py

class DepositRequest(RequestBase, CamelModel):
    """Deposit Request type."""

    pubkey: BLSPublicKey
    """
    The public key of the beacon chain validator.
    """
    withdrawal_credentials: Hash
    """
    The withdrawal credentials of the beacon chain validator.
    """
    amount: HexNumber
    """
    The amount in gwei of the deposit.
    """
    signature: BLSSignature
    """
    The signature of the deposit using the validator's private key that matches the
    `pubkey`.
    """
    index: HexNumber
    """
    The index of the deposit.
    """

    type: ClassVar[int] = 0

    def __bytes__(self) -> bytes:
        """Return deposit's attributes as bytes."""
        return (
            bytes(self.pubkey)
            + bytes(self.withdrawal_credentials)
            + self.amount.to_bytes(8, "little")
            + bytes(self.signature)
            + self.index.to_bytes(8, "little")
        )

pubkey: BLSPublicKey instance-attribute

The public key of the beacon chain validator.

withdrawal_credentials: Hash instance-attribute

The withdrawal credentials of the beacon chain validator.

amount: HexNumber instance-attribute

The amount in gwei of the deposit.

signature: BLSSignature instance-attribute

The signature of the deposit using the validator's private key that matches the pubkey.

index: HexNumber instance-attribute

The index of the deposit.

__bytes__()

Return deposit's attributes as bytes.

Source code in src/ethereum_test_types/types.py

def __bytes__(self) -> bytes:
    """Return deposit's attributes as bytes."""
    return (
        bytes(self.pubkey)
        + bytes(self.withdrawal_credentials)
        + self.amount.to_bytes(8, "little")
        + bytes(self.signature)
        + self.index.to_bytes(8, "little")
    )

Environment

Bases: EnvironmentGeneric[ZeroPaddedHexNumber]

Structure used to keep track of the context in which a block must be executed.

Source code in src/ethereum_test_types/types.py

class Environment(EnvironmentGeneric[ZeroPaddedHexNumber]):
    """
    Structure used to keep track of the context in which a block
    must be executed.
    """

    blob_gas_used: ZeroPaddedHexNumber | None = Field(None, alias="currentBlobGasUsed")
    parent_ommers_hash: Hash = Field(Hash(EmptyOmmersRoot), alias="parentUncleHash")
    parent_blob_gas_used: ZeroPaddedHexNumber | None = Field(None)
    parent_excess_blob_gas: ZeroPaddedHexNumber | None = Field(None)
    parent_beacon_block_root: Hash | None = Field(None)

    block_hashes: Dict[Number, Hash] = Field(default_factory=dict)
    ommers: List[Hash] = Field(default_factory=list)
    withdrawals: List[Withdrawal] | None = Field(None)
    extra_data: Bytes = Field(Bytes(b"\x00"), exclude=True)

    @computed_field  # type: ignore[misc]
    @cached_property
    def parent_hash(self) -> Hash | None:
        """
        Obtains the latest hash according to the highest block number in
        `block_hashes`.
        """
        if len(self.block_hashes) == 0:
            return None

        last_index = max(self.block_hashes.keys())
        return Hash(self.block_hashes[last_index])

    def set_fork_requirements(self, fork: Fork) -> "Environment":
        """Fill required fields in an environment depending on the fork."""
        number = self.number
        timestamp = self.timestamp

        updated_values: Dict[str, Any] = {}

        if fork.header_prev_randao_required(number, timestamp) and self.prev_randao is None:
            updated_values["prev_randao"] = 0

        if fork.header_withdrawals_required(number, timestamp) and self.withdrawals is None:
            updated_values["withdrawals"] = []

        if (
            fork.header_base_fee_required(number, timestamp)
            and self.base_fee_per_gas is None
            and self.parent_base_fee_per_gas is None
        ):
            updated_values["base_fee_per_gas"] = DEFAULT_BASE_FEE

        if fork.header_zero_difficulty_required(number, timestamp):
            updated_values["difficulty"] = 0
        elif self.difficulty is None and self.parent_difficulty is None:
            updated_values["difficulty"] = 0x20000

        if (
            fork.header_excess_blob_gas_required(number, timestamp)
            and self.excess_blob_gas is None
            and self.parent_excess_blob_gas is None
        ):
            updated_values["excess_blob_gas"] = 0

        if (
            fork.header_blob_gas_used_required(number, timestamp)
            and self.blob_gas_used is None
            and self.parent_blob_gas_used is None
        ):
            updated_values["blob_gas_used"] = 0

        if (
            fork.header_beacon_root_required(number, timestamp)
            and self.parent_beacon_block_root is None
        ):
            updated_values["parent_beacon_block_root"] = 0

        return self.copy(**updated_values)

parent_hash: Hash | None cached property

Obtains the latest hash according to the highest block number in block_hashes.

set_fork_requirements(fork)

Fill required fields in an environment depending on the fork.

Source code in src/ethereum_test_types/types.py

def set_fork_requirements(self, fork: Fork) -> "Environment":
    """Fill required fields in an environment depending on the fork."""
    number = self.number
    timestamp = self.timestamp

    updated_values: Dict[str, Any] = {}

    if fork.header_prev_randao_required(number, timestamp) and self.prev_randao is None:
        updated_values["prev_randao"] = 0

    if fork.header_withdrawals_required(number, timestamp) and self.withdrawals is None:
        updated_values["withdrawals"] = []

    if (
        fork.header_base_fee_required(number, timestamp)
        and self.base_fee_per_gas is None
        and self.parent_base_fee_per_gas is None
    ):
        updated_values["base_fee_per_gas"] = DEFAULT_BASE_FEE

    if fork.header_zero_difficulty_required(number, timestamp):
        updated_values["difficulty"] = 0
    elif self.difficulty is None and self.parent_difficulty is None:
        updated_values["difficulty"] = 0x20000

    if (
        fork.header_excess_blob_gas_required(number, timestamp)
        and self.excess_blob_gas is None
        and self.parent_excess_blob_gas is None
    ):
        updated_values["excess_blob_gas"] = 0

    if (
        fork.header_blob_gas_used_required(number, timestamp)
        and self.blob_gas_used is None
        and self.parent_blob_gas_used is None
    ):
        updated_values["blob_gas_used"] = 0

    if (
        fork.header_beacon_root_required(number, timestamp)
        and self.parent_beacon_block_root is None
    ):
        updated_values["parent_beacon_block_root"] = 0

    return self.copy(**updated_values)

NetworkWrappedTransaction

Bases: CamelModel, RLPSerializable

Network wrapped transaction as defined in EIP-4844.

Source code in src/ethereum_test_types/types.py

class NetworkWrappedTransaction(CamelModel, RLPSerializable):
    """
    Network wrapped transaction as defined in
    [EIP-4844](https://eips.ethereum.org/EIPS/eip-4844#networking).
    """

    tx: Transaction

    blobs: Sequence[Bytes] | None = Field(None, exclude=True)
    blob_kzg_commitments: Sequence[Bytes] | None = Field(None, exclude=True)
    blob_kzg_proofs: Sequence[Bytes] | None = Field(None, exclude=True)

    rlp_fields: ClassVar[List[str]] = ["tx", "blobs", "blob_kzg_commitments", "blob_kzg_proofs"]

    def get_rlp_prefix(self) -> bytes:
        """
        Return the transaction type as bytes to be appended at the beginning of the
        serialized transaction if type is not 0.
        """
        if self.tx.ty > 0:
            return bytes([self.tx.ty])
        return b""

get_rlp_prefix()

Return the transaction type as bytes to be appended at the beginning of the serialized transaction if type is not 0.

Source code in src/ethereum_test_types/types.py

def get_rlp_prefix(self) -> bytes:
    """
    Return the transaction type as bytes to be appended at the beginning of the
    serialized transaction if type is not 0.
    """
    if self.tx.ty > 0:
        return bytes([self.tx.ty])
    return b""

Removable

Sentinel class to detect if a parameter should be removed. (None normally means "do not modify").

Source code in src/ethereum_test_types/types.py

class Removable:
    """
    Sentinel class to detect if a parameter should be removed.
    (`None` normally means "do not modify").
    """

    def __eq__(self, other: Any) -> bool:
        """Return True for all Removable."""
        if not isinstance(other, Removable):
            return NotImplemented
        return True

__eq__(other)

Return True for all Removable.

Source code in src/ethereum_test_types/types.py

def __eq__(self, other: Any) -> bool:
    """Return True for all Removable."""
    if not isinstance(other, Removable):
        return NotImplemented
    return True

Requests

Requests for the transition tool.

Source code in src/ethereum_test_types/types.py

class Requests:
    """Requests for the transition tool."""

    requests_list: List[Bytes]

    def __init__(
        self,
        *requests: RequestBase,
        requests_lists: List[List[RequestBase] | Bytes] | None = None,
    ):
        """Initialize requests object."""
        if requests_lists is not None:
            assert len(requests) == 0, "requests must be empty if list is provided"
            self.requests_list = []
            for requests_list in requests_lists:
                self.requests_list.append(requests_list_to_bytes(requests_list))
            return
        else:
            lists: Dict[int, List[RequestBase]] = defaultdict(list)
            for r in requests:
                lists[r.type].append(r)

            self.requests_list = [
                Bytes(bytes([request_type]) + requests_list_to_bytes(lists[request_type]))
                for request_type in sorted(lists.keys())
            ]

    def __bytes__(self) -> bytes:
        """Return requests hash."""
        s: bytes = b"".join(r.sha256() for r in self.requests_list)
        return Bytes(s).sha256()

__init__(*requests, requests_lists=None)

Initialize requests object.

Source code in src/ethereum_test_types/types.py

def __init__(
    self,
    *requests: RequestBase,
    requests_lists: List[List[RequestBase] | Bytes] | None = None,
):
    """Initialize requests object."""
    if requests_lists is not None:
        assert len(requests) == 0, "requests must be empty if list is provided"
        self.requests_list = []
        for requests_list in requests_lists:
            self.requests_list.append(requests_list_to_bytes(requests_list))
        return
    else:
        lists: Dict[int, List[RequestBase]] = defaultdict(list)
        for r in requests:
            lists[r.type].append(r)

        self.requests_list = [
            Bytes(bytes([request_type]) + requests_list_to_bytes(lists[request_type]))
            for request_type in sorted(lists.keys())
        ]

__bytes__()

Return requests hash.

Source code in src/ethereum_test_types/types.py

def __bytes__(self) -> bytes:
    """Return requests hash."""
    s: bytes = b"".join(r.sha256() for r in self.requests_list)
    return Bytes(s).sha256()

Storage

Bases: EthereumTestRootModel[Dict[StorageKeyValueType, StorageKeyValueType]]

Definition of contract storage in the pre or post state of a test.

This model accepts a dictionary with keys and values as any of: str, int, bytes, or any type that supports conversion to bytes, and automatically casts them to HashInt.

Source code in src/ethereum_test_base_types/composite_types.py

class Storage(EthereumTestRootModel[Dict[StorageKeyValueType, StorageKeyValueType]]):
    """
    Definition of contract storage in the `pre` or `post` state of a test.

    This model accepts a dictionary with keys and values as any of: str, int,
    bytes, or any type that supports conversion to bytes, and automatically
    casts them to `HashInt`.
    """

    # internal storage is maintained as a dict with HashInt keys and values.
    root: Dict[StorageKeyValueType, StorageKeyValueType] = Field(default_factory=dict)

    _current_slot: int = PrivateAttr(0)
    _hint_map: Dict[StorageKeyValueType, str] = PrivateAttr(default_factory=dict)
    _any_map: Dict[StorageKeyValueType, bool] = PrivateAttr(default_factory=dict)

    StorageDictType: ClassVar[TypeAlias] = Dict[
        str | int | bytes | SupportsBytes, str | int | bytes | SupportsBytes
    ]
    """
    Dictionary type to be used when defining an input to initialize a storage.
    """

    @dataclass(kw_only=True)
    class InvalidTypeError(Exception):
        """Invalid type used when describing test's expected storage key or value."""

        key_or_value: Any

        def __init__(self, key_or_value: Any, *args):
            """Initialize the exception with the invalid type."""
            super().__init__(args)
            self.key_or_value = key_or_value

        def __str__(self):
            """Print exception string."""
            return f"invalid type for key/value: {self.key_or_value}"

    @dataclass(kw_only=True)
    class InvalidValueError(Exception):
        """
        Invalid value used when describing test's expected storage key or
        value.
        """

        key_or_value: Any

        def __init__(self, key_or_value: Any, *args):
            """Initialize the exception with the invalid value."""
            super().__init__(args)
            self.key_or_value = key_or_value

        def __str__(self):
            """Print exception string."""
            return f"invalid value for key/value: {self.key_or_value}"

    @dataclass(kw_only=True)
    class MissingKeyError(Exception):
        """Test expected to find a storage key set but key was missing."""

        key: int

        def __init__(self, key: int, *args):
            """Initialize the exception with the missing key."""
            super().__init__(args)
            self.key = key

        def __str__(self):
            """Print exception string."""
            return "key {0} not found in storage".format(Hash(self.key))

    @dataclass(kw_only=True)
    class KeyValueMismatchError(Exception):
        """
        Test expected a certain value in a storage key but value found
        was different.
        """

        address: Address
        key: int
        want: int
        got: int
        hint: str

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

        def __str__(self):
            """Print exception string."""
            label_str = ""
            if self.address.label is not None:
                label_str = f" ({self.address.label})"
            return (
                f"incorrect value in address {self.address}{label_str} for "
                + f"key {Hash(self.key)}{f' ({self.hint})' if self.hint else ''}:"
                + f" want {HexNumber(self.want)} (dec:{int(self.want)}),"
                + f" got {HexNumber(self.got)} (dec:{int(self.got)})"
            )

    def __contains__(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType) -> bool:
        """Check for an item in the storage."""
        return StorageKeyValueTypeAdapter.validate_python(key) in self.root

    def __getitem__(
        self, key: StorageKeyValueTypeConvertible | StorageKeyValueType
    ) -> StorageKeyValueType:
        """Return an item from the storage."""
        return self.root[StorageKeyValueTypeAdapter.validate_python(key)]

    def __setitem__(
        self,
        key: StorageKeyValueTypeConvertible | StorageKeyValueType,
        value: StorageKeyValueTypeConvertible | StorageKeyValueType,
    ):  # noqa: SC200
        """Set an item in the storage."""
        self.root[StorageKeyValueTypeAdapter.validate_python(key)] = (
            StorageKeyValueTypeAdapter.validate_python(value)
        )

    def __delitem__(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType):
        """Delete an item from the storage."""
        del self.root[StorageKeyValueTypeAdapter.validate_python(key)]

    def __iter__(self):
        """Return an iterator over the storage."""
        return iter(self.root)

    def __eq__(self, other) -> bool:
        """Return True if both storages are equal."""
        if not isinstance(other, Storage):
            return False
        return self.root == other.root

    def __ne__(self, other) -> bool:
        """Return True if both storages are not equal."""
        if not isinstance(other, Storage):
            return False
        return self.root != other.root

    def __bool__(self) -> bool:
        """Return True if the storage is not empty."""
        return any(v for v in self.root.values())

    def __add__(self, other: "Storage") -> "Storage":
        """Return a new storage that is the sum of two storages."""
        return Storage({**self.root, **other.root})

    def keys(self) -> set[StorageKeyValueType]:
        """Return the keys of the storage."""
        return set(self.root.keys())

    def set_next_slot(self, slot: int) -> "Storage":
        """Set the next slot to be used by `store_next`."""
        self._current_slot = slot
        return self

    def items(self):
        """Return the items of the storage."""
        return self.root.items()

    def set_expect_any(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType):
        """Mark key to be able to have any expected value when comparing storages."""
        self._any_map[StorageKeyValueTypeAdapter.validate_python(key)] = True

    def store_next(
        self, value: StorageKeyValueTypeConvertible | StorageKeyValueType | bool, hint: str = ""
    ) -> StorageKeyValueType:
        """
        Store a value in the storage and returns the key where the value is stored.

        Increments the key counter so the next time this function is called,
        the next key is used.
        """
        slot = StorageKeyValueTypeAdapter.validate_python(self._current_slot)
        self._current_slot += 1
        if hint:
            self._hint_map[slot] = hint
        self[slot] = StorageKeyValueTypeAdapter.validate_python(value)
        return slot

    def peek_slot(self) -> int:
        """Peek the next slot that will be used by `store_next`."""
        return self._current_slot

    def contains(self, other: "Storage") -> bool:
        """
        Return True if self contains all keys with equal value as
        contained by second storage.
        Used for comparison with test expected post state and alloc returned
        by the transition tool.
        """
        for key in other.keys():
            if key not in self:
                return False
            if self[key] != other[key]:
                return False
        return True

    def must_contain(self, address: Address, other: "Storage"):
        """
        Succeeds only if self contains all keys with equal value as
        contained by second storage.
        Used for comparison with test expected post state and alloc returned
        by the transition tool.
        Raises detailed exception when a difference is found.
        """
        for key in other.keys():
            if key not in self:
                # storage[key]==0 is equal to missing storage
                if other[key] != 0:
                    raise Storage.MissingKeyError(key=key)
            elif self[key] != other[key]:
                raise Storage.KeyValueMismatchError(
                    address=address,
                    key=key,
                    want=self[key],
                    got=other[key],
                    hint=self._hint_map.get(key, ""),
                )

    def must_be_equal(self, address: Address, other: "Storage | None"):
        """Succeed only if "self" is equal to "other" storage."""
        # Test keys contained in both storage objects
        if other is None:
            other = Storage({})
        for key in self.keys() & other.keys():
            if self[key] != other[key]:
                raise Storage.KeyValueMismatchError(
                    address=address,
                    key=key,
                    want=self[key],
                    got=other[key],
                    hint=self._hint_map.get(key, ""),
                )

        # Test keys contained in either one of the storage objects
        for key in self.keys() ^ other.keys():
            if key in self:
                if self[key] != 0:
                    raise Storage.KeyValueMismatchError(
                        address=address,
                        key=key,
                        want=self[key],
                        got=0,
                        hint=self._hint_map.get(key, ""),
                    )

            elif other[key] != 0:
                # Skip key verification if we allow this key to be ANY
                if self._any_map.get(key) is True:
                    continue
                raise Storage.KeyValueMismatchError(
                    address=address,
                    key=key,
                    want=0,
                    got=other[key],
                    hint=self._hint_map.get(key, ""),
                )

    def canary(self) -> "Storage":
        """
        Return a canary storage filled with non-zero values where the current storage expects
        zero values, to guarantee that the test overwrites the storage.
        """
        return Storage({key: HashInt(0xBA5E) for key in self.keys() if self[key] == 0})

StorageDictType: TypeAlias = Dict[str | int | bytes | SupportsBytes, str | int | bytes | SupportsBytes] class-attribute

Dictionary type to be used when defining an input to initialize a storage.

InvalidTypeError dataclass

Bases: Exception

Invalid type used when describing test's expected storage key or value.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class InvalidTypeError(Exception):
    """Invalid type used when describing test's expected storage key or value."""

    key_or_value: Any

    def __init__(self, key_or_value: Any, *args):
        """Initialize the exception with the invalid type."""
        super().__init__(args)
        self.key_or_value = key_or_value

    def __str__(self):
        """Print exception string."""
        return f"invalid type for key/value: {self.key_or_value}"

__init__(key_or_value, *args)

Initialize the exception with the invalid type.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, key_or_value: Any, *args):
    """Initialize the exception with the invalid type."""
    super().__init__(args)
    self.key_or_value = key_or_value

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    return f"invalid type for key/value: {self.key_or_value}"

InvalidValueError dataclass

Bases: Exception

Invalid value used when describing test's expected storage key or value.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class InvalidValueError(Exception):
    """
    Invalid value used when describing test's expected storage key or
    value.
    """

    key_or_value: Any

    def __init__(self, key_or_value: Any, *args):
        """Initialize the exception with the invalid value."""
        super().__init__(args)
        self.key_or_value = key_or_value

    def __str__(self):
        """Print exception string."""
        return f"invalid value for key/value: {self.key_or_value}"

__init__(key_or_value, *args)

Initialize the exception with the invalid value.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, key_or_value: Any, *args):
    """Initialize the exception with the invalid value."""
    super().__init__(args)
    self.key_or_value = key_or_value

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    return f"invalid value for key/value: {self.key_or_value}"

MissingKeyError dataclass

Bases: Exception

Test expected to find a storage key set but key was missing.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class MissingKeyError(Exception):
    """Test expected to find a storage key set but key was missing."""

    key: int

    def __init__(self, key: int, *args):
        """Initialize the exception with the missing key."""
        super().__init__(args)
        self.key = key

    def __str__(self):
        """Print exception string."""
        return "key {0} not found in storage".format(Hash(self.key))

__init__(key, *args)

Initialize the exception with the missing key.

Source code in src/ethereum_test_base_types/composite_types.py

def __init__(self, key: int, *args):
    """Initialize the exception with the missing key."""
    super().__init__(args)
    self.key = key

__str__()

Print exception string.

Source code in src/ethereum_test_base_types/composite_types.py

def __str__(self):
    """Print exception string."""
    return "key {0} not found in storage".format(Hash(self.key))

KeyValueMismatchError dataclass

Bases: Exception

Test expected a certain value in a storage key but value found was different.

Source code in src/ethereum_test_base_types/composite_types.py

@dataclass(kw_only=True)
class KeyValueMismatchError(Exception):
    """
    Test expected a certain value in a storage key but value found
    was different.
    """

    address: Address
    key: int
    want: int
    got: int
    hint: str

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

    def __str__(self):
        """Print exception string."""
        label_str = ""
        if self.address.label is not None:
            label_str = f" ({self.address.label})"
        return (
            f"incorrect value in address {self.address}{label_str} for "
            + f"key {Hash(self.key)}{f' ({self.hint})' if self.hint else ''}:"
            + f" want {HexNumber(self.want)} (dec:{int(self.want)}),"
            + f" got {HexNumber(self.got)} (dec:{int(self.got)})"
        )

__init__(address, key, want, got, hint='', *args)

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

Source code in src/ethereum_test_base_types/composite_types.py

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

__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"incorrect value in address {self.address}{label_str} for "
        + f"key {Hash(self.key)}{f' ({self.hint})' if self.hint else ''}:"
        + f" want {HexNumber(self.want)} (dec:{int(self.want)}),"
        + f" got {HexNumber(self.got)} (dec:{int(self.got)})"
    )

__contains__(key)

Check for an item in the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def __contains__(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType) -> bool:
    """Check for an item in the storage."""
    return StorageKeyValueTypeAdapter.validate_python(key) in self.root

__getitem__(key)

Return an item from the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def __getitem__(
    self, key: StorageKeyValueTypeConvertible | StorageKeyValueType
) -> StorageKeyValueType:
    """Return an item from the storage."""
    return self.root[StorageKeyValueTypeAdapter.validate_python(key)]

__setitem__(key, value)

Set an item in the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def __setitem__(
    self,
    key: StorageKeyValueTypeConvertible | StorageKeyValueType,
    value: StorageKeyValueTypeConvertible | StorageKeyValueType,
):  # noqa: SC200
    """Set an item in the storage."""
    self.root[StorageKeyValueTypeAdapter.validate_python(key)] = (
        StorageKeyValueTypeAdapter.validate_python(value)
    )

__delitem__(key)

Delete an item from the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def __delitem__(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType):
    """Delete an item from the storage."""
    del self.root[StorageKeyValueTypeAdapter.validate_python(key)]

__iter__()

Return an iterator over the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def __iter__(self):
    """Return an iterator over the storage."""
    return iter(self.root)

__eq__(other)

Return True if both storages are equal.

Source code in src/ethereum_test_base_types/composite_types.py

def __eq__(self, other) -> bool:
    """Return True if both storages are equal."""
    if not isinstance(other, Storage):
        return False
    return self.root == other.root

__ne__(other)

Return True if both storages are not equal.

Source code in src/ethereum_test_base_types/composite_types.py

def __ne__(self, other) -> bool:
    """Return True if both storages are not equal."""
    if not isinstance(other, Storage):
        return False
    return self.root != other.root

__bool__()

Return True if the storage is not empty.

Source code in src/ethereum_test_base_types/composite_types.py

def __bool__(self) -> bool:
    """Return True if the storage is not empty."""
    return any(v for v in self.root.values())

__add__(other)

Return a new storage that is the sum of two storages.

Source code in src/ethereum_test_base_types/composite_types.py

def __add__(self, other: "Storage") -> "Storage":
    """Return a new storage that is the sum of two storages."""
    return Storage({**self.root, **other.root})

keys()

Return the keys of the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def keys(self) -> set[StorageKeyValueType]:
    """Return the keys of the storage."""
    return set(self.root.keys())

set_next_slot(slot)

Set the next slot to be used by store_next.

Source code in src/ethereum_test_base_types/composite_types.py

def set_next_slot(self, slot: int) -> "Storage":
    """Set the next slot to be used by `store_next`."""
    self._current_slot = slot
    return self

items()

Return the items of the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def items(self):
    """Return the items of the storage."""
    return self.root.items()

set_expect_any(key)

Mark key to be able to have any expected value when comparing storages.

Source code in src/ethereum_test_base_types/composite_types.py

def set_expect_any(self, key: StorageKeyValueTypeConvertible | StorageKeyValueType):
    """Mark key to be able to have any expected value when comparing storages."""
    self._any_map[StorageKeyValueTypeAdapter.validate_python(key)] = True

store_next(value, hint='')

Store a value in the storage and returns the key where the value is stored.

Increments the key counter so the next time this function is called, the next key is used.

Source code in src/ethereum_test_base_types/composite_types.py

def store_next(
    self, value: StorageKeyValueTypeConvertible | StorageKeyValueType | bool, hint: str = ""
) -> StorageKeyValueType:
    """
    Store a value in the storage and returns the key where the value is stored.

    Increments the key counter so the next time this function is called,
    the next key is used.
    """
    slot = StorageKeyValueTypeAdapter.validate_python(self._current_slot)
    self._current_slot += 1
    if hint:
        self._hint_map[slot] = hint
    self[slot] = StorageKeyValueTypeAdapter.validate_python(value)
    return slot

peek_slot()

Peek the next slot that will be used by store_next.

Source code in src/ethereum_test_base_types/composite_types.py

def peek_slot(self) -> int:
    """Peek the next slot that will be used by `store_next`."""
    return self._current_slot

contains(other)

Return True if self contains all keys with equal value as contained by second storage. Used for comparison with test expected post state and alloc returned by the transition tool.

Source code in src/ethereum_test_base_types/composite_types.py

def contains(self, other: "Storage") -> bool:
    """
    Return True if self contains all keys with equal value as
    contained by second storage.
    Used for comparison with test expected post state and alloc returned
    by the transition tool.
    """
    for key in other.keys():
        if key not in self:
            return False
        if self[key] != other[key]:
            return False
    return True

must_contain(address, other)

Succeeds only if self contains all keys with equal value as contained by second storage. Used for comparison with test expected post state and alloc returned by the transition tool. Raises detailed exception when a difference is found.

Source code in src/ethereum_test_base_types/composite_types.py

def must_contain(self, address: Address, other: "Storage"):
    """
    Succeeds only if self contains all keys with equal value as
    contained by second storage.
    Used for comparison with test expected post state and alloc returned
    by the transition tool.
    Raises detailed exception when a difference is found.
    """
    for key in other.keys():
        if key not in self:
            # storage[key]==0 is equal to missing storage
            if other[key] != 0:
                raise Storage.MissingKeyError(key=key)
        elif self[key] != other[key]:
            raise Storage.KeyValueMismatchError(
                address=address,
                key=key,
                want=self[key],
                got=other[key],
                hint=self._hint_map.get(key, ""),
            )

must_be_equal(address, other)

Succeed only if "self" is equal to "other" storage.

Source code in src/ethereum_test_base_types/composite_types.py

def must_be_equal(self, address: Address, other: "Storage | None"):
    """Succeed only if "self" is equal to "other" storage."""
    # Test keys contained in both storage objects
    if other is None:
        other = Storage({})
    for key in self.keys() & other.keys():
        if self[key] != other[key]:
            raise Storage.KeyValueMismatchError(
                address=address,
                key=key,
                want=self[key],
                got=other[key],
                hint=self._hint_map.get(key, ""),
            )

    # Test keys contained in either one of the storage objects
    for key in self.keys() ^ other.keys():
        if key in self:
            if self[key] != 0:
                raise Storage.KeyValueMismatchError(
                    address=address,
                    key=key,
                    want=self[key],
                    got=0,
                    hint=self._hint_map.get(key, ""),
                )

        elif other[key] != 0:
            # Skip key verification if we allow this key to be ANY
            if self._any_map.get(key) is True:
                continue
            raise Storage.KeyValueMismatchError(
                address=address,
                key=key,
                want=0,
                got=other[key],
                hint=self._hint_map.get(key, ""),
            )

canary()

Return a canary storage filled with non-zero values where the current storage expects zero values, to guarantee that the test overwrites the storage.

Source code in src/ethereum_test_base_types/composite_types.py

def canary(self) -> "Storage":
    """
    Return a canary storage filled with non-zero values where the current storage expects
    zero values, to guarantee that the test overwrites the storage.
    """
    return Storage({key: HashInt(0xBA5E) for key in self.keys() if self[key] == 0})

Transaction

Bases: TransactionGeneric[HexNumber], TransactionTransitionToolConverter, SignableRLPSerializable

Generic object that can represent all Ethereum transaction types.

Source code in src/ethereum_test_types/types.py

class Transaction(
    TransactionGeneric[HexNumber], TransactionTransitionToolConverter, SignableRLPSerializable
):
    """Generic object that can represent all Ethereum transaction types."""

    gas_limit: HexNumber = Field(HexNumber(21_000), serialization_alias="gas")
    to: Address | None = Field(Address(0xAA))
    data: Bytes = Field(Bytes(b""), alias="input")

    authorization_list: List[AuthorizationTuple] | None = None

    initcodes: List[Bytes] | None = None

    secret_key: Hash | None = None
    error: List[TransactionException] | TransactionException | None = Field(None, exclude=True)

    protected: bool = Field(True, exclude=True)

    expected_receipt: TransactionReceipt | None = Field(None, exclude=True)

    zero: ClassVar[Literal[0]] = 0

    model_config = ConfigDict(validate_assignment=True)

    class InvalidFeePaymentError(Exception):
        """Transaction described more than one fee payment type."""

        def __str__(self):
            """Print exception string."""
            return "only one type of fee payment field can be used in a single tx"

    class InvalidSignaturePrivateKeyError(Exception):
        """
        Transaction describes both the signature and private key of
        source account.
        """

        def __str__(self):
            """Print exception string."""
            return "can't define both 'signature' and 'private_key'"

    def model_post_init(self, __context):
        """Ensure transaction has no conflicting properties."""
        super().model_post_init(__context)

        if self.gas_price is not None and (
            self.max_fee_per_gas is not None
            or self.max_priority_fee_per_gas is not None
            or self.max_fee_per_blob_gas is not None
        ):
            raise Transaction.InvalidFeePaymentError()

        if "ty" not in self.model_fields_set:
            # Try to deduce transaction type from included fields
            if self.initcodes is not None:
                self.ty = 6
            elif self.authorization_list is not None:
                self.ty = 4
            elif self.max_fee_per_blob_gas is not None or self.blob_versioned_hashes is not None:
                self.ty = 3
            elif self.max_fee_per_gas is not None or self.max_priority_fee_per_gas is not None:
                self.ty = 2
            elif self.access_list is not None:
                self.ty = 1
            else:
                self.ty = 0

        if "v" in self.model_fields_set and self.secret_key is not None:
            raise Transaction.InvalidSignaturePrivateKeyError()

        if "v" not in self.model_fields_set and self.secret_key is None:
            if self.sender is not None:
                self.secret_key = self.sender.key
            else:
                self.secret_key = Hash(TestPrivateKey)
                self.sender = EOA(address=TestAddress, key=self.secret_key, nonce=0)

        # Set default values for fields that are required for certain tx types
        if self.ty <= 1 and self.gas_price is None:
            self.gas_price = TransactionDefaults.gas_price
        if self.ty >= 1 and self.access_list is None:
            self.access_list = []
        if self.ty < 1:
            assert self.access_list is None, "access_list must be None"

        if self.ty >= 2 and self.max_fee_per_gas is None:
            self.max_fee_per_gas = TransactionDefaults.max_fee_per_gas
        if self.ty >= 2 and self.max_priority_fee_per_gas is None:
            self.max_priority_fee_per_gas = TransactionDefaults.max_priority_fee_per_gas
        if self.ty < 2:
            assert self.max_fee_per_gas is None, "max_fee_per_gas must be None"
            assert self.max_priority_fee_per_gas is None, "max_priority_fee_per_gas must be None"

        if self.ty == 3 and self.max_fee_per_blob_gas is None:
            self.max_fee_per_blob_gas = 1
        if self.ty != 3:
            assert self.blob_versioned_hashes is None, "blob_versioned_hashes must be None"
            assert self.max_fee_per_blob_gas is None, "max_fee_per_blob_gas must be None"

        if self.ty == 4 and self.authorization_list is None:
            self.authorization_list = []
        if self.ty != 4:
            assert self.authorization_list is None, "authorization_list must be None"

        if self.ty == 6 and self.initcodes is None:
            self.initcodes = []
        if self.ty != 6:
            assert self.initcodes is None, "initcodes must be None"

        if "nonce" not in self.model_fields_set and self.sender is not None:
            self.nonce = HexNumber(self.sender.get_nonce())

    def with_error(
        self, error: List[TransactionException] | TransactionException
    ) -> "Transaction":
        """Create a copy of the transaction with an added error."""
        return self.copy(error=error)

    def with_nonce(self, nonce: int) -> "Transaction":
        """Create a copy of the transaction with a modified nonce."""
        return self.copy(nonce=nonce)

    @cached_property
    def signature_bytes(self) -> Bytes:
        """Returns the serialized bytes of the transaction signature."""
        assert "v" in self.model_fields_set, "transaction must be signed"
        v = int(self.v)
        if self.ty == 0:
            if self.protected:
                assert self.chain_id is not None
                v -= 35 + (self.chain_id * 2)
            else:
                v -= 27
        return Bytes(
            self.r.to_bytes(32, byteorder="big")
            + self.s.to_bytes(32, byteorder="big")
            + bytes([v])
        )

    def sign(self: "Transaction"):
        """Signs the authorization tuple with a private key."""
        signature_bytes: bytes | None = None
        rlp_signing_bytes = self.rlp_signing_bytes()
        if (
            "v" not in self.model_fields_set
            and "r" not in self.model_fields_set
            and "s" not in self.model_fields_set
        ):
            signing_key: Hash | None = None
            if self.secret_key is not None:
                signing_key = self.secret_key
                self.secret_key = None
            elif self.sender is not None:
                eoa = self.sender
                assert eoa is not None, "signer must be set"
                signing_key = eoa.key
            assert signing_key is not None, "secret_key or signer must be set"

            signature_bytes = PrivateKey(secret=signing_key).sign_recoverable(
                rlp_signing_bytes, hasher=keccak256
            )
            v, r, s = (
                signature_bytes[64],
                int.from_bytes(signature_bytes[0:32], byteorder="big"),
                int.from_bytes(signature_bytes[32:64], byteorder="big"),
            )
            if self.ty == 0:
                if self.protected:
                    v += 35 + (self.chain_id * 2)
                else:  # not protected
                    v += 27
            self.v, self.r, self.s = (HexNumber(v), HexNumber(r), HexNumber(s))
            self.model_fields_set.add("v")
            self.model_fields_set.add("r")
            self.model_fields_set.add("s")

        if self.sender is None:
            try:
                if not signature_bytes:
                    signature_bytes = (
                        int(self.r).to_bytes(32, byteorder="big")
                        + int(self.s).to_bytes(32, byteorder="big")
                        + bytes([self.v])
                    )
                public_key = PublicKey.from_signature_and_message(
                    signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
                )
                self.signer = EOA(
                    address=Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
                )
            except Exception:
                # Signer remains `None` in this case
                pass

    def with_signature_and_sender(self, *, keep_secret_key: bool = False) -> "Transaction":
        """Return signed version of the transaction using the private key."""
        updated_values: Dict[str, Any] = {}

        if (
            "v" in self.model_fields_set
            or "r" in self.model_fields_set
            or "s" in self.model_fields_set
        ):
            # Transaction already signed
            if self.sender is not None:
                return self

            public_key = PublicKey.from_signature_and_message(
                self.signature_bytes, self.rlp_signing_bytes().keccak256(), hasher=None
            )
            updated_values["sender"] = Address(
                keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
            )
            return self.copy(**updated_values)

        if self.secret_key is None:
            raise ValueError("secret_key must be set to sign a transaction")

        # Get the signing bytes
        signing_hash = self.rlp_signing_bytes().keccak256()

        # Sign the bytes
        signature_bytes = PrivateKey(secret=self.secret_key).sign_recoverable(
            signing_hash, hasher=None
        )
        public_key = PublicKey.from_signature_and_message(
            signature_bytes, signing_hash, hasher=None
        )

        sender = keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
        updated_values["sender"] = Address(sender)

        v, r, s = (
            signature_bytes[64],
            int.from_bytes(signature_bytes[0:32], byteorder="big"),
            int.from_bytes(signature_bytes[32:64], byteorder="big"),
        )
        if self.ty == 0:
            if self.protected:
                v += 35 + (self.chain_id * 2)
            else:  # not protected
                v += 27

        updated_values["v"] = HexNumber(v)
        updated_values["r"] = HexNumber(r)
        updated_values["s"] = HexNumber(s)

        updated_values["secret_key"] = None

        updated_tx: "Transaction" = self.model_copy(update=updated_values)

        # Remove the secret key if requested
        if keep_secret_key:
            updated_tx.secret_key = self.secret_key
        return updated_tx

    def get_rlp_signing_fields(self) -> List[str]:
        """
        Return the list of values included in the envelope used for signing depending on
        the transaction type.
        """
        field_list: List[str]
        if self.ty == 6:
            # EIP-7873: https://eips.ethereum.org/EIPS/eip-7873
            field_list = [
                "chain_id",
                "nonce",
                "max_priority_fee_per_gas",
                "max_fee_per_gas",
                "gas_limit",
                "to",
                "value",
                "data",
                "access_list",
                "initcodes",
            ]
        elif self.ty == 4:
            # EIP-7702: https://eips.ethereum.org/EIPS/eip-7702
            field_list = [
                "chain_id",
                "nonce",
                "max_priority_fee_per_gas",
                "max_fee_per_gas",
                "gas_limit",
                "to",
                "value",
                "data",
                "access_list",
                "authorization_list",
            ]
        elif self.ty == 3:
            # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
            field_list = [
                "chain_id",
                "nonce",
                "max_priority_fee_per_gas",
                "max_fee_per_gas",
                "gas_limit",
                "to",
                "value",
                "data",
                "access_list",
                "max_fee_per_blob_gas",
                "blob_versioned_hashes",
            ]
        elif self.ty == 2:
            # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
            field_list = [
                "chain_id",
                "nonce",
                "max_priority_fee_per_gas",
                "max_fee_per_gas",
                "gas_limit",
                "to",
                "value",
                "data",
                "access_list",
            ]
        elif self.ty == 1:
            # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
            field_list = [
                "chain_id",
                "nonce",
                "gas_price",
                "gas_limit",
                "to",
                "value",
                "data",
                "access_list",
            ]
        elif self.ty == 0:
            field_list = ["nonce", "gas_price", "gas_limit", "to", "value", "data"]
            if self.protected:
                # EIP-155: https://eips.ethereum.org/EIPS/eip-155
                field_list.extend(["chain_id", "zero", "zero"])
        else:
            raise NotImplementedError(f"signing for transaction type {self.ty} not implemented")

        for field in field_list:
            if field != "to":
                assert getattr(self, field) is not None, (
                    f"{field} must be set for type {self.ty} tx"
                )
        return field_list

    def get_rlp_fields(self) -> List[str]:
        """
        Return the list of values included in the list used for rlp encoding depending on
        the transaction type.
        """
        fields = self.get_rlp_signing_fields()
        if self.ty == 0 and self.protected:
            fields = fields[:-3]
        return fields + ["v", "r", "s"]

    def get_rlp_prefix(self) -> bytes:
        """
        Return the transaction type as bytes to be appended at the beginning of the
        serialized transaction if type is not 0.
        """
        if self.ty > 0:
            return bytes([self.ty])
        return b""

    def get_rlp_signing_prefix(self) -> bytes:
        """
        Return the transaction type as bytes to be appended at the beginning of the
        serialized transaction signing envelope if type is not 0.
        """
        if self.ty > 0:
            return bytes([self.ty])
        return b""

    @cached_property
    def hash(self) -> Hash:
        """Returns hash of the transaction."""
        return self.rlp().keccak256()

    @cached_property
    def serializable_list(self) -> Any:
        """Return list of values included in the transaction as a serializable object."""
        return self.rlp() if self.ty > 0 else self.to_list(signing=False)

    @staticmethod
    def list_root(input_txs: List["Transaction"]) -> Hash:
        """Return transactions root of a list of transactions."""
        t = HexaryTrie(db={})
        for i, tx in enumerate(input_txs):
            t.set(eth_rlp.encode(Uint(i)), tx.rlp())
        return Hash(t.root_hash)

    @staticmethod
    def list_blob_versioned_hashes(input_txs: List["Transaction"]) -> List[Hash]:
        """Get list of ordered blob versioned hashes from a list of transactions."""
        return [
            blob_versioned_hash
            for tx in input_txs
            if tx.blob_versioned_hashes is not None
            for blob_versioned_hash in tx.blob_versioned_hashes
        ]

    @cached_property
    def created_contract(self) -> Address:
        """Return address of the contract created by the transaction."""
        if self.to is not None:
            raise ValueError("transaction is not a contract creation")
        if self.sender is None:
            raise ValueError("sender address is None")
        hash_bytes = Bytes(eth_rlp.encode([self.sender, int_to_bytes(self.nonce)])).keccak256()
        return Address(hash_bytes[-20:])

InvalidFeePaymentError

Bases: Exception

Transaction described more than one fee payment type.

Source code in src/ethereum_test_types/types.py

class InvalidFeePaymentError(Exception):
    """Transaction described more than one fee payment type."""

    def __str__(self):
        """Print exception string."""
        return "only one type of fee payment field can be used in a single tx"

__str__()

Print exception string.

Source code in src/ethereum_test_types/types.py

def __str__(self):
    """Print exception string."""
    return "only one type of fee payment field can be used in a single tx"

InvalidSignaturePrivateKeyError

Bases: Exception

Transaction describes both the signature and private key of source account.

Source code in src/ethereum_test_types/types.py

class InvalidSignaturePrivateKeyError(Exception):
    """
    Transaction describes both the signature and private key of
    source account.
    """

    def __str__(self):
        """Print exception string."""
        return "can't define both 'signature' and 'private_key'"

__str__()

Print exception string.

Source code in src/ethereum_test_types/types.py

def __str__(self):
    """Print exception string."""
    return "can't define both 'signature' and 'private_key'"

model_post_init(__context)

Ensure transaction has no conflicting properties.

Source code in src/ethereum_test_types/types.py

def model_post_init(self, __context):
    """Ensure transaction has no conflicting properties."""
    super().model_post_init(__context)

    if self.gas_price is not None and (
        self.max_fee_per_gas is not None
        or self.max_priority_fee_per_gas is not None
        or self.max_fee_per_blob_gas is not None
    ):
        raise Transaction.InvalidFeePaymentError()

    if "ty" not in self.model_fields_set:
        # Try to deduce transaction type from included fields
        if self.initcodes is not None:
            self.ty = 6
        elif self.authorization_list is not None:
            self.ty = 4
        elif self.max_fee_per_blob_gas is not None or self.blob_versioned_hashes is not None:
            self.ty = 3
        elif self.max_fee_per_gas is not None or self.max_priority_fee_per_gas is not None:
            self.ty = 2
        elif self.access_list is not None:
            self.ty = 1
        else:
            self.ty = 0

    if "v" in self.model_fields_set and self.secret_key is not None:
        raise Transaction.InvalidSignaturePrivateKeyError()

    if "v" not in self.model_fields_set and self.secret_key is None:
        if self.sender is not None:
            self.secret_key = self.sender.key
        else:
            self.secret_key = Hash(TestPrivateKey)
            self.sender = EOA(address=TestAddress, key=self.secret_key, nonce=0)

    # Set default values for fields that are required for certain tx types
    if self.ty <= 1 and self.gas_price is None:
        self.gas_price = TransactionDefaults.gas_price
    if self.ty >= 1 and self.access_list is None:
        self.access_list = []
    if self.ty < 1:
        assert self.access_list is None, "access_list must be None"

    if self.ty >= 2 and self.max_fee_per_gas is None:
        self.max_fee_per_gas = TransactionDefaults.max_fee_per_gas
    if self.ty >= 2 and self.max_priority_fee_per_gas is None:
        self.max_priority_fee_per_gas = TransactionDefaults.max_priority_fee_per_gas
    if self.ty < 2:
        assert self.max_fee_per_gas is None, "max_fee_per_gas must be None"
        assert self.max_priority_fee_per_gas is None, "max_priority_fee_per_gas must be None"

    if self.ty == 3 and self.max_fee_per_blob_gas is None:
        self.max_fee_per_blob_gas = 1
    if self.ty != 3:
        assert self.blob_versioned_hashes is None, "blob_versioned_hashes must be None"
        assert self.max_fee_per_blob_gas is None, "max_fee_per_blob_gas must be None"

    if self.ty == 4 and self.authorization_list is None:
        self.authorization_list = []
    if self.ty != 4:
        assert self.authorization_list is None, "authorization_list must be None"

    if self.ty == 6 and self.initcodes is None:
        self.initcodes = []
    if self.ty != 6:
        assert self.initcodes is None, "initcodes must be None"

    if "nonce" not in self.model_fields_set and self.sender is not None:
        self.nonce = HexNumber(self.sender.get_nonce())

with_error(error)

Create a copy of the transaction with an added error.

Source code in src/ethereum_test_types/types.py

def with_error(
    self, error: List[TransactionException] | TransactionException
) -> "Transaction":
    """Create a copy of the transaction with an added error."""
    return self.copy(error=error)

with_nonce(nonce)

Create a copy of the transaction with a modified nonce.

Source code in src/ethereum_test_types/types.py

def with_nonce(self, nonce: int) -> "Transaction":
    """Create a copy of the transaction with a modified nonce."""
    return self.copy(nonce=nonce)

signature_bytes: Bytes cached property

Returns the serialized bytes of the transaction signature.

sign()

Signs the authorization tuple with a private key.

Source code in src/ethereum_test_types/types.py

def sign(self: "Transaction"):
    """Signs the authorization tuple with a private key."""
    signature_bytes: bytes | None = None
    rlp_signing_bytes = self.rlp_signing_bytes()
    if (
        "v" not in self.model_fields_set
        and "r" not in self.model_fields_set
        and "s" not in self.model_fields_set
    ):
        signing_key: Hash | None = None
        if self.secret_key is not None:
            signing_key = self.secret_key
            self.secret_key = None
        elif self.sender is not None:
            eoa = self.sender
            assert eoa is not None, "signer must be set"
            signing_key = eoa.key
        assert signing_key is not None, "secret_key or signer must be set"

        signature_bytes = PrivateKey(secret=signing_key).sign_recoverable(
            rlp_signing_bytes, hasher=keccak256
        )
        v, r, s = (
            signature_bytes[64],
            int.from_bytes(signature_bytes[0:32], byteorder="big"),
            int.from_bytes(signature_bytes[32:64], byteorder="big"),
        )
        if self.ty == 0:
            if self.protected:
                v += 35 + (self.chain_id * 2)
            else:  # not protected
                v += 27
        self.v, self.r, self.s = (HexNumber(v), HexNumber(r), HexNumber(s))
        self.model_fields_set.add("v")
        self.model_fields_set.add("r")
        self.model_fields_set.add("s")

    if self.sender is None:
        try:
            if not signature_bytes:
                signature_bytes = (
                    int(self.r).to_bytes(32, byteorder="big")
                    + int(self.s).to_bytes(32, byteorder="big")
                    + bytes([self.v])
                )
            public_key = PublicKey.from_signature_and_message(
                signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
            )
            self.signer = EOA(
                address=Address(keccak256(public_key.format(compressed=False)[1:])[32 - 20 :])
            )
        except Exception:
            # Signer remains `None` in this case
            pass

with_signature_and_sender(*, keep_secret_key=False)

Return signed version of the transaction using the private key.

Source code in src/ethereum_test_types/types.py

def with_signature_and_sender(self, *, keep_secret_key: bool = False) -> "Transaction":
    """Return signed version of the transaction using the private key."""
    updated_values: Dict[str, Any] = {}

    if (
        "v" in self.model_fields_set
        or "r" in self.model_fields_set
        or "s" in self.model_fields_set
    ):
        # Transaction already signed
        if self.sender is not None:
            return self

        public_key = PublicKey.from_signature_and_message(
            self.signature_bytes, self.rlp_signing_bytes().keccak256(), hasher=None
        )
        updated_values["sender"] = Address(
            keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
        )
        return self.copy(**updated_values)

    if self.secret_key is None:
        raise ValueError("secret_key must be set to sign a transaction")

    # Get the signing bytes
    signing_hash = self.rlp_signing_bytes().keccak256()

    # Sign the bytes
    signature_bytes = PrivateKey(secret=self.secret_key).sign_recoverable(
        signing_hash, hasher=None
    )
    public_key = PublicKey.from_signature_and_message(
        signature_bytes, signing_hash, hasher=None
    )

    sender = keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
    updated_values["sender"] = Address(sender)

    v, r, s = (
        signature_bytes[64],
        int.from_bytes(signature_bytes[0:32], byteorder="big"),
        int.from_bytes(signature_bytes[32:64], byteorder="big"),
    )
    if self.ty == 0:
        if self.protected:
            v += 35 + (self.chain_id * 2)
        else:  # not protected
            v += 27

    updated_values["v"] = HexNumber(v)
    updated_values["r"] = HexNumber(r)
    updated_values["s"] = HexNumber(s)

    updated_values["secret_key"] = None

    updated_tx: "Transaction" = self.model_copy(update=updated_values)

    # Remove the secret key if requested
    if keep_secret_key:
        updated_tx.secret_key = self.secret_key
    return updated_tx

get_rlp_signing_fields()

Return the list of values included in the envelope used for signing depending on the transaction type.

Source code in src/ethereum_test_types/types.py

def get_rlp_signing_fields(self) -> List[str]:
    """
    Return the list of values included in the envelope used for signing depending on
    the transaction type.
    """
    field_list: List[str]
    if self.ty == 6:
        # EIP-7873: https://eips.ethereum.org/EIPS/eip-7873
        field_list = [
            "chain_id",
            "nonce",
            "max_priority_fee_per_gas",
            "max_fee_per_gas",
            "gas_limit",
            "to",
            "value",
            "data",
            "access_list",
            "initcodes",
        ]
    elif self.ty == 4:
        # EIP-7702: https://eips.ethereum.org/EIPS/eip-7702
        field_list = [
            "chain_id",
            "nonce",
            "max_priority_fee_per_gas",
            "max_fee_per_gas",
            "gas_limit",
            "to",
            "value",
            "data",
            "access_list",
            "authorization_list",
        ]
    elif self.ty == 3:
        # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
        field_list = [
            "chain_id",
            "nonce",
            "max_priority_fee_per_gas",
            "max_fee_per_gas",
            "gas_limit",
            "to",
            "value",
            "data",
            "access_list",
            "max_fee_per_blob_gas",
            "blob_versioned_hashes",
        ]
    elif self.ty == 2:
        # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
        field_list = [
            "chain_id",
            "nonce",
            "max_priority_fee_per_gas",
            "max_fee_per_gas",
            "gas_limit",
            "to",
            "value",
            "data",
            "access_list",
        ]
    elif self.ty == 1:
        # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
        field_list = [
            "chain_id",
            "nonce",
            "gas_price",
            "gas_limit",
            "to",
            "value",
            "data",
            "access_list",
        ]
    elif self.ty == 0:
        field_list = ["nonce", "gas_price", "gas_limit", "to", "value", "data"]
        if self.protected:
            # EIP-155: https://eips.ethereum.org/EIPS/eip-155
            field_list.extend(["chain_id", "zero", "zero"])
    else:
        raise NotImplementedError(f"signing for transaction type {self.ty} not implemented")

    for field in field_list:
        if field != "to":
            assert getattr(self, field) is not None, (
                f"{field} must be set for type {self.ty} tx"
            )
    return field_list

get_rlp_fields()

Return the list of values included in the list used for rlp encoding depending on the transaction type.

Source code in src/ethereum_test_types/types.py

def get_rlp_fields(self) -> List[str]:
    """
    Return the list of values included in the list used for rlp encoding depending on
    the transaction type.
    """
    fields = self.get_rlp_signing_fields()
    if self.ty == 0 and self.protected:
        fields = fields[:-3]
    return fields + ["v", "r", "s"]

get_rlp_prefix()

Return the transaction type as bytes to be appended at the beginning of the serialized transaction if type is not 0.

Source code in src/ethereum_test_types/types.py

def get_rlp_prefix(self) -> bytes:
    """
    Return the transaction type as bytes to be appended at the beginning of the
    serialized transaction if type is not 0.
    """
    if self.ty > 0:
        return bytes([self.ty])
    return b""

get_rlp_signing_prefix()

Return the transaction type as bytes to be appended at the beginning of the serialized transaction signing envelope if type is not 0.

Source code in src/ethereum_test_types/types.py

def get_rlp_signing_prefix(self) -> bytes:
    """
    Return the transaction type as bytes to be appended at the beginning of the
    serialized transaction signing envelope if type is not 0.
    """
    if self.ty > 0:
        return bytes([self.ty])
    return b""

hash: Hash cached property

Returns hash of the transaction.

serializable_list: Any cached property

Return list of values included in the transaction as a serializable object.

list_root(input_txs) staticmethod

Return transactions root of a list of transactions.

Source code in src/ethereum_test_types/types.py

@staticmethod
def list_root(input_txs: List["Transaction"]) -> Hash:
    """Return transactions root of a list of transactions."""
    t = HexaryTrie(db={})
    for i, tx in enumerate(input_txs):
        t.set(eth_rlp.encode(Uint(i)), tx.rlp())
    return Hash(t.root_hash)

list_blob_versioned_hashes(input_txs) staticmethod

Get list of ordered blob versioned hashes from a list of transactions.

Source code in src/ethereum_test_types/types.py

@staticmethod
def list_blob_versioned_hashes(input_txs: List["Transaction"]) -> List[Hash]:
    """Get list of ordered blob versioned hashes from a list of transactions."""
    return [
        blob_versioned_hash
        for tx in input_txs
        if tx.blob_versioned_hashes is not None
        for blob_versioned_hash in tx.blob_versioned_hashes
    ]

created_contract: Address cached property

Return address of the contract created by the transaction.

TransactionDefaults dataclass

Default values for transactions.

Source code in src/ethereum_test_types/types.py

@dataclass
class TransactionDefaults:
    """Default values for transactions."""

    chain_id: int = 1
    gas_price = 10
    max_fee_per_gas = 7
    max_priority_fee_per_gas: int = 0

TransactionReceipt

Bases: CamelModel

Transaction receipt.

Source code in src/ethereum_test_types/types.py

class TransactionReceipt(CamelModel):
    """Transaction receipt."""

    transaction_hash: Hash | None = None
    gas_used: HexNumber | None = None
    root: Bytes | None = None
    status: HexNumber | None = None
    cumulative_gas_used: HexNumber | None = None
    logs_bloom: Bloom | None = None
    logs: List[TransactionLog] | None = None
    contract_address: Address | None = None
    effective_gas_price: HexNumber | None = None
    block_hash: Hash | None = None
    transaction_index: HexNumber | None = None
    blob_gas_used: HexNumber | None = None
    blob_gas_price: HexNumber | None = None
    delegations: List[ReceiptDelegation] | None = None

Withdrawal

Bases: WithdrawalGeneric[HexNumber]

Withdrawal type.

Source code in src/ethereum_test_types/types.py

class Withdrawal(WithdrawalGeneric[HexNumber]):
    """Withdrawal type."""

    pass

WithdrawalRequest

Bases: RequestBase, CamelModel

Withdrawal Request type.

Source code in src/ethereum_test_types/types.py

class WithdrawalRequest(RequestBase, CamelModel):
    """Withdrawal Request type."""

    source_address: Address = Address(0)
    """
    The address of the execution layer account that made the withdrawal request.
    """
    validator_pubkey: BLSPublicKey
    """
    The current public key of the validator as it currently is in the beacon state.
    """
    amount: HexNumber
    """
    The amount in gwei to be withdrawn on the beacon chain.
    """

    type: ClassVar[int] = 1

    def __bytes__(self) -> bytes:
        """Return withdrawal's attributes as bytes."""
        return (
            bytes(self.source_address)
            + bytes(self.validator_pubkey)
            + self.amount.to_bytes(8, "little")
        )

source_address: Address = Address(0) class-attribute instance-attribute

The address of the execution layer account that made the withdrawal request.

validator_pubkey: BLSPublicKey instance-attribute

The current public key of the validator as it currently is in the beacon state.

amount: HexNumber instance-attribute

The amount in gwei to be withdrawn on the beacon chain.

__bytes__()

Return withdrawal's attributes as bytes.

Source code in src/ethereum_test_types/types.py

def __bytes__(self) -> bytes:
    """Return withdrawal's attributes as bytes."""
    return (
        bytes(self.source_address)
        + bytes(self.validator_pubkey)
        + self.amount.to_bytes(8, "little")
    )

keccak256(data)

Calculate keccak256 hash of the given data.

Source code in src/ethereum_test_types/types.py

def keccak256(data: bytes) -> Hash:
    """Calculate keccak256 hash of the given data."""
    return Bytes(data).keccak256()