Ethereum Test Types package

Common definitions and types.

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 packages/testing/src/execution_testing/test_types/account_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,
    ) -> "EOA":
        """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) -> Self:
        """Return copy of the EOA."""
        return self.__class__(Address(self), key=self.key, nonce=self.nonce)

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

Init the EOA.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def __new__(
    cls,
    address: "FixedSizeBytesConvertible | Address | EOA | None" = None,
    *,
    key: FixedSizeBytesConvertible | None = None,
    nonce: NumberConvertible = 0,
) -> "EOA":
    """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 packages/testing/src/execution_testing/test_types/account_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 packages/testing/src/execution_testing/test_types/account_types.py

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

Alloc

Bases: Alloc

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

Source code in packages/testing/src/execution_testing/test_types/account_types.py

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

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

        address: Address
        account: Account | None

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

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

        address: Address

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

    @dataclass(kw_only=True)
    class CollisionError(Exception):
        """Different accounts at the same address."""

        address: Address
        account_1: Account | None
        account_2: Account | None

        def to_json(self) -> Dict[str, Any]:
            """Dump to json object."""
            return {
                "address": self.address.hex(),
                "account_1": self.account_1.model_dump(mode="json")
                if self.account_1 is not None
                else None,
                "account_2": self.account_2.model_dump(mode="json")
                if self.account_2 is not None
                else None,
            }

        @classmethod
        def from_json(cls, obj: Dict[str, Any]) -> Self:
            """Parse from a json dict."""
            return cls(
                address=Address(obj["address"]),
                account_1=Account.model_validate(obj["account_1"])
                if obj["account_1"] is not None
                else None,
                account_2=Account.model_validate(obj["account_2"])
                if obj["account_2"] is not None
                else None,
            )

        def __str__(self) -> str:
            """Print exception string."""
            return (
                "Overlapping key defining different accounts detected:\n"
                f"{json.dumps(self.to_json(), indent=2)}"
            )

    class KeyCollisionMode(Enum):
        """Mode for handling key collisions when merging allocations."""

        ERROR = auto()
        OVERWRITE = auto()
        ALLOW_IDENTICAL_ACCOUNTS = auto()

    @classmethod
    def merge(
        cls,
        alloc_1: "Alloc",
        alloc_2: "Alloc",
        key_collision_mode: KeyCollisionMode = KeyCollisionMode.OVERWRITE,
    ) -> "Alloc":
        """Return merged allocation of two sources."""
        overlapping_keys = alloc_1.root.keys() & alloc_2.root.keys()
        if overlapping_keys:
            if key_collision_mode == cls.KeyCollisionMode.ERROR:
                raise Exception(
                    f"Overlapping keys detected: "
                    f"{[key.hex() for key in overlapping_keys]}"
                )
            elif (
                key_collision_mode
                == cls.KeyCollisionMode.ALLOW_IDENTICAL_ACCOUNTS
            ):
                # The overlapping keys must point to the exact same account
                for key in overlapping_keys:
                    account_1 = alloc_1[key]
                    account_2 = alloc_2[key]
                    if account_1 != account_2:
                        raise Alloc.CollisionError(
                            address=key,
                            account_1=account_1,
                            account_2=account_2,
                        )
        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) -> Iterator[Address]:  # type: ignore [override]
        """Return iterator over the allocation."""
        return iter(self.root)

    def items(self) -> ItemsView[Address, Account | None]:
        """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,
    ) -> 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
    ) -> None:
        """Delete account associated with an address."""
        if not isinstance(address, Address):
            address = Address(address)
        self.root.pop(address, None)

    def __eq__(self, other: object) -> 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) -> Hash:
        """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 Hash(state_root(state))

    def verify_post_alloc(self, got_alloc: "Alloc") -> None:
        """
        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=address, account=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=address)

    def deterministic_deploy_contract(
        self,
        *,
        deploy_code: BytesConvertible,
        salt: Hash | int = 0,
        initcode: BytesConvertible | None = None,
        storage: Storage | StorageRootType | None = None,
        label: str | None = None,
    ) -> Address:
        """
        Deploy a contract to the allocation at a deterministic location
        using a deterministic deployment proxy.

        The initcode is not executed during test filling; it is executed only
        when the tests run on live networks. Therefore, if the initcode
        performs modifications to the storage, these must be specified using
        the `storage` parameter.

        Args:
            deploy_code: Contract code to deploy.
            salt: Salt to use for deterministic deployment.
            initcode: Initcode to use for deterministic deployment.
                      If `None`, the initcode is derived from `deploy_code`.
            storage: The expected storage state of the deployed contract after
                     initcode execution.
            label: Label to use for the contract.

        """
        raise NotImplementedError(
            "deterministic_deploy_contract is not implemented in the base "
            "class"
        )

    def deploy_contract(
        self,
        code: BytesConvertible,
        *,
        storage: Storage | StorageRootType | None = None,
        balance: NumberConvertible = 0,
        nonce: NumberConvertible = 1,
        address: Address | None = None,
        label: str | None = None,
        stub: 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,
        code: BytesConvertible | 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,
        *,
        minimum_balance: bool = False,
    ) -> None:
        """
        Fund an address with a given amount.

        Add a funded account to the pre-allocation.
        The address must not already exist in the pre-allocation. To set the
        balance of an account, use the `amount` parameter in `fund_eoa()` or
        the `balance` parameter in `deploy_contract()` at creation time.

        Args:
            address: Address to fund
            amount: Amount to fund in Wei
            minimum_balance: If set to True, account will be checked to have a
                minimum balance of `amount` and only fund if the balance is
                insufficient

        """
        raise NotImplementedError(
            "fund_address is not implemented in the base class"
        )

    def empty_account(self) -> Address:
        """
        Return a previously unused account guaranteed to be empty.

        This ensures the account has zero balance, zero nonce, no code, and no
        storage. The account is not a precompile or a system contract.
        """
        raise NotImplementedError(
            "empty_account is not implemented in the base class"
        )

UnexpectedAccountError dataclass

Bases: Exception

Unexpected account found in the allocation.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

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

    address: Address
    account: Account | None

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

__str__()

Print exception string.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

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

MissingAccountError dataclass

Bases: Exception

Expected account not found in the allocation.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

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

    address: Address

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

__str__()

Print exception string.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

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

CollisionError dataclass

Bases: Exception

Different accounts at the same address.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

@dataclass(kw_only=True)
class CollisionError(Exception):
    """Different accounts at the same address."""

    address: Address
    account_1: Account | None
    account_2: Account | None

    def to_json(self) -> Dict[str, Any]:
        """Dump to json object."""
        return {
            "address": self.address.hex(),
            "account_1": self.account_1.model_dump(mode="json")
            if self.account_1 is not None
            else None,
            "account_2": self.account_2.model_dump(mode="json")
            if self.account_2 is not None
            else None,
        }

    @classmethod
    def from_json(cls, obj: Dict[str, Any]) -> Self:
        """Parse from a json dict."""
        return cls(
            address=Address(obj["address"]),
            account_1=Account.model_validate(obj["account_1"])
            if obj["account_1"] is not None
            else None,
            account_2=Account.model_validate(obj["account_2"])
            if obj["account_2"] is not None
            else None,
        )

    def __str__(self) -> str:
        """Print exception string."""
        return (
            "Overlapping key defining different accounts detected:\n"
            f"{json.dumps(self.to_json(), indent=2)}"
        )

to_json()

Dump to json object.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def to_json(self) -> Dict[str, Any]:
    """Dump to json object."""
    return {
        "address": self.address.hex(),
        "account_1": self.account_1.model_dump(mode="json")
        if self.account_1 is not None
        else None,
        "account_2": self.account_2.model_dump(mode="json")
        if self.account_2 is not None
        else None,
    }

from_json(obj) classmethod

Parse from a json dict.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

@classmethod
def from_json(cls, obj: Dict[str, Any]) -> Self:
    """Parse from a json dict."""
    return cls(
        address=Address(obj["address"]),
        account_1=Account.model_validate(obj["account_1"])
        if obj["account_1"] is not None
        else None,
        account_2=Account.model_validate(obj["account_2"])
        if obj["account_2"] is not None
        else None,
    )

__str__()

Print exception string.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def __str__(self) -> str:
    """Print exception string."""
    return (
        "Overlapping key defining different accounts detected:\n"
        f"{json.dumps(self.to_json(), indent=2)}"
    )

KeyCollisionMode

Bases: Enum

Mode for handling key collisions when merging allocations.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

class KeyCollisionMode(Enum):
    """Mode for handling key collisions when merging allocations."""

    ERROR = auto()
    OVERWRITE = auto()
    ALLOW_IDENTICAL_ACCOUNTS = auto()

merge(alloc_1, alloc_2, key_collision_mode=KeyCollisionMode.OVERWRITE) classmethod

Return merged allocation of two sources.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

@classmethod
def merge(
    cls,
    alloc_1: "Alloc",
    alloc_2: "Alloc",
    key_collision_mode: KeyCollisionMode = KeyCollisionMode.OVERWRITE,
) -> "Alloc":
    """Return merged allocation of two sources."""
    overlapping_keys = alloc_1.root.keys() & alloc_2.root.keys()
    if overlapping_keys:
        if key_collision_mode == cls.KeyCollisionMode.ERROR:
            raise Exception(
                f"Overlapping keys detected: "
                f"{[key.hex() for key in overlapping_keys]}"
            )
        elif (
            key_collision_mode
            == cls.KeyCollisionMode.ALLOW_IDENTICAL_ACCOUNTS
        ):
            # The overlapping keys must point to the exact same account
            for key in overlapping_keys:
                account_1 = alloc_1[key]
                account_2 = alloc_2[key]
                if account_1 != account_2:
                    raise Alloc.CollisionError(
                        address=key,
                        account_1=account_1,
                        account_2=account_2,
                    )
    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 packages/testing/src/execution_testing/test_types/account_types.py

def __iter__(self) -> Iterator[Address]:  # type: ignore [override]
    """Return iterator over the allocation."""
    return iter(self.root)

items()

Return iterator over the allocation items.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def items(self) -> ItemsView[Address, Account | None]:
    """Return iterator over the allocation items."""
    return self.root.items()

__getitem__(address)

Return account associated with an address.

Source code in packages/testing/src/execution_testing/test_types/account_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 packages/testing/src/execution_testing/test_types/account_types.py

def __setitem__(
    self,
    address: Address | FixedSizeBytesConvertible,
    account: Account | None,
) -> 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 packages/testing/src/execution_testing/test_types/account_types.py

def __delitem__(
    self, address: Address | FixedSizeBytesConvertible
) -> None:
    """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 packages/testing/src/execution_testing/test_types/account_types.py

def __eq__(self, other: object) -> 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 packages/testing/src/execution_testing/test_types/account_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 packages/testing/src/execution_testing/test_types/account_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 packages/testing/src/execution_testing/test_types/account_types.py

def state_root(self) -> Hash:
    """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 Hash(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 packages/testing/src/execution_testing/test_types/account_types.py

def verify_post_alloc(self, got_alloc: "Alloc") -> None:
    """
    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=address, account=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=address)

deterministic_deploy_contract(*, deploy_code, salt=0, initcode=None, storage=None, label=None)

Deploy a contract to the allocation at a deterministic location using a deterministic deployment proxy.

The initcode is not executed during test filling; it is executed only when the tests run on live networks. Therefore, if the initcode performs modifications to the storage, these must be specified using the storage parameter.

Parameters:

Name	Type	Description	Default
deploy_code	BytesConvertible	Contract code to deploy.	required
salt	Hash | int	Salt to use for deterministic deployment.	0
initcode	BytesConvertible | None	Initcode to use for deterministic deployment. If None, the initcode is derived from deploy_code.	None
storage	Storage | StorageRootType | None	The expected storage state of the deployed contract after initcode execution.	None
label	str | None	Label to use for the contract.	None

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def deterministic_deploy_contract(
    self,
    *,
    deploy_code: BytesConvertible,
    salt: Hash | int = 0,
    initcode: BytesConvertible | None = None,
    storage: Storage | StorageRootType | None = None,
    label: str | None = None,
) -> Address:
    """
    Deploy a contract to the allocation at a deterministic location
    using a deterministic deployment proxy.

    The initcode is not executed during test filling; it is executed only
    when the tests run on live networks. Therefore, if the initcode
    performs modifications to the storage, these must be specified using
    the `storage` parameter.

    Args:
        deploy_code: Contract code to deploy.
        salt: Salt to use for deterministic deployment.
        initcode: Initcode to use for deterministic deployment.
                  If `None`, the initcode is derived from `deploy_code`.
        storage: The expected storage state of the deployed contract after
                 initcode execution.
        label: Label to use for the contract.

    """
    raise NotImplementedError(
        "deterministic_deploy_contract is not implemented in the base "
        "class"
    )

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

Deploy a contract to the allocation.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def deploy_contract(
    self,
    code: BytesConvertible,
    *,
    storage: Storage | StorageRootType | None = None,
    balance: NumberConvertible = 0,
    nonce: NumberConvertible = 1,
    address: Address | None = None,
    label: str | None = None,
    stub: 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, code=None, delegation=None, nonce=None)

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

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def fund_eoa(
    self,
    amount: NumberConvertible | None = None,
    label: str | None = None,
    storage: Storage | None = None,
    code: BytesConvertible | 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, *, minimum_balance=False)

Fund an address with a given amount.

Add a funded account to the pre-allocation. The address must not already exist in the pre-allocation. To set the balance of an account, use the amount parameter in fund_eoa() or the balance parameter in deploy_contract() at creation time.

Parameters:

Name	Type	Description	Default
address	Address	Address to fund	required
amount	NumberConvertible	Amount to fund in Wei	required
minimum_balance	bool	If set to True, account will be checked to have a minimum balance of amount and only fund if the balance is insufficient	False

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def fund_address(
    self,
    address: Address,
    amount: NumberConvertible,
    *,
    minimum_balance: bool = False,
) -> None:
    """
    Fund an address with a given amount.

    Add a funded account to the pre-allocation.
    The address must not already exist in the pre-allocation. To set the
    balance of an account, use the `amount` parameter in `fund_eoa()` or
    the `balance` parameter in `deploy_contract()` at creation time.

    Args:
        address: Address to fund
        amount: Amount to fund in Wei
        minimum_balance: If set to True, account will be checked to have a
            minimum balance of `amount` and only fund if the balance is
            insufficient

    """
    raise NotImplementedError(
        "fund_address is not implemented in the base class"
    )

empty_account()

Return a previously unused account guaranteed to be empty.

This ensures the account has zero balance, zero nonce, no code, and no storage. The account is not a precompile or a system contract.

