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.

Doubles as a PreState provider for the spec's state transition: once any PreState method is called the instance transitions from CONSTRUCTION to LIVE (a code-hash → bytes cache is built once) and further free mutations via __setitem__/__delitem__ are rejected. The only mutation entry point in LIVE is apply_diff, which patches self.root and updates the cache in lockstep. freeze locks the allocation for read-only assertion use.

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.

    Doubles as a `PreState` provider for the spec's state transition: once
    any `PreState` method is called the instance transitions from
    `CONSTRUCTION` to `LIVE` (a code-hash → bytes cache is built once) and
    further free mutations via `__setitem__`/`__delitem__` are rejected.
    The only mutation entry point in `LIVE` is `apply_diff`, which patches
    `self.root` and updates the cache in lockstep. `freeze` locks the
    allocation for read-only assertion use.
    """

    _phase: _Phase = PrivateAttr(default=_Phase.CONSTRUCTION)
    _code_store: Dict[Hash32, Bytes] = PrivateAttr(default_factory=dict)
    _state_commitment: StateCommitment | None = PrivateAttr(default=None)
    """
    Commitment scheme this allocation's state root is computed under.

    Unset by default: it must be seeded from the accompanying fork before any
    state-root computation.
    """

    @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,
        state_commitment: StateCommitment | None = None,
    ) -> "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_copy(deep=True)

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

        if state_commitment is not None:
            merged.migrate_state_commitment(state_commitment)
        else:
            # By default, state commitment of the second alloc takes precedence
            merged.migrate_state_commitment(alloc_2.state_commitment())
        return 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."""
        self._require_construction("__setitem__")
        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."""
        self._require_construction("__delitem__")
        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 get(self, address: Address) -> Account | None:
        """Get an account if it's present in the allocation, otherwise None."""
        account = self.root.get(address)
        if not account:
            return None
        return account

    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."""
        return Hash(self._state_module().state_root(self._materialize_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
                got_account = got_alloc.get(address)
                if got_account:
                    raise Alloc.UnexpectedAccountError(
                        address=address, account=got_account
                    )
            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 get_alloc_grouping_hash(self) -> AllocGroupHash | None:
        """
        Return the grouping hash if the allocation belongs to a particular
        group, otherwise `None`.

        Method can be overloaded by other implementations of the Alloc to
        return the appropriate group.
        """
        return None

    def calculate_diff(self, base_alloc: "Alloc") -> "Alloc":
        """
        Calculate the state difference between self and a base.

        Returns an Alloc containing only the accounts that:
        - Changed between base and self (balance, nonce, storage, code)
        - Were created during test execution (new accounts)
        - Were deleted during test execution (represented as None)

        Args:
            base_alloc: Genesis pre-allocation state

        Returns:
            Alloc containing only the state differences for efficient storage

        """
        diff: Dict[Address, Account | None] = {}

        # Find all addresses that exist in either state
        all_addresses = set(self.root.keys()) | set(base_alloc.root.keys())

        for address in all_addresses:
            genesis_account = base_alloc.root.get(address)
            post_account = self.root.get(address)

            # Account was deleted (exists in genesis but not in post)
            if genesis_account is not None and post_account is None:
                diff[address] = None

            # Account was created (doesn't exist in genesis but exists in post)
            elif genesis_account is None and post_account is not None:
                diff[address] = post_account

            # Account was modified (exists in both but different)
            elif genesis_account != post_account:
                diff[address] = post_account

            # Account unchanged - don't include in diff

        return Alloc(diff)

    # ------------------------------------------------------------------
    # PreState protocol implementation
    # ------------------------------------------------------------------

    def _require_construction(self, operation: str) -> None:
        """Reject mutations once the allocation has left construction."""
        if self._phase is not _Phase.CONSTRUCTION:
            raise RuntimeError(
                f"{operation} not allowed: Alloc is in phase "
                f"{self._phase.name}. Mutate via apply_diff during LIVE, "
                f"or call freeze() to lock the allocation."
            )

    def _build_cache(self) -> None:
        """Populate the code-hash → bytes cache from `self.root`."""
        self._code_store = {spec_state.EMPTY_CODE_HASH: Bytes(b"")}
        for account in self.root.values():
            if account is None:
                continue
            code = bytes(account.code) if account.code else b""
            if not code:
                continue
            self._code_store[spec_keccak256(code)] = Bytes(code)

    def _ensure_live(self) -> None:
        """Transition from `CONSTRUCTION` to `LIVE`, building the cache."""
        if self._phase is _Phase.CONSTRUCTION:
            self._build_cache()
            self._phase = _Phase.LIVE

    def _state_module(self) -> ModuleType:
        """
        Return the spec state module implementing `self._state_commitment`.
        """
        if self._state_commitment is None:
            raise ValueError(
                "Alloc state commitment is unset; seed it from the "
                "accompanying fork."
            )
        if self._state_commitment is StateCommitment.MPT:
            return spec_state_mpt
        raise NotImplementedError("State commitment type not yet implemented.")

    def _materialize_state(self) -> spec_state.PreState:
        """
        Build a spec-side `PreState` mirror of `self.root` using the
        implementation module for this allocation's commitment scheme.

        Used as the trie-backed delegate for `compute_state_root` (a
        cold, once-per-block call). The materialized state is not
        retained.
        """
        mod = self._state_module()
        state: spec_state.PreState = mod.State()
        for address, account in self.root.items():
            if account is None:
                continue
            addr = Bytes20(address)
            code = bytes(account.code) if account.code else b""
            if code:
                code_hash = mod.store_code(state, code)
            else:
                code_hash = spec_state.EMPTY_CODE_HASH
            mod.set_account(
                state,
                addr,
                spec_state.Account(
                    nonce=Uint(int(account.nonce)),
                    balance=U256(int(account.balance)),
                    code_hash=code_hash,
                ),
            )
            for key_hi, value_hi in account.storage.root.items():
                value_int = int(value_hi)
                if value_int == 0:
                    continue
                mod.set_storage(
                    state,
                    addr,
                    Bytes32(int(key_hi).to_bytes(32, "big")),
                    U256(value_int),
                )
        return state

    def get_account_optional(
        self, address: Bytes20
    ) -> Optional[spec_state.Account]:
        """
        Return the spec-side `Account` at `address`, or `None`.

        Conforms to `ethereum.state.PreState.get_account_optional`.
        """
        self._ensure_live()
        account = self.root.get(Address(address))
        if account is None:
            return None
        code = bytes(account.code) if account.code else b""
        code_hash = (
            spec_keccak256(code) if code else spec_state.EMPTY_CODE_HASH
        )
        return spec_state.Account(
            nonce=Uint(int(account.nonce)),
            balance=U256(int(account.balance)),
            code_hash=code_hash,
        )

    def get_storage(self, address: Bytes20, key: Bytes32) -> U256:
        """
        Return the storage value at `key` for `address`, or `U256(0)`.

        Conforms to `ethereum.state.PreState.get_storage`.
        """
        self._ensure_live()
        account = self.root.get(Address(address))
        if account is None:
            return U256(0)
        key_int = int.from_bytes(bytes(key), "big")
        value_hi = account.storage.root.get(HashInt(key_int))
        if value_hi is None:
            return U256(0)
        return U256(int(value_hi))

    def get_code(self, code_hash: Hash32) -> Bytes:
        """
        Return the bytecode for `code_hash`.

        Conforms to `ethereum.state.PreState.get_code`.
        """
        self._ensure_live()
        if code_hash == spec_state.EMPTY_CODE_HASH:
            return Bytes(b"")
        return self._code_store[code_hash]

    def compute_state_root(self, block_diff: spec_state.BlockDiff) -> Hash32:
        """
        Compute the state root after applying `block_diff` to the
        pre-state.

        Conforms to `ethereum.state.PreState.compute_state_root`.
        Builds the trie inline; `Alloc` does not cache `Trie`
        instances.
        """
        self._ensure_live()
        state = self._materialize_state()
        return state.compute_state_root(block_diff)

    # ------------------------------------------------------------------
    # Lifecycle: apply_diff and freeze
    # ------------------------------------------------------------------

    def apply_diff(self, diff: spec_state.BlockDiff) -> None:
        """
        Apply a `BlockDiff` to mutate the allocation in place.

        The only mutation entry point in the `LIVE` phase. Writes bypass
        `__setitem__` intentionally — `_code_store` is updated additively
        in lockstep with `self.root`.
        """
        if self._phase is _Phase.FROZEN:
            raise RuntimeError("apply_diff not allowed: Alloc is FROZEN")
        if self._phase is _Phase.CONSTRUCTION:
            raise RuntimeError(
                "apply_diff not allowed in CONSTRUCTION: the allocation "
                "has not been used as a PreState yet, so its cache is not "
                "built. Trigger a PreState method (or hand it to a "
                "BlockState) before calling apply_diff."
            )

        for code_hash, code in diff.code_changes.items():
            self._code_store[Hash32(code_hash)] = Bytes(code)

        for address in diff.storage_clears:
            addr = Address(address)
            current = self.root.get(addr)
            if current is not None and current.storage.root:
                self.root[addr] = current.model_copy(
                    update={"storage": Storage(root={})}
                )

        for address, spec_account in diff.account_changes.items():
            addr = Address(address)
            if spec_account is None:
                self.root.pop(addr, None)
                continue
            code_hash = Hash32(spec_account.code_hash)
            if code_hash == spec_state.EMPTY_CODE_HASH:
                code = Bytes(b"")
            else:
                code = self._code_store[code_hash]
            existing = self.root.get(addr)
            existing_storage = (
                existing.storage if existing is not None else Storage(root={})
            )
            self.root[addr] = Account(
                nonce=int(spec_account.nonce),
                balance=int(spec_account.balance),
                code=code,
                storage=existing_storage,
            )

        for address, slots in diff.storage_changes.items():
            addr = Address(address)
            account = self.root.get(addr)
            if account is None:
                continue
            merged: Dict[HashInt, HashInt] = dict(account.storage.root)
            for key, value in slots.items():
                key_int = HashInt(int.from_bytes(bytes(key), "big"))
                value_int = int(value)
                if value_int == 0:
                    merged.pop(key_int, None)
                else:
                    merged[key_int] = HashInt(value_int)
            self.root[addr] = account.model_copy(
                update={"storage": Storage(root=merged)}
            )

        # Drop zero-valued storage entries from every account. Ethereum
        # treats an absent slot as zero, so a literal ``{0x00: 0x00}``
        # pair carried over untouched from the pre-state JSON would
        # otherwise survive into the post-state dump and produce noise
        # the spec-state-backed pipeline never had (the spec's
        # ``set_storage`` drops zeros on insert).
        for addr, account in list(self.root.items()):
            if account is None or not account.storage.root:
                continue
            cleaned = {
                key: value
                for key, value in account.storage.root.items()
                if int(value) != 0
            }
            if len(cleaned) != len(account.storage.root):
                self.root[addr] = account.model_copy(
                    update={"storage": Storage(root=cleaned)}
                )

    def freeze(self) -> None:
        """Lock the allocation: no further mutations allowed."""
        self._phase = _Phase.FROZEN

    def state_commitment(self) -> StateCommitment | None:
        """
        Return the commitment scheme this allocation is committed under, or
        `None` if it has not been seeded from a fork yet.
        """
        return self._state_commitment

    def migrate_state_commitment(
        self, commitment: StateCommitment | None
    ) -> None:
        """
        Switch the commitment scheme used to compute the state root.
        """
        if self._phase is _Phase.FROZEN:
            raise RuntimeError(
                "migrate_state_commitment not allowed: Alloc is FROZEN"
            )
        self._state_commitment = commitment

    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 stub_eoa(self, label: str) -> EOA:
        """Return the EOA for a key-bearing stub."""
        raise NotImplementedError(
            "stub_eoa 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 nonexistent_account(self) -> Address:
        """
        Return the address of a previously unused nonexistent account.

        The address is guaranteed to not be a precompile or a system contract.
        No account is created — it remains nonexistent in the pre-state.
        """
        raise NotImplementedError(
            "nonexistent_account is not implemented in the base class"
        )

    def expect_account_state(
        self,
        addresses: Address | Sequence[Address],
        *,
        is_existing_account: bool = True,
        is_contract: bool = False,
        min_balance: int | None = None,
        code_prefix: bytes | None = None,
    ) -> None:
        """
        Register start-block expectation(s) for predeployed account(s).

        Accepts a single address or a range; labels ride on the addresses
        themselves. Used only by fill-stateful; ignored by other
        allocations.
        """

    def verify_deployed_accounts(self, block_number: int) -> None:
        """
        Verify predeployed-account expectations at block_number.

        No-op unless fill-stateful allocation.
        """

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, state_commitment=None) 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,
    state_commitment: StateCommitment | None = None,
) -> "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_copy(deep=True)

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

    if state_commitment is not None:
        merged.migrate_state_commitment(state_commitment)
    else:
        # By default, state commitment of the second alloc takes precedence
        merged.migrate_state_commitment(alloc_2.state_commitment())
    return 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."""
    self._require_construction("__setitem__")
    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."""
    self._require_construction("__delitem__")
    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

get(address)

Get an account if it's present in the allocation, otherwise None.

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

def get(self, address: Address) -> Account | None:
    """Get an account if it's present in the allocation, otherwise None."""
    account = self.root.get(address)
    if not account:
        return None
    return account

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."""
    return Hash(self._state_module().state_root(self._materialize_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
            got_account = got_alloc.get(address)
            if got_account:
                raise Alloc.UnexpectedAccountError(
                    address=address, account=got_account
                )
        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)

get_alloc_grouping_hash()

Return the grouping hash if the allocation belongs to a particular group, otherwise None.

Method can be overloaded by other implementations of the Alloc to return the appropriate group.

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

def get_alloc_grouping_hash(self) -> AllocGroupHash | None:
    """
    Return the grouping hash if the allocation belongs to a particular
    group, otherwise `None`.

    Method can be overloaded by other implementations of the Alloc to
    return the appropriate group.
    """
    return None

calculate_diff(base_alloc)

Calculate the state difference between self and a base.

Returns an Alloc containing only the accounts that: - Changed between base and self (balance, nonce, storage, code) - Were created during test execution (new accounts) - Were deleted during test execution (represented as None)

Parameters:

Name	Type	Description	Default
base_alloc	Alloc	Genesis pre-allocation state	required

Returns:

Type	Description
Alloc	Alloc containing only the state differences for efficient storage

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

def calculate_diff(self, base_alloc: "Alloc") -> "Alloc":
    """
    Calculate the state difference between self and a base.

    Returns an Alloc containing only the accounts that:
    - Changed between base and self (balance, nonce, storage, code)
    - Were created during test execution (new accounts)
    - Were deleted during test execution (represented as None)

    Args:
        base_alloc: Genesis pre-allocation state

    Returns:
        Alloc containing only the state differences for efficient storage

    """
    diff: Dict[Address, Account | None] = {}

    # Find all addresses that exist in either state
    all_addresses = set(self.root.keys()) | set(base_alloc.root.keys())

    for address in all_addresses:
        genesis_account = base_alloc.root.get(address)
        post_account = self.root.get(address)

        # Account was deleted (exists in genesis but not in post)
        if genesis_account is not None and post_account is None:
            diff[address] = None

        # Account was created (doesn't exist in genesis but exists in post)
        elif genesis_account is None and post_account is not None:
            diff[address] = post_account

        # Account was modified (exists in both but different)
        elif genesis_account != post_account:
            diff[address] = post_account

        # Account unchanged - don't include in diff

    return Alloc(diff)

get_account_optional(address)

Return the spec-side Account at address, or None.

Conforms to ethereum.state.PreState.get_account_optional.

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

def get_account_optional(
    self, address: Bytes20
) -> Optional[spec_state.Account]:
    """
    Return the spec-side `Account` at `address`, or `None`.

    Conforms to `ethereum.state.PreState.get_account_optional`.
    """
    self._ensure_live()
    account = self.root.get(Address(address))
    if account is None:
        return None
    code = bytes(account.code) if account.code else b""
    code_hash = (
        spec_keccak256(code) if code else spec_state.EMPTY_CODE_HASH
    )
    return spec_state.Account(
        nonce=Uint(int(account.nonce)),
        balance=U256(int(account.balance)),
        code_hash=code_hash,
    )

get_storage(address, key)

Return the storage value at key for address, or U256(0).

Conforms to ethereum.state.PreState.get_storage.

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

def get_storage(self, address: Bytes20, key: Bytes32) -> U256:
    """
    Return the storage value at `key` for `address`, or `U256(0)`.

    Conforms to `ethereum.state.PreState.get_storage`.
    """
    self._ensure_live()
    account = self.root.get(Address(address))
    if account is None:
        return U256(0)
    key_int = int.from_bytes(bytes(key), "big")
    value_hi = account.storage.root.get(HashInt(key_int))
    if value_hi is None:
        return U256(0)
    return U256(int(value_hi))

get_code(code_hash)

Return the bytecode for code_hash.

Conforms to ethereum.state.PreState.get_code.

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

def get_code(self, code_hash: Hash32) -> Bytes:
    """
    Return the bytecode for `code_hash`.

    Conforms to `ethereum.state.PreState.get_code`.
    """
    self._ensure_live()
    if code_hash == spec_state.EMPTY_CODE_HASH:
        return Bytes(b"")
    return self._code_store[code_hash]

compute_state_root(block_diff)

Compute the state root after applying block_diff to the pre-state.

Conforms to ethereum.state.PreState.compute_state_root. Builds the trie inline; Alloc does not cache Trie instances.

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

def compute_state_root(self, block_diff: spec_state.BlockDiff) -> Hash32:
    """
    Compute the state root after applying `block_diff` to the
    pre-state.

    Conforms to `ethereum.state.PreState.compute_state_root`.
    Builds the trie inline; `Alloc` does not cache `Trie`
    instances.
    """
    self._ensure_live()
    state = self._materialize_state()
    return state.compute_state_root(block_diff)

apply_diff(diff)

Apply a BlockDiff to mutate the allocation in place.

The only mutation entry point in the LIVE phase. Writes bypass __setitem__ intentionally — _code_store is updated additively in lockstep with self.root.

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

def apply_diff(self, diff: spec_state.BlockDiff) -> None:
    """
    Apply a `BlockDiff` to mutate the allocation in place.

    The only mutation entry point in the `LIVE` phase. Writes bypass
    `__setitem__` intentionally — `_code_store` is updated additively
    in lockstep with `self.root`.
    """
    if self._phase is _Phase.FROZEN:
        raise RuntimeError("apply_diff not allowed: Alloc is FROZEN")
    if self._phase is _Phase.CONSTRUCTION:
        raise RuntimeError(
            "apply_diff not allowed in CONSTRUCTION: the allocation "
            "has not been used as a PreState yet, so its cache is not "
            "built. Trigger a PreState method (or hand it to a "
            "BlockState) before calling apply_diff."
        )

    for code_hash, code in diff.code_changes.items():
        self._code_store[Hash32(code_hash)] = Bytes(code)

    for address in diff.storage_clears:
        addr = Address(address)
        current = self.root.get(addr)
        if current is not None and current.storage.root:
            self.root[addr] = current.model_copy(
                update={"storage": Storage(root={})}
            )

    for address, spec_account in diff.account_changes.items():
        addr = Address(address)
        if spec_account is None:
            self.root.pop(addr, None)
            continue
        code_hash = Hash32(spec_account.code_hash)
        if code_hash == spec_state.EMPTY_CODE_HASH:
            code = Bytes(b"")
        else:
            code = self._code_store[code_hash]
        existing = self.root.get(addr)
        existing_storage = (
            existing.storage if existing is not None else Storage(root={})
        )
        self.root[addr] = Account(
            nonce=int(spec_account.nonce),
            balance=int(spec_account.balance),
            code=code,
            storage=existing_storage,
        )

    for address, slots in diff.storage_changes.items():
        addr = Address(address)
        account = self.root.get(addr)
        if account is None:
            continue
        merged: Dict[HashInt, HashInt] = dict(account.storage.root)
        for key, value in slots.items():
            key_int = HashInt(int.from_bytes(bytes(key), "big"))
            value_int = int(value)
            if value_int == 0:
                merged.pop(key_int, None)
            else:
                merged[key_int] = HashInt(value_int)
        self.root[addr] = account.model_copy(
            update={"storage": Storage(root=merged)}
        )

    # Drop zero-valued storage entries from every account. Ethereum
    # treats an absent slot as zero, so a literal ``{0x00: 0x00}``
    # pair carried over untouched from the pre-state JSON would
    # otherwise survive into the post-state dump and produce noise
    # the spec-state-backed pipeline never had (the spec's
    # ``set_storage`` drops zeros on insert).
    for addr, account in list(self.root.items()):
        if account is None or not account.storage.root:
            continue
        cleaned = {
            key: value
            for key, value in account.storage.root.items()
            if int(value) != 0
        }
        if len(cleaned) != len(account.storage.root):
            self.root[addr] = account.model_copy(
                update={"storage": Storage(root=cleaned)}
            )

freeze()

Lock the allocation: no further mutations allowed.

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

def freeze(self) -> None:
    """Lock the allocation: no further mutations allowed."""
    self._phase = _Phase.FROZEN

state_commitment()

Return the commitment scheme this allocation is committed under, or None if it has not been seeded from a fork yet.

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

def state_commitment(self) -> StateCommitment | None:
    """
    Return the commitment scheme this allocation is committed under, or
    `None` if it has not been seeded from a fork yet.
    """
    return self._state_commitment

migrate_state_commitment(commitment)

Switch the commitment scheme used to compute the state root.

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

def migrate_state_commitment(
    self, commitment: StateCommitment | None
) -> None:
    """
    Switch the commitment scheme used to compute the state root.
    """
    if self._phase is _Phase.FROZEN:
        raise RuntimeError(
            "migrate_state_commitment not allowed: Alloc is FROZEN"
        )
    self._state_commitment = commitment

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

stub_eoa(label)

Return the EOA for a key-bearing stub.

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

def stub_eoa(self, label: str) -> EOA:
    """Return the EOA for a key-bearing stub."""
    raise NotImplementedError(
        "stub_eoa 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"
    )

nonexistent_account()

Return the address of a previously unused nonexistent account.

The address is guaranteed to not be a precompile or a system contract. No account is created — it remains nonexistent in the pre-state.

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

def nonexistent_account(self) -> Address:
    """
    Return the address of a previously unused nonexistent account.

    The address is guaranteed to not be a precompile or a system contract.
    No account is created — it remains nonexistent in the pre-state.
    """
    raise NotImplementedError(
        "nonexistent_account is not implemented in the base class"
    )

expect_account_state(addresses, *, is_existing_account=True, is_contract=False, min_balance=None, code_prefix=None)

Register start-block expectation(s) for predeployed account(s).

Accepts a single address or a range; labels ride on the addresses themselves. Used only by fill-stateful; ignored by other allocations.

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

def expect_account_state(
    self,
    addresses: Address | Sequence[Address],
    *,
    is_existing_account: bool = True,
    is_contract: bool = False,
    min_balance: int | None = None,
    code_prefix: bytes | None = None,
) -> None:
    """
    Register start-block expectation(s) for predeployed account(s).

    Accepts a single address or a range; labels ride on the addresses
    themselves. Used only by fill-stateful; ignored by other
    allocations.
    """

verify_deployed_accounts(block_number)

Verify predeployed-account expectations at block_number.

No-op unless fill-stateful allocation.

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

def verify_deployed_accounts(self, block_number: int) -> None:
    """
    Verify predeployed-account expectations at block_number.

    No-op unless fill-stateful allocation.
    """

AllocGroupHash

Bases: FixedSizeBytes[8]

Class that helps represent hashes used to group allocs.

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

class AllocGroupHash(FixedSizeBytes[8]):  # type: ignore
    """Class that helps represent hashes used to group allocs."""

    @classmethod
    def from_preimage(cls, x: str | bytes) -> "AllocGroupHash":
        """
        Perform a hash (sha256) then truncate the output to get the alloc
        hash.
        """
        if isinstance(x, str):
            x = x.encode("utf-8")
        return cls(sha256(x).digest()[:8])

    def __xor__(self, other: "int | AllocGroupHash") -> "AllocGroupHash":
        """
        Alloc hashes are usually combination of multiple inputs via
        XOR operation.
        """
        if isinstance(other, int):
            other = AllocGroupHash(other)
        return AllocGroupHash(
            bytes(a ^ b for a, b in zip(self, other, strict=True))
        )

from_preimage(x) classmethod

Perform a hash (sha256) then truncate the output to get the alloc hash.

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

@classmethod
def from_preimage(cls, x: str | bytes) -> "AllocGroupHash":
    """
    Perform a hash (sha256) then truncate the output to get the alloc
    hash.
    """
    if isinstance(x, str):
        x = x.encode("utf-8")
    return cls(sha256(x).digest()[:8])

__xor__(other)

Alloc hashes are usually combination of multiple inputs via XOR operation.

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

def __xor__(self, other: "int | AllocGroupHash") -> "AllocGroupHash":
    """
    Alloc hashes are usually combination of multiple inputs via
    XOR operation.
    """
    if isinstance(other, int):
        other = AllocGroupHash(other)
    return AllocGroupHash(
        bytes(a ^ b for a, b in zip(self, other, strict=True))
    )

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 = fork.get_blob_constant("AMOUNT_CELL_PROOFS")
        return (
            "blob_"
            + str(seed)
            + "_"
            + fork.name().lower()
            + "_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.
        """
        return CACHED_BLOBS_DIRECTORY / Blob.get_filename(fork, seed)

    @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(fork.get_blob_constant("BLS_MODULUS"))
                for _ in range(
                    fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
                )
            ]

            encoded: list[bytes] = [
                i.to_bytes(
                    fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"),
                    "big",
                )
                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 = fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
            bytes_per_field = 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(
                    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 = 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 = 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 = 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 = fork.get_blob_constant("AMOUNT_CELL_PROOFS")
    return (
        "blob_"
        + str(seed)
        + "_"
        + fork.name().lower()
        + "_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.
    """
    return CACHED_BLOBS_DIRECTORY / Blob.get_filename(fork, seed)

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(fork.get_blob_constant("BLS_MODULUS"))
            for _ in range(
                fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
            )
        ]

        encoded: list[bytes] = [
            i.to_bytes(
                fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"),
                "big",
            )
            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 = fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")
        bytes_per_field = 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(
                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 = 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 = 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 = 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)
        )

BalAccountAbsentValues

Bases: CamelModel

Represents explicit absent value expectations for a specific account in a block.

This class specifies specific changes that should NOT exist in the BAL for a given account.

IMPORTANT: This class is for checking that specific values are absent, NOT for checking that entire categories are empty. For complete absence checks (e.g., "no nonce changes at all"), use BalAccountExpectation with empty lists instead.

The validation works by checking that none of the specified explicit changes exist in the actual BAL.

Example

Forbid specific nonce change at tx 1 with post_nonce=5, and specific

storage change

absent_values = BalAccountAbsentValues( nonce_changes=[ # Forbid exact nonce change at this tx BalNonceChange(block_access_index=1, post_nonce=5), ], storage_changes=[ BalStorageSlot( slot=0x42, slot_changes=[ # Forbid exact storage change at this slot and tx BalStorageChange(block_access_index=2, post_value=0x99) ], ) ], )

For checking complete absence

Use BalAccountExpectation with empty lists instead

expectation = BalAccountExpectation( nonce_changes=[], # Expect NO nonce changes at all storage_changes=[], # Expect NO storage changes at all )

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

class BalAccountAbsentValues(CamelModel):
    """
    Represents explicit absent value expectations for a specific account
    in a block.

    This class specifies specific changes that should NOT exist in the BAL
    for a given account.

    IMPORTANT: This class is for checking that specific values are absent,
    NOT for checking that entire categories are empty. For complete
    absence checks (e.g., "no nonce changes at all"), use
    BalAccountExpectation with empty lists instead.

    The validation works by checking that none of the specified
    explicit changes exist in the actual BAL.

    Example:
        # Forbid specific nonce change at tx 1 with post_nonce=5, and specific
        # storage change
        absent_values = BalAccountAbsentValues(
            nonce_changes=[
                # Forbid exact nonce change at this tx
                BalNonceChange(block_access_index=1, post_nonce=5),
            ],
            storage_changes=[
                BalStorageSlot(
                    slot=0x42,
                    slot_changes=[
                        # Forbid exact storage change at this slot and tx
                        BalStorageChange(block_access_index=2, post_value=0x99)
                    ],
                )
            ],
        )

    For checking complete absence:
        # Use BalAccountExpectation with empty lists instead
        expectation = BalAccountExpectation(
            nonce_changes=[],  # Expect NO nonce changes at all
            storage_changes=[],  # Expect NO storage changes at all
        )

    """

    nonce_changes: List[BalNonceChange] = Field(
        default_factory=list,
        description="List of nonce changes that should NOT exist in the BAL. "
        "Validates that none of these changes are present.",
    )
    balance_changes: List[BalBalanceChange] = Field(
        default_factory=list,
        description=(
            "List of balance changes that should NOT exist in the BAL. "
            "Validates that none of these changes are present."
        ),
    )
    code_changes: List[BalCodeChange] = Field(
        default_factory=list,
        description="List of code changes that should NOT exist in the BAL. "
        "Validates that none of these changes are present.",
    )
    storage_changes: List[BalStorageSlot] = Field(
        default_factory=list,
        description=(
            "List of storage slots/changes that should NOT exist in the BAL. "
            "Validates that none of these changes are present."
        ),
    )
    storage_reads: List[StorageKey] = Field(
        default_factory=list,
        description="List of storage slots that should NOT be read.",
    )

    @model_validator(mode="after")
    def validate_specific_absences_only(self) -> "BalAccountAbsentValues":
        """Ensure absence fields contain specific values, not empty checks."""
        # at least one field must have content
        if not any(
            [
                self.nonce_changes,
                self.balance_changes,
                self.code_changes,
                self.storage_changes,
                self.storage_reads,
            ]
        ):
            raise ValueError(
                "At least one absence field must be specified. "
                "`BalAccountAbsentValues` is for checking specific forbidden "
                f"values. {EMPTY_LIST_ERROR_MSG}"
            )

        # check that no fields are explicitly set to empty lists
        field_checks = [
            ("nonce_changes", self.nonce_changes),
            ("balance_changes", self.balance_changes),
            ("code_changes", self.code_changes),
            ("storage_changes", self.storage_changes),
            ("storage_reads", self.storage_reads),
        ]

        for field_name, field_value in field_checks:
            if field_name in self.model_fields_set and field_value == []:
                raise ValueError(
                    f"`BalAccountAbsentValues.{field_name}` cannot be an "
                    f"empty list. {EMPTY_LIST_ERROR_MSG}"
                )

        # validate that storage_changes don't have empty slot_changes
        for storage_slot in self.storage_changes:
            if not storage_slot.slot_changes:
                raise ValueError(
                    f"`BalAccountAbsentValues.storage_changes"
                    f"[{storage_slot.slot}].slot_changes` cannot be an empty "
                    f"list. {EMPTY_LIST_ERROR_MSG}"
                )

        return self

    @staticmethod
    def _validate_forbidden_changes(
        actual_changes: List,
        forbidden_changes: List,
        match_fn: Any,
        error_msg_fn: Any,
    ) -> None:
        for actual in actual_changes:
            for forbidden in forbidden_changes:
                if match_fn(actual, forbidden):
                    raise BlockAccessListValidationError(error_msg_fn(actual))

    def validate_against(self, account: BalAccountChange) -> None:
        """
        Validate that the account does not contain the forbidden changes
        specified in this object.

        Args:
            account: The BalAccountChange to validate against

        Raises:
            BlockAccessListValidationError: If any forbidden changes are found

        """
        self._validate_forbidden_changes(
            account.nonce_changes,
            self.nonce_changes,
            lambda a, f: (
                a.block_access_index == f.block_access_index
                and a.post_nonce == f.post_nonce
            ),
            lambda a: (
                f"Unexpected nonce change found at tx {a.block_access_index}"
            ),
        )
        self._validate_forbidden_changes(
            account.balance_changes,
            self.balance_changes,
            lambda a, f: (
                a.block_access_index == f.block_access_index
                and a.post_balance == f.post_balance
            ),
            lambda a: (
                f"Unexpected balance change found at tx {a.block_access_index}"
            ),
        )
        self._validate_forbidden_changes(
            account.code_changes,
            self.code_changes,
            lambda a, f: (
                a.block_access_index == f.block_access_index
                and a.new_code == f.new_code
            ),
            lambda a: (
                f"Unexpected code change found at tx {a.block_access_index}"
            ),
        )

        for forbidden_storage_slot in self.storage_changes:
            for actual_storage_slot in account.storage_changes:
                if actual_storage_slot.slot == forbidden_storage_slot.slot:
                    slot_id = actual_storage_slot.slot
                    self._validate_forbidden_changes(
                        actual_storage_slot.slot_changes,
                        forbidden_storage_slot.slot_changes,
                        lambda a, f: (
                            a.block_access_index == f.block_access_index
                            and a.post_value == f.post_value
                        ),
                        lambda a, slot=slot_id: (
                            f"Unexpected storage change found at slot {slot} "
                            f"in tx {a.block_access_index}"
                        ),
                    )

        for forbidden_read in self.storage_reads:
            for actual_read in account.storage_reads:
                if actual_read == forbidden_read:
                    raise BlockAccessListValidationError(
                        f"Unexpected storage read found at slot {actual_read}"
                    )

validate_specific_absences_only()

Ensure absence fields contain specific values, not empty checks.

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

@model_validator(mode="after")
def validate_specific_absences_only(self) -> "BalAccountAbsentValues":
    """Ensure absence fields contain specific values, not empty checks."""
    # at least one field must have content
    if not any(
        [
            self.nonce_changes,
            self.balance_changes,
            self.code_changes,
            self.storage_changes,
            self.storage_reads,
        ]
    ):
        raise ValueError(
            "At least one absence field must be specified. "
            "`BalAccountAbsentValues` is for checking specific forbidden "
            f"values. {EMPTY_LIST_ERROR_MSG}"
        )

    # check that no fields are explicitly set to empty lists
    field_checks = [
        ("nonce_changes", self.nonce_changes),
        ("balance_changes", self.balance_changes),
        ("code_changes", self.code_changes),
        ("storage_changes", self.storage_changes),
        ("storage_reads", self.storage_reads),
    ]

    for field_name, field_value in field_checks:
        if field_name in self.model_fields_set and field_value == []:
            raise ValueError(
                f"`BalAccountAbsentValues.{field_name}` cannot be an "
                f"empty list. {EMPTY_LIST_ERROR_MSG}"
            )

    # validate that storage_changes don't have empty slot_changes
    for storage_slot in self.storage_changes:
        if not storage_slot.slot_changes:
            raise ValueError(
                f"`BalAccountAbsentValues.storage_changes"
                f"[{storage_slot.slot}].slot_changes` cannot be an empty "
                f"list. {EMPTY_LIST_ERROR_MSG}"
            )

    return self

validate_against(account)

Validate that the account does not contain the forbidden changes specified in this object.

Parameters:

Name	Type	Description	Default
account	BalAccountChange	The BalAccountChange to validate against	required

Raises:

Type	Description
BlockAccessListValidationError	If any forbidden changes are found

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

def validate_against(self, account: BalAccountChange) -> None:
    """
    Validate that the account does not contain the forbidden changes
    specified in this object.

    Args:
        account: The BalAccountChange to validate against

    Raises:
        BlockAccessListValidationError: If any forbidden changes are found

    """
    self._validate_forbidden_changes(
        account.nonce_changes,
        self.nonce_changes,
        lambda a, f: (
            a.block_access_index == f.block_access_index
            and a.post_nonce == f.post_nonce
        ),
        lambda a: (
            f"Unexpected nonce change found at tx {a.block_access_index}"
        ),
    )
    self._validate_forbidden_changes(
        account.balance_changes,
        self.balance_changes,
        lambda a, f: (
            a.block_access_index == f.block_access_index
            and a.post_balance == f.post_balance
        ),
        lambda a: (
            f"Unexpected balance change found at tx {a.block_access_index}"
        ),
    )
    self._validate_forbidden_changes(
        account.code_changes,
        self.code_changes,
        lambda a, f: (
            a.block_access_index == f.block_access_index
            and a.new_code == f.new_code
        ),
        lambda a: (
            f"Unexpected code change found at tx {a.block_access_index}"
        ),
    )

    for forbidden_storage_slot in self.storage_changes:
        for actual_storage_slot in account.storage_changes:
            if actual_storage_slot.slot == forbidden_storage_slot.slot:
                slot_id = actual_storage_slot.slot
                self._validate_forbidden_changes(
                    actual_storage_slot.slot_changes,
                    forbidden_storage_slot.slot_changes,
                    lambda a, f: (
                        a.block_access_index == f.block_access_index
                        and a.post_value == f.post_value
                    ),
                    lambda a, slot=slot_id: (
                        f"Unexpected storage change found at slot {slot} "
                        f"in tx {a.block_access_index}"
                    ),
                )

    for forbidden_read in self.storage_reads:
        for actual_read in account.storage_reads:
            if actual_read == forbidden_read:
                raise BlockAccessListValidationError(
                    f"Unexpected storage read found at slot {actual_read}"
                )

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)

    _rlp_override: Bytes | None = PrivateAttr(default=None)

    @classmethod
    def from_rlp(cls, data: Bytes) -> "BlockAccessList":
        """
        Decode an RLP-encoded block access list into a BlockAccessList.

        The RLP structure per EIP-7928 is:
        [
          [address, storage_changes, storage_reads,
           balance_changes, nonce_changes, code_changes],
          ...
        ]
        """
        decoded = _seq_from_rlp(eth_rlp.decode(data))
        accounts = []
        for account_rlp in decoded:
            fields = _seq_from_rlp(account_rlp)

            storage_changes = []
            for slot_entry in _seq_from_rlp(fields[1]):
                slot_fields = _seq_from_rlp(slot_entry)
                storage_changes.append(
                    BalStorageSlot(
                        slot=_hex_from_rlp(slot_fields[0]),
                        slot_changes=_decode_indexed_changes(
                            slot_fields[1],
                            BalStorageChange,
                            "post_value",
                        ),
                    )
                )

            accounts.append(
                BalAccountChange(
                    address=Address(_bytes_from_rlp(fields[0])),
                    storage_changes=storage_changes,
                    storage_reads=[
                        _hex_from_rlp(sr) for sr in _seq_from_rlp(fields[2])
                    ],
                    balance_changes=_decode_indexed_changes(
                        fields[3], BalBalanceChange, "post_balance"
                    ),
                    nonce_changes=_decode_indexed_changes(
                        fields[4], BalNonceChange, "post_nonce"
                    ),
                    code_changes=_decode_indexed_changes(
                        fields[5],
                        BalCodeChange,
                        "new_code",
                        value_fn=lambda v: Bytes(_bytes_from_rlp(v)),
                    ),
                )
            )

        return cls(root=accounts)

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

    @validate_call
    def with_rlp_override(self, rlp: Bytes) -> Self:
        """
        Return a BAL with the same contents whose serialization is ``rlp``,
        mirroring ``RLPSerializable.rlp_override``.

        A fresh instance is built rather than a copy so that no cached
        canonical encoding is carried over.
        """
        new_instance = type(self)(root=self.root)
        new_instance._rlp_override = rlp
        return new_instance

    @property
    def has_rlp_override(self) -> bool:
        """Return whether ``rlp`` is an override rather than the encoding."""
        return self._rlp_override is not None

    @cached_property
    def rlp(self) -> Bytes:
        """Return the RLP encoded block access list for hash verification."""
        if self._rlp_override is not None:
            return self._rlp_override
        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]}"
                    )

from_rlp(data) classmethod

Decode an RLP-encoded block access list into a BlockAccessList.

The RLP structure per EIP-7928 is: [ [address, storage_changes, storage_reads, balance_changes, nonce_changes, code_changes], ... ]

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

@classmethod
def from_rlp(cls, data: Bytes) -> "BlockAccessList":
    """
    Decode an RLP-encoded block access list into a BlockAccessList.

    The RLP structure per EIP-7928 is:
    [
      [address, storage_changes, storage_reads,
       balance_changes, nonce_changes, code_changes],
      ...
    ]
    """
    decoded = _seq_from_rlp(eth_rlp.decode(data))
    accounts = []
    for account_rlp in decoded:
        fields = _seq_from_rlp(account_rlp)

        storage_changes = []
        for slot_entry in _seq_from_rlp(fields[1]):
            slot_fields = _seq_from_rlp(slot_entry)
            storage_changes.append(
                BalStorageSlot(
                    slot=_hex_from_rlp(slot_fields[0]),
                    slot_changes=_decode_indexed_changes(
                        slot_fields[1],
                        BalStorageChange,
                        "post_value",
                    ),
                )
            )

        accounts.append(
            BalAccountChange(
                address=Address(_bytes_from_rlp(fields[0])),
                storage_changes=storage_changes,
                storage_reads=[
                    _hex_from_rlp(sr) for sr in _seq_from_rlp(fields[2])
                ],
                balance_changes=_decode_indexed_changes(
                    fields[3], BalBalanceChange, "post_balance"
                ),
                nonce_changes=_decode_indexed_changes(
                    fields[4], BalNonceChange, "post_nonce"
                ),
                code_changes=_decode_indexed_changes(
                    fields[5],
                    BalCodeChange,
                    "new_code",
                    value_fn=lambda v: Bytes(_bytes_from_rlp(v)),
                ),
            )
        )

    return cls(root=accounts)

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)

with_rlp_override(rlp)

Return a BAL with the same contents whose serialization is rlp, mirroring RLPSerializable.rlp_override.

A fresh instance is built rather than a copy so that no cached canonical encoding is carried over.

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

@validate_call
def with_rlp_override(self, rlp: Bytes) -> Self:
    """
    Return a BAL with the same contents whose serialization is ``rlp``,
    mirroring ``RLPSerializable.rlp_override``.

    A fresh instance is built rather than a copy so that no cached
    canonical encoding is carried over.
    """
    new_instance = type(self)(root=self.root)
    new_instance._rlp_override = rlp
    return new_instance

has_rlp_override property

Return whether rlp is an override rather than the encoding.

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)
    )
    _rlp_modifier: Callable[["BlockAccessList"], Bytes] | 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))

        """
        if self._modifier is not None:
            raise ValueError(
                "This expectation already has a content modifier; a second "
                "`modify` call would replace it. Pass every modifier to a "
                "single `modify` call instead."
            )
        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 is None:
            return t8n_bal
        modified = self._modifier(t8n_bal)
        if not isinstance(modified, BlockAccessList):
            raise TypeError(
                "`modify` expects a content modifier returning a "
                f"BlockAccessList, got {type(modified).__name__}. Use "
                "`modify_rlp` or `override_rlp` for encoders."
            )
        return modified

    def modify_rlp(
        self, modifier: Callable[["BlockAccessList"], Bytes]
    ) -> "BlockAccessListExpectation":
        """
        Create a new expectation that re-encodes the BAL carried by the
        engine payload, leaving the header commitment canonical.

        Use this for encoding-level invalid cases; `modify` changes the
        BAL's contents and so also moves the header hash.

        Only the engine payload can carry the re-encoding, so the test must
        be marked `blockchain_test_engine_only`.
        """
        if self._rlp_modifier is not None:
            raise ValueError(
                "This expectation already re-encodes the block access list; "
                "a second `modify_rlp` call would replace the first. Keep a "
                "single `modify_rlp` call."
            )
        new_instance = self.model_copy(deep=True)
        new_instance._rlp_modifier = modifier
        return new_instance

    def modified_rlp(self, bal: "BlockAccessList") -> Bytes | None:
        """Return the re-encoded payload BAL, or None if unmodified."""
        if self._rlp_modifier is None:
            return None
        encoded = self._rlp_modifier(bal)
        if not isinstance(encoded, bytes):
            raise TypeError(
                "`modify_rlp` expects an encoder returning bytes, got "
                f"{type(encoded).__name__}. Use `modify` for content "
                "modifiers."
            )
        return Bytes(encoded)

    @property
    def has_modifier(self) -> bool:
        """Return whether this expectation rewrites the BAL."""
        return self._modifier is not None or self._rlp_modifier is not None

    @property
    def has_rlp_modifier(self) -> bool:
        """Return whether this expectation re-encodes the payload BAL."""
        return self._rlp_modifier is not None

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

    """
    if self._modifier is not None:
        raise ValueError(
            "This expectation already has a content modifier; a second "
            "`modify` call would replace it. Pass every modifier to a "
            "single `modify` call instead."
        )
    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 is None:
        return t8n_bal
    modified = self._modifier(t8n_bal)
    if not isinstance(modified, BlockAccessList):
        raise TypeError(
            "`modify` expects a content modifier returning a "
            f"BlockAccessList, got {type(modified).__name__}. Use "
            "`modify_rlp` or `override_rlp` for encoders."
        )
    return modified

modify_rlp(modifier)

Create a new expectation that re-encodes the BAL carried by the engine payload, leaving the header commitment canonical.

Use this for encoding-level invalid cases; modify changes the BAL's contents and so also moves the header hash.

Only the engine payload can carry the re-encoding, so the test must be marked blockchain_test_engine_only.

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

def modify_rlp(
    self, modifier: Callable[["BlockAccessList"], Bytes]
) -> "BlockAccessListExpectation":
    """
    Create a new expectation that re-encodes the BAL carried by the
    engine payload, leaving the header commitment canonical.

    Use this for encoding-level invalid cases; `modify` changes the
    BAL's contents and so also moves the header hash.

    Only the engine payload can carry the re-encoding, so the test must
    be marked `blockchain_test_engine_only`.
    """
    if self._rlp_modifier is not None:
        raise ValueError(
            "This expectation already re-encodes the block access list; "
            "a second `modify_rlp` call would replace the first. Keep a "
            "single `modify_rlp` call."
        )
    new_instance = self.model_copy(deep=True)
    new_instance._rlp_modifier = modifier
    return new_instance

modified_rlp(bal)

Return the re-encoded payload BAL, or None if unmodified.

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

def modified_rlp(self, bal: "BlockAccessList") -> Bytes | None:
    """Return the re-encoded payload BAL, or None if unmodified."""
    if self._rlp_modifier is None:
        return None
    encoded = self._rlp_modifier(bal)
    if not isinstance(encoded, bytes):
        raise TypeError(
            "`modify_rlp` expects an encoder returning bytes, got "
            f"{type(encoded).__name__}. Use `modify` for content "
            "modifiers."
        )
    return Bytes(encoded)

has_modifier property

Return whether this expectation rewrites the BAL.

has_rlp_modifier property

Return whether this expectation re-encodes the payload 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_slot_number: 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
    block_access_list_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."""
        updated_values: Dict[str, Any] = {}

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

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

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

        if fork.header_slot_number_required() and self.slot_number is None:
            updated_values["slot_number"] = (
                int(self.parent_slot_number) + 1
                if self.parent_slot_number is not None
                else 0
            )

        return self.copy(**updated_values)

    @classmethod
    def for_fork(cls, fork: Fork, **kwargs: Any) -> "Environment":
        """
        Build an environment from only the fields the fork's header has.

        Use it when one pinned context spans forks whose headers differ:
        the pins the fork lacks are dropped instead of raising in
        `check_fork_fields`.
        """
        return cls(
            **{
                name: value
                for name, value in kwargs.items()
                if name not in FORK_GATED_FIELDS
                or FORK_GATED_FIELDS[name](fork)
            }
        )

    def check_fork_fields(self, fork: Fork) -> None:
        """
        Raise if a set field is one the fork's block header lacks.

        Serializing such a field into a fixture would misstate the block
        context, so the test has to state its intent instead.
        """
        unsupported = [
            name
            for name, required in FORK_GATED_FIELDS.items()
            if getattr(self, name) is not None and not required(fork)
        ]
        if not unsupported:
            return
        raise ValueError(
            f"{', '.join(unsupported)} set for {fork.name()}, whose block "
            "header has no such field. Remove the field from the test, "
            "build the environment with Environment.for_fork(fork, ...) "
            "to keep only the fields the fork has, or, to check that "
            "clients reject the field, set it on the block through "
            "Block(rlp_modifier=Header(...))."
        )

    def canonical_json(self) -> str:
        """
        Return the canonical JSON encoding of this model.

        Keys are alias-cased and sorted, and unset fields are excluded, so
        two equal models encode identically and the encoding is stable
        across processes: usable as a grouping key or a hash pre-image.
        """
        return json.dumps(
            self.model_dump(mode="json", by_alias=True, exclude_none=True),
            sort_keys=True,
        )

    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."""
    updated_values: Dict[str, Any] = {}

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

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

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

    if fork.header_slot_number_required() and self.slot_number is None:
        updated_values["slot_number"] = (
            int(self.parent_slot_number) + 1
            if self.parent_slot_number is not None
            else 0
        )

    return self.copy(**updated_values)

for_fork(fork, **kwargs) classmethod

Build an environment from only the fields the fork's header has.

Use it when one pinned context spans forks whose headers differ: the pins the fork lacks are dropped instead of raising in check_fork_fields.

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

@classmethod
def for_fork(cls, fork: Fork, **kwargs: Any) -> "Environment":
    """
    Build an environment from only the fields the fork's header has.

    Use it when one pinned context spans forks whose headers differ:
    the pins the fork lacks are dropped instead of raising in
    `check_fork_fields`.
    """
    return cls(
        **{
            name: value
            for name, value in kwargs.items()
            if name not in FORK_GATED_FIELDS
            or FORK_GATED_FIELDS[name](fork)
        }
    )

check_fork_fields(fork)

Raise if a set field is one the fork's block header lacks.

Serializing such a field into a fixture would misstate the block context, so the test has to state its intent instead.

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

def check_fork_fields(self, fork: Fork) -> None:
    """
    Raise if a set field is one the fork's block header lacks.

    Serializing such a field into a fixture would misstate the block
    context, so the test has to state its intent instead.
    """
    unsupported = [
        name
        for name, required in FORK_GATED_FIELDS.items()
        if getattr(self, name) is not None and not required(fork)
    ]
    if not unsupported:
        return
    raise ValueError(
        f"{', '.join(unsupported)} set for {fork.name()}, whose block "
        "header has no such field. Remove the field from the test, "
        "build the environment with Environment.for_fork(fork, ...) "
        "to keep only the fields the fork has, or, to check that "
        "clients reject the field, set it on the block through "
        "Block(rlp_modifier=Header(...))."
    )

canonical_json()

Return the canonical JSON encoding of this model.

Keys are alias-cased and sorted, and unset fields are excluded, so two equal models encode identically and the encoding is stable across processes: usable as a grouping key or a hash pre-image.

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

def canonical_json(self) -> str:
    """
    Return the canonical JSON encoding of this model.

    Keys are alias-cased and sorted, and unset fields are excluded, so
    two equal models encode identically and the encoding is stable
    across processes: usable as a grouping key or a hash pre-image.
    """
    return json.dumps(
        self.model_dump(mode="json", by_alias=True, exclude_none=True),
        sort_keys=True,
    )

__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], SSZModel

Withdrawal type; also the consensus-layer SSZ container.

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

class Withdrawal(WithdrawalGeneric[HexNumber], SSZModel):
    """Withdrawal type; also the consensus-layer SSZ container."""

    index: Uint64
    validator_index: Uint64
    amount: Uint64

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 = DEFAULT_CHAIN_ID

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

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

    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 / RPC output not part of model."""
        if isinstance(data, dict):
            data = dict(data)
            # geth (1.16+) returns extra fields in receipts
            data.pop("type", None)
            data.pop("blockNumber", None)
            # Fields eth_getTransactionReceipt returns that the fixture
            # schema does not model. Fill-stateful fetches receipts live,
            # so we tolerate these even though the t8n path never emits
            # them.
            data.pop("from", None)
            data.pop("to", None)
            root = data.get("root")
            root_is_empty = root in (None, "", "0x", b"", bytearray())
            if not root_is_empty:
                # geth's t8n JSON uses `root` for pre-Byzantium receipts while
                # also populating `status`. For fixture re-encoding, a
                # non-empty root must take precedence over status.
                data.setdefault("post_state", root)
                data.pop("status", 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 / RPC 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 / RPC output not part of model."""
    if isinstance(data, dict):
        data = dict(data)
        # geth (1.16+) returns extra fields in receipts
        data.pop("type", None)
        data.pop("blockNumber", None)
        # Fields eth_getTransactionReceipt returns that the fixture
        # schema does not model. Fill-stateful fetches receipts live,
        # so we tolerate these even though the t8n path never emits
        # them.
        data.pop("from", None)
        data.pop("to", None)
        root = data.get("root")
        root_is_empty = root in (None, "", "0x", b"", bytearray())
        if not root_is_empty:
            # geth's t8n JSON uses `root` for pre-Byzantium receipts while
            # also populating `status`. For fixture re-encoding, a
            # non-empty root must take precedence over status.
            data.setdefault("post_state", root)
            data.pop("status", None)
    return data

SystemContractInteractionBase dataclass

Base class for all types of request transactions we want to test.

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

@dataclass(kw_only=True, frozen=True)
class SystemContractInteractionBase:
    """Base class for all types of request transactions we want to test."""

    sender_account: EOA | None = None
    """Account that sends the transaction."""
    requests: Sequence[SystemContractRequest]
    """Requests to be included in the block."""
    gas_limits: List[int | None] | None = None
    """
    Optional per-request gas overrides, aligned with `requests`. Only set by
    the out-of-gas test functions; left `None` for normal tests so the
    automatic transaction gas-limit (and full inner-call gas) applies.
    """

    @property
    def request_source_address(self) -> Address | None:
        """Address recorded as the source of the requests."""
        raise NotImplementedError

    def transactions(self) -> List[Transaction]:
        """Return the transactions that trigger the requests."""
        raise NotImplementedError

    def update_pre(self, pre: Alloc) -> Self:
        """
        Allocate accounts/contracts in `pre` and return a new instance with
        the allocated state populated. Does not mutate `self`, so the
        parametrize value remains pristine across fixture format runs.
        """
        raise NotImplementedError

sender_account = None class-attribute instance-attribute

Account that sends the transaction.

requests instance-attribute

Requests to be included in the block.

gas_limits = None class-attribute instance-attribute

Optional per-request gas overrides, aligned with requests. Only set by the out-of-gas test functions; left None for normal tests so the automatic transaction gas-limit (and full inner-call gas) applies.

request_source_address property

Address recorded as the source of the requests.

transactions()

Return the transactions that trigger the requests.

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

def transactions(self) -> List[Transaction]:
    """Return the transactions that trigger the requests."""
    raise NotImplementedError

update_pre(pre)

Allocate accounts/contracts in pre and return a new instance with the allocated state populated. Does not mutate self, so the parametrize value remains pristine across fixture format runs.

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

def update_pre(self, pre: Alloc) -> Self:
    """
    Allocate accounts/contracts in `pre` and return a new instance with
    the allocated state populated. Does not mutate `self`, so the
    parametrize value remains pristine across fixture format runs.
    """
    raise NotImplementedError

SystemContractInteractionContract dataclass

Bases: SystemContractInteractionBase

Describe requests originated from a relay contract.

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

@dataclass(kw_only=True, frozen=True)
class SystemContractInteractionContract(SystemContractInteractionBase):
    """Describe requests originated from a relay contract."""

    tx_value: int = 0
    """Value to send with the transaction."""

    contract_balance: int | None = None
    """
    Balance of the relay contract that sends the requests. `None` (the
    default) funds the contract with the sum of the request values, which is
    always enough to forward each request's value.
    """
    contract_address: Address | None = None
    """Address of the relay contract that sends the requests."""
    entry_address: Address | None = None
    """Address to send the transaction to."""

    call_type: Op = field(default_factory=lambda: Op.CALL)
    """Type of call to be made to the system contract."""
    call_depth: int = 2
    """Frame depth of the system contract when it processes the requests."""
    extra_code: Bytecode = field(default_factory=Bytecode)
    """Extra code to be included in the relay contract."""

    @property
    def request_source_address(self) -> Address | None:
        """The relay contract is the source of the requests."""
        return self.contract_address

    @property
    def contract_code(self) -> Bytecode:
        """Code used by the relay contract."""
        return relay_contract_code(
            self.requests,
            call_type=self.call_type,
            extra_code=self.extra_code,
            gas_limits=self.gas_limits,
        )

    def transactions(self) -> List[Transaction]:
        """Return the single transaction that drives the relay contract."""
        assert self.entry_address is not None, "Entry address not initialized"
        return [
            Transaction(
                to=self.entry_address,
                value=self.tx_value,
                data=b"".join(r.calldata for r in self.requests),
                sender=self.sender_account,
            )
        ]

    def update_pre(self, pre: Alloc) -> Self:
        """
        Return a copy of self with the allocated sender/contract/entry
        addresses populated.
        """
        sender_account = pre.fund_eoa()
        contract_balance = (
            self.contract_balance
            if self.contract_balance is not None
            else sum(r.value for r in self.requests)
        )
        contract_address = pre.deploy_contract(
            code=self.contract_code, balance=contract_balance
        )
        entry_address = contract_address
        if self.call_depth > 2:
            for _ in range(1, self.call_depth - 1):
                entry_address = pre.deploy_contract(
                    code=Op.CALLDATACOPY(0, 0, Op.CALLDATASIZE)
                    + Op.POP(
                        Op.CALL(
                            Op.GAS,
                            entry_address,
                            0,
                            0,
                            Op.CALLDATASIZE,
                            0,
                            0,
                        )
                    ),
                )
        return replace(
            self,
            sender_account=sender_account,
            contract_address=contract_address,
            entry_address=entry_address,
        )

tx_value = 0 class-attribute instance-attribute

Value to send with the transaction.

contract_balance = None class-attribute instance-attribute

Balance of the relay contract that sends the requests. None (the default) funds the contract with the sum of the request values, which is always enough to forward each request's value.

contract_address = None class-attribute instance-attribute

Address of the relay contract that sends the requests.

entry_address = None class-attribute instance-attribute

Address to send the transaction to.

call_type = field(default_factory=(lambda: Op.CALL)) class-attribute instance-attribute

Type of call to be made to the system contract.

call_depth = 2 class-attribute instance-attribute

Frame depth of the system contract when it processes the requests.

extra_code = field(default_factory=Bytecode) class-attribute instance-attribute

Extra code to be included in the relay contract.

request_source_address property

The relay contract is the source of the requests.

contract_code property

Code used by the relay contract.

transactions()

Return the single transaction that drives the relay contract.

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

def transactions(self) -> List[Transaction]:
    """Return the single transaction that drives the relay contract."""
    assert self.entry_address is not None, "Entry address not initialized"
    return [
        Transaction(
            to=self.entry_address,
            value=self.tx_value,
            data=b"".join(r.calldata for r in self.requests),
            sender=self.sender_account,
        )
    ]

update_pre(pre)

Return a copy of self with the allocated sender/contract/entry addresses populated.

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

def update_pre(self, pre: Alloc) -> Self:
    """
    Return a copy of self with the allocated sender/contract/entry
    addresses populated.
    """
    sender_account = pre.fund_eoa()
    contract_balance = (
        self.contract_balance
        if self.contract_balance is not None
        else sum(r.value for r in self.requests)
    )
    contract_address = pre.deploy_contract(
        code=self.contract_code, balance=contract_balance
    )
    entry_address = contract_address
    if self.call_depth > 2:
        for _ in range(1, self.call_depth - 1):
            entry_address = pre.deploy_contract(
                code=Op.CALLDATACOPY(0, 0, Op.CALLDATASIZE)
                + Op.POP(
                    Op.CALL(
                        Op.GAS,
                        entry_address,
                        0,
                        0,
                        Op.CALLDATASIZE,
                        0,
                        0,
                    )
                ),
            )
    return replace(
        self,
        sender_account=sender_account,
        contract_address=contract_address,
        entry_address=entry_address,
    )

SystemContractInteractionMeasuredOutOfGasContract dataclass

Bases: SystemContractInteractionContract

Relay-contract interaction that self-measures each request's gas cost and forces the requests marked invalid (valid=False) out of gas by forwarding one gas less than required, independent of the fork's gas schedule.

Reuses SystemContractInteractionContract for the driving transaction and pre-state allocation.

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

@dataclass(kw_only=True, frozen=True)
class SystemContractInteractionMeasuredOutOfGasContract(
    SystemContractInteractionContract
):
    """
    Relay-contract interaction that self-measures each request's gas cost and
    forces the requests marked invalid (`valid=False`) out of gas by forwarding
    one gas less than required, independent of the fork's gas schedule.

    Reuses `SystemContractInteractionContract` for the driving transaction and
    pre-state allocation.
    """

    @property
    def contract_code(self) -> Bytecode:
        """
        Build a relay contract that measures, at runtime, the gas each system
        contract request needs, then forces the requests marked invalid out of
        gas by forwarding one gas less than required.

        Like `relay_contract_code`, the contract reads the concatenated request
        calldata from its own calldata. It then issues, in order:

        1. A warm-up call for the first valid request, warming the predeploy
           account and its storage slots so the measurement reflects the warm
           cost.
        2. A measured call for the second valid request, capturing `GAS` around
           the call to record its total cost (CALL base cost + value transfer
           + the gas value stipend + the callee's own consumption).
        3. Full-gas calls for any remaining valid requests.
        4. An overhead probe: a call identical to (2) that forwards only
           minimal gas so the callee runs out, isolating everything except the
           callee's own consumption. Subtracting it from (2) yields the gas
           that must be forwarded for the callee to succeed.
        5. A call for each invalid request forwarding `(total - overhead) - 1`
           gas, one short of the requirement, so it runs out of gas and is not
           enqueued.

        Because steps (2) and (4) use byte-identical op sequences (including a
        same-width gas `PUSH`), every base-opcode cost cancels in the
        subtraction, making the forced out-of-gas independent of the fork's
        gas schedule.

        All requests must share the same calldata length and a non-zero call
        value so the measured overhead applies uniformly to each call.
        """
        valid_indices = [i for i, r in enumerate(self.requests) if r.valid]
        invalid_indices = [
            i for i, r in enumerate(self.requests) if not r.valid
        ]
        assert len(valid_indices) >= 2, (
            "measured_out_of_gas_relay_code needs at least two valid requests"
        )
        assert invalid_indices, (
            "measured_out_of_gas_relay_code needs at least one invalid request"
        )
        assert len({len(r.calldata) for r in self.requests}) == 1, (
            "all requests must share the same calldata length"
        )
        assert all(r.value > 0 for r in self.requests), (
            "all requests must have a non-zero call value"
        )

        offsets: List[int] = []
        current = 0
        for r in self.requests:
            offsets.append(current)
            current += len(r.calldata)

        def issue(
            *,
            index: int,
            gas_argument: Bytecode | Op,
            measure_into: int | None = None,
        ) -> Bytecode:
            r = self.requests[index]
            copy = Op.CALLDATACOPY(0, offsets[index], len(r.calldata))
            call = Op.CALL(
                gas_argument,
                r.system_contract_address,
                r.value,
                0,
                len(r.calldata),
                0,
                0,
            )
            if measure_into is None:
                return copy + Op.POP(call)
            return (
                copy
                + Op.GAS
                + call
                + Op.POP
                + Op.GAS
                + Op.SWAP1
                + Op.SUB
                + Op.PUSH2(measure_into)
                + Op.MSTORE
            )

        warmup_index = valid_indices[0]
        probe_index = valid_indices[1]

        code = issue(index=warmup_index, gas_argument=Op.GAS)
        code += issue(
            index=probe_index,
            # Forward all gas using the same PUSH opcode in both calls:
            # Op.GAS and Op.PUSH0 could diverge in gas cost in the future.
            gas_argument=Op.PUSH4[0xFFFFFFFF],
            measure_into=_MEASURE_TOTAL_SLOT,
        )
        for i in valid_indices[2:]:
            code += issue(index=i, gas_argument=Op.GAS)
        # The overhead probe reuses the second valid request so its call value
        # and calldata size (hence CALL base cost) match the total measurement
        # exactly.
        code += issue(
            index=probe_index,
            gas_argument=Op.PUSH4[0],
            measure_into=_MEASURE_OVERHEAD_SLOT,
        )
        forwarded_gas = Op.SUB(
            Op.SUB(
                Op.MLOAD(_MEASURE_TOTAL_SLOT), Op.MLOAD(_MEASURE_OVERHEAD_SLOT)
            ),
            1,
        )
        for i in invalid_indices:
            code += issue(index=i, gas_argument=forwarded_gas)
        return code + self.extra_code

contract_code property

Build a relay contract that measures, at runtime, the gas each system contract request needs, then forces the requests marked invalid out of gas by forwarding one gas less than required.

Like relay_contract_code, the contract reads the concatenated request calldata from its own calldata. It then issues, in order:

1. A warm-up call for the first valid request, warming the predeploy account and its storage slots so the measurement reflects the warm cost.
2. A measured call for the second valid request, capturing GAS around the call to record its total cost (CALL base cost + value transfer
3. the gas value stipend + the callee's own consumption).
4. Full-gas calls for any remaining valid requests.
5. An overhead probe: a call identical to (2) that forwards only minimal gas so the callee runs out, isolating everything except the callee's own consumption. Subtracting it from (2) yields the gas that must be forwarded for the callee to succeed.
6. A call for each invalid request forwarding (total - overhead) - 1 gas, one short of the requirement, so it runs out of gas and is not enqueued.

Because steps (2) and (4) use byte-identical op sequences (including a same-width gas PUSH), every base-opcode cost cancels in the subtraction, making the forced out-of-gas independent of the fork's gas schedule.

All requests must share the same calldata length and a non-zero call value so the measured overhead applies uniformly to each call.

SystemContractInteractionTransaction dataclass

Bases: SystemContractInteractionBase

Describe requests originated from an externally owned account, one transaction per request.

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

@dataclass(kw_only=True, frozen=True)
class SystemContractInteractionTransaction(SystemContractInteractionBase):
    """
    Describe requests originated from an externally owned account, one
    transaction per request.
    """

    @property
    def request_source_address(self) -> Address | None:
        """The sender account is the source of the requests."""
        return self.sender_account

    def transactions(self) -> List[Transaction]:
        """Return one transaction per request."""
        assert self.sender_account is not None, (
            "Sender account not initialized"
        )
        txs: List[Transaction] = []
        for i, request in enumerate(self.requests):
            gas_limit = (
                self.gas_limits[i] if self.gas_limits is not None else None
            )
            txs.append(
                Transaction(
                    gas_limit=gas_limit,
                    to=request.system_contract_address,
                    value=request.value,
                    data=request.calldata,
                    sender=self.sender_account,
                )
            )
        return txs

    def update_pre(self, pre: Alloc) -> Self:
        """Return a copy of self with `sender_account` populated."""
        return replace(self, sender_account=pre.fund_eoa())

request_source_address property

The sender account is the source of the requests.

transactions()

Return one transaction per request.

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

def transactions(self) -> List[Transaction]:
    """Return one transaction per request."""
    assert self.sender_account is not None, (
        "Sender account not initialized"
    )
    txs: List[Transaction] = []
    for i, request in enumerate(self.requests):
        gas_limit = (
            self.gas_limits[i] if self.gas_limits is not None else None
        )
        txs.append(
            Transaction(
                gas_limit=gas_limit,
                to=request.system_contract_address,
                value=request.value,
                data=request.calldata,
                sender=self.sender_account,
            )
        )
    return txs

update_pre(pre)

Return a copy of self with sender_account populated.

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

def update_pre(self, pre: Alloc) -> Self:
    """Return a copy of self with `sender_account` populated."""
    return replace(self, sender_account=pre.fund_eoa())

fee_increment_blocks(request_class, n)

Return N blocks such that each subsequent block has an increasing fee for the requests of request_class.

Each block contains the number of requests required to reach the next fee increment (plus the per-block target), built from sequential indices via from_index and wrapped in a relay-contract interaction.

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

def fee_increment_blocks(
    request_class: Type[FeeSystemContractRequest], n: int
) -> List[List[SystemContractInteractionContract]]:
    """
    Return N blocks such that each subsequent block has an increasing fee for
    the requests of `request_class`.

    Each block contains the number of requests required to reach the next fee
    increment (plus the per-block target), built from sequential indices via
    `from_index` and wrapped in a relay-contract interaction.
    """
    blocks = []
    previous_excess = 0
    request_index = 0
    previous_fee = 0
    for required_excess_requests in request_class.get_n_fee_increments(n):
        requests_required = (
            required_excess_requests
            + request_class.target_per_block
            - previous_excess
        )
        fee = request_class.get_fee(previous_excess)
        assert fee > previous_fee
        blocks.append(
            [
                SystemContractInteractionContract(
                    requests=[
                        request_class.from_index(i)
                        for i in range(
                            request_index,
                            request_index + requests_required,
                        )
                    ],
                )
            ]
        )
        previous_fee = fee
        request_index += requests_required
        previous_excess = required_excess_requests

    return blocks

relay_contract_code(requests, *, call_type, extra_code, gas_limits=None)

Build the code of a relay contract that issues each request by calling its system contract with the request's calldata.

The contract reads the concatenated request calldata from its own calldata, copies each request payload into memory and issues call_type to the corresponding system contract.

gas_limits is an optional list, aligned with requests, that overrides the gas forwarded to each inner call. It is only used by the out-of-gas test functions; a None entry (or omitting the list) forwards all available gas via Op.GAS.

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

def relay_contract_code(
    requests: Sequence[SystemContractRequest],
    *,
    call_type: Op,
    extra_code: Bytecode,
    gas_limits: List[int | None] | None = None,
) -> Bytecode:
    """
    Build the code of a relay contract that issues each request by calling its
    system contract with the request's calldata.

    The contract reads the concatenated request calldata from its own calldata,
    copies each request payload into memory and issues `call_type` to the
    corresponding system contract.

    `gas_limits` is an optional list, aligned with `requests`, that overrides
    the gas forwarded to each inner call. It is only used by the out-of-gas
    test functions; a `None` entry (or omitting the list) forwards all
    available gas via `Op.GAS`.
    """
    if gas_limits is not None:
        assert len(gas_limits) == len(requests), (
            "gas_limits must be aligned with requests"
        )
    code = Bytecode()
    current_offset = 0
    for i, r in enumerate(requests):
        gas_limit = gas_limits[i] if gas_limits is not None else None
        value_arg = [r.value] if call_type in (Op.CALL, Op.CALLCODE) else []
        code += Op.CALLDATACOPY(0, current_offset, len(r.calldata)) + Op.POP(
            call_type(
                Op.GAS if gas_limit is None else gas_limit,
                r.system_contract_address,
                *value_arg,
                0,
                len(r.calldata),
                0,
                0,
            )
        )
        current_offset += len(r.calldata)
    return code + extra_code

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
    creates_account: bool = Field(False, exclude=True)
    writes_delegation: bool = Field(True, exclude=True)
    # Whether applying this authorization is the transaction's first
    # write to the authority's account leaf. False for a self-sponsored
    # authority (the sender is written at inclusion), for repeated
    # authorizations on one authority, and for invalid authorizations.
    first_write: bool = Field(True, exclude=True)

    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(signing_key).sign_recoverable(
                rlp_signing_bytes.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()
                )
                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(signing_key).sign_recoverable(
            rlp_signing_bytes.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()
            )
            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 set_gas_limit(
        self,
        *,
        max_gas_limit: int,
        transaction_gas_limit_cap: int | None,
        state_gas_reservoir_enabled: bool = False,
    ) -> None:
        """Set the transaction gas limit if unset."""
        self.tx.set_gas_limit(
            max_gas_limit=max_gas_limit,
            transaction_gas_limit_cap=transaction_gas_limit_cap,
            state_gas_reservoir_enabled=state_gas_reservoir_enabled,
        )

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

set_gas_limit(*, max_gas_limit, transaction_gas_limit_cap, state_gas_reservoir_enabled=False)

Set the transaction gas limit if unset.

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

def set_gas_limit(
    self,
    *,
    max_gas_limit: int,
    transaction_gas_limit_cap: int | None,
    state_gas_reservoir_enabled: bool = False,
) -> None:
    """Set the transaction gas limit if unset."""
    self.tx.set_gas_limit(
        max_gas_limit=max_gas_limit,
        transaction_gas_limit_cap=transaction_gas_limit_cap,
        state_gas_reservoir_enabled=state_gas_reservoir_enabled,
    )

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

    @model_validator(mode="before")
    @classmethod
    def treat_none_gas_limit_as_unset(cls, data: Any) -> Any:
        """Treat a `None` gas limit (any alias) as unset."""
        if isinstance(data, dict):
            for alias in ("gas_limit", "gasLimit", "gas"):
                if alias in data and data[alias] is None:
                    del data[alias]
        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)

    state_gas_reservoir: int = Field(
        0,
        exclude=True,
        description=(
            "Extra gas on top of the transaction gas limit cap, reserved "
            "for state gas (EIP-8037). Only takes effect when `gas_limit` "
            "is unset and the fork enables the state gas reservoir: "
            "leaving it unset keeps the full implicit gas limit, an "
            "explicit 0 pins the gas limit to exactly the cap (no "
            "reservoir), and a positive value pins it to the cap plus the "
            "requested reservoir. Requesting a positive reservoir on a "
            "fork without the state gas reservoir raises an error."
        ),
    )

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

        FEE_FIELD_LABELS = {
            "gas_price": "legacy/type-1",
            "max_fee_per_gas": "type-2+",
            "max_priority_fee_per_gas": "type-2+",
            "max_fee_per_blob_gas": "type-3+",
        }

        def __init__(self, *conflicting_fields: str) -> None:
            """Store the conflicting fee fields used in the transaction."""
            self.conflicting_fields = conflicting_fields

        def __str__(self) -> str:
            """Print exception string."""
            if not self.conflicting_fields:
                return "cannot mix fee fields in a single tx"
            labels = ", ".join(
                f"'{f}' ({self.FEE_FIELD_LABELS[f]})"
                for f in self.conflicting_fields
            )
            return f"cannot mix fee fields in a single tx: {labels}"

    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)

        conflicting_fee_fields = [
            field_name
            for field_name in (
                "gas_price",
                "max_fee_per_gas",
                "max_priority_fee_per_gas",
                "max_fee_per_blob_gas",
            )
            if getattr(self, field_name) is not None
        ]
        if self.gas_price is not None and len(conflicting_fee_fields) > 1:
            raise Transaction.InvalidFeePaymentError(*conflicting_fee_fields)

        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 >= 2:
            assert self.gas_price is None, (
                f"gas_price not supported for tx type {int(self.ty)}"
            )
        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(signing_key).sign_recoverable(
                rlp_signing_bytes.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()
                )
                self.sender = EOA(
                    address=Address(
                        keccak256(public_key.format(compressed=False)[1:])[
                            32 - 20 :
                        ]
                    )
                )
            except Exception:
                # Signer remains `None` in this case
                pass

    def _calculate_implicit_gas_limit(
        self,
        *,
        max_gas_limit: int,
        transaction_gas_limit_cap: int | None,
        state_gas_reservoir_enabled: bool = False,
    ) -> HexNumber:
        """
        Calculate the gas limit given the current external factors.

        The implicit gas limit defaults to `max_gas_limit`, clamped to
        the fork's transaction gas limit cap if there is one. On forks
        with the state gas reservoir enabled (EIP-8037),
        `state_gas_reservoir` refines this: unset keeps the full
        `max_gas_limit` (any excess above the cap acts as an implicit
        reservoir), an explicit 0 pins the gas limit to exactly the
        cap, and a positive value pins it to the cap plus the requested
        reservoir.
        """
        tx_gas_limit = max_gas_limit
        if state_gas_reservoir_enabled:
            if "state_gas_reservoir" in self.model_fields_set:
                assert transaction_gas_limit_cap is not None, (
                    "state_gas_reservoir_enabled is True but "
                    "transaction_gas_limit_cap is None; the state "
                    "gas reservoir is defined as gas above the cap "
                    "(EIP-8037 builds on EIP-7825), so a fork that "
                    "enables it must also define a cap"
                )
                if self.state_gas_reservoir > 0:
                    minimum_gas_with_reservoir = (
                        transaction_gas_limit_cap + self.state_gas_reservoir
                    )
                    if tx_gas_limit < minimum_gas_with_reservoir:
                        raise Exception(
                            "test correctness: the requested state "
                            "gas reservoir of "
                            f"{self.state_gas_reservoir} requires a "
                            f"gas limit of {minimum_gas_with_reservoir} "
                            "(transaction gas limit cap of "
                            f"{transaction_gas_limit_cap} plus "
                            "reservoir), but only "
                            f"{tx_gas_limit} gas is available for "
                            "this transaction."
                        )
                    tx_gas_limit = minimum_gas_with_reservoir
                else:
                    if tx_gas_limit > transaction_gas_limit_cap:
                        tx_gas_limit = transaction_gas_limit_cap
        else:
            if (
                transaction_gas_limit_cap is not None
                and tx_gas_limit > transaction_gas_limit_cap
            ):
                tx_gas_limit = transaction_gas_limit_cap
        return HexNumber(tx_gas_limit)

    def _check_state_gas_reservoir_supported(
        self, *, state_gas_reservoir_enabled: bool
    ) -> None:
        """Raise if a positive reservoir is requested but unsupported."""
        if not state_gas_reservoir_enabled and self.state_gas_reservoir > 0:
            raise Exception(
                "test correctness: transaction requests a state gas "
                f"reservoir of {self.state_gas_reservoir} but the fork "
                "does not enable the state gas reservoir; the request "
                "would be silently ignored."
            )

    def with_gas_limit(
        self,
        *,
        max_gas_limit: int,
        transaction_gas_limit_cap: int | None,
        state_gas_reservoir_enabled: bool = False,
    ) -> Self:
        """Return copy of the transaction with the set gas limit."""
        updated_values: Dict[str, Any] = {}

        self._check_state_gas_reservoir_supported(
            state_gas_reservoir_enabled=state_gas_reservoir_enabled
        )
        if "gas_limit" not in self.model_fields_set:
            updated_values["gas_limit"] = self._calculate_implicit_gas_limit(
                max_gas_limit=max_gas_limit,
                transaction_gas_limit_cap=transaction_gas_limit_cap,
                state_gas_reservoir_enabled=state_gas_reservoir_enabled,
            )

        return self.model_copy(update=updated_values)

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

        if "gas_limit" not in self.model_fields_set:
            raise ValueError("gas_limit must be set to sign a transaction")

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

        # Sign the bytes
        signing_key = PrivateKey(self.secret_key)
        signature_bytes = signing_key.sign_recoverable(signing_hash)

        # The key comparison is what makes reusing `sender` safe: it is
        # reassignable and `EOA.key` may be `None`, so a sender that does not
        # hold the signing key would contradict its own signature.
        if self.sender is not None and self.sender.key == self.secret_key:
            updated_values["sender"] = self.sender
        else:
            # The address the public key recovery would return, since the
            # signature was produced by this key, but derived with one point
            # multiplication instead.
            public_key = signing_key.public_key
            updated_values["sender"] = Address(
                keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
            )

        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
        ]

    @staticmethod
    def calculate_max_gas_limit(
        *,
        txs: List["Transaction"],
        env_gas_limit: int,
        transaction_gas_limit_cap: int | None,
        state_gas_reservoir_enabled: bool,
    ) -> int:
        """
        Calculate the maximum gas limit that can be set in a transaction
        given a list of transactions with and without gas-limits set
        and a maximum available environment gas.
        """
        available_gas = env_gas_limit
        unset_gas_limit_tx_count = 0
        for tx in txs:
            if "gas_limit" not in tx.model_fields_set:
                unset_gas_limit_tx_count += 1
            else:
                available_gas -= int(tx.gas_limit)

        if unset_gas_limit_tx_count == 0:
            return 0

        if available_gas <= 0:
            raise Exception(
                "test correctness: unable to automatically calculate gas "
                "limit for transactions (no remaining gas: explicit "
                "transaction gas limits already consume the full "
                f"environment gas limit of {env_gas_limit})."
            )

        max_tx_gas_limit = available_gas // unset_gas_limit_tx_count
        if state_gas_reservoir_enabled:
            transaction_gas_limit_cap = None
        if transaction_gas_limit_cap:
            max_tx_gas_limit = min(max_tx_gas_limit, transaction_gas_limit_cap)
        return max_tx_gas_limit

    @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 set_gas_limit(
        self,
        *,
        max_gas_limit: int,
        transaction_gas_limit_cap: int | None,
        state_gas_reservoir_enabled: bool = False,
    ) -> None:
        """Set the transaction gas limit if unset."""
        self._check_state_gas_reservoir_supported(
            state_gas_reservoir_enabled=state_gas_reservoir_enabled
        )
        if "gas_limit" not in self.model_fields_set:
            self.gas_limit = self._calculate_implicit_gas_limit(
                max_gas_limit=max_gas_limit,
                transaction_gas_limit_cap=transaction_gas_limit_cap,
                state_gas_reservoir_enabled=state_gas_reservoir_enabled,
            )

    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"
        )
        assert "gas_limit" in self.model_fields_set, (
            "Impossible to calculate minimum balance without a set 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 * self.gas_limit
                + self.value
                + max_fee_per_blob_gas
                * (fork.blob_gas_per_blob() * len(self.blob_versioned_hashes))
            )
        else:
            return gas_price * self.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

treat_none_gas_limit_as_unset(data) classmethod

Treat a None gas limit (any alias) as unset.

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

@model_validator(mode="before")
@classmethod
def treat_none_gas_limit_as_unset(cls, data: Any) -> Any:
    """Treat a `None` gas limit (any alias) as unset."""
    if isinstance(data, dict):
        for alias in ("gas_limit", "gasLimit", "gas"):
            if alias in data and data[alias] is None:
                del data[alias]
    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."""

    FEE_FIELD_LABELS = {
        "gas_price": "legacy/type-1",
        "max_fee_per_gas": "type-2+",
        "max_priority_fee_per_gas": "type-2+",
        "max_fee_per_blob_gas": "type-3+",
    }

    def __init__(self, *conflicting_fields: str) -> None:
        """Store the conflicting fee fields used in the transaction."""
        self.conflicting_fields = conflicting_fields

    def __str__(self) -> str:
        """Print exception string."""
        if not self.conflicting_fields:
            return "cannot mix fee fields in a single tx"
        labels = ", ".join(
            f"'{f}' ({self.FEE_FIELD_LABELS[f]})"
            for f in self.conflicting_fields
        )
        return f"cannot mix fee fields in a single tx: {labels}"

__init__(*conflicting_fields)

Store the conflicting fee fields used in the transaction.

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

def __init__(self, *conflicting_fields: str) -> None:
    """Store the conflicting fee fields used in the transaction."""
    self.conflicting_fields = conflicting_fields

__str__()

Print exception string.

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

def __str__(self) -> str:
    """Print exception string."""
    if not self.conflicting_fields:
        return "cannot mix fee fields in a single tx"
    labels = ", ".join(
        f"'{f}' ({self.FEE_FIELD_LABELS[f]})"
        for f in self.conflicting_fields
    )
    return f"cannot mix fee fields in a single tx: {labels}"

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)

    conflicting_fee_fields = [
        field_name
        for field_name in (
            "gas_price",
            "max_fee_per_gas",
            "max_priority_fee_per_gas",
            "max_fee_per_blob_gas",
        )
        if getattr(self, field_name) is not None
    ]
    if self.gas_price is not None and len(conflicting_fee_fields) > 1:
        raise Transaction.InvalidFeePaymentError(*conflicting_fee_fields)

    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 >= 2:
        assert self.gas_price is None, (
            f"gas_price not supported for tx type {int(self.ty)}"
        )
    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(signing_key).sign_recoverable(
            rlp_signing_bytes.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()
            )
            self.sender = EOA(
                address=Address(
                    keccak256(public_key.format(compressed=False)[1:])[
                        32 - 20 :
                    ]
                )
            )
        except Exception:
            # Signer remains `None` in this case
            pass

with_gas_limit(*, max_gas_limit, transaction_gas_limit_cap, state_gas_reservoir_enabled=False)

Return copy of the transaction with the set gas limit.

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

def with_gas_limit(
    self,
    *,
    max_gas_limit: int,
    transaction_gas_limit_cap: int | None,
    state_gas_reservoir_enabled: bool = False,
) -> Self:
    """Return copy of the transaction with the set gas limit."""
    updated_values: Dict[str, Any] = {}

    self._check_state_gas_reservoir_supported(
        state_gas_reservoir_enabled=state_gas_reservoir_enabled
    )
    if "gas_limit" not in self.model_fields_set:
        updated_values["gas_limit"] = self._calculate_implicit_gas_limit(
            max_gas_limit=max_gas_limit,
            transaction_gas_limit_cap=transaction_gas_limit_cap,
            state_gas_reservoir_enabled=state_gas_reservoir_enabled,
        )

    return self.model_copy(update=updated_values)

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

    if "gas_limit" not in self.model_fields_set:
        raise ValueError("gas_limit must be set to sign a transaction")

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

    # Sign the bytes
    signing_key = PrivateKey(self.secret_key)
    signature_bytes = signing_key.sign_recoverable(signing_hash)

    # The key comparison is what makes reusing `sender` safe: it is
    # reassignable and `EOA.key` may be `None`, so a sender that does not
    # hold the signing key would contradict its own signature.
    if self.sender is not None and self.sender.key == self.secret_key:
        updated_values["sender"] = self.sender
    else:
        # The address the public key recovery would return, since the
        # signature was produced by this key, but derived with one point
        # multiplication instead.
        public_key = signing_key.public_key
        updated_values["sender"] = Address(
            keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
        )

    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
    ]

calculate_max_gas_limit(*, txs, env_gas_limit, transaction_gas_limit_cap, state_gas_reservoir_enabled) staticmethod

Calculate the maximum gas limit that can be set in a transaction given a list of transactions with and without gas-limits set and a maximum available environment gas.

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

@staticmethod
def calculate_max_gas_limit(
    *,
    txs: List["Transaction"],
    env_gas_limit: int,
    transaction_gas_limit_cap: int | None,
    state_gas_reservoir_enabled: bool,
) -> int:
    """
    Calculate the maximum gas limit that can be set in a transaction
    given a list of transactions with and without gas-limits set
    and a maximum available environment gas.
    """
    available_gas = env_gas_limit
    unset_gas_limit_tx_count = 0
    for tx in txs:
        if "gas_limit" not in tx.model_fields_set:
            unset_gas_limit_tx_count += 1
        else:
            available_gas -= int(tx.gas_limit)

    if unset_gas_limit_tx_count == 0:
        return 0

    if available_gas <= 0:
        raise Exception(
            "test correctness: unable to automatically calculate gas "
            "limit for transactions (no remaining gas: explicit "
            "transaction gas limits already consume the full "
            f"environment gas limit of {env_gas_limit})."
        )

    max_tx_gas_limit = available_gas // unset_gas_limit_tx_count
    if state_gas_reservoir_enabled:
        transaction_gas_limit_cap = None
    if transaction_gas_limit_cap:
        max_tx_gas_limit = min(max_tx_gas_limit, transaction_gas_limit_cap)
    return max_tx_gas_limit

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)

set_gas_limit(*, max_gas_limit, transaction_gas_limit_cap, state_gas_reservoir_enabled=False)

Set the transaction gas limit if unset.

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

def set_gas_limit(
    self,
    *,
    max_gas_limit: int,
    transaction_gas_limit_cap: int | None,
    state_gas_reservoir_enabled: bool = False,
) -> None:
    """Set the transaction gas limit if unset."""
    self._check_state_gas_reservoir_supported(
        state_gas_reservoir_enabled=state_gas_reservoir_enabled
    )
    if "gas_limit" not in self.model_fields_set:
        self.gas_limit = self._calculate_implicit_gas_limit(
            max_gas_limit=max_gas_limit,
            transaction_gas_limit_cap=transaction_gas_limit_cap,
            state_gas_reservoir_enabled=state_gas_reservoir_enabled,
        )

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"
    )
    assert "gas_limit" in self.model_fields_set, (
        "Impossible to calculate minimum balance without a set 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 * self.gas_limit
            + self.value
            + max_fee_per_blob_gas
            * (fork.blob_gas_per_blob() * len(self.blob_versioned_hashes))
        )
    else:
        return gas_price * self.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()