Source code in packages/testing/src/execution_testing/test_types/account_types.py

def empty_account(self) -> Address:
    """
    Return a previously unused account guaranteed to be empty.

    This ensures the account has zero balance, zero nonce, no code, and no
    storage. The account is not a precompile or a system contract.
    """
    raise NotImplementedError(
        "empty_account is not implemented in the base class"
    )

Blob

Bases: CamelModel

Class representing a full blob.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

class Blob(CamelModel):
    """Class representing a full blob."""

    data: Bytes
    commitment: Bytes
    proof: List[Bytes] | Bytes  # Bytes < Osaka, List[Bytes] >= Osaka
    # None (in json: null) < Osaka, List[Bytes] >= Osaka
    cells: List[Bytes] | None

    versioned_hash: Hash
    name: str
    fork: Fork
    seed: int
    timestamp: int  # fork transitions require timestamp >= 15000 to occur

    _trusted_setup: ClassVar[Any | None] = None

    @classmethod
    def trusted_setup(cls) -> Any:
        """Set trusted setup if it is not already set."""
        if cls._trusted_setup is None:
            trusted_setup_path = (
                Path(realpath(__file__)).parent / "kzg_trusted_setup.txt"
            )
            trusted_setup = ckzg.load_trusted_setup(str(trusted_setup_path), 0)
            cls._trusted_setup = trusted_setup

        return cls._trusted_setup

    @staticmethod
    def get_filename(fork: Fork, seed: int) -> str:
        """
        Return filename this blob would have as string (with .json extension).
        """
        amount_cell_proofs: int = cast(
            int, fork.get_blob_constant("AMOUNT_CELL_PROOFS")
        )
        return (
            "blob_"
            + str(seed)
            + "_cell_proofs_"
            + str(amount_cell_proofs)
            + ".json"
        )

    @staticmethod
    def get_filepath(fork: Fork, seed: int) -> Path:
        """
        Return the Path to the blob that would be created with these
        parameters.
        """
        # determine amount of cell proofs for this fork (0 or 128)
        would_be_filename: str = Blob.get_filename(fork, seed)

        # return path to blob
        return CACHED_BLOBS_DIRECTORY / would_be_filename

    @staticmethod
    def from_fork(fork: Fork, seed: int = 0, timestamp: int = 0) -> "Blob":
        """
        Construct Blob instances. Fork logic is encapsulated within nested
        functions.
        """

        def generate_blob_data(rng_seed: int = 0) -> Bytes:
            """Calculate blob data deterministically via provided seed."""
            # create local (independent) RNG object seeded with rng_seed
            rng = random.Random(rng_seed)

            # generate blob
            ints: list[int] = [
                rng.randrange(cast(int, fork.get_blob_constant("BLS_MODULUS")))
                for _ in range(
                    cast(
                        int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
                    )
                )
            ]

            encoded: list[bytes] = [
                i.to_bytes(
                    cast(
                        int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
                    ),
                    cast(
                        Literal["big"],
                        fork.get_blob_constant("KZG_ENDIANNESS"),
                    ),
                )
                for i in ints
            ]
            blob: bytes = b"".join(encoded)  # without 0x

            return Bytes(blob)

        def get_versioned_hash(commitment: Bytes, version: int = 1) -> Hash:
            """Calculate versioned hash for a given blob."""
            return Hash(bytes([version]) + sha256(commitment).digest()[1:])

        def get_commitment(data: Bytes) -> Bytes:
            """
            Take a blob and returns a cryptographic commitment to it.

            Note: Each cell holds the exact same copy of this commitment.
            """
            # sanity check
            field_elements: int = cast(
                int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
            )
            bytes_per_field: int = cast(
                int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
            )
            assert len(data) == field_elements * bytes_per_field, (
                f"Expected blob of length "
                f"{field_elements * bytes_per_field} but got blob of length "
                f"{len(data)}"
            )

            # calculate commitment
            commitment = ckzg.blob_to_kzg_commitment(
                data, Blob.trusted_setup()
            )
            assert len(commitment) == fork.get_blob_constant(
                "BYTES_PER_COMMITMENT"
            ), (
                f"Expected {fork.get_blob_constant('BYTES_PER_COMMITMENT')} "
                f"resulting commitments but got {len(commitment)} commitments"
            )

            return commitment

        def get_proof(fork: Fork, data: Bytes) -> List[Bytes] | Bytes:
            # determine whether this fork is <osaka or >= osaka by looking at
            # amount of cell_proofs
            amount_cell_proofs = fork.get_blob_constant("AMOUNT_CELL_PROOFS")

            # cancun, prague
            if amount_cell_proofs == 0:
                z = 2  # 2 is one of many possible valid field elements z
                # https://github.com/ethereum/consensus-specs/blob/ad884507f
                #  7a1d5962cd3dfb5f7b3e41aab728c55/tests/core/pyspec/eth2spec/
                #  test/utils/kzg_tests.py#L58-L66)
                z_valid_size: bytes = z.to_bytes(
                    cast(
                        int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
                    ),
                    byteorder="big",
                )
                proof, _ = ckzg.compute_kzg_proof(
                    data, z_valid_size, Blob.trusted_setup()
                )
                return proof

            # >=osaka
            if amount_cell_proofs == 128:
                _, proofs = ckzg.compute_cells_and_kzg_proofs(
                    data, Blob.trusted_setup()
                )  # returns List[byte] of length 128
                return proofs

            raise AssertionError(
                f"get_proof() has not been implemented yet for fork: "
                f"{fork.name()}. Got amount of cell proofs "
                f"{amount_cell_proofs} but expected 128."
            )

        def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
            # determine whether this fork is <osaka or >= osaka by looking at
            # amount of cell_proofs
            amount_cell_proofs = fork.get_blob_constant("AMOUNT_CELL_PROOFS")

            # cancun, prague
            if amount_cell_proofs == 0:
                return None

            # >=osaka
            if amount_cell_proofs == 128:
                cells, _ = ckzg.compute_cells_and_kzg_proofs(
                    data, Blob.trusted_setup()
                )  # returns List[byte] of length 128
                return cells  # List[bytes]

            raise AssertionError(
                f"get_cells() has not been implemented yet for fork: "
                f"{fork.name()}. Got amount of cell proofs "
                f"{amount_cell_proofs} but expected 128."
            )

        # first, create cached blobs dir if necessary
        if not CACHED_BLOBS_DIRECTORY.exists():
            CACHED_BLOBS_DIRECTORY.mkdir(
                parents=True, exist_ok=True
            )  # create all necessary dirs on the way

        # handle transition forks (blob related constants are needed and only
        # available for normal forks)
        fork = fork.fork_at(timestamp=timestamp)

        # if this blob already exists then load from file. use lock
        blob_location: Path = Blob.get_filepath(fork, seed)

        # use lock to avoid race conditions
        lock_file_path = blob_location.with_suffix(".lock")
        with FileLock(lock_file_path):
            if blob_location.exists():
                logger.debug(
                    f"Blob exists already, reading it from file "
                    f"{blob_location}"
                )
                return Blob.from_file(Blob.get_filename(fork, seed))

        assert fork.supports_blobs(), (
            f"Provided fork {fork.name()} does not support blobs!"
        )

        # get data for blob parameters
        data: Bytes = generate_blob_data(seed)
        commitment: Bytes = get_commitment(data)
        proof: List[Bytes] | Bytes = get_proof(fork, data)
        cells: List[Bytes] | None = get_cells(fork, data)
        versioned_hash: Hash = get_versioned_hash(commitment)
        name: str = Blob.get_filename(fork, seed)

        blob = Blob(
            data=data,
            commitment=commitment,
            proof=proof,
            cells=cells,
            versioned_hash=versioned_hash,
            name=name,
            fork=fork,
            seed=seed,
            timestamp=timestamp,
        )
        # for most effective caching temporarily persist every blob that is
        # created in cache
        blob.write_to_file()

        return blob

    @staticmethod
    def from_file(file_name: str) -> "Blob":
        """
        Read a .json file and reconstruct object it represents.

        You can load a blob only via its filename (with or without .json
        extension).
        """
        # ensure filename was passed
        assert file_name.startswith("blob_"), (
            f"You provided an invalid blob filename. Expected it to start "
            f"with 'blob_' but got: {file_name}"
        )

        if ".json" not in file_name:
            file_name = file_name + ".json"

        # determine path where this blob would be stored if it existed
        blob_file_location = CACHED_BLOBS_DIRECTORY / file_name

        # check whether blob exists
        assert blob_file_location.exists(), (
            f"Tried to load blob from file but {blob_file_location} does not "
            "exist"
        )

        # read blob from file
        with open(blob_file_location, "r", encoding="utf-8") as f:
            json_str: str = f.read()

        # reconstruct and return blob object
        return Blob.model_validate_json(json_str)

    def write_to_file(self) -> None:
        """Take a blob object, serialize it and write it to disk as json."""
        json_str = self.model_dump_json()
        output_location = Blob.get_filepath(self.fork, self.seed)

        # use lock to avoid race conditions
        lock_file_path = output_location.with_suffix(".lock")
        with FileLock(lock_file_path):
            # warn if existing static_blob gets overwritten
            if output_location.exists():
                logger.debug(
                    f"Blob {output_location} already exists. It will be "
                    "overwritten."
                )

            # overwrite existing
            with open(output_location, "w", encoding="utf-8") as f:
                f.write(json_str)

    def verify_cell_kzg_proof_batch(self, cell_indices: list) -> bool:
        """
        Check whether all cell proofs are valid and returns True only if that
        is the case.
        """
        amount_cell_proofs: int = cast(
            int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
        )

        assert amount_cell_proofs > 0, (
            f"verify_cell_kzg_proof_batch() is not available for your fork: "
            f"{self.fork.name()}."
        )

        assert self.cells is not None, "self.cells is None, critical error."

        assert len(cell_indices) == len(self.cells), (
            f"Cell Indices list (detected length {len(cell_indices)}) and "
            f"Cell list (detected length {len(self.cells)}) should have same "
            "length."
        )

        # each cell refers to the same commitment
        commitments: list[bytes] = [self.commitment] * len(cell_indices)

        is_valid = ckzg.verify_cell_kzg_proof_batch(
            commitments,
            cell_indices,
            self.cells,
            self.proof,
            Blob.trusted_setup(),
        )

        return is_valid

    def delete_cells_then_recover_them(
        self, deletion_indices: list[int]
    ) -> None:
        """
        Simulate the cell recovery process in user-specified scenario.

        Note: Requirement for successful reconstruction is having at least N of
        the 2N cells.

        Theoretical Usage: You pass a cell list with to 128 elements to this
        function along with a list of deletion indices. These cells will be
        deleted and then the ckzg recovery mechanism is used to repair the
        missing cells. If no assertion is triggered the reconstruction was
        successful.
        """
        amount_cell_proofs: int = cast(
            int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
        )

        assert amount_cell_proofs > 0, (
            f"delete_cells_then_recover_them() is not available for fork: "
            f"{self.fork.name()}"
        )

        assert self.cells is not None, "self.cells is None, critical problem."

        assert isinstance(self.proof, list), (
            "This function only works when self.proof is a list, but it seems "
            "to be of type bytes (not a list)"
        )

        assert len(self.cells) == 128, (
            f"You are supposed to pass a full cell list with 128 elements to "
            f"this function, but got list of length {len(self.cells)}"
        )

        assert len(deletion_indices) < 129, (
            f"You can't delete more than every cell (max len of deletion "
            f"indices list is 128), but you passed a deletion indices list of "
            f"length {len(deletion_indices)}"
        )
        for i in deletion_indices:
            assert 0 <= i <= 127, (
                f"Expected integers in range [0, 127], but got: {i}"
            )

        # delete cells
        all_cell_indices: list[int] = list(range(128))
        remaining_indices: list[int] = [
            i for i in all_cell_indices if i not in deletion_indices
        ]
        remaining_cells = [
            c for i, c in enumerate(self.cells) if i not in deletion_indices
        ]

        recovered_cells, recovered_proofs = ckzg.recover_cells_and_kzg_proofs(
            remaining_indices, remaining_cells, Blob.trusted_setup()
        )  # on success returns two lists of len 128

        # determine success/failure
        assert len(recovered_cells) == len(self.cells), (
            f"Failed to recover cell list. Original cell list had length "
            f"{len(self.cells)} but recovered cell list has length "
            f"{len(recovered_cells)}"
        )
        assert len(recovered_proofs) == len(self.proof), (
            f"Failed to recover proofs list. Original proofs list had length "
            f"{len(self.proof)} but recovered proofs list has length "
            f"{len(recovered_proofs)}"
        )

        for i in range(len(recovered_cells)):
            assert self.cells[i] == recovered_cells[i], (
                f"Failed to correctly restore missing cells. At index {i} "
                f"original cell was 0x{self.cells[i].hex()} but reconstructed "
                f"cell does not match: 0x{recovered_cells[i].hex()}"
            )
            assert self.proof[i] == recovered_proofs[i], (
                f"Failed to correctly restore missing proofs. At index {i} "
                f"original proof was 0x{self.proof[i].hex()} but "
                f"reconstructed proof does not match: "
                f"0x{recovered_proofs[i].hex()}"
            )

    class ProofCorruptionMode(Enum):
        """
        Define what the proof corruption modes do.

        For Osaka and later each Bytes object in the list is manipulated this
        way.
        """

        CORRUPT_FIRST_BYTE = 1  # corrupts a single byte (index 0)
        CORRUPT_LAST_BYTE = 2  # corrupts a single byte (last valid index)
        CORRUPT_TO_ALL_ZEROES = 3  # sets all proof bytes to 0
        CORRUPT_ALL_BYTES = 4  # corrupts all bytes

    def corrupt_proof(self, mode: ProofCorruptionMode) -> None:
        """Corrupt the proof field, supports different corruption modes."""

        def corrupt_byte(b: bytes) -> Bytes:
            """
            Bit-flip all bits of provided byte using XOR to guarantee change.
            """
            if len(b) != 1:
                raise ValueError("Input must be a single byte")
            return Bytes(bytes([b[0] ^ 0xFF]))

        # >=osaka
        amount_cell_proofs: int = cast(
            int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
        )
        if amount_cell_proofs > 0:
            assert isinstance(self.proof, list), (
                "proof was expected to be a list but it isn't"
            )

            if mode == self.ProofCorruptionMode.CORRUPT_FIRST_BYTE:
                for i in range(len(self.proof)):
                    b: Bytes = self.proof[i]
                    corrupted: Bytes = Bytes(corrupt_byte(b[:1]) + b[1:])
                    self.proof[i] = corrupted
            elif mode == self.ProofCorruptionMode.CORRUPT_LAST_BYTE:
                for i in range(len(self.proof)):
                    b = self.proof[i]
                    corrupted = Bytes(b[:-1] + corrupt_byte(b[-1:]))
                    self.proof[i] = corrupted
            elif mode == self.ProofCorruptionMode.CORRUPT_TO_ALL_ZEROES:
                for i in range(len(self.proof)):
                    self.proof[i] = Bytes(bytes(len(self.proof[i])))
            elif mode == self.ProofCorruptionMode.CORRUPT_ALL_BYTES:
                for i in range(len(self.proof)):
                    b = self.proof[i]
                    corrupted_bytes = Bytes(
                        b"".join(corrupt_byte(bytes([byte])) for byte in b)
                    )
                    self.proof[i] = corrupted_bytes
            return

        # pre-osaka (cancun and prague)
        assert amount_cell_proofs == 0, (
            f"You need to adjust corrupt_proof to handle fork "
            f"{self.fork.name()}"
        )
        assert isinstance(self.proof, Bytes), (
            "proof was expected to be Bytes but it isn't"
        )

        if mode == self.ProofCorruptionMode.CORRUPT_FIRST_BYTE:
            self.proof = Bytes(corrupt_byte(self.proof[:1]) + self.proof[1:])
        elif mode == self.ProofCorruptionMode.CORRUPT_LAST_BYTE:
            self.proof = Bytes(self.proof[:-1] + corrupt_byte(self.proof[-1:]))
        elif mode == self.ProofCorruptionMode.CORRUPT_TO_ALL_ZEROES:
            self.proof = Bytes(bytes(len(self.proof)))
        elif mode == self.ProofCorruptionMode.CORRUPT_ALL_BYTES:
            self.proof = Bytes(
                b"".join(corrupt_byte(bytes([byte])) for byte in self.proof)
            )

trusted_setup() classmethod

Set trusted setup if it is not already set.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

@classmethod
def trusted_setup(cls) -> Any:
    """Set trusted setup if it is not already set."""
    if cls._trusted_setup is None:
        trusted_setup_path = (
            Path(realpath(__file__)).parent / "kzg_trusted_setup.txt"
        )
        trusted_setup = ckzg.load_trusted_setup(str(trusted_setup_path), 0)
        cls._trusted_setup = trusted_setup

    return cls._trusted_setup

get_filename(fork, seed) staticmethod

Return filename this blob would have as string (with .json extension).

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

@staticmethod
def get_filename(fork: Fork, seed: int) -> str:
    """
    Return filename this blob would have as string (with .json extension).
    """
    amount_cell_proofs: int = cast(
        int, fork.get_blob_constant("AMOUNT_CELL_PROOFS")
    )
    return (
        "blob_"
        + str(seed)
        + "_cell_proofs_"
        + str(amount_cell_proofs)
        + ".json"
    )

get_filepath(fork, seed) staticmethod

Return the Path to the blob that would be created with these parameters.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

@staticmethod
def get_filepath(fork: Fork, seed: int) -> Path:
    """
    Return the Path to the blob that would be created with these
    parameters.
    """
    # determine amount of cell proofs for this fork (0 or 128)
    would_be_filename: str = Blob.get_filename(fork, seed)

    # return path to blob
    return CACHED_BLOBS_DIRECTORY / would_be_filename

from_fork(fork, seed=0, timestamp=0) staticmethod

Construct Blob instances. Fork logic is encapsulated within nested functions.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

@staticmethod
def from_fork(fork: Fork, seed: int = 0, timestamp: int = 0) -> "Blob":
    """
    Construct Blob instances. Fork logic is encapsulated within nested
    functions.
    """

    def generate_blob_data(rng_seed: int = 0) -> Bytes:
        """Calculate blob data deterministically via provided seed."""
        # create local (independent) RNG object seeded with rng_seed
        rng = random.Random(rng_seed)

        # generate blob
        ints: list[int] = [
            rng.randrange(cast(int, fork.get_blob_constant("BLS_MODULUS")))
            for _ in range(
                cast(
                    int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
                )
            )
        ]

        encoded: list[bytes] = [
            i.to_bytes(
                cast(
                    int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
                ),
                cast(
                    Literal["big"],
                    fork.get_blob_constant("KZG_ENDIANNESS"),
                ),
            )
            for i in ints
        ]
        blob: bytes = b"".join(encoded)  # without 0x

        return Bytes(blob)

    def get_versioned_hash(commitment: Bytes, version: int = 1) -> Hash:
        """Calculate versioned hash for a given blob."""
        return Hash(bytes([version]) + sha256(commitment).digest()[1:])

    def get_commitment(data: Bytes) -> Bytes:
        """
        Take a blob and returns a cryptographic commitment to it.

        Note: Each cell holds the exact same copy of this commitment.
        """
        # sanity check
        field_elements: int = cast(
            int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
        )
        bytes_per_field: int = cast(
            int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
        )
        assert len(data) == field_elements * bytes_per_field, (
            f"Expected blob of length "
            f"{field_elements * bytes_per_field} but got blob of length "
            f"{len(data)}"
        )

        # calculate commitment
        commitment = ckzg.blob_to_kzg_commitment(
            data, Blob.trusted_setup()
        )
        assert len(commitment) == fork.get_blob_constant(
            "BYTES_PER_COMMITMENT"
        ), (
            f"Expected {fork.get_blob_constant('BYTES_PER_COMMITMENT')} "
            f"resulting commitments but got {len(commitment)} commitments"
        )

        return commitment

    def get_proof(fork: Fork, data: Bytes) -> List[Bytes] | Bytes:
        # determine whether this fork is <osaka or >= osaka by looking at
        # amount of cell_proofs
        amount_cell_proofs = fork.get_blob_constant("AMOUNT_CELL_PROOFS")

        # cancun, prague
        if amount_cell_proofs == 0:
            z = 2  # 2 is one of many possible valid field elements z
            # https://github.com/ethereum/consensus-specs/blob/ad884507f
            #  7a1d5962cd3dfb5f7b3e41aab728c55/tests/core/pyspec/eth2spec/
            #  test/utils/kzg_tests.py#L58-L66)
            z_valid_size: bytes = z.to_bytes(
                cast(
                    int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")
                ),
                byteorder="big",
            )
            proof, _ = ckzg.compute_kzg_proof(
                data, z_valid_size, Blob.trusted_setup()
            )
            return proof

        # >=osaka
        if amount_cell_proofs == 128:
            _, proofs = ckzg.compute_cells_and_kzg_proofs(
                data, Blob.trusted_setup()
            )  # returns List[byte] of length 128
            return proofs

        raise AssertionError(
            f"get_proof() has not been implemented yet for fork: "
            f"{fork.name()}. Got amount of cell proofs "
            f"{amount_cell_proofs} but expected 128."
        )

    def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
        # determine whether this fork is <osaka or >= osaka by looking at
        # amount of cell_proofs
        amount_cell_proofs = fork.get_blob_constant("AMOUNT_CELL_PROOFS")

        # cancun, prague
        if amount_cell_proofs == 0:
            return None

        # >=osaka
        if amount_cell_proofs == 128:
            cells, _ = ckzg.compute_cells_and_kzg_proofs(
                data, Blob.trusted_setup()
            )  # returns List[byte] of length 128
            return cells  # List[bytes]

        raise AssertionError(
            f"get_cells() has not been implemented yet for fork: "
            f"{fork.name()}. Got amount of cell proofs "
            f"{amount_cell_proofs} but expected 128."
        )

    # first, create cached blobs dir if necessary
    if not CACHED_BLOBS_DIRECTORY.exists():
        CACHED_BLOBS_DIRECTORY.mkdir(
            parents=True, exist_ok=True
        )  # create all necessary dirs on the way

    # handle transition forks (blob related constants are needed and only
    # available for normal forks)
    fork = fork.fork_at(timestamp=timestamp)

    # if this blob already exists then load from file. use lock
    blob_location: Path = Blob.get_filepath(fork, seed)

    # use lock to avoid race conditions
    lock_file_path = blob_location.with_suffix(".lock")
    with FileLock(lock_file_path):
        if blob_location.exists():
            logger.debug(
                f"Blob exists already, reading it from file "
                f"{blob_location}"
            )
            return Blob.from_file(Blob.get_filename(fork, seed))

    assert fork.supports_blobs(), (
        f"Provided fork {fork.name()} does not support blobs!"
    )

    # get data for blob parameters
    data: Bytes = generate_blob_data(seed)
    commitment: Bytes = get_commitment(data)
    proof: List[Bytes] | Bytes = get_proof(fork, data)
    cells: List[Bytes] | None = get_cells(fork, data)
    versioned_hash: Hash = get_versioned_hash(commitment)
    name: str = Blob.get_filename(fork, seed)

    blob = Blob(
        data=data,
        commitment=commitment,
        proof=proof,
        cells=cells,
        versioned_hash=versioned_hash,
        name=name,
        fork=fork,
        seed=seed,
        timestamp=timestamp,
    )
    # for most effective caching temporarily persist every blob that is
    # created in cache
    blob.write_to_file()

    return blob

from_file(file_name) staticmethod

Read a .json file and reconstruct object it represents.

You can load a blob only via its filename (with or without .json extension).

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

@staticmethod
def from_file(file_name: str) -> "Blob":
    """
    Read a .json file and reconstruct object it represents.

    You can load a blob only via its filename (with or without .json
    extension).
    """
    # ensure filename was passed
    assert file_name.startswith("blob_"), (
        f"You provided an invalid blob filename. Expected it to start "
        f"with 'blob_' but got: {file_name}"
    )

    if ".json" not in file_name:
        file_name = file_name + ".json"

    # determine path where this blob would be stored if it existed
    blob_file_location = CACHED_BLOBS_DIRECTORY / file_name

    # check whether blob exists
    assert blob_file_location.exists(), (
        f"Tried to load blob from file but {blob_file_location} does not "
        "exist"
    )

    # read blob from file
    with open(blob_file_location, "r", encoding="utf-8") as f:
        json_str: str = f.read()

    # reconstruct and return blob object
    return Blob.model_validate_json(json_str)

write_to_file()

Take a blob object, serialize it and write it to disk as json.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

def write_to_file(self) -> None:
    """Take a blob object, serialize it and write it to disk as json."""
    json_str = self.model_dump_json()
    output_location = Blob.get_filepath(self.fork, self.seed)

    # use lock to avoid race conditions
    lock_file_path = output_location.with_suffix(".lock")
    with FileLock(lock_file_path):
        # warn if existing static_blob gets overwritten
        if output_location.exists():
            logger.debug(
                f"Blob {output_location} already exists. It will be "
                "overwritten."
            )

        # overwrite existing
        with open(output_location, "w", encoding="utf-8") as f:
            f.write(json_str)

verify_cell_kzg_proof_batch(cell_indices)

Check whether all cell proofs are valid and returns True only if that is the case.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

def verify_cell_kzg_proof_batch(self, cell_indices: list) -> bool:
    """
    Check whether all cell proofs are valid and returns True only if that
    is the case.
    """
    amount_cell_proofs: int = cast(
        int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
    )

    assert amount_cell_proofs > 0, (
        f"verify_cell_kzg_proof_batch() is not available for your fork: "
        f"{self.fork.name()}."
    )

    assert self.cells is not None, "self.cells is None, critical error."

    assert len(cell_indices) == len(self.cells), (
        f"Cell Indices list (detected length {len(cell_indices)}) and "
        f"Cell list (detected length {len(self.cells)}) should have same "
        "length."
    )

    # each cell refers to the same commitment
    commitments: list[bytes] = [self.commitment] * len(cell_indices)

    is_valid = ckzg.verify_cell_kzg_proof_batch(
        commitments,
        cell_indices,
        self.cells,
        self.proof,
        Blob.trusted_setup(),
    )

    return is_valid

delete_cells_then_recover_them(deletion_indices)

Simulate the cell recovery process in user-specified scenario.

Note: Requirement for successful reconstruction is having at least N of the 2N cells.

Theoretical Usage: You pass a cell list with to 128 elements to this function along with a list of deletion indices. These cells will be deleted and then the ckzg recovery mechanism is used to repair the missing cells. If no assertion is triggered the reconstruction was successful.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

def delete_cells_then_recover_them(
    self, deletion_indices: list[int]
) -> None:
    """
    Simulate the cell recovery process in user-specified scenario.

    Note: Requirement for successful reconstruction is having at least N of
    the 2N cells.

    Theoretical Usage: You pass a cell list with to 128 elements to this
    function along with a list of deletion indices. These cells will be
    deleted and then the ckzg recovery mechanism is used to repair the
    missing cells. If no assertion is triggered the reconstruction was
    successful.
    """
    amount_cell_proofs: int = cast(
        int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
    )

    assert amount_cell_proofs > 0, (
        f"delete_cells_then_recover_them() is not available for fork: "
        f"{self.fork.name()}"
    )

    assert self.cells is not None, "self.cells is None, critical problem."

    assert isinstance(self.proof, list), (
        "This function only works when self.proof is a list, but it seems "
        "to be of type bytes (not a list)"
    )

    assert len(self.cells) == 128, (
        f"You are supposed to pass a full cell list with 128 elements to "
        f"this function, but got list of length {len(self.cells)}"
    )

    assert len(deletion_indices) < 129, (
        f"You can't delete more than every cell (max len of deletion "
        f"indices list is 128), but you passed a deletion indices list of "
        f"length {len(deletion_indices)}"
    )
    for i in deletion_indices:
        assert 0 <= i <= 127, (
            f"Expected integers in range [0, 127], but got: {i}"
        )

    # delete cells
    all_cell_indices: list[int] = list(range(128))
    remaining_indices: list[int] = [
        i for i in all_cell_indices if i not in deletion_indices
    ]
    remaining_cells = [
        c for i, c in enumerate(self.cells) if i not in deletion_indices
    ]

    recovered_cells, recovered_proofs = ckzg.recover_cells_and_kzg_proofs(
        remaining_indices, remaining_cells, Blob.trusted_setup()
    )  # on success returns two lists of len 128

    # determine success/failure
    assert len(recovered_cells) == len(self.cells), (
        f"Failed to recover cell list. Original cell list had length "
        f"{len(self.cells)} but recovered cell list has length "
        f"{len(recovered_cells)}"
    )
    assert len(recovered_proofs) == len(self.proof), (
        f"Failed to recover proofs list. Original proofs list had length "
        f"{len(self.proof)} but recovered proofs list has length "
        f"{len(recovered_proofs)}"
    )

    for i in range(len(recovered_cells)):
        assert self.cells[i] == recovered_cells[i], (
            f"Failed to correctly restore missing cells. At index {i} "
            f"original cell was 0x{self.cells[i].hex()} but reconstructed "
            f"cell does not match: 0x{recovered_cells[i].hex()}"
        )
        assert self.proof[i] == recovered_proofs[i], (
            f"Failed to correctly restore missing proofs. At index {i} "
            f"original proof was 0x{self.proof[i].hex()} but "
            f"reconstructed proof does not match: "
            f"0x{recovered_proofs[i].hex()}"
        )

ProofCorruptionMode

Bases: Enum

Define what the proof corruption modes do.

For Osaka and later each Bytes object in the list is manipulated this way.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

class ProofCorruptionMode(Enum):
    """
    Define what the proof corruption modes do.

    For Osaka and later each Bytes object in the list is manipulated this
    way.
    """

    CORRUPT_FIRST_BYTE = 1  # corrupts a single byte (index 0)
    CORRUPT_LAST_BYTE = 2  # corrupts a single byte (last valid index)
    CORRUPT_TO_ALL_ZEROES = 3  # sets all proof bytes to 0
    CORRUPT_ALL_BYTES = 4  # corrupts all bytes

corrupt_proof(mode)

Corrupt the proof field, supports different corruption modes.

Source code in packages/testing/src/execution_testing/test_types/blob_types.py

def corrupt_proof(self, mode: ProofCorruptionMode) -> None:
    """Corrupt the proof field, supports different corruption modes."""

    def corrupt_byte(b: bytes) -> Bytes:
        """
        Bit-flip all bits of provided byte using XOR to guarantee change.
        """
        if len(b) != 1:
            raise ValueError("Input must be a single byte")
        return Bytes(bytes([b[0] ^ 0xFF]))

    # >=osaka
    amount_cell_proofs: int = cast(
        int, self.fork.get_blob_constant("AMOUNT_CELL_PROOFS")
    )
    if amount_cell_proofs > 0:
        assert isinstance(self.proof, list), (
            "proof was expected to be a list but it isn't"
        )

        if mode == self.ProofCorruptionMode.CORRUPT_FIRST_BYTE:
            for i in range(len(self.proof)):
                b: Bytes = self.proof[i]
                corrupted: Bytes = Bytes(corrupt_byte(b[:1]) + b[1:])
                self.proof[i] = corrupted
        elif mode == self.ProofCorruptionMode.CORRUPT_LAST_BYTE:
            for i in range(len(self.proof)):
                b = self.proof[i]
                corrupted = Bytes(b[:-1] + corrupt_byte(b[-1:]))
                self.proof[i] = corrupted
        elif mode == self.ProofCorruptionMode.CORRUPT_TO_ALL_ZEROES:
            for i in range(len(self.proof)):
                self.proof[i] = Bytes(bytes(len(self.proof[i])))
        elif mode == self.ProofCorruptionMode.CORRUPT_ALL_BYTES:
            for i in range(len(self.proof)):
                b = self.proof[i]
                corrupted_bytes = Bytes(
                    b"".join(corrupt_byte(bytes([byte])) for byte in b)
                )
                self.proof[i] = corrupted_bytes
        return

    # pre-osaka (cancun and prague)
    assert amount_cell_proofs == 0, (
        f"You need to adjust corrupt_proof to handle fork "
        f"{self.fork.name()}"
    )
    assert isinstance(self.proof, Bytes), (
        "proof was expected to be Bytes but it isn't"
    )

    if mode == self.ProofCorruptionMode.CORRUPT_FIRST_BYTE:
        self.proof = Bytes(corrupt_byte(self.proof[:1]) + self.proof[1:])
    elif mode == self.ProofCorruptionMode.CORRUPT_LAST_BYTE:
        self.proof = Bytes(self.proof[:-1] + corrupt_byte(self.proof[-1:]))
    elif mode == self.ProofCorruptionMode.CORRUPT_TO_ALL_ZEROES:
        self.proof = Bytes(bytes(len(self.proof)))
    elif mode == self.ProofCorruptionMode.CORRUPT_ALL_BYTES:
        self.proof = Bytes(
            b"".join(corrupt_byte(bytes([byte])) for byte in self.proof)
        )

BalAccountChange

Bases: CamelModel, RLPSerializable

Represents all changes to a specific account in a block.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalAccountChange(CamelModel, RLPSerializable):
    """Represents all changes to a specific account in a block."""

    address: Address = Field(..., description="Account address")
    nonce_changes: List[BalNonceChange] = Field(
        default_factory=list, description="List of nonce changes"
    )
    balance_changes: List[BalBalanceChange] = Field(
        default_factory=list, description="List of balance changes"
    )
    code_changes: List[BalCodeChange] = Field(
        default_factory=list, description="List of code changes"
    )
    storage_changes: List[BalStorageSlot] = Field(
        default_factory=list, description="List of storage changes"
    )
    storage_reads: List[ZeroPaddedHexNumber] = Field(
        default_factory=list,
        description="List of storage slots that were read",
    )

    rlp_fields: ClassVar[List[str]] = [
        "address",
        "storage_changes",
        "storage_reads",
        "balance_changes",
        "nonce_changes",
        "code_changes",
    ]

BalAccountExpectation

Bases: CamelModel

Represents expected changes to a specific account in a block.

Same as BalAccountChange but without the address field, used for expectations.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

class BalAccountExpectation(CamelModel):
    """
    Represents expected changes to a specific account in a block.

    Same as BalAccountChange but without the address field,
    used for expectations.
    """

    nonce_changes: List[BalNonceChange] = Field(
        default_factory=list, description="List of expected nonce changes"
    )
    balance_changes: List[BalBalanceChange] = Field(
        default_factory=list, description="List of expected balance changes"
    )
    code_changes: List[BalCodeChange] = Field(
        default_factory=list, description="List of expected code changes"
    )
    storage_changes: List[BalStorageSlot] = Field(
        default_factory=list, description="List of expected storage changes"
    )
    storage_reads: List[StorageKey] = Field(
        default_factory=list, description="List of expected read storage slots"
    )
    absent_values: Optional[BalAccountAbsentValues] = Field(
        default=None,
        description=(
            "Explicit absent value expectations using BalAccountAbsentValues"
        ),
    )

    _EMPTY: ClassVar[Optional["BalAccountExpectation"]] = None

    @classmethod
    def empty(cls) -> "BalAccountExpectation":
        """
        Create an expectation that validates the account has NO changes.

        This is distinct from `BalAccountExpectation()` with no fields set,
        which is ambiguous and clashes with `model_fields_set` logic, and
        will raise a clarifying error if used in expectations.

        Returns:
            A BalAccountExpectation instance with all change lists empty.
            This uses a classvar to facilitate identity checks across
            multiple expectation instances.

        """
        if cls._EMPTY is None:
            cls._EMPTY = cls(
                nonce_changes=[],
                balance_changes=[],
                code_changes=[],
                storage_changes=[],
                storage_reads=[],
            )
        return cls._EMPTY

empty() classmethod

Create an expectation that validates the account has NO changes.

This is distinct from BalAccountExpectation() with no fields set, which is ambiguous and clashes with model_fields_set logic, and will raise a clarifying error if used in expectations.

Returns:

Type	Description
BalAccountExpectation	A BalAccountExpectation instance with all change lists empty.
BalAccountExpectation	This uses a classvar to facilitate identity checks across
BalAccountExpectation	multiple expectation instances.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

@classmethod
def empty(cls) -> "BalAccountExpectation":
    """
    Create an expectation that validates the account has NO changes.

    This is distinct from `BalAccountExpectation()` with no fields set,
    which is ambiguous and clashes with `model_fields_set` logic, and
    will raise a clarifying error if used in expectations.

    Returns:
        A BalAccountExpectation instance with all change lists empty.
        This uses a classvar to facilitate identity checks across
        multiple expectation instances.

    """
    if cls._EMPTY is None:
        cls._EMPTY = cls(
            nonce_changes=[],
            balance_changes=[],
            code_changes=[],
            storage_changes=[],
            storage_reads=[],
        )
    return cls._EMPTY

BalBalanceChange

Bases: CamelModel, RLPSerializable

Represents a balance change in the block access list.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalBalanceChange(CamelModel, RLPSerializable):
    """Represents a balance change in the block access list."""

    block_access_index: ZeroPaddedHexNumber = Field(
        ZeroPaddedHexNumber(1),
        description="Transaction index where the change occurred",
    )
    post_balance: ZeroPaddedHexNumber = Field(
        ..., description="Balance after the transaction"
    )

    rlp_fields: ClassVar[List[str]] = ["block_access_index", "post_balance"]

BalCodeChange

Bases: CamelModel, RLPSerializable

Represents a code change in the block access list.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalCodeChange(CamelModel, RLPSerializable):
    """Represents a code change in the block access list."""

    block_access_index: ZeroPaddedHexNumber = Field(
        ZeroPaddedHexNumber(1),
        description="Transaction index where the change occurred",
    )
    new_code: Bytes = Field(..., description="New code bytes")

    rlp_fields: ClassVar[List[str]] = ["block_access_index", "new_code"]

BalNonceChange

Bases: CamelModel, RLPSerializable

Represents a nonce change in the block access list.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalNonceChange(CamelModel, RLPSerializable):
    """Represents a nonce change in the block access list."""

    block_access_index: ZeroPaddedHexNumber = Field(
        ZeroPaddedHexNumber(1),
        description="Transaction index where the change occurred",
    )
    post_nonce: ZeroPaddedHexNumber = Field(
        ..., description="Nonce value after the transaction"
    )

    rlp_fields: ClassVar[List[str]] = ["block_access_index", "post_nonce"]

BalStorageChange

Bases: CamelModel, RLPSerializable

Represents a change to a specific storage slot.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalStorageChange(CamelModel, RLPSerializable):
    """Represents a change to a specific storage slot."""

    block_access_index: ZeroPaddedHexNumber = Field(
        ZeroPaddedHexNumber(1),
        description="Transaction index where the change occurred",
    )
    post_value: ZeroPaddedHexNumber = Field(
        ..., description="Value after the transaction"
    )

    rlp_fields: ClassVar[List[str]] = ["block_access_index", "post_value"]

BalStorageSlot

Bases: CamelModel, RLPSerializable

Represents all changes to a specific storage slot.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/account_changes.py

class BalStorageSlot(CamelModel, RLPSerializable):
    """Represents all changes to a specific storage slot."""

    slot: ZeroPaddedHexNumber = Field(..., description="Storage slot key")
    slot_changes: List[BalStorageChange] = Field(
        default_factory=list, description="List of changes to this slot"
    )
    validate_any_change: bool = Field(
        default=False,
        description=(
            "If True, asserts at least one change exists in this slot "
            "without validating specific values. Mutually exclusive with "
            "non-empty slot_changes."
        ),
        exclude=True,
    )

    rlp_fields: ClassVar[List[str]] = ["slot", "slot_changes"]

    @model_validator(mode="after")
    def _check_mutual_exclusion(self) -> Self:
        if self.validate_any_change and self.slot_changes:
            raise ValueError(
                "Cannot set both validate_any_change=True and slot_changes. "
                "Use validate_any_change=True to assert at least one change "
                "exists, or slot_changes=[...] to validate specific changes."
            )
        return self

BlockAccessList

Bases: EthereumTestRootModel[List[BalAccountChange]]

Block Access List for t8n tool communication and fixtures.

This model represents the BAL exactly as defined in EIP-7928 - it is itself a list of account changes (root model), not a container.

Used for: - Communication with t8n tools - Fixture generation - RLP encoding for hash verification

Example

bal = BlockAccessList([ BalAccountChange(address=alice, nonce_changes=[...]), BalAccountChange(address=bob, balance_changes=[...]) ])

Source code in packages/testing/src/execution_testing/test_types/block_access_list/t8n.py

class BlockAccessList(EthereumTestRootModel[List[BalAccountChange]]):
    """
    Block Access List for t8n tool communication and fixtures.

    This model represents the BAL exactly as defined in EIP-7928
    - it is itself a list of account changes (root model), not a container.

    Used for:
    - Communication with t8n tools
    - Fixture generation
    - RLP encoding for hash verification

    Example:
        bal = BlockAccessList([
            BalAccountChange(address=alice, nonce_changes=[...]),
            BalAccountChange(address=bob, balance_changes=[...])
        ])

    """

    root: List[BalAccountChange] = Field(default_factory=list)

    def to_list(self) -> List[Any]:
        """Return the list for RLP encoding per EIP-7928."""
        return to_serializable_element(self.root)

    @cached_property
    def rlp(self) -> Bytes:
        """Return the RLP encoded block access list for hash verification."""
        return Bytes(eth_rlp.encode(self.to_list()))

    @cached_property
    def rlp_hash(self) -> Bytes:
        """Return the hash of the RLP encoded block access list."""
        return self.rlp.keccak256()

    def validate_structure(self) -> None:
        """
        Validate BAL structure follows EIP-7928 requirements.

        Checks:
        - Addresses are in lexicographic (ascending) order
        - Transaction indices are sorted and unique within each change list
        - Storage slots are in ascending order
        - Storage reads are in ascending order

        Raises:
            BlockAccessListValidationError: If validation fails

        """
        # Check address ordering (ascending)
        for i in range(1, len(self.root)):
            if self.root[i - 1].address >= self.root[i].address:
                raise BlockAccessListValidationError(
                    f"BAL addresses are not in lexicographic order: "
                    f"{self.root[i - 1].address} >= {self.root[i].address}"
                )

        # Check transaction index ordering and uniqueness within accounts
        for account in self.root:
            changes_to_check: List[tuple[str, List[Any]]] = [
                ("nonce_changes", account.nonce_changes),
                ("balance_changes", account.balance_changes),
                ("code_changes", account.code_changes),
            ]

            for field_name, change_list in changes_to_check:
                if not change_list:
                    continue

                bal_indices = [c.block_access_index for c in change_list]

                # Check both ordering and duplicates
                if bal_indices != sorted(bal_indices):
                    raise BlockAccessListValidationError(
                        f"Block access indices not in ascending order in "
                        f"{field_name} of account {account.address}. Got: "
                        f"{bal_indices}, Expected: {sorted(bal_indices)}"
                    )

                if len(bal_indices) != len(set(bal_indices)):
                    duplicates = sorted(
                        {
                            idx
                            for idx in bal_indices
                            if bal_indices.count(idx) > 1
                        }
                    )
                    raise BlockAccessListValidationError(
                        f"Duplicate transaction indices in {field_name} of "
                        f"account {account.address}. Duplicates: {duplicates}"
                    )

            # Check storage slot ordering
            for i in range(1, len(account.storage_changes)):
                if (
                    account.storage_changes[i - 1].slot
                    >= account.storage_changes[i].slot
                ):
                    raise BlockAccessListValidationError(
                        f"Storage slots not in ascending order in account "
                        f"{account.address}: "
                        f"{account.storage_changes[i - 1].slot} >= "
                        f"{account.storage_changes[i].slot}"
                    )

            # Check bal index ordering and uniqueness within storage slots
            for storage_slot in account.storage_changes:
                if not storage_slot.slot_changes:
                    continue

                bal_indices = [
                    c.block_access_index for c in storage_slot.slot_changes
                ]

                # Check both ordering and duplicates
                if bal_indices != sorted(bal_indices):
                    raise BlockAccessListValidationError(
                        f"Transaction indices not in ascending order in "
                        f"storage slot {storage_slot.slot} of account "
                        f"{account.address}. Got: {bal_indices}, Expected: "
                        f"{sorted(bal_indices)}"
                    )

                if len(bal_indices) != len(set(bal_indices)):
                    duplicates = sorted(
                        {
                            idx
                            for idx in bal_indices
                            if bal_indices.count(idx) > 1
                        }
                    )
                    raise BlockAccessListValidationError(
                        f"Duplicate transaction indices in storage slot "
                        f"{storage_slot.slot} of account {account.address}. "
                        f"Duplicates: {duplicates}"
                    )

            # Check storage reads ordering
            for i in range(1, len(account.storage_reads)):
                if account.storage_reads[i - 1] >= account.storage_reads[i]:
                    raise BlockAccessListValidationError(
                        f"Storage reads not in ascending order in account "
                        f"{account.address}: "
                        f"{account.storage_reads[i - 1]} >= "
                        f"{account.storage_reads[i]}"
                    )

to_list()

Return the list for RLP encoding per EIP-7928.

Source code in packages/testing/src/execution_testing/test_types/block_access_list/t8n.py

def to_list(self) -> List[Any]:
    """Return the list for RLP encoding per EIP-7928."""
    return to_serializable_element(self.root)

rlp cached property

Return the RLP encoded block access list for hash verification.

rlp_hash cached property

Return the hash of the RLP encoded block access list.

validate_structure()

Validate BAL structure follows EIP-7928 requirements.

Checks: - Addresses are in lexicographic (ascending) order - Transaction indices are sorted and unique within each change list - Storage slots are in ascending order - Storage reads are in ascending order

Raises:

Type	Description
BlockAccessListValidationError	If validation fails

Source code in packages/testing/src/execution_testing/test_types/block_access_list/t8n.py

def validate_structure(self) -> None:
    """
    Validate BAL structure follows EIP-7928 requirements.

    Checks:
    - Addresses are in lexicographic (ascending) order
    - Transaction indices are sorted and unique within each change list
    - Storage slots are in ascending order
    - Storage reads are in ascending order

    Raises:
        BlockAccessListValidationError: If validation fails

    """
    # Check address ordering (ascending)
    for i in range(1, len(self.root)):
        if self.root[i - 1].address >= self.root[i].address:
            raise BlockAccessListValidationError(
                f"BAL addresses are not in lexicographic order: "
                f"{self.root[i - 1].address} >= {self.root[i].address}"
            )

    # Check transaction index ordering and uniqueness within accounts
    for account in self.root:
        changes_to_check: List[tuple[str, List[Any]]] = [
            ("nonce_changes", account.nonce_changes),
            ("balance_changes", account.balance_changes),
            ("code_changes", account.code_changes),
        ]

        for field_name, change_list in changes_to_check:
            if not change_list:
                continue

            bal_indices = [c.block_access_index for c in change_list]

            # Check both ordering and duplicates
            if bal_indices != sorted(bal_indices):
                raise BlockAccessListValidationError(
                    f"Block access indices not in ascending order in "
                    f"{field_name} of account {account.address}. Got: "
                    f"{bal_indices}, Expected: {sorted(bal_indices)}"
                )

            if len(bal_indices) != len(set(bal_indices)):
                duplicates = sorted(
                    {
                        idx
                        for idx in bal_indices
                        if bal_indices.count(idx) > 1
                    }
                )
                raise BlockAccessListValidationError(
                    f"Duplicate transaction indices in {field_name} of "
                    f"account {account.address}. Duplicates: {duplicates}"
                )

        # Check storage slot ordering
        for i in range(1, len(account.storage_changes)):
            if (
                account.storage_changes[i - 1].slot
                >= account.storage_changes[i].slot
            ):
                raise BlockAccessListValidationError(
                    f"Storage slots not in ascending order in account "
                    f"{account.address}: "
                    f"{account.storage_changes[i - 1].slot} >= "
                    f"{account.storage_changes[i].slot}"
                )

        # Check bal index ordering and uniqueness within storage slots
        for storage_slot in account.storage_changes:
            if not storage_slot.slot_changes:
                continue

            bal_indices = [
                c.block_access_index for c in storage_slot.slot_changes
            ]

            # Check both ordering and duplicates
            if bal_indices != sorted(bal_indices):
                raise BlockAccessListValidationError(
                    f"Transaction indices not in ascending order in "
                    f"storage slot {storage_slot.slot} of account "
                    f"{account.address}. Got: {bal_indices}, Expected: "
                    f"{sorted(bal_indices)}"
                )

            if len(bal_indices) != len(set(bal_indices)):
                duplicates = sorted(
                    {
                        idx
                        for idx in bal_indices
                        if bal_indices.count(idx) > 1
                    }
                )
                raise BlockAccessListValidationError(
                    f"Duplicate transaction indices in storage slot "
                    f"{storage_slot.slot} of account {account.address}. "
                    f"Duplicates: {duplicates}"
                )

        # Check storage reads ordering
        for i in range(1, len(account.storage_reads)):
            if account.storage_reads[i - 1] >= account.storage_reads[i]:
                raise BlockAccessListValidationError(
                    f"Storage reads not in ascending order in account "
                    f"{account.address}: "
                    f"{account.storage_reads[i - 1]} >= "
                    f"{account.storage_reads[i]}"
                )

BlockAccessListExpectation

Bases: CamelModel

Block Access List expectation model for test writing.

This model is used to define expected BAL values in tests. It supports: - Partial validation (only checks explicitly set fields) - Convenient test syntax with named parameters - Verification against actual BAL from t8n - Explicit exclusion of addresses (using None values)

Example

In test definition

expected_block_access_list = BlockAccessListExpectation( account_expectations={ alice: BalAccountExpectation( nonce_changes=[ BalNonceChange(block_access_index=1, post_nonce=1) ] ), bob: None, # Bob should NOT be in the BAL } )

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

class BlockAccessListExpectation(CamelModel):
    """
    Block Access List expectation model for test writing.

    This model is used to define expected BAL values in tests. It supports:
    - Partial validation (only checks explicitly set fields)
    - Convenient test syntax with named parameters
    - Verification against actual BAL from t8n
    - Explicit exclusion of addresses (using None values)

    Example:
        # In test definition
        expected_block_access_list = BlockAccessListExpectation(
            account_expectations={
                alice: BalAccountExpectation(
                    nonce_changes=[
                        BalNonceChange(block_access_index=1, post_nonce=1)
                    ]
                ),
                bob: None,  # Bob should NOT be in the BAL
            }
        )

    """

    account_expectations: Dict[Address, BalAccountExpectation | None] = Field(
        default_factory=dict,
        description="Expected account changes or exclusions to verify",
    )

    _modifier: Callable[["BlockAccessList"], "BlockAccessList"] | None = (
        PrivateAttr(default=None)
    )

    def modify(
        self, *modifiers: Callable[["BlockAccessList"], "BlockAccessList"]
    ) -> "BlockAccessListExpectation":
        """
        Create a new expectation with a modifier for invalid test cases.

        Args:
            modifiers: One or more functions that take and return
                       a BlockAccessList

        Returns:
            A new BlockAccessListExpectation instance with
            the modifiers applied

        Example:
            from execution_testing.test_types.block_access_list.
            modifiers import remove_nonces

            expectation = BlockAccessListExpectation(
                account_expectations={...}
            ).modify(remove_nonces(alice))

        """
        new_instance = self.model_copy(deep=True)
        new_instance._modifier = compose(*modifiers)
        return new_instance

    def modify_if_invalid_test(
        self, t8n_bal: "BlockAccessList"
    ) -> "BlockAccessList":
        """
        Apply the modifier to the given BAL if this is an invalid test case.

        Args:
            t8n_bal: The BlockAccessList from t8n tool

        Returns:
            The potentially transformed BlockAccessList for the fixture

        """
        if self._modifier:
            return self._modifier(t8n_bal)
        return t8n_bal

    def verify_against(self, actual_bal: "BlockAccessList") -> None:
        """
        Verify that the actual BAL from the client matches this expected BAL.

        Validation steps:
        1. Verify address expectations - presence or explicit absence
        2. Verify expected changes within accounts match actual changes

        Args:
            actual_bal: The BlockAccessList model from the client

        Raises:
            BlockAccessListValidationError: If verification fails

        """
        actual_accounts_by_addr = {acc.address: acc for acc in actual_bal.root}
        for address, expectation in self.account_expectations.items():
            if expectation is None:
                # check explicit exclusion of address when set to `None`
                if address in actual_accounts_by_addr:
                    raise BlockAccessListValidationError(
                        f"Address {address} should not be in BAL but was found"
                    )
            elif not expectation.model_fields_set:
                # Disallow ambiguous BalAccountExpectation() with no fields set
                raise BlockAccessListValidationError(
                    f"Address {address}: BalAccountExpectation() with no "
                    "fields set is ambiguous. Use BalAccountExpectation."
                    "empty() to validate no changes, or explicitly set the "
                    "fields to validate (e.g., nonce_changes=[...])."
                )
            else:
                # check address is present and validate changes
                if address not in actual_accounts_by_addr:
                    raise BlockAccessListValidationError(
                        f"Expected address {address} not found in actual BAL"
                    )

                if expectation is BalAccountExpectation.empty():
                    # explicit check for "no changes" validation w/ .empty()
                    if actual_accounts_by_addr.get(
                        address
                    ) != BalAccountChange(address=address):
                        raise BlockAccessListValidationError(
                            f"No account changes expected for {address} but "
                            f"found changes: "
                            f"{actual_accounts_by_addr[address]}"
                        )

                actual_account = actual_accounts_by_addr[address]
                try:
                    self._compare_account_expectations(
                        expectation, actual_account
                    )
                except AssertionError as e:
                    raise BlockAccessListValidationError(
                        f"Account {address}: {str(e)}"
                    ) from e

    @staticmethod
    def _compare_account_expectations(
        expected: BalAccountExpectation, actual: BalAccountChange
    ) -> None:
        """
        Compare expected and actual account changes using
        subsequence validation.

        Args:
            expected: The expected account changes
            actual: The actual account changes from the BAL

        Raises:
            AssertionError: If validation fails

        """
        # Check absence expectations first if defined
        if expected.absent_values is not None:
            expected.absent_values.validate_against(actual)

        # Validate expected changes using subsequence validation
        field_pairs: List[tuple[str, Any, Any]] = [
            ("nonce_changes", expected.nonce_changes, actual.nonce_changes),
            (
                "balance_changes",
                expected.balance_changes,
                actual.balance_changes,
            ),
            ("code_changes", expected.code_changes, actual.code_changes),
            (
                "storage_changes",
                expected.storage_changes,
                actual.storage_changes,
            ),
            ("storage_reads", expected.storage_reads, actual.storage_reads),
        ]

        for field_name, expected_list, actual_list in field_pairs:
            # Only validate fields that were explicitly set
            if field_name not in expected.model_fields_set:
                continue

            # Check if explicitly set to empty but actual has values
            if not expected_list and actual_list:
                raise BlockAccessListValidationError(
                    f"Expected {field_name} to be empty but found "
                    f"{actual_list}"
                )

            if field_name == "storage_reads":
                # storage_reads is a simple list of StorageKey
                actual_idx = 0
                for expected_read in expected_list:
                    found = False
                    while actual_idx < len(actual_list):
                        if actual_list[actual_idx] == expected_read:
                            found = True
                            actual_idx += 1
                            break
                        actual_idx += 1

                    if not found:
                        raise BlockAccessListValidationError(
                            f"Storage read {expected_read} not found or not "
                            f"in correct order. Actual reads: {actual_list}"
                        )

            elif field_name == "storage_changes":
                # storage_changes is a list of BalStorageSlot
                actual_idx = 0
                for expected_slot in expected_list:
                    found = False
                    while actual_idx < len(actual_list):
                        if actual_list[actual_idx].slot == expected_slot.slot:
                            # Found matching slot, now validate slot_changes
                            actual_slot_changes = actual_list[
                                actual_idx
                            ].slot_changes
                            expected_slot_changes = expected_slot.slot_changes

                            if expected_slot.validate_any_change:
                                # Assert at least one change exists
                                if not actual_slot_changes:
                                    raise BlockAccessListValidationError(
                                        f"Expected at least one change "
                                        f"in slot {expected_slot.slot} "
                                        f"(validate_any_change=True) "
                                        f"but found none"
                                    )
                            elif not expected_slot_changes:
                                # Explicitly empty = assert no changes
                                if actual_slot_changes:
                                    raise BlockAccessListValidationError(
                                        f"Expected no changes in slot "
                                        f"{expected_slot.slot} but found "
                                        f"{actual_slot_changes}"
                                    )
                            else:
                                # Validate slot_changes as subsequence
                                slot_actual_idx = 0
                                for expected_change in expected_slot_changes:
                                    slot_found = False
                                    while slot_actual_idx < len(
                                        actual_slot_changes
                                    ):
                                        actual_change = actual_slot_changes[
                                            slot_actual_idx
                                        ]
                                        actual_ba_idx = (
                                            actual_change.block_access_index
                                        )
                                        expected_ba_idx = (
                                            expected_change.block_access_index
                                        )
                                        idx_match = (
                                            actual_ba_idx == expected_ba_idx
                                        )
                                        val_match = (
                                            actual_change.post_value
                                            == expected_change.post_value
                                        )
                                        if idx_match and val_match:
                                            slot_found = True
                                            slot_actual_idx += 1
                                            break
                                        slot_actual_idx += 1

                                    if not slot_found:
                                        raise BlockAccessListValidationError(
                                            f"Storage change "
                                            f"{expected_change} not found or "
                                            f"not in correct order in slot "
                                            f"{expected_slot.slot}. Actual "
                                            f"slot changes: "
                                            f"{actual_slot_changes}"
                                        )

                            found = True
                            actual_idx += 1
                            break
                        actual_idx += 1

                    if not found:
                        raise BlockAccessListValidationError(
                            f"Storage slot {expected_slot.slot} not found "
                            f"or not in correct order. Actual slots: "
                            f"{[s.slot for s in actual_list]}"
                        )

            else:
                # Handle nonce_changes, balance_changes, code_changes
                # Create tuples for comparison (ordering already validated)
                if field_name == "nonce_changes":
                    expected_tuples = [
                        (c.block_access_index, c.post_nonce)
                        for c in expected_list
                    ]
                    actual_tuples = [
                        (c.block_access_index, c.post_nonce)
                        for c in actual_list
                    ]
                    item_type = "nonce"
                elif field_name == "balance_changes":
                    expected_tuples = [
                        (c.block_access_index, int(c.post_balance))
                        for c in expected_list
                    ]
                    actual_tuples = [
                        (c.block_access_index, int(c.post_balance))
                        for c in actual_list
                    ]
                    item_type = "balance"
                elif field_name == "code_changes":
                    expected_tuples = [
                        (c.block_access_index, bytes(c.new_code))
                        for c in expected_list
                    ]
                    actual_tuples = [
                        (c.block_access_index, bytes(c.new_code))
                        for c in actual_list
                    ]
                    item_type = "code"
                else:
                    # sanity check
                    raise ValueError(f"Unexpected field type: {field_name}")

                # Check that expected forms a subsequence of actual
                actual_idx = 0
                for exp_tuple in expected_tuples:
                    found = False
                    while actual_idx < len(actual_tuples):
                        if actual_tuples[actual_idx] == exp_tuple:
                            found = True
                            actual_idx += 1
                            break
                        actual_idx += 1

                    if not found:
                        raise BlockAccessListValidationError(
                            f"{item_type.capitalize()} change {exp_tuple} not "
                            f"found or not in correct order. Actual changes: "
                            f"{actual_tuples}"
                        )

modify(*modifiers)

Create a new expectation with a modifier for invalid test cases.

Parameters:

Name	Type	Description	Default
modifiers	Callable[[BlockAccessList], BlockAccessList]	One or more functions that take and return a BlockAccessList	()

Returns:

Type	Description
BlockAccessListExpectation	A new BlockAccessListExpectation instance with
BlockAccessListExpectation	the modifiers applied

Example

from execution_testing.test_types.block_access_list. modifiers import remove_nonces

expectation = BlockAccessListExpectation( account_expectations={...} ).modify(remove_nonces(alice))

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

def modify(
    self, *modifiers: Callable[["BlockAccessList"], "BlockAccessList"]
) -> "BlockAccessListExpectation":
    """
    Create a new expectation with a modifier for invalid test cases.

    Args:
        modifiers: One or more functions that take and return
                   a BlockAccessList

    Returns:
        A new BlockAccessListExpectation instance with
        the modifiers applied

    Example:
        from execution_testing.test_types.block_access_list.
        modifiers import remove_nonces

        expectation = BlockAccessListExpectation(
            account_expectations={...}
        ).modify(remove_nonces(alice))

    """
    new_instance = self.model_copy(deep=True)
    new_instance._modifier = compose(*modifiers)
    return new_instance

modify_if_invalid_test(t8n_bal)

Apply the modifier to the given BAL if this is an invalid test case.

Parameters:

Name	Type	Description	Default
t8n_bal	BlockAccessList	The BlockAccessList from t8n tool	required

Returns:

Type	Description
BlockAccessList	The potentially transformed BlockAccessList for the fixture

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

def modify_if_invalid_test(
    self, t8n_bal: "BlockAccessList"
) -> "BlockAccessList":
    """
    Apply the modifier to the given BAL if this is an invalid test case.

    Args:
        t8n_bal: The BlockAccessList from t8n tool

    Returns:
        The potentially transformed BlockAccessList for the fixture

    """
    if self._modifier:
        return self._modifier(t8n_bal)
    return t8n_bal

verify_against(actual_bal)

Verify that the actual BAL from the client matches this expected BAL.

Validation steps: 1. Verify address expectations - presence or explicit absence 2. Verify expected changes within accounts match actual changes

Parameters:

Name	Type	Description	Default
actual_bal	BlockAccessList	The BlockAccessList model from the client	required

Raises:

Type	Description
BlockAccessListValidationError	If verification fails

Source code in packages/testing/src/execution_testing/test_types/block_access_list/expectations.py

def verify_against(self, actual_bal: "BlockAccessList") -> None:
    """
    Verify that the actual BAL from the client matches this expected BAL.

    Validation steps:
    1. Verify address expectations - presence or explicit absence
    2. Verify expected changes within accounts match actual changes

    Args:
        actual_bal: The BlockAccessList model from the client

    Raises:
        BlockAccessListValidationError: If verification fails

    """
    actual_accounts_by_addr = {acc.address: acc for acc in actual_bal.root}
    for address, expectation in self.account_expectations.items():
        if expectation is None:
            # check explicit exclusion of address when set to `None`
            if address in actual_accounts_by_addr:
                raise BlockAccessListValidationError(
                    f"Address {address} should not be in BAL but was found"
                )
        elif not expectation.model_fields_set:
            # Disallow ambiguous BalAccountExpectation() with no fields set
            raise BlockAccessListValidationError(
                f"Address {address}: BalAccountExpectation() with no "
                "fields set is ambiguous. Use BalAccountExpectation."
                "empty() to validate no changes, or explicitly set the "
                "fields to validate (e.g., nonce_changes=[...])."
            )
        else:
            # check address is present and validate changes
            if address not in actual_accounts_by_addr:
                raise BlockAccessListValidationError(
                    f"Expected address {address} not found in actual BAL"
                )

            if expectation is BalAccountExpectation.empty():
                # explicit check for "no changes" validation w/ .empty()
                if actual_accounts_by_addr.get(
                    address
                ) != BalAccountChange(address=address):
                    raise BlockAccessListValidationError(
                        f"No account changes expected for {address} but "
                        f"found changes: "
                        f"{actual_accounts_by_addr[address]}"
                    )

            actual_account = actual_accounts_by_addr[address]
            try:
                self._compare_account_expectations(
                    expectation, actual_account
                )
            except AssertionError as e:
                raise BlockAccessListValidationError(
                    f"Account {address}: {str(e)}"
                ) from e

Environment

Bases: EnvironmentGeneric[ZeroPaddedHexNumber]

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

Source code in packages/testing/src/execution_testing/test_types/block_types.py

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

    @model_validator(mode="before")
    @classmethod
    def strip_computed_fields(cls, data: Any) -> Any:
        """Strip computed fields that are not valid input fields."""
        if isinstance(data, dict):
            data.pop("parent_hash", None)
            data.pop("parentHash", None)
        return data

    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[ZeroPaddedHexNumber, 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)

    # EIP-7928: Block-level access lists
    bal_hash: Hash | None = Field(None)
    block_access_lists: Bytes | None = Field(None)

    @computed_field  # type: ignore[prop-decorator]
    @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(
                block_number=number, timestamp=timestamp
            )
            and self.prev_randao is None
        ):
            updated_values["prev_randao"] = 0

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

        if (
            fork.header_base_fee_required(
                block_number=number, timestamp=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(
            block_number=number, timestamp=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(
                block_number=number, timestamp=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(
                block_number=number, timestamp=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(
                block_number=number, timestamp=timestamp
            )
            and self.parent_beacon_block_root is None
        ):
            updated_values["parent_beacon_block_root"] = 0

        return self.copy(**updated_values)

    def __hash__(self) -> int:
        """Hashes the environment object."""
        hash_dict = self.model_dump(exclude_none=True, by_alias=True)

        sorted_items = sorted(hash_dict.items())
        hash_string = str(sorted_items)

        digest = hashlib.sha256(hash_string.encode("utf-8")).digest()
        return int.from_bytes(digest[:8], byteorder="big")

    def __eq__(self, other: object) -> bool:
        """Check if two environment objects are equal."""
        if not isinstance(other, Environment):
            return False

        self_dict = self.model_dump(exclude_none=True, by_alias=True)
        self_dict["extra_data"] = self.extra_data.hex()

        other_dict = other.model_dump(exclude_none=True, by_alias=True)
        other_dict["extra_data"] = other.extra_data.hex()

        return self_dict == other_dict

strip_computed_fields(data) classmethod

Strip computed fields that are not valid input fields.

Source code in packages/testing/src/execution_testing/test_types/block_types.py

@model_validator(mode="before")
@classmethod
def strip_computed_fields(cls, data: Any) -> Any:
    """Strip computed fields that are not valid input fields."""
    if isinstance(data, dict):
        data.pop("parent_hash", None)
        data.pop("parentHash", None)
    return data

parent_hash 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 packages/testing/src/execution_testing/test_types/block_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(
            block_number=number, timestamp=timestamp
        )
        and self.prev_randao is None
    ):
        updated_values["prev_randao"] = 0

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

    if (
        fork.header_base_fee_required(
            block_number=number, timestamp=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(
        block_number=number, timestamp=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(
            block_number=number, timestamp=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(
            block_number=number, timestamp=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(
            block_number=number, timestamp=timestamp
        )
        and self.parent_beacon_block_root is None
    ):
        updated_values["parent_beacon_block_root"] = 0

    return self.copy(**updated_values)

__hash__()

Hashes the environment object.

Source code in packages/testing/src/execution_testing/test_types/block_types.py

def __hash__(self) -> int:
    """Hashes the environment object."""
    hash_dict = self.model_dump(exclude_none=True, by_alias=True)

    sorted_items = sorted(hash_dict.items())
    hash_string = str(sorted_items)

    digest = hashlib.sha256(hash_string.encode("utf-8")).digest()
    return int.from_bytes(digest[:8], byteorder="big")

__eq__(other)

Check if two environment objects are equal.

Source code in packages/testing/src/execution_testing/test_types/block_types.py

def __eq__(self, other: object) -> bool:
    """Check if two environment objects are equal."""
    if not isinstance(other, Environment):
        return False

    self_dict = self.model_dump(exclude_none=True, by_alias=True)
    self_dict["extra_data"] = self.extra_data.hex()

    other_dict = other.model_dump(exclude_none=True, by_alias=True)
    other_dict["extra_data"] = other.extra_data.hex()

    return self_dict == other_dict

EnvironmentDefaults dataclass

Default environment values.

Source code in packages/testing/src/execution_testing/test_types/block_types.py

@dataclass
class EnvironmentDefaults:
    """Default environment values."""

    # By default, the constant `DEFAULT_BLOCK_GAS_LIMIT` is used.
    # Other libraries (pytest plugins) may override this value by modifying the
    # `EnvironmentDefaults.gas_limit` class attribute.
    gas_limit: int = DEFAULT_BLOCK_GAS_LIMIT

Withdrawal

Bases: WithdrawalGeneric[HexNumber]

Withdrawal type.

Source code in packages/testing/src/execution_testing/test_types/block_types.py

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

    pass

ChainConfig

Bases: CamelModel

Chain configuration.

Source code in packages/testing/src/execution_testing/test_types/chain_config_types.py

class ChainConfig(CamelModel):
    """Chain configuration."""

    chain_id: int = Field(
        default_factory=lambda: ChainConfigDefaults.chain_id,
        validate_default=True,
    )

ChainConfigDefaults

Default values for the chain configuration.

Can be modified by modules that import this module and want to override the default values.

Source code in packages/testing/src/execution_testing/test_types/chain_config_types.py

class ChainConfigDefaults:
    """
    Default values for the chain configuration.

    Can be modified by modules that import this module and want to override the
    default values.
    """

    chain_id: int = 1

TestParameterGroup

Bases: BaseModel

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 packages/testing/src/execution_testing/test_types/helpers.py

class TestParameterGroup(BaseModel):
    """
    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

    model_config = ConfigDict(frozen=True, validate_default=True)

    def __repr__(self) -> str:
        """
        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 = [
            f"{field}_{value}"
            # Include the field only if it is not optional or not set to its
            # default value
            for field, value in self.model_dump(
                exclude_defaults=True, exclude_unset=True
            ).items()
        ]

        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 packages/testing/src/execution_testing/test_types/helpers.py

def __repr__(self) -> str:
    """
    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 = [
        f"{field}_{value}"
        # Include the field only if it is not optional or not set to its
        # default value
        for field, value in self.model_dump(
            exclude_defaults=True, exclude_unset=True
        ).items()
    ]

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

add_kzg_version(b_hashes, kzg_version)

Add KZG version to each blob hash.

Source code in packages/testing/src/execution_testing/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 packages/testing/src/execution_testing/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 packages/testing/src/execution_testing/test_types/helpers.py

def compute_create2_address(
    address: FixedSizeBytesConvertible,
    salt: FixedSizeBytesConvertible,
    initcode: Bytecode | BytesConvertible,
) -> Address:
    """
    Compute address of the resulting contract created using the `CREATE2`
    opcode.
    """
    if isinstance(initcode, Bytecode):
        initcode_hash = initcode.keccak256()
    else:
        initcode_hash = Bytes(initcode).keccak256()
    hash_bytes = Bytes(
        b"\xff" + Address(address) + Hash(salt) + initcode_hash
    ).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 packages/testing/src/execution_testing/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_deterministic_create2_address(*, salt, initcode, fork)

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

Source code in packages/testing/src/execution_testing/test_types/helpers.py

def compute_deterministic_create2_address(
    *,
    salt: FixedSizeBytesConvertible,
    initcode: BytesConvertible,
    fork: Fork,
) -> Address:
    """
    Compute address of the resulting contract created using the `CREATE2`
    opcode using the `DETERMINISTIC_FACTORY_ADDRESS`.
    """
    factory_address = (
        fork.deterministic_factory_predeploy_address()
        or DETERMINISTIC_FACTORY_ADDRESS
    )
    return compute_create2_address(
        address=factory_address,
        salt=salt,
        initcode=initcode,
    )

contract_address_from_hash(account_hash, salt)

Calculate an address from a given (account) hash plus a salt.

Useful to not duplicate accounts in the pre-allocation when grouping many tests.

Source code in packages/testing/src/execution_testing/test_types/helpers.py

def contract_address_from_hash(account_hash: Hash, salt: int) -> Address:
    """
    Calculate an address from a given (account) hash plus a salt.

    Useful to not duplicate accounts in the pre-allocation when grouping
    many tests.
    """
    return Address(
        Bytes(account_hash + salt.to_bytes(64, "big")).sha256()[12:]
    )

eoa_from_hash(account_hash, salt)

Calculate an EOA from a given (account) hash plus a salt.

Useful to not duplicate accounts in the pre-allocation when grouping many tests.

Source code in packages/testing/src/execution_testing/test_types/helpers.py

def eoa_from_hash(account_hash: Hash, salt: int) -> EOA:
    """
    Calculate an EOA from a given (account) hash plus a salt.

    Useful to not duplicate accounts in the pre-allocation when grouping
    many tests.
    """
    return EOA(key=Bytes(account_hash + salt.to_bytes(64, "big")).sha256())

TestPhase

Bases: str, Enum

Test phase for state and blockchain tests.

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

class TestPhase(str, Enum):
    """Test phase for state and blockchain tests."""

    SETUP = "setup"
    # TODO: Change string to "execution", remain as "testing" for backwards
    # compatibility
    EXECUTION = "testing"
    CLEANUP = "cleanup"

TestPhaseManager

Manages test phases for transactions and blocks.

This singleton class provides context managers for SETUP and EXECUTION phases. Transactions automatically detect and tag themselves with the current phase.

Usage

with TestPhaseManager.setup(): # Transactions created here have test_phase = SETUP setup_tx = Transaction(...)

with TestPhaseManager.execution(): # Transactions created here have test_phase = EXECUTION benchmark_tx = Transaction(...)

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

class TestPhaseManager:
    """
    Manages test phases for transactions and blocks.

    This singleton class provides context managers for SETUP and
    EXECUTION phases. Transactions automatically detect and tag
    themselves with the current phase.

    Usage:
        with TestPhaseManager.setup():
            # Transactions created here have test_phase = SETUP
            setup_tx = Transaction(...)

        with TestPhaseManager.execution():
            # Transactions created here have test_phase = EXECUTION
            benchmark_tx = Transaction(...)
    """

    _current_phase: ClassVar[Optional[TestPhase]] = None

    @classmethod
    @contextmanager
    def setup(cls) -> Iterator[None]:
        """Context manager for the setup phase of a benchmark test."""
        old_phase = cls._current_phase
        cls._current_phase = TestPhase.SETUP
        try:
            yield
        finally:
            cls._current_phase = old_phase

    @classmethod
    @contextmanager
    def execution(cls) -> Iterator[None]:
        """Context manager for the execution phase of a test."""
        old_phase = cls._current_phase
        cls._current_phase = TestPhase.EXECUTION
        try:
            yield
        finally:
            cls._current_phase = old_phase

    @classmethod
    def get_current_phase(cls) -> Optional[TestPhase]:
        """Get the current test phase."""
        return cls._current_phase

    @classmethod
    def reset(cls) -> None:
        """Reset the phase state to None (primarily for testing)."""
        cls._current_phase = None

setup() classmethod

Context manager for the setup phase of a benchmark test.

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

@classmethod
@contextmanager
def setup(cls) -> Iterator[None]:
    """Context manager for the setup phase of a benchmark test."""
    old_phase = cls._current_phase
    cls._current_phase = TestPhase.SETUP
    try:
        yield
    finally:
        cls._current_phase = old_phase

execution() classmethod

Context manager for the execution phase of a test.

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

@classmethod
@contextmanager
def execution(cls) -> Iterator[None]:
    """Context manager for the execution phase of a test."""
    old_phase = cls._current_phase
    cls._current_phase = TestPhase.EXECUTION
    try:
        yield
    finally:
        cls._current_phase = old_phase

get_current_phase() classmethod

Get the current test phase.

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

@classmethod
def get_current_phase(cls) -> Optional[TestPhase]:
    """Get the current test phase."""
    return cls._current_phase

reset() classmethod

Reset the phase state to None (primarily for testing).

Source code in packages/testing/src/execution_testing/test_types/phase_manager.py

@classmethod
def reset(cls) -> None:
    """Reset the phase state to None (primarily for testing)."""
    cls._current_phase = None

TransactionLog

Bases: CamelModel

Transaction log.

Source code in packages/testing/src/execution_testing/test_types/receipt_types.py

class TransactionLog(CamelModel):
    """Transaction log."""

    address: Address | None = None
    topics: List[Hash] | None = None
    data: Bytes | None = None
    block_number: HexNumber | None = None
    transaction_hash: Hash | None = None
    transaction_index: HexNumber | None = None
    block_hash: Hash | None = None
    log_index: HexNumber | None = None
    removed: bool | None = None
    block_timestamp: HexNumber | None = None

TransactionReceipt

Bases: CamelModel

Transaction receipt.

Source code in packages/testing/src/execution_testing/test_types/receipt_types.py

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

    @model_validator(mode="before")
    @classmethod
    def strip_extra_fields(cls, data: Any) -> Any:
        """Strip extra fields from t8n tool output not part of model."""
        if isinstance(data, dict):
            # geth (1.16+) returns extra fields in receipts
            data.pop("type", None)
            data.pop("blockNumber", None)
        return data

    transaction_hash: Hash | None = None
    post_state: Hash | None = Field(
        None, validation_alias=AliasChoices("post_state", "postState")
    )
    root: Bytes | None = None
    status: HexNumber | None = Field(
        None, validation_alias=AliasChoices("status", "succeeded")
    )
    cumulative_gas_used: HexNumber | None = None
    bloom: Bloom | None = Field(
        None, validation_alias=AliasChoices("logs_bloom", "logsBloom", "bloom")
    )
    logs: List[TransactionLog] | None = None
    gas_used: HexNumber | 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

strip_extra_fields(data) classmethod

Strip extra fields from t8n tool output not part of model.

Source code in packages/testing/src/execution_testing/test_types/receipt_types.py

@model_validator(mode="before")
@classmethod
def strip_extra_fields(cls, data: Any) -> Any:
    """Strip extra fields from t8n tool output not part of model."""
    if isinstance(data, dict):
        # geth (1.16+) returns extra fields in receipts
        data.pop("type", None)
        data.pop("blockNumber", None)
    return data

ConsolidationRequest

Bases: RequestBase, CamelModel

Consolidation Request type.

Source code in packages/testing/src/execution_testing/test_types/request_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(0) class-attribute instance-attribute

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

source_pubkey instance-attribute

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

target_pubkey 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 packages/testing/src/execution_testing/test_types/request_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 packages/testing/src/execution_testing/test_types/request_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 instance-attribute

The public key of the beacon chain validator.

withdrawal_credentials instance-attribute

The withdrawal credentials of the beacon chain validator.

amount instance-attribute

The amount in gwei of the deposit.

signature instance-attribute

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

index instance-attribute

The index of the deposit.

__bytes__()

Return deposit's attributes as bytes.

Source code in packages/testing/src/execution_testing/test_types/request_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")
    )

Requests

Requests for the transition tool.

Source code in packages/testing/src/execution_testing/test_types/request_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,
    ) -> 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 packages/testing/src/execution_testing/test_types/request_types.py

def __init__(
    self,
    *requests: RequestBase,
    requests_lists: List[List[RequestBase] | Bytes] | None = 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 packages/testing/src/execution_testing/test_types/request_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()

WithdrawalRequest

Bases: RequestBase, CamelModel

Withdrawal Request type.

Source code in packages/testing/src/execution_testing/test_types/request_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(0) class-attribute instance-attribute

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

validator_pubkey instance-attribute

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

amount instance-attribute

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

__bytes__()

Return withdrawal's attributes as bytes.

Source code in packages/testing/src/execution_testing/test_types/request_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")
    )

AuthorizationTuple

Bases: AuthorizationTupleGeneric[HexNumber]

Authorization tuple for transactions.

Source code in packages/testing/src/execution_testing/test_types/transaction_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") -> None:
        """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 packages/testing/src/execution_testing/test_types/transaction_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 packages/testing/src/execution_testing/test_types/transaction_types.py

def sign(self: "AuthorizationTuple") -> None:
    """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

NetworkWrappedTransaction

Bases: CamelModel, RLPSerializable

Network wrapped transaction as defined in EIP-4844.

< Osaka: rlp([tx_payload_body, blobs, commitments, proofs])

= Osaka: rlp([tx_payload_body, wrapper_version, blobs, commitments, cell_proofs])

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

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

    < Osaka: rlp([tx_payload_body, blobs, commitments, proofs])

    >= Osaka: rlp([tx_payload_body, wrapper_version,  blobs, commitments,
                   cell_proofs])
    """

    tx: Transaction
    blob_objects: Sequence[Blob]
    wrapper_version: int | None = None  # only exists in >= osaka

    @computed_field  # type: ignore[prop-decorator]
    @property
    def blobs(self) -> Sequence[Bytes]:
        """Return a list of blob data as bytes."""
        return [blob.data for blob in self.blob_objects]

    @computed_field  # type: ignore[prop-decorator]
    @property
    def commitments(self) -> Sequence[Bytes]:
        """Return a list of kzg commitments."""
        return [blob.commitment for blob in self.blob_objects]

    @computed_field  # type: ignore[prop-decorator]
    @property
    def proofs(self) -> Sequence[Bytes] | None:
        """Return a list of kzg proofs (returns None >= Osaka)."""
        if self.wrapper_version is not None:
            return None

        proofs: list[Bytes] = []
        for blob in self.blob_objects:
            assert isinstance(blob.proof, Bytes)
            proofs.append(blob.proof)

        return proofs

    @computed_field  # type: ignore[prop-decorator]
    @property
    def cell_proofs(self) -> Sequence[Bytes] | None:
        """Return a list of cells (returns None < Osaka)."""
        if self.wrapper_version is None:
            return None

        cells: list[Bytes] = []
        for blob in self.blob_objects:
            assert isinstance(blob.proof, list)
            cells.extend(blob.proof)

        return cells

    def get_rlp_fields(self) -> List[str]:
        """
        Return an ordered list of field names to be included in RLP
        serialization.

        Function can be overridden to customize the logic to return the fields.

        By default, rlp_fields class variable is used.

        The list can be nested list up to one extra level to represent nested
        fields.
        """
        # only put a wrapper_version field for >=osaka (value 1), otherwise
        # omit field
        wrapper = []
        if self.wrapper_version is not None:
            wrapper = ["wrapper_version"]

        rlp_proofs: list[str] = []
        if self.proofs is not None:
            rlp_proofs = ["proofs"]

        rlp_cell_proofs: list[str] = []
        if self.cell_proofs is not None:
            rlp_cell_proofs = ["cell_proofs"]

        rlp_fields: List[str] = [  # structure explained in
            # https://eips.ethereum.org/EIPS/eip-7594#Networking
            "tx",  # tx_payload_body
            *wrapper,  # wrapper_version, which is always 1 for osaka (was non-
            # existing before)
            "blobs",  # Blob.data
            "commitments",
            *rlp_proofs,
            *rlp_cell_proofs,
        ]

        assert ("proofs" in rlp_fields) or ("cell_proofs" in rlp_fields), (
            "Neither proofs nor cell_proofs are in rlp_fields. Critical error!"
        )

        # logger.debug(f"Ended up with this rlp field list: {rlp_fields}")

        return rlp_fields

    @property
    def hash(self) -> Hash:
        """Return the hash of the transaction."""
        return self.tx.hash

    @property
    def sender(self) -> EOA | None:
        """Return the sender of the transaction."""
        return self.tx.sender

    @property
    def to(self) -> Address | None:
        """Return the to address of the transaction."""
        return self.tx.to

    def set_gas_price(
        self,
        *,
        gas_price: int,
        max_fee_per_gas: int,
        max_priority_fee_per_gas: int,
        max_fee_per_blob_gas: int,
    ) -> None:
        """Set gas price to values of the current execution environment."""
        self.tx.set_gas_price(
            gas_price=gas_price,
            max_fee_per_gas=max_fee_per_gas,
            max_priority_fee_per_gas=max_priority_fee_per_gas,
            max_fee_per_blob_gas=max_fee_per_blob_gas,
        )

    def signer_minimum_balance(self, *, fork: Fork) -> int:
        """Return minimum balance of the signer."""
        return self.tx.signer_minimum_balance(fork=fork)

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

    def with_signature_and_sender(
        self, *, keep_secret_key: bool = False
    ) -> Self:
        """Return a new transaction with the signature and sender added."""
        return self.model_copy(
            update={
                "tx": self.tx.with_signature_and_sender(
                    keep_secret_key=keep_secret_key
                ),
            }
        )

blobs property

Return a list of blob data as bytes.

commitments property

Return a list of kzg commitments.

proofs property

Return a list of kzg proofs (returns None >= Osaka).

cell_proofs property

Return a list of cells (returns None < Osaka).

get_rlp_fields()

Return an ordered list of field names to be included in RLP serialization.

Function can be overridden to customize the logic to return the fields.

By default, rlp_fields class variable is used.

The list can be nested list up to one extra level to represent nested fields.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def get_rlp_fields(self) -> List[str]:
    """
    Return an ordered list of field names to be included in RLP
    serialization.

    Function can be overridden to customize the logic to return the fields.

    By default, rlp_fields class variable is used.

    The list can be nested list up to one extra level to represent nested
    fields.
    """
    # only put a wrapper_version field for >=osaka (value 1), otherwise
    # omit field
    wrapper = []
    if self.wrapper_version is not None:
        wrapper = ["wrapper_version"]

    rlp_proofs: list[str] = []
    if self.proofs is not None:
        rlp_proofs = ["proofs"]

    rlp_cell_proofs: list[str] = []
    if self.cell_proofs is not None:
        rlp_cell_proofs = ["cell_proofs"]

    rlp_fields: List[str] = [  # structure explained in
        # https://eips.ethereum.org/EIPS/eip-7594#Networking
        "tx",  # tx_payload_body
        *wrapper,  # wrapper_version, which is always 1 for osaka (was non-
        # existing before)
        "blobs",  # Blob.data
        "commitments",
        *rlp_proofs,
        *rlp_cell_proofs,
    ]

    assert ("proofs" in rlp_fields) or ("cell_proofs" in rlp_fields), (
        "Neither proofs nor cell_proofs are in rlp_fields. Critical error!"
    )

    # logger.debug(f"Ended up with this rlp field list: {rlp_fields}")

    return rlp_fields

hash property

Return the hash of the transaction.

sender property

Return the sender of the transaction.

to property

Return the to address of the transaction.

set_gas_price(*, gas_price, max_fee_per_gas, max_priority_fee_per_gas, max_fee_per_blob_gas)

Set gas price to values of the current execution environment.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def set_gas_price(
    self,
    *,
    gas_price: int,
    max_fee_per_gas: int,
    max_priority_fee_per_gas: int,
    max_fee_per_blob_gas: int,
) -> None:
    """Set gas price to values of the current execution environment."""
    self.tx.set_gas_price(
        gas_price=gas_price,
        max_fee_per_gas=max_fee_per_gas,
        max_priority_fee_per_gas=max_priority_fee_per_gas,
        max_fee_per_blob_gas=max_fee_per_blob_gas,
    )

signer_minimum_balance(*, fork)

Return minimum balance of the signer.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def signer_minimum_balance(self, *, fork: Fork) -> int:
    """Return minimum balance of the signer."""
    return self.tx.signer_minimum_balance(fork=fork)

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 packages/testing/src/execution_testing/test_types/transaction_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""

with_signature_and_sender(*, keep_secret_key=False)

Return a new transaction with the signature and sender added.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def with_signature_and_sender(
    self, *, keep_secret_key: bool = False
) -> Self:
    """Return a new transaction with the signature and sender added."""
    return self.model_copy(
        update={
            "tx": self.tx.with_signature_and_sender(
                keep_secret_key=keep_secret_key
            ),
        }
    )

Transaction

Bases: TransactionGeneric[HexNumber], TransactionTransitionToolConverter, SignableRLPSerializable

Generic object that can represent all Ethereum transaction types.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

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

    @model_validator(mode="before")
    @classmethod
    def strip_hash_from_t8n_output(cls, data: Any) -> Any:
        """Strip the hash field which may be included in t8n tool output."""
        if isinstance(data, dict):
            # If the class has a transaction_hash, keep it. This is likely
            # an internal RPC class that extends from Transaction.
            if "transaction_hash" not in cls.model_fields:
                data.pop("hash", None)
        return data

    gas_limit: HexNumber = Field(
        HexNumber(21_000),
        serialization_alias="gas",
        validation_alias=AliasChoices("gas_limit", "gasLimit", "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

    metadata: TransactionTestMetadata | None = Field(None, exclude=True)
    test_phase: TestPhase | None = Field(
        default_factory=TestPhaseManager.get_current_phase, exclude=True
    )

    model_config = ConfigDict(validate_assignment=True)

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

        def __str__(self) -> str:
            """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) -> str:
            """Print exception string."""
            return "can't define both 'signature' and 'private_key'"

    def model_post_init(self, __context: Any) -> None:
        """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 = HexNumber(6)
            elif self.authorization_list is not None:
                self.ty = HexNumber(4)
            elif (
                self.max_fee_per_blob_gas is not None
                or self.blob_versioned_hashes is not None
            ):
                self.ty = HexNumber(3)
            elif (
                self.max_fee_per_gas is not None
                or self.max_priority_fee_per_gas is not None
            ):
                self.ty = HexNumber(2)
            elif self.access_list is not None:
                self.ty = HexNumber(1)
            else:
                self.ty = HexNumber(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 = HexNumber(TransactionDefaults.gas_price)
            self.model_fields_set.remove("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 = HexNumber(
                TransactionDefaults.max_fee_per_gas
            )
            self.model_fields_set.remove("max_fee_per_gas")
        if self.ty >= 2 and self.max_priority_fee_per_gas is None:
            self.max_priority_fee_per_gas = HexNumber(
                TransactionDefaults.max_priority_fee_per_gas
            )
            self.model_fields_set.remove("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 = HexNumber(1)
            self.model_fields_set.remove("max_fee_per_blob_gas")
        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") -> None:
        """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:
                    v = self.v
                    if self.ty == 0:
                        if v > 28:
                            v -= 35 + (self.chain_id * 2)
                        else:  # not protected
                            v -= 27
                    signature_bytes = (
                        int(self.r).to_bytes(32, byteorder="big")
                        + int(self.s).to_bytes(32, byteorder="big")
                        + bytes([v])
                    )
                public_key = PublicKey.from_signature_and_message(
                    signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
                )
                self.sender = 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
    ) -> Self:
        """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 = 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""

    def metadata_string(self) -> str | None:
        """Return the metadata field as a formatted json string or None."""
        if self.metadata is None:
            return None
        return self.metadata.to_json()

    @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_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:])

    def set_gas_price(
        self,
        *,
        gas_price: int,
        max_fee_per_gas: int,
        max_priority_fee_per_gas: int,
        max_fee_per_blob_gas: int,
    ) -> None:
        """
        Set the gas price to the appropriate values of the current
        execution environment.

        Values are only set if they were not
         set during instance creation.
        """
        if self.ty <= 1:
            if "gas_price" not in self.model_fields_set:
                self.gas_price = HexNumber(gas_price)
        else:
            if "max_fee_per_gas" not in self.model_fields_set:
                self.max_fee_per_gas = HexNumber(max_fee_per_gas)
            if "max_priority_fee_per_gas" not in self.model_fields_set:
                self.max_priority_fee_per_gas = HexNumber(
                    max_priority_fee_per_gas
                )
            if self.ty == 3:
                if "max_fee_per_blob_gas" not in self.model_fields_set:
                    self.max_fee_per_blob_gas = HexNumber(max_fee_per_blob_gas)

    def signer_minimum_balance(self, *, fork: Fork) -> int:
        """Return minimum balance of the signer."""
        gas_price = self.gas_price or self.max_fee_per_gas
        assert gas_price is not None, (
            "Impossible to calculate minimum balance without gas price"
        )
        gas_limit = self.gas_limit
        if self.ty == 3 and self.blob_versioned_hashes is not None:
            max_fee_per_blob_gas = self.max_fee_per_blob_gas
            assert max_fee_per_blob_gas is not None, (
                "Impossible to calculate minimum balance without "
                "max_fee_per_blob_gas"
            )
            return (
                gas_price * gas_limit
                + self.value
                + max_fee_per_blob_gas
                * (fork.blob_gas_per_blob() * len(self.blob_versioned_hashes))
            )
        else:
            return gas_price * gas_limit + self.value

    def _format_field_value(self, value: Any) -> str:
        """
        Format a field value for string representation.

        Uses decimal for numeric values (int, HexNumber, etc.) and
        hex encoding for Address, Bytes, Hash, etc.
        """
        if value is None:
            return "None"

        # fields like 'value' should be shown as decimal number
        if isinstance(value, numbers.Number):
            # Force decimal representation for int subclasses like HexNumber
            if isinstance(value, int):
                return str(int(value))

            return str(value)

        # fields like 'to' should be shown as hex string
        if hasattr(value, "hex") and callable(value.hex):
            return f'"{value.hex()}"'

        return repr(value)

    def __repr__(self) -> str:
        """
        Return string representation with hex-encoded values for
        applicable fields.
        """
        field_strs = []
        for field_name in self.__class__.model_fields:
            value = getattr(self, field_name)
            formatted_value = self._format_field_value(value)
            field_strs.append(f"{field_name}={formatted_value}")

        return f"{self.__class__.__name__}({', '.join(field_strs)})"

    def __str__(self) -> str:
        """Return the repr string representation."""
        return self.__repr__()

strip_hash_from_t8n_output(data) classmethod

Strip the hash field which may be included in t8n tool output.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

@model_validator(mode="before")
@classmethod
def strip_hash_from_t8n_output(cls, data: Any) -> Any:
    """Strip the hash field which may be included in t8n tool output."""
    if isinstance(data, dict):
        # If the class has a transaction_hash, keep it. This is likely
        # an internal RPC class that extends from Transaction.
        if "transaction_hash" not in cls.model_fields:
            data.pop("hash", None)
    return data

InvalidFeePaymentError

Bases: Exception

Transaction described more than one fee payment type.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

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

    def __str__(self) -> str:
        """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 packages/testing/src/execution_testing/test_types/transaction_types.py

def __str__(self) -> str:
    """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 packages/testing/src/execution_testing/test_types/transaction_types.py

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

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

__str__()

Print exception string.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def __str__(self) -> str:
    """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 packages/testing/src/execution_testing/test_types/transaction_types.py

def model_post_init(self, __context: Any) -> None:
    """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 = HexNumber(6)
        elif self.authorization_list is not None:
            self.ty = HexNumber(4)
        elif (
            self.max_fee_per_blob_gas is not None
            or self.blob_versioned_hashes is not None
        ):
            self.ty = HexNumber(3)
        elif (
            self.max_fee_per_gas is not None
            or self.max_priority_fee_per_gas is not None
        ):
            self.ty = HexNumber(2)
        elif self.access_list is not None:
            self.ty = HexNumber(1)
        else:
            self.ty = HexNumber(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 = HexNumber(TransactionDefaults.gas_price)
        self.model_fields_set.remove("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 = HexNumber(
            TransactionDefaults.max_fee_per_gas
        )
        self.model_fields_set.remove("max_fee_per_gas")
    if self.ty >= 2 and self.max_priority_fee_per_gas is None:
        self.max_priority_fee_per_gas = HexNumber(
            TransactionDefaults.max_priority_fee_per_gas
        )
        self.model_fields_set.remove("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 = HexNumber(1)
        self.model_fields_set.remove("max_fee_per_blob_gas")
    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 packages/testing/src/execution_testing/test_types/transaction_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 packages/testing/src/execution_testing/test_types/transaction_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 cached property

Returns the serialized bytes of the transaction signature.

sign()

Signs the authorization tuple with a private key.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def sign(self: "Transaction") -> None:
    """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:
                v = self.v
                if self.ty == 0:
                    if v > 28:
                        v -= 35 + (self.chain_id * 2)
                    else:  # not protected
                        v -= 27
                signature_bytes = (
                    int(self.r).to_bytes(32, byteorder="big")
                    + int(self.s).to_bytes(32, byteorder="big")
                    + bytes([v])
                )
            public_key = PublicKey.from_signature_and_message(
                signature_bytes, rlp_signing_bytes.keccak256(), hasher=None
            )
            self.sender = 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 packages/testing/src/execution_testing/test_types/transaction_types.py

def with_signature_and_sender(
    self, *, keep_secret_key: bool = False
) -> Self:
    """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 = 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 packages/testing/src/execution_testing/test_types/transaction_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 packages/testing/src/execution_testing/test_types/transaction_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 packages/testing/src/execution_testing/test_types/transaction_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 packages/testing/src/execution_testing/test_types/transaction_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""

metadata_string()

Return the metadata field as a formatted json string or None.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def metadata_string(self) -> str | None:
    """Return the metadata field as a formatted json string or None."""
    if self.metadata is None:
        return None
    return self.metadata.to_json()

hash cached property

Returns hash of the transaction.

serializable_list cached property

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

list_blob_versioned_hashes(input_txs) staticmethod

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

Source code in packages/testing/src/execution_testing/test_types/transaction_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 cached property

Return address of the contract created by the transaction.

set_gas_price(*, gas_price, max_fee_per_gas, max_priority_fee_per_gas, max_fee_per_blob_gas)

Set the gas price to the appropriate values of the current execution environment.

Values are only set if they were not set during instance creation.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def set_gas_price(
    self,
    *,
    gas_price: int,
    max_fee_per_gas: int,
    max_priority_fee_per_gas: int,
    max_fee_per_blob_gas: int,
) -> None:
    """
    Set the gas price to the appropriate values of the current
    execution environment.

    Values are only set if they were not
     set during instance creation.
    """
    if self.ty <= 1:
        if "gas_price" not in self.model_fields_set:
            self.gas_price = HexNumber(gas_price)
    else:
        if "max_fee_per_gas" not in self.model_fields_set:
            self.max_fee_per_gas = HexNumber(max_fee_per_gas)
        if "max_priority_fee_per_gas" not in self.model_fields_set:
            self.max_priority_fee_per_gas = HexNumber(
                max_priority_fee_per_gas
            )
        if self.ty == 3:
            if "max_fee_per_blob_gas" not in self.model_fields_set:
                self.max_fee_per_blob_gas = HexNumber(max_fee_per_blob_gas)

signer_minimum_balance(*, fork)

Return minimum balance of the signer.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def signer_minimum_balance(self, *, fork: Fork) -> int:
    """Return minimum balance of the signer."""
    gas_price = self.gas_price or self.max_fee_per_gas
    assert gas_price is not None, (
        "Impossible to calculate minimum balance without gas price"
    )
    gas_limit = self.gas_limit
    if self.ty == 3 and self.blob_versioned_hashes is not None:
        max_fee_per_blob_gas = self.max_fee_per_blob_gas
        assert max_fee_per_blob_gas is not None, (
            "Impossible to calculate minimum balance without "
            "max_fee_per_blob_gas"
        )
        return (
            gas_price * gas_limit
            + self.value
            + max_fee_per_blob_gas
            * (fork.blob_gas_per_blob() * len(self.blob_versioned_hashes))
        )
    else:
        return gas_price * gas_limit + self.value

__repr__()

Return string representation with hex-encoded values for applicable fields.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def __repr__(self) -> str:
    """
    Return string representation with hex-encoded values for
    applicable fields.
    """
    field_strs = []
    for field_name in self.__class__.model_fields:
        value = getattr(self, field_name)
        formatted_value = self._format_field_value(value)
        field_strs.append(f"{field_name}={formatted_value}")

    return f"{self.__class__.__name__}({', '.join(field_strs)})"

__str__()

Return the repr string representation.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def __str__(self) -> str:
    """Return the repr string representation."""
    return self.__repr__()

TransactionDefaults dataclass

Default values for transactions.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

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

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

TransactionTestMetadata

Bases: CamelModel

Represents the metadata for a transaction.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

class TransactionTestMetadata(CamelModel):
    """Represents the metadata for a transaction."""

    test_id: str | None = None
    phase: TestPhase | None = None
    action: str | None = None  # e.g. deploy / fund / execute
    target: str | None = None  # account/contract label
    tx_index: int | None = None  # index within this phase

    def to_json(self) -> str:
        """
        Convert the transaction metadata into json string for it to be embedded
        in the request id.
        """
        return self.model_dump_json(exclude_none=True, by_alias=True)

to_json()

Convert the transaction metadata into json string for it to be embedded in the request id.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

def to_json(self) -> str:
    """
    Convert the transaction metadata into json string for it to be embedded
    in the request id.
    """
    return self.model_dump_json(exclude_none=True, by_alias=True)

TransactionType

Bases: IntEnum

Transaction types.

Source code in packages/testing/src/execution_testing/test_types/transaction_types.py

class TransactionType(IntEnum):
    """Transaction types."""

    LEGACY = 0
    ACCESS_LIST = 1
    BASE_FEE = 2
    BLOB_TRANSACTION = 3
    SET_CODE = 4

Removable

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

Source code in packages/testing/src/execution_testing/test_types/utils.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 packages/testing/src/execution_testing/test_types/utils.py

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

keccak256(data)

Calculate keccak256 hash of the given data.

Source code in packages/testing/src/execution_testing/test_types/utils.py

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