Ethereum Test Tools Package

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

Code dataclass

Generic code object.

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

@dataclass(kw_only=True)
class Code:
    """
    Generic code object.
    """

    bytecode: Optional[bytes] = None
    """
    bytes array that represents the bytecode of this object.
    """
    name: Optional[str] = None
    """
    Name used to describe this code.
    Usually used to add extra information to a test case.
    """

    def assemble(self) -> bytes:
        """
        Transform the Code object into bytes.
        Normally will be overriden by the classes that inherit this class.
        """
        if self.bytecode is None:
            return bytes()
        else:
            return self.bytecode

    def __add__(self, other: Union[str, bytes, "Code"]) -> "Code":
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(bytecode=(code_to_bytes(self) + code_to_bytes(other)))

    def __radd__(self, other: Union[str, bytes, "Code"]) -> "Code":
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(bytecode=(code_to_bytes(other) + code_to_bytes(self)))

bytecode: Optional[bytes] = None instance-attribute class-attribute

bytes array that represents the bytecode of this object.

name: Optional[str] = None instance-attribute class-attribute

Name used to describe this code. Usually used to add extra information to a test case.

assemble()

Transform the Code object into bytes. Normally will be overriden by the classes that inherit this class.

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

def assemble(self) -> bytes:
    """
    Transform the Code object into bytes.
    Normally will be overriden by the classes that inherit this class.
    """
    if self.bytecode is None:
        return bytes()
    else:
        return self.bytecode

__add__(other)

Adds two code objects together, by converting both to bytes first.

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

def __add__(self, other: Union[str, bytes, "Code"]) -> "Code":
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(bytecode=(code_to_bytes(self) + code_to_bytes(other)))

__radd__(other)

Adds two code objects together, by converting both to bytes first.

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

def __radd__(self, other: Union[str, bytes, "Code"]) -> "Code":
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(bytecode=(code_to_bytes(other) + code_to_bytes(self)))

Initcode

Bases: Code

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

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

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

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

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

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

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

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

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

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

    def __init__(
        self,
        *,
        deploy_code: str | bytes | Code,
        initcode_length: Optional[int] = None,
        padding_byte: int = 0x00,
        name: Optional[str] = None,
    ):
        """
        Generate legacy initcode that inits a contract with the specified code.
        The initcode can be padded to a specified length for testing purposes.
        """
        self.execution_gas = 0
        self.deploy_code = deploy_code
        deploy_code_bytes = code_to_bytes(self.deploy_code)
        code_length = len(deploy_code_bytes)

        initcode = bytearray()

        # PUSH2: length=<bytecode length>
        initcode.append(0x61)
        initcode += code_length.to_bytes(length=2, byteorder="big")
        self.execution_gas += 3

        # PUSH1: offset=0
        initcode.append(0x60)
        initcode.append(0x00)
        self.execution_gas += 3

        # DUP2
        initcode.append(0x81)
        self.execution_gas += 3

        # PUSH1: initcode_length=11 (constant)
        initcode.append(0x60)
        initcode.append(0x0B)
        self.execution_gas += 3

        # DUP3
        initcode.append(0x82)
        self.execution_gas += 3

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

        # RETURN: offset=0, length
        initcode.append(0xF3)
        self.execution_gas += 0

        pre_padding_bytes = bytes(initcode) + deploy_code_bytes

        if initcode_length is not None:
            if len(pre_padding_bytes) > initcode_length:
                raise Exception("Invalid specified length for initcode")

            padding_bytes = bytes([padding_byte] * (initcode_length - len(pre_padding_bytes)))
        else:
            padding_bytes = bytes()

        self.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(deploy_code_bytes)

        super().__init__(bytecode=pre_padding_bytes + padding_bytes, name=name)

__init__(*, deploy_code, initcode_length=None, padding_byte=0, name=None)

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

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

def __init__(
    self,
    *,
    deploy_code: str | bytes | Code,
    initcode_length: Optional[int] = None,
    padding_byte: int = 0x00,
    name: Optional[str] = None,
):
    """
    Generate legacy initcode that inits a contract with the specified code.
    The initcode can be padded to a specified length for testing purposes.
    """
    self.execution_gas = 0
    self.deploy_code = deploy_code
    deploy_code_bytes = code_to_bytes(self.deploy_code)
    code_length = len(deploy_code_bytes)

    initcode = bytearray()

    # PUSH2: length=<bytecode length>
    initcode.append(0x61)
    initcode += code_length.to_bytes(length=2, byteorder="big")
    self.execution_gas += 3

    # PUSH1: offset=0
    initcode.append(0x60)
    initcode.append(0x00)
    self.execution_gas += 3

    # DUP2
    initcode.append(0x81)
    self.execution_gas += 3

    # PUSH1: initcode_length=11 (constant)
    initcode.append(0x60)
    initcode.append(0x0B)
    self.execution_gas += 3

    # DUP3
    initcode.append(0x82)
    self.execution_gas += 3

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

    # RETURN: offset=0, length
    initcode.append(0xF3)
    self.execution_gas += 0

    pre_padding_bytes = bytes(initcode) + deploy_code_bytes

    if initcode_length is not None:
        if len(pre_padding_bytes) > initcode_length:
            raise Exception("Invalid specified length for initcode")

        padding_bytes = bytes([padding_byte] * (initcode_length - len(pre_padding_bytes)))
    else:
        padding_bytes = bytes()

    self.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(deploy_code_bytes)

    super().__init__(bytecode=pre_padding_bytes + padding_bytes, name=name)

execution_gas: int = 0 instance-attribute

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

deploy_code: bytes | str | Code = deploy_code instance-attribute

Bytecode to be deployed by the initcode.

deployment_gas: int = GAS_PER_DEPLOYED_CODE_BYTE * len(deploy_code_bytes) instance-attribute

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

ceiling_division(a, b)

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

Source code in src/ethereum_test_tools/common/helpers.py

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

fill_test(t8n, test_spec, fork, engine, spec, eips=None)

Fills fixtures for the specified fork.

Source code in src/ethereum_test_tools/filling/fill.py

def fill_test(
    t8n: TransitionTool,
    test_spec: BaseTest,
    fork: Fork,
    engine: str,
    spec: ReferenceSpec | None,
    eips: Optional[List[int]] = None,
) -> Fixture:
    """
    Fills fixtures for the specified fork.
    """
    t8n.reset_traces()

    genesis_rlp, genesis = test_spec.make_genesis(t8n, fork)

    (blocks, head, alloc) = test_spec.make_blocks(
        t8n,
        genesis,
        fork,
        eips=eips,
    )

    fork_name = fork.name()
    fixture = Fixture(
        blocks=blocks,
        genesis=genesis,
        genesis_rlp=genesis_rlp,
        head=head,
        fork="+".join([fork_name] + [str(eip) for eip in eips]) if eips is not None else fork_name,
        pre_state=copy(test_spec.pre),
        post_state=alloc_to_accounts(alloc),
        seal_engine=engine,
        name=test_spec.tag,
    )
    fixture.fill_info(t8n, spec)

    return fixture

compute_create_address(address, nonce)

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

Source code in src/ethereum_test_tools/common/helpers.py

def compute_create_address(address: str | int, nonce: int) -> str:
    """
    Compute address of the resulting contract created using a transaction
    or the `CREATE` opcode.
    """
    if type(address) is str:
        if address.startswith("0x"):
            address = address[2:]
        address_bytes = bytes.fromhex(address)
    elif type(address) is int:
        address_bytes = address.to_bytes(length=20, byteorder="big")
    if nonce == 0:
        nonce_bytes = bytes()
    else:
        nonce_bytes = nonce.to_bytes(length=1, byteorder="big")
    hash = keccak256(encode([address_bytes, nonce_bytes]))
    return "0x" + hash[-20:].hex()

EngineAPIError

Bases: IntEnum

List of Engine API errors

Source code in src/ethereum_test_tools/common/constants.py

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

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

Opcode

Bases: bytes

Represents a single Opcode instruction in the EVM, with extra metadata useful to parametrize tests.

Parameters

• popped_stack_items: number of items the opcode pops from the stack
• pushed_stack_items: number of items the opcode pushes to the stack
• min_stack_height: minimum stack height required by the opcode
• data_portion_length: number of bytes after the opcode in the bytecode that represent data

Source code in src/ethereum_test_tools/vm/opcode.py

class Opcode(bytes):
    """
    Represents a single Opcode instruction in the EVM, with extra
    metadata useful to parametrize tests.

    Parameters
    ----------
    - popped_stack_items: number of items the opcode pops from the stack
    - pushed_stack_items: number of items the opcode pushes to the stack
    - min_stack_height: minimum stack height required by the opcode
    - data_portion_length: number of bytes after the opcode in the bytecode
        that represent data
    """

    popped_stack_items: int
    pushed_stack_items: int
    min_stack_height: int
    data_portion_length: int
    _name_: str

    def __new__(
        cls,
        opcode_or_byte: Union[int, "Opcode"],
        *,
        popped_stack_items: int = 0,
        pushed_stack_items: int = 0,
        min_stack_height: int = 0,
        data_portion_length: int = 0,
    ):
        """
        Creates a new opcode instance.
        """
        if type(opcode_or_byte) is Opcode:
            # Required because Enum class calls the base class with the
            # instantiated object as parameter.
            return opcode_or_byte
        elif isinstance(opcode_or_byte, int):
            obj = super().__new__(cls, [opcode_or_byte])
            obj.popped_stack_items = popped_stack_items
            obj.pushed_stack_items = pushed_stack_items
            obj.min_stack_height = min_stack_height
            obj.data_portion_length = data_portion_length
            return obj

    def __call__(self, *args_t: Union[int, bytes, "Opcode"]) -> bytes:
        """
        Makes all opcode instances callable to return formatted bytecode,
        which constitutes a data portion, that is located after the opcode
        byte, and pre-opcode bytecode, which is normally used to set up the
        stack.

        This useful to automatically format, e.g., push opcodes and their
        data sections as `Opcodes.PUSH1(0x00)`.

        Data sign is automatically detected but for this reason the range
        of the input must be:
        `[-2^(data_portion_bits-1), 2^(data_portion_bits)]`
        where:
        `data_portion_bits == data_portion_length * 8`

        For the stack, the arguments are set up in the opposite order they are
        given, so the first argument is the last item pushed to the stack.

        The resulting stack arrangement does not take into account opcode stack
        element consumption, so the stack height is not guaranteed to be
        correct and the user must take this into consideration.

        Integers can also be used as stack elements, in which case they are
        automatically converted to PUSH operations, and negative numbers always
        use a PUSH32 operation.


        """
        args: List[Union[int, bytes, "Opcode"]] = list(args_t)
        pre_opcode_bytecode = bytes()
        data_portion = bytes()

        if self.data_portion_length > 0:
            # For opcodes with a data portion, the first argument is the data
            # and the rest of the arguments form the stack.
            if len(args) == 0:
                raise ValueError("Opcode with data portion requires at least one argument")
            data = args.pop(0)
            if isinstance(data, bytes):
                data_portion = data
            elif isinstance(data, int):
                signed = data < 0
                data_portion = data.to_bytes(
                    length=self.data_portion_length,
                    byteorder="big",
                    signed=signed,
                )
            else:
                raise TypeError("Opcode data portion must be either an int or a bytes")

        # The rest of the arguments conform the stack.
        while len(args) > 0:
            data = args.pop()
            if isinstance(data, bytes):
                pre_opcode_bytecode += data
            elif isinstance(data, int):
                # We are going to push a constant to the stack.
                signed = data < 0
                data_size = _get_int_size(data)
                if data_size > 32:
                    raise ValueError("Opcode stack data must be less than 32 bytes")
                elif data_size == 0:
                    # Pushing 0 is done with the PUSH1 opcode for compatibility
                    # reasons.
                    data_size = 1

                pre_opcode_bytecode += _push_opcodes_byte_list[data_size]
                pre_opcode_bytecode += data.to_bytes(
                    length=data_size,
                    byteorder="big",
                    signed=signed,
                )

            else:
                raise TypeError("Opcode stack data must be either an int or a bytes")

        return pre_opcode_bytecode + self + data_portion

    def __len__(self) -> int:
        """
        Returns the total bytecode length of the opcode, taking into account
        its data portion.
        """
        return self.data_portion_length + 1

    def int(self) -> int:
        """
        Returns the integer representation of the opcode.
        """
        return int.from_bytes(bytes=self, byteorder="big")

    def __str__(self) -> str:
        """
        Return the name of the opcode, assigned at Enum creation.
        """
        return self._name_

__new__(opcode_or_byte, *, popped_stack_items=0, pushed_stack_items=0, min_stack_height=0, data_portion_length=0)

Creates a new opcode instance.

Source code in src/ethereum_test_tools/vm/opcode.py

def __new__(
    cls,
    opcode_or_byte: Union[int, "Opcode"],
    *,
    popped_stack_items: int = 0,
    pushed_stack_items: int = 0,
    min_stack_height: int = 0,
    data_portion_length: int = 0,
):
    """
    Creates a new opcode instance.
    """
    if type(opcode_or_byte) is Opcode:
        # Required because Enum class calls the base class with the
        # instantiated object as parameter.
        return opcode_or_byte
    elif isinstance(opcode_or_byte, int):
        obj = super().__new__(cls, [opcode_or_byte])
        obj.popped_stack_items = popped_stack_items
        obj.pushed_stack_items = pushed_stack_items
        obj.min_stack_height = min_stack_height
        obj.data_portion_length = data_portion_length
        return obj

__call__(*args_t)

Makes all opcode instances callable to return formatted bytecode, which constitutes a data portion, that is located after the opcode byte, and pre-opcode bytecode, which is normally used to set up the stack.

This useful to automatically format, e.g., push opcodes and their data sections as Opcodes.PUSH1(0x00).

Data sign is automatically detected but for this reason the range of the input must be: [-2^(data_portion_bits-1), 2^(data_portion_bits)] where: data_portion_bits == data_portion_length * 8

For the stack, the arguments are set up in the opposite order they are given, so the first argument is the last item pushed to the stack.

The resulting stack arrangement does not take into account opcode stack element consumption, so the stack height is not guaranteed to be correct and the user must take this into consideration.

Integers can also be used as stack elements, in which case they are automatically converted to PUSH operations, and negative numbers always use a PUSH32 operation.

Source code in src/ethereum_test_tools/vm/opcode.py

def __call__(self, *args_t: Union[int, bytes, "Opcode"]) -> bytes:
    """
    Makes all opcode instances callable to return formatted bytecode,
    which constitutes a data portion, that is located after the opcode
    byte, and pre-opcode bytecode, which is normally used to set up the
    stack.

    This useful to automatically format, e.g., push opcodes and their
    data sections as `Opcodes.PUSH1(0x00)`.

    Data sign is automatically detected but for this reason the range
    of the input must be:
    `[-2^(data_portion_bits-1), 2^(data_portion_bits)]`
    where:
    `data_portion_bits == data_portion_length * 8`

    For the stack, the arguments are set up in the opposite order they are
    given, so the first argument is the last item pushed to the stack.

    The resulting stack arrangement does not take into account opcode stack
    element consumption, so the stack height is not guaranteed to be
    correct and the user must take this into consideration.

    Integers can also be used as stack elements, in which case they are
    automatically converted to PUSH operations, and negative numbers always
    use a PUSH32 operation.


    """
    args: List[Union[int, bytes, "Opcode"]] = list(args_t)
    pre_opcode_bytecode = bytes()
    data_portion = bytes()

    if self.data_portion_length > 0:
        # For opcodes with a data portion, the first argument is the data
        # and the rest of the arguments form the stack.
        if len(args) == 0:
            raise ValueError("Opcode with data portion requires at least one argument")
        data = args.pop(0)
        if isinstance(data, bytes):
            data_portion = data
        elif isinstance(data, int):
            signed = data < 0
            data_portion = data.to_bytes(
                length=self.data_portion_length,
                byteorder="big",
                signed=signed,
            )
        else:
            raise TypeError("Opcode data portion must be either an int or a bytes")

    # The rest of the arguments conform the stack.
    while len(args) > 0:
        data = args.pop()
        if isinstance(data, bytes):
            pre_opcode_bytecode += data
        elif isinstance(data, int):
            # We are going to push a constant to the stack.
            signed = data < 0
            data_size = _get_int_size(data)
            if data_size > 32:
                raise ValueError("Opcode stack data must be less than 32 bytes")
            elif data_size == 0:
                # Pushing 0 is done with the PUSH1 opcode for compatibility
                # reasons.
                data_size = 1

            pre_opcode_bytecode += _push_opcodes_byte_list[data_size]
            pre_opcode_bytecode += data.to_bytes(
                length=data_size,
                byteorder="big",
                signed=signed,
            )

        else:
            raise TypeError("Opcode stack data must be either an int or a bytes")

    return pre_opcode_bytecode + self + data_portion

__len__()

Returns the total bytecode length of the opcode, taking into account its data portion.

Source code in src/ethereum_test_tools/vm/opcode.py

def __len__(self) -> int:
    """
    Returns the total bytecode length of the opcode, taking into account
    its data portion.
    """
    return self.data_portion_length + 1

int()

Returns the integer representation of the opcode.

Source code in src/ethereum_test_tools/vm/opcode.py

def int(self) -> int:
    """
    Returns the integer representation of the opcode.
    """
    return int.from_bytes(bytes=self, byteorder="big")

__str__()

Return the name of the opcode, assigned at Enum creation.

Source code in src/ethereum_test_tools/vm/opcode.py

def __str__(self) -> str:
    """
    Return the name of the opcode, assigned at Enum creation.
    """
    return self._name_

BlockchainTest dataclass

Bases: BaseTest

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

Source code in src/ethereum_test_tools/spec/blockchain_test.py

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

    pre: Mapping[str, Account]
    post: Mapping[str, Account]
    blocks: List[Block]
    genesis_environment: Environment = field(default_factory=Environment)
    tag: str = ""

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Returns the parameter name used to identify this filler in a test.
        """
        return "blockchain_test"

    def make_genesis(
        self,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[bytes, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = self.genesis_environment.set_fork_requirements(fork)

        genesis = FixtureHeader(
            parent_hash=EmptyHash,
            ommers_hash=EmptyOmmersRoot,
            coinbase=ZeroAddress,
            state_root=t8n.calc_state_root(
                to_json(self.pre),
                fork,
            ),
            transactions_root=EmptyTrieRoot,
            receipt_root=EmptyTrieRoot,
            bloom=EmptyBloom,
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=0,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data=bytes([0]),
            mix_digest=EmptyHash,
            nonce=EmptyNonce,
            base_fee=env.base_fee,
            data_gas_used=env.data_gas_used,
            excess_data_gas=env.excess_data_gas,
            withdrawals_root=t8n.calc_withdrawals_root(env.withdrawals, fork)
            if env.withdrawals is not None
            else None,
        )

        genesis_rlp, genesis.hash = genesis.build(
            txs=[],
            ommers=[],
            withdrawals=env.withdrawals,
        )

        return genesis_rlp, genesis

    def make_block(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Dict[str, Any],
        previous_head: bytes,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[FixtureBlock, Environment, Dict[str, Any], bytes]:
        """
        Produces a block based on the previous environment and allocation.
        If the block is an invalid block, the environment and allocation
        returned are the same as passed as parameters.
        Raises exception on invalid test behavior.

        Returns
        -------
            FixtureBlock: Block to be appended to the fixture.
            Environment: Environment for the next block to produce.
                If the produced block is invalid, this is exactly the same
                environment as the one passed as parameter.
            Dict[str, Any]: Allocation for the next block to produce.
                If the produced block is invalid, this is exactly the same
                allocation as the one passed as parameter.
            str: Hash of the head of the chain, only updated if the produced
                block is not invalid.

        """
        if block.rlp and block.exception is not None:
            raise Exception(
                "test correctness: post-state cannot be verified if the "
                + "block's rlp is supplied and the block is not supposed "
                + "to produce an exception"
            )

        if block.rlp is None:
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            env = block.set_environment(previous_env)
            env = env.set_fork_requirements(fork)

            txs = (
                [tx.with_signature_and_sender() for tx in block.txs]
                if block.txs is not None
                else []
            )

            next_alloc, result = t8n.evaluate(
                alloc=previous_alloc,
                txs=to_json_or_none(txs),
                env=to_json(env),
                fork=fork,
                chain_id=chain_id,
                reward=fork.get_reward(env.number, env.timestamp),
                eips=eips,
            )
            try:
                rejected_txs = verify_transactions(txs, result)
            except Exception as e:
                print_traces(t8n.get_traces())
                pprint(result)
                pprint(previous_alloc)
                pprint(next_alloc)
                raise e

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

            header = FixtureHeader.from_dict(
                result
                | {
                    "parentHash": env.parent_hash(),
                    "miner": env.coinbase,
                    "transactionsRoot": result.get("txRoot"),
                    "difficulty": str_or_none(result.get("currentDifficulty"), "0"),
                    "number": str(env.number),
                    "gasLimit": str(env.gas_limit),
                    "timestamp": str(env.timestamp),
                    "extraData": block.extra_data
                    if block.extra_data is not None and len(block.extra_data) != 0
                    else "0x",
                    "sha3Uncles": "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",  # noqa: E501
                    "mixHash": "0x0000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
                    "nonce": "0x0000000000000000",
                    "baseFeePerGas": result.get("currentBaseFee"),
                    "excessDataGas": result.get("currentExcessDataGas"),
                }
            )

            assert len(header.state_root) == 32

            if block.rlp_modifier is not None:
                # Modify any parameter specified in the `rlp_modifier` after
                # transition tool processing.
                header = header.join(block.rlp_modifier)

            rlp, header.hash = header.build(
                txs=txs,
                ommers=[],
                withdrawals=env.withdrawals,
            )

            new_payload = FixtureEngineNewPayload.from_fixture_header(
                fork=fork,
                header=header,
                transactions=txs,
                withdrawals=env.withdrawals,
                error_code=block.engine_api_error_code,
            )

            if block.exception is None:
                # Return environment and allocation of the following block
                return (
                    FixtureBlock(
                        rlp=rlp,
                        new_payload=new_payload,
                        block_header=header,
                        block_number=header.number,
                        txs=txs,
                        ommers=[],
                        withdrawals=env.withdrawals,
                    ),
                    env.apply_new_parent(header),
                    next_alloc,
                    header.hash,
                )
            else:
                return (
                    FixtureBlock(
                        rlp=rlp,
                        new_payload=new_payload,
                        expected_exception=block.exception,
                        block_number=header.number,
                    ),
                    previous_env,
                    previous_alloc,
                    previous_head,
                )
        else:
            return (
                FixtureBlock(
                    rlp=block.rlp,
                    expected_exception=block.exception,
                ),
                previous_env,
                previous_alloc,
                previous_head,
            )

    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        fork: Fork,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
        """
        Create a block list from the blockchain test definition.
        Performs checks against the expected behavior of the test.
        Raises exception on invalid test behavior.
        """
        alloc = to_json(self.pre)
        env = Environment.from_parent_header(genesis)
        blocks: List[FixtureBlock] = []
        head = genesis.hash if genesis.hash is not None else bytes([0] * 32)
        for block in self.blocks:
            fixture_block, env, alloc, head = self.make_block(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                previous_head=head,
                chain_id=chain_id,
                eips=eips,
            )
            blocks.append(fixture_block)

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

        return (blocks, head, alloc)

pytest_parameter_name() classmethod

Returns the parameter name used to identify this filler in a test.

Source code in src/ethereum_test_tools/spec/blockchain_test.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Returns the parameter name used to identify this filler in a test.
    """
    return "blockchain_test"

make_genesis(t8n, fork)

Create a genesis block from the state test definition.

Source code in src/ethereum_test_tools/spec/blockchain_test.py

def make_genesis(
    self,
    t8n: TransitionTool,
    fork: Fork,
) -> Tuple[bytes, FixtureHeader]:
    """
    Create a genesis block from the state test definition.
    """
    env = self.genesis_environment.set_fork_requirements(fork)

    genesis = FixtureHeader(
        parent_hash=EmptyHash,
        ommers_hash=EmptyOmmersRoot,
        coinbase=ZeroAddress,
        state_root=t8n.calc_state_root(
            to_json(self.pre),
            fork,
        ),
        transactions_root=EmptyTrieRoot,
        receipt_root=EmptyTrieRoot,
        bloom=EmptyBloom,
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=0,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data=bytes([0]),
        mix_digest=EmptyHash,
        nonce=EmptyNonce,
        base_fee=env.base_fee,
        data_gas_used=env.data_gas_used,
        excess_data_gas=env.excess_data_gas,
        withdrawals_root=t8n.calc_withdrawals_root(env.withdrawals, fork)
        if env.withdrawals is not None
        else None,
    )

    genesis_rlp, genesis.hash = genesis.build(
        txs=[],
        ommers=[],
        withdrawals=env.withdrawals,
    )

    return genesis_rlp, genesis

make_block(t8n, fork, block, previous_env, previous_alloc, previous_head, chain_id=1, eips=None)

Produces a block based on the previous environment and allocation. If the block is an invalid block, the environment and allocation returned are the same as passed as parameters. Raises exception on invalid test behavior.

Returns

FixtureBlock: Block to be appended to the fixture.
Environment: Environment for the next block to produce.
    If the produced block is invalid, this is exactly the same
    environment as the one passed as parameter.
Dict[str, Any]: Allocation for the next block to produce.
    If the produced block is invalid, this is exactly the same
    allocation as the one passed as parameter.
str: Hash of the head of the chain, only updated if the produced
    block is not invalid.

Source code in src/ethereum_test_tools/spec/blockchain_test.py

def make_block(
    self,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Dict[str, Any],
    previous_head: bytes,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[FixtureBlock, Environment, Dict[str, Any], bytes]:
    """
    Produces a block based on the previous environment and allocation.
    If the block is an invalid block, the environment and allocation
    returned are the same as passed as parameters.
    Raises exception on invalid test behavior.

    Returns
    -------
        FixtureBlock: Block to be appended to the fixture.
        Environment: Environment for the next block to produce.
            If the produced block is invalid, this is exactly the same
            environment as the one passed as parameter.
        Dict[str, Any]: Allocation for the next block to produce.
            If the produced block is invalid, this is exactly the same
            allocation as the one passed as parameter.
        str: Hash of the head of the chain, only updated if the produced
            block is not invalid.

    """
    if block.rlp and block.exception is not None:
        raise Exception(
            "test correctness: post-state cannot be verified if the "
            + "block's rlp is supplied and the block is not supposed "
            + "to produce an exception"
        )

    if block.rlp is None:
        # This is the most common case, the RLP needs to be constructed
        # based on the transactions to be included in the block.
        # Set the environment according to the block to execute.
        env = block.set_environment(previous_env)
        env = env.set_fork_requirements(fork)

        txs = (
            [tx.with_signature_and_sender() for tx in block.txs]
            if block.txs is not None
            else []
        )

        next_alloc, result = t8n.evaluate(
            alloc=previous_alloc,
            txs=to_json_or_none(txs),
            env=to_json(env),
            fork=fork,
            chain_id=chain_id,
            reward=fork.get_reward(env.number, env.timestamp),
            eips=eips,
        )
        try:
            rejected_txs = verify_transactions(txs, result)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(result)
            pprint(previous_alloc)
            pprint(next_alloc)
            raise e

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

        header = FixtureHeader.from_dict(
            result
            | {
                "parentHash": env.parent_hash(),
                "miner": env.coinbase,
                "transactionsRoot": result.get("txRoot"),
                "difficulty": str_or_none(result.get("currentDifficulty"), "0"),
                "number": str(env.number),
                "gasLimit": str(env.gas_limit),
                "timestamp": str(env.timestamp),
                "extraData": block.extra_data
                if block.extra_data is not None and len(block.extra_data) != 0
                else "0x",
                "sha3Uncles": "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",  # noqa: E501
                "mixHash": "0x0000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
                "nonce": "0x0000000000000000",
                "baseFeePerGas": result.get("currentBaseFee"),
                "excessDataGas": result.get("currentExcessDataGas"),
            }
        )

        assert len(header.state_root) == 32

        if block.rlp_modifier is not None:
            # Modify any parameter specified in the `rlp_modifier` after
            # transition tool processing.
            header = header.join(block.rlp_modifier)

        rlp, header.hash = header.build(
            txs=txs,
            ommers=[],
            withdrawals=env.withdrawals,
        )

        new_payload = FixtureEngineNewPayload.from_fixture_header(
            fork=fork,
            header=header,
            transactions=txs,
            withdrawals=env.withdrawals,
            error_code=block.engine_api_error_code,
        )

        if block.exception is None:
            # Return environment and allocation of the following block
            return (
                FixtureBlock(
                    rlp=rlp,
                    new_payload=new_payload,
                    block_header=header,
                    block_number=header.number,
                    txs=txs,
                    ommers=[],
                    withdrawals=env.withdrawals,
                ),
                env.apply_new_parent(header),
                next_alloc,
                header.hash,
            )
        else:
            return (
                FixtureBlock(
                    rlp=rlp,
                    new_payload=new_payload,
                    expected_exception=block.exception,
                    block_number=header.number,
                ),
                previous_env,
                previous_alloc,
                previous_head,
            )
    else:
        return (
            FixtureBlock(
                rlp=block.rlp,
                expected_exception=block.exception,
            ),
            previous_env,
            previous_alloc,
            previous_head,
        )

make_blocks(t8n, genesis, fork, chain_id=1, eips=None)

Create a block list from the blockchain test definition. Performs checks against the expected behavior of the test. Raises exception on invalid test behavior.

Source code in src/ethereum_test_tools/spec/blockchain_test.py

def make_blocks(
    self,
    t8n: TransitionTool,
    genesis: FixtureHeader,
    fork: Fork,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
    """
    Create a block list from the blockchain test definition.
    Performs checks against the expected behavior of the test.
    Raises exception on invalid test behavior.
    """
    alloc = to_json(self.pre)
    env = Environment.from_parent_header(genesis)
    blocks: List[FixtureBlock] = []
    head = genesis.hash if genesis.hash is not None else bytes([0] * 32)
    for block in self.blocks:
        fixture_block, env, alloc, head = self.make_block(
            t8n=t8n,
            fork=fork,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
            previous_head=head,
            chain_id=chain_id,
            eips=eips,
        )
        blocks.append(fixture_block)

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

    return (blocks, head, alloc)

ReferenceSpec

Reference Specification Description Abstract Class.

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

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

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

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

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

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

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

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

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

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

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

name() abstractmethod

Returns the name of the spec.

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

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

has_known_version() abstractmethod

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

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

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

known_version() abstractmethod

Returns the latest known version in the reference.

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

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

api_url() abstractmethod

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

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

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

latest_version() abstractmethod

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

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

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

is_outdated() abstractmethod

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

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

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

write_info(info) abstractmethod

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

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

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

parseable_from_module(module_dict) staticmethod abstractmethod

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

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

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

parse_from_module(module_dict) staticmethod abstractmethod

Parses the module's dict into a reference spec.

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

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

StateTest dataclass

Bases: BaseTest

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

Source code in src/ethereum_test_tools/spec/state_test.py

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

    env: Environment
    pre: Mapping[str, Account]
    post: Mapping[str, Account]
    txs: List[Transaction]
    engine_api_error_code: Optional[EngineAPIError] = None
    tag: str = ""

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Returns the parameter name used to identify this filler in a test.
        """
        return "state_test"

    def make_genesis(
        self,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[bytes, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = self.env.set_fork_requirements(fork)

        genesis = FixtureHeader(
            parent_hash=EmptyHash,
            ommers_hash=EmptyOmmersRoot,
            coinbase=ZeroAddress,
            state_root=t8n.calc_state_root(
                to_json(self.pre),
                fork,
            ),
            transactions_root=EmptyTrieRoot,
            receipt_root=EmptyTrieRoot,
            bloom=EmptyBloom,
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=env.number - 1,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data=bytes([0]),
            mix_digest=EmptyHash,
            nonce=EmptyNonce,
            base_fee=env.base_fee,
            data_gas_used=env.data_gas_used,
            excess_data_gas=env.excess_data_gas,
            withdrawals_root=t8n.calc_withdrawals_root(env.withdrawals, fork)
            if env.withdrawals is not None
            else None,
        )

        genesis_rlp, genesis.hash = genesis.build(
            txs=[],
            ommers=[],
            withdrawals=env.withdrawals,
        )

        return genesis_rlp, genesis

    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        fork: Fork,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
        """
        Create a block from the state test definition.
        Performs checks against the expected behavior of the test.
        Raises exception on invalid test behavior.
        """
        env = self.env.apply_new_parent(genesis)
        env = env.set_fork_requirements(fork)

        txs = [tx.with_signature_and_sender() for tx in self.txs] if self.txs is not None else []

        alloc, result = t8n.evaluate(
            alloc=to_json(self.pre),
            txs=to_json(txs),
            env=to_json(env),
            fork=fork,
            chain_id=chain_id,
            reward=fork.get_reward(env.number, env.timestamp),
            eips=eips,
        )

        rejected_txs = verify_transactions(txs, result)
        if len(rejected_txs) > 0:
            raise Exception(
                "one or more transactions in `StateTest` are "
                + "intrinsically invalid, which are not allowed. "
                + "Use `BlockchainTest` to verify rejection of blocks "
                + "that include invalid transactions."
            )

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

        header = FixtureHeader.from_dict(
            result
            | {
                "parentHash": genesis.hash,
                "miner": env.coinbase,
                "transactionsRoot": result.get("txRoot"),
                "difficulty": str_or_none(env.difficulty, result.get("currentDifficulty")),
                "number": str(env.number),
                "gasLimit": str(env.gas_limit),
                "timestamp": str(env.timestamp),
                "extraData": "0x00",
                "sha3Uncles": "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
                "mixHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
                "nonce": "0x0000000000000000",
                "baseFeePerGas": result.get("currentBaseFee"),
                "excessDataGas": result.get("currentExcessDataGas"),
            }
        )

        block, header.hash = header.build(
            txs=txs,
            ommers=[],
            withdrawals=env.withdrawals,
        )

        new_payload = FixtureEngineNewPayload.from_fixture_header(
            fork=fork,
            header=header,
            transactions=txs,
            withdrawals=env.withdrawals,
            error_code=self.engine_api_error_code,
        )

        return (
            [
                FixtureBlock(
                    rlp=block,
                    new_payload=new_payload,
                    block_header=header,
                    txs=txs,
                    ommers=[],
                    withdrawals=env.withdrawals,
                )
            ],
            header.hash,
            alloc,
        )

pytest_parameter_name() classmethod

Returns the parameter name used to identify this filler in a test.

Source code in src/ethereum_test_tools/spec/state_test.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Returns the parameter name used to identify this filler in a test.
    """
    return "state_test"

make_genesis(t8n, fork)

Create a genesis block from the state test definition.

Source code in src/ethereum_test_tools/spec/state_test.py

def make_genesis(
    self,
    t8n: TransitionTool,
    fork: Fork,
) -> Tuple[bytes, FixtureHeader]:
    """
    Create a genesis block from the state test definition.
    """
    env = self.env.set_fork_requirements(fork)

    genesis = FixtureHeader(
        parent_hash=EmptyHash,
        ommers_hash=EmptyOmmersRoot,
        coinbase=ZeroAddress,
        state_root=t8n.calc_state_root(
            to_json(self.pre),
            fork,
        ),
        transactions_root=EmptyTrieRoot,
        receipt_root=EmptyTrieRoot,
        bloom=EmptyBloom,
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=env.number - 1,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data=bytes([0]),
        mix_digest=EmptyHash,
        nonce=EmptyNonce,
        base_fee=env.base_fee,
        data_gas_used=env.data_gas_used,
        excess_data_gas=env.excess_data_gas,
        withdrawals_root=t8n.calc_withdrawals_root(env.withdrawals, fork)
        if env.withdrawals is not None
        else None,
    )

    genesis_rlp, genesis.hash = genesis.build(
        txs=[],
        ommers=[],
        withdrawals=env.withdrawals,
    )

    return genesis_rlp, genesis

make_blocks(t8n, genesis, fork, chain_id=1, eips=None)

Create a block from the state test definition. Performs checks against the expected behavior of the test. Raises exception on invalid test behavior.

Source code in src/ethereum_test_tools/spec/state_test.py

def make_blocks(
    self,
    t8n: TransitionTool,
    genesis: FixtureHeader,
    fork: Fork,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
    """
    Create a block from the state test definition.
    Performs checks against the expected behavior of the test.
    Raises exception on invalid test behavior.
    """
    env = self.env.apply_new_parent(genesis)
    env = env.set_fork_requirements(fork)

    txs = [tx.with_signature_and_sender() for tx in self.txs] if self.txs is not None else []

    alloc, result = t8n.evaluate(
        alloc=to_json(self.pre),
        txs=to_json(txs),
        env=to_json(env),
        fork=fork,
        chain_id=chain_id,
        reward=fork.get_reward(env.number, env.timestamp),
        eips=eips,
    )

    rejected_txs = verify_transactions(txs, result)
    if len(rejected_txs) > 0:
        raise Exception(
            "one or more transactions in `StateTest` are "
            + "intrinsically invalid, which are not allowed. "
            + "Use `BlockchainTest` to verify rejection of blocks "
            + "that include invalid transactions."
        )

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

    header = FixtureHeader.from_dict(
        result
        | {
            "parentHash": genesis.hash,
            "miner": env.coinbase,
            "transactionsRoot": result.get("txRoot"),
            "difficulty": str_or_none(env.difficulty, result.get("currentDifficulty")),
            "number": str(env.number),
            "gasLimit": str(env.gas_limit),
            "timestamp": str(env.timestamp),
            "extraData": "0x00",
            "sha3Uncles": "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
            "mixHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
            "nonce": "0x0000000000000000",
            "baseFeePerGas": result.get("currentBaseFee"),
            "excessDataGas": result.get("currentExcessDataGas"),
        }
    )

    block, header.hash = header.build(
        txs=txs,
        ommers=[],
        withdrawals=env.withdrawals,
    )

    new_payload = FixtureEngineNewPayload.from_fixture_header(
        fork=fork,
        header=header,
        transactions=txs,
        withdrawals=env.withdrawals,
        error_code=self.engine_api_error_code,
    )

    return (
        [
            FixtureBlock(
                rlp=block,
                new_payload=new_payload,
                block_header=header,
                txs=txs,
                ommers=[],
                withdrawals=env.withdrawals,
            )
        ],
        header.hash,
        alloc,
    )

Yul

Bases: Code

Yul compiler. Compiles Yul source code into bytecode.

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

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

    source: str
    compiled: Optional[bytes] = None

    def __init__(
        self,
        source: str,
        fork: Optional[Fork] = None,
        binary: Optional[Path | str] = None,
    ):
        self.source = source
        self.evm_version = get_evm_version_from_fork(fork)
        if binary is None:
            which_path = which("solc")
            if which_path is not None:
                binary = Path(which_path)
        if binary is None or not Path(binary).exists():
            raise Exception(
                """`solc` binary executable not found, please refer to
                https://docs.soliditylang.org/en/latest/installing-solidity.html
                for help downloading and installing `solc`"""
            )
        self.binary = Path(binary)

    def assemble(self) -> bytes:
        """
        Assembles using `solc --assemble`.
        """
        if not self.compiled:
            solc_args: Tuple[Union[Path, str], ...] = ()
            if self.evm_version:
                solc_args = (
                    self.binary,
                    "--evm-version",
                    self.evm_version,
                    *DEFAULT_SOLC_ARGS,
                )
            else:
                solc_args = (self.binary, *DEFAULT_SOLC_ARGS)
            result = run(
                solc_args,
                input=str.encode(self.source),
                stdout=PIPE,
                stderr=PIPE,
            )

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

            lines = result.stdout.decode().split("\n")

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

            self.compiled = bytes.fromhex(hex_str)
        return self.compiled

    def version(self) -> str:
        """
        Return solc's version string
        """
        result = run(
            [self.binary, "--version"],
            stdout=PIPE,
            stderr=PIPE,
        )
        solc_output = result.stdout.decode().split("\n")
        version_pattern = r"0\.\d+\.\d+\+\S+"
        solc_version_string = None
        for line in solc_output:
            match = re.search(version_pattern, line)
            if match:
                solc_version_string = match.group(0)
                break
        if not solc_version_string:
            warnings.warn("Unable to determine solc version.")
            solc_version_string = "unknown"
        return solc_version_string

assemble()

Assembles using solc --assemble.

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

def assemble(self) -> bytes:
    """
    Assembles using `solc --assemble`.
    """
    if not self.compiled:
        solc_args: Tuple[Union[Path, str], ...] = ()
        if self.evm_version:
            solc_args = (
                self.binary,
                "--evm-version",
                self.evm_version,
                *DEFAULT_SOLC_ARGS,
            )
        else:
            solc_args = (self.binary, *DEFAULT_SOLC_ARGS)
        result = run(
            solc_args,
            input=str.encode(self.source),
            stdout=PIPE,
            stderr=PIPE,
        )

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

        lines = result.stdout.decode().split("\n")

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

        self.compiled = bytes.fromhex(hex_str)
    return self.compiled

version()

Return solc's version string

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

def version(self) -> str:
    """
    Return solc's version string
    """
    result = run(
        [self.binary, "--version"],
        stdout=PIPE,
        stderr=PIPE,
    )
    solc_output = result.stdout.decode().split("\n")
    version_pattern = r"0\.\d+\.\d+\+\S+"
    solc_version_string = None
    for line in solc_output:
        match = re.search(version_pattern, line)
        if match:
            solc_version_string = match.group(0)
            break
    if not solc_version_string:
        warnings.warn("Unable to determine solc version.")
        solc_version_string = "unknown"
    return solc_version_string

compute_create2_address(address, salt, initcode)

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

Source code in src/ethereum_test_tools/common/helpers.py

def compute_create2_address(address: str | int, salt: int, initcode: bytes) -> str:
    """
    Compute address of the resulting contract created using the `CREATE2`
    opcode.
    """
    ff = bytes([0xFF])
    if type(address) is str:
        if address.startswith("0x"):
            address = address[2:]
        address_bytes = bytes.fromhex(address)
    elif type(address) is int:
        address_bytes = address.to_bytes(length=20, byteorder="big")
    salt_bytes = salt.to_bytes(length=32, byteorder="big")
    initcode_hash = keccak256(initcode)
    hash = keccak256(ff + address_bytes + salt_bytes + initcode_hash)
    return "0x" + hash[-20:].hex()

cost_memory_bytes(new_bytes, previous_bytes)

Calculates the cost of memory expansion, based on the costs specified in the yellow paper: https://ethereum.github.io/yellowpaper/paper.pdf

Source code in src/ethereum_test_tools/common/helpers.py

def cost_memory_bytes(new_bytes: int, previous_bytes: int) -> int:
    """
    Calculates the cost of memory expansion, based on the costs specified in
    the yellow paper: https://ethereum.github.io/yellowpaper/paper.pdf
    """
    if new_bytes <= previous_bytes:
        return 0
    new_words = ceiling_division(new_bytes, 32)
    previous_words = ceiling_division(previous_bytes, 32)

    def c(w: int) -> int:
        g_memory = 3
        return (g_memory * w) + ((w * w) // 512)

    return c(new_words) - c(previous_words)

Auto

Class to use as a sentinel value for parameters that should be automatically calculated.

Source code in src/ethereum_test_tools/common/types.py

class Auto:
    """
    Class to use as a sentinel value for parameters that should be
    automatically calculated.
    """

    def __repr__(self) -> str:
        """Print the correct test id."""
        return "auto"

__repr__()

Print the correct test id.

Source code in src/ethereum_test_tools/common/types.py

def __repr__(self) -> str:
    """Print the correct test id."""
    return "auto"

BaseTest

Represents a base Ethereum test which must return a genesis and a blockchain.

Source code in src/ethereum_test_tools/spec/base_test.py

class BaseTest:
    """
    Represents a base Ethereum test which must return a genesis and a
    blockchain.
    """

    pre: Mapping[str, Account]
    tag: str = ""

    @abstractmethod
    def make_genesis(
        self,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[bytes, FixtureHeader]:
        """
        Create a genesis block from the test definition.
        """
        pass

    @abstractmethod
    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        fork: Fork,
        chain_id: int = 1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
        """
        Generate the blockchain that must be executed sequentially during test.
        """
        pass

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

make_genesis(t8n, fork) abstractmethod

Create a genesis block from the test definition.

Source code in src/ethereum_test_tools/spec/base_test.py

@abstractmethod
def make_genesis(
    self,
    t8n: TransitionTool,
    fork: Fork,
) -> Tuple[bytes, FixtureHeader]:
    """
    Create a genesis block from the test definition.
    """
    pass

make_blocks(t8n, genesis, fork, chain_id=1, eips=None) abstractmethod

Generate the blockchain that must be executed sequentially during test.

Source code in src/ethereum_test_tools/spec/base_test.py

@abstractmethod
def make_blocks(
    self,
    t8n: TransitionTool,
    genesis: FixtureHeader,
    fork: Fork,
    chain_id: int = 1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], bytes, Dict[str, Any]]:
    """
    Generate the blockchain that must be executed sequentially during test.
    """
    pass

pytest_parameter_name() classmethod abstractmethod

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

Source code in src/ethereum_test_tools/spec/base_test.py

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

copy_opcode_cost(length)

Calculates the cost of the COPY opcodes, assuming memory expansion from empty memory, based on the costs specified in the yellow paper: https://ethereum.github.io/yellowpaper/paper.pdf

Source code in src/ethereum_test_tools/common/helpers.py

def copy_opcode_cost(length: int) -> int:
    """
    Calculates the cost of the COPY opcodes, assuming memory expansion from
    empty memory, based on the costs specified in the yellow paper:
    https://ethereum.github.io/yellowpaper/paper.pdf
    """
    return 3 + (ceiling_division(length, 32) * 3) + cost_memory_bytes(length, 0)

Storage

Definition of a storage in pre or post state of a test

Source code in src/ethereum_test_tools/common/types.py

class Storage:
    """
    Definition of a storage in pre or post state of a test
    """

    data: Dict[int, int]

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

    class InvalidType(Exception):
        """
        Invalid type used when describing test's expected storage key or value.
        """

        key_or_value: Any

        def __init__(self, key_or_value: Any, *args):
            super().__init__(args)
            self.key_or_value = key_or_value

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

    class InvalidValue(Exception):
        """
        Invalid value used when describing test's expected storage key or
        value.
        """

        key_or_value: Any

        def __init__(self, key_or_value: Any, *args):
            super().__init__(args)
            self.key_or_value = key_or_value

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

    class AmbiguousKeyValue(Exception):
        """
        Key is represented twice in the storage.
        """

        key_1: str | int
        val_1: str | int
        key_2: str | int
        val_2: str | int

        def __init__(
            self,
            key_1: str | int,
            val_1: str | int,
            key_2: str | int,
            val_2: str | int,
            *args,
        ):
            super().__init__(args)
            self.key_1 = key_1
            self.val_1 = val_1
            self.key_2 = key_2
            self.val_2 = val_2

        def __str__(self):
            """Print exception string"""
            return f"""
            Key is represented twice (due to negative numbers) with different
            values in storage:
            s[{self.key_1}] = {self.val_1} and s[{self.key_2}] = {self.val_2}
            """

    class MissingKey(Exception):
        """
        Test expected to find a storage key set but key was missing.
        """

        key: int

        def __init__(self, key: int, *args):
            super().__init__(args)
            self.key = key

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

    class KeyValueMismatch(Exception):
        """
        Test expected a certain value in a storage key but value found
        was different.
        """

        key: int
        want: int
        got: int

        def __init__(self, key: int, want: int, got: int, *args):
            super().__init__(args)
            self.key = key
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string"""
            return (
                "incorrect value for key {0}: want {1} (dec:{2})," + " got {3} (dec:{4})"
            ).format(
                Storage.key_value_to_string(self.key),
                Storage.key_value_to_string(self.want),
                self.want,
                Storage.key_value_to_string(self.got),
                self.got,
            )

    @staticmethod
    def parse_key_value(input: str | int | bytes) -> int:
        """
        Parses a key or value to a valid int key for storage.
        """
        if type(input) is str:
            input = int(input, 0)
        elif type(input) is int:
            pass
        elif type(input) is bytes:
            input = int.from_bytes(input, "big")
        else:
            raise Storage.InvalidType(input)

        if input > MAX_STORAGE_KEY_VALUE or input < MIN_STORAGE_KEY_VALUE:
            raise Storage.InvalidValue(input)
        return input

    @staticmethod
    def key_value_to_string(value: int) -> str:
        """
        Transforms a key or value into an hex string.
        """
        hex_str = value.to_bytes(32, "big", signed=(value < 0)).hex().lstrip("0")
        if hex_str == "":
            hex_str = "00"
        if len(hex_str) % 2 != 0:
            hex_str = "0" + hex_str
        return "0x" + hex_str

    def __init__(self, input: StorageDictType):
        """
        Initializes the storage using a given mapping which can have
        keys and values either as string or int.
        Strings must be valid decimal or hexadecimal (starting with 0x)
        numbers.
        """
        self.data = {}
        for key in input:
            value = Storage.parse_key_value(input[key])
            key = Storage.parse_key_value(key)
            self.data[key] = value
        pass

    def __len__(self) -> int:
        """Returns number of elements in the storage"""
        return len(self.data)

    def __contains__(self, key: str | int) -> bool:
        """Checks for an item in the storage"""
        key = Storage.parse_key_value(key)
        return key in self.data

    def __getitem__(self, key: str | int) -> int:
        """Returns an item from the storage"""
        key = Storage.parse_key_value(key)
        if key not in self.data:
            raise KeyError()
        return self.data[key]

    def __setitem__(self, key: str | int, value: str | int):  # noqa: SC200
        """Sets an item in the storage"""
        self.data[Storage.parse_key_value(key)] = Storage.parse_key_value(value)

    def __delitem__(self, key: str | int):
        """Deletes an item from the storage"""
        del self.data[Storage.parse_key_value(key)]

    def to_dict(self) -> Mapping[str, str]:
        """
        Converts the storage into a string dict with appropriate 32-byte
        hex string formatting.
        """
        res: Dict[str, str] = {}
        for key in self.data:
            key_repr = Storage.key_value_to_string(key)
            val_repr = Storage.key_value_to_string(self.data[key])
            if key_repr in res and val_repr != res[key_repr]:
                raise Storage.AmbiguousKeyValue(key_repr, res[key_repr], key, val_repr)
            res[key_repr] = val_repr
        return res

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

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

    def must_be_equal(self, other: "Storage"):
        """
        Succeeds only if "self" is equal to "other" storage.
        """
        # Test keys contained in both storage objects
        for key in self.data.keys() & other.data.keys():
            if self.data[key] != other.data[key]:
                raise Storage.KeyValueMismatch(key, self.data[key], other.data[key])

        # Test keys contained in either one of the storage objects
        for key in self.data.keys() ^ other.data.keys():
            if key in self.data:
                if self.data[key] != 0:
                    raise Storage.KeyValueMismatch(key, self.data[key], 0)

            elif other.data[key] != 0:
                raise Storage.KeyValueMismatch(key, 0, other.data[key])

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

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

InvalidType

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class InvalidType(Exception):
    """
    Invalid type used when describing test's expected storage key or value.
    """

    key_or_value: Any

    def __init__(self, key_or_value: Any, *args):
        super().__init__(args)
        self.key_or_value = key_or_value

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

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

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

InvalidValue

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class InvalidValue(Exception):
    """
    Invalid value used when describing test's expected storage key or
    value.
    """

    key_or_value: Any

    def __init__(self, key_or_value: Any, *args):
        super().__init__(args)
        self.key_or_value = key_or_value

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

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

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

AmbiguousKeyValue

Bases: Exception

Key is represented twice in the storage.

Source code in src/ethereum_test_tools/common/types.py

class AmbiguousKeyValue(Exception):
    """
    Key is represented twice in the storage.
    """

    key_1: str | int
    val_1: str | int
    key_2: str | int
    val_2: str | int

    def __init__(
        self,
        key_1: str | int,
        val_1: str | int,
        key_2: str | int,
        val_2: str | int,
        *args,
    ):
        super().__init__(args)
        self.key_1 = key_1
        self.val_1 = val_1
        self.key_2 = key_2
        self.val_2 = val_2

    def __str__(self):
        """Print exception string"""
        return f"""
        Key is represented twice (due to negative numbers) with different
        values in storage:
        s[{self.key_1}] = {self.val_1} and s[{self.key_2}] = {self.val_2}
        """

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

def __str__(self):
    """Print exception string"""
    return f"""
    Key is represented twice (due to negative numbers) with different
    values in storage:
    s[{self.key_1}] = {self.val_1} and s[{self.key_2}] = {self.val_2}
    """

MissingKey

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class MissingKey(Exception):
    """
    Test expected to find a storage key set but key was missing.
    """

    key: int

    def __init__(self, key: int, *args):
        super().__init__(args)
        self.key = key

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

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

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

KeyValueMismatch

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class KeyValueMismatch(Exception):
    """
    Test expected a certain value in a storage key but value found
    was different.
    """

    key: int
    want: int
    got: int

    def __init__(self, key: int, want: int, got: int, *args):
        super().__init__(args)
        self.key = key
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string"""
        return (
            "incorrect value for key {0}: want {1} (dec:{2})," + " got {3} (dec:{4})"
        ).format(
            Storage.key_value_to_string(self.key),
            Storage.key_value_to_string(self.want),
            self.want,
            Storage.key_value_to_string(self.got),
            self.got,
        )

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

def __str__(self):
    """Print exception string"""
    return (
        "incorrect value for key {0}: want {1} (dec:{2})," + " got {3} (dec:{4})"
    ).format(
        Storage.key_value_to_string(self.key),
        Storage.key_value_to_string(self.want),
        self.want,
        Storage.key_value_to_string(self.got),
        self.got,
    )

parse_key_value(input) staticmethod

Parses a key or value to a valid int key for storage.

Source code in src/ethereum_test_tools/common/types.py

@staticmethod
def parse_key_value(input: str | int | bytes) -> int:
    """
    Parses a key or value to a valid int key for storage.
    """
    if type(input) is str:
        input = int(input, 0)
    elif type(input) is int:
        pass
    elif type(input) is bytes:
        input = int.from_bytes(input, "big")
    else:
        raise Storage.InvalidType(input)

    if input > MAX_STORAGE_KEY_VALUE or input < MIN_STORAGE_KEY_VALUE:
        raise Storage.InvalidValue(input)
    return input

key_value_to_string(value) staticmethod

Transforms a key or value into an hex string.

Source code in src/ethereum_test_tools/common/types.py

@staticmethod
def key_value_to_string(value: int) -> str:
    """
    Transforms a key or value into an hex string.
    """
    hex_str = value.to_bytes(32, "big", signed=(value < 0)).hex().lstrip("0")
    if hex_str == "":
        hex_str = "00"
    if len(hex_str) % 2 != 0:
        hex_str = "0" + hex_str
    return "0x" + hex_str

__init__(input)

Initializes the storage using a given mapping which can have keys and values either as string or int. Strings must be valid decimal or hexadecimal (starting with 0x) numbers.

Source code in src/ethereum_test_tools/common/types.py

def __init__(self, input: StorageDictType):
    """
    Initializes the storage using a given mapping which can have
    keys and values either as string or int.
    Strings must be valid decimal or hexadecimal (starting with 0x)
    numbers.
    """
    self.data = {}
    for key in input:
        value = Storage.parse_key_value(input[key])
        key = Storage.parse_key_value(key)
        self.data[key] = value
    pass

__len__()

Returns number of elements in the storage

Source code in src/ethereum_test_tools/common/types.py

def __len__(self) -> int:
    """Returns number of elements in the storage"""
    return len(self.data)

__contains__(key)

Checks for an item in the storage

Source code in src/ethereum_test_tools/common/types.py

def __contains__(self, key: str | int) -> bool:
    """Checks for an item in the storage"""
    key = Storage.parse_key_value(key)
    return key in self.data

__getitem__(key)

Returns an item from the storage

Source code in src/ethereum_test_tools/common/types.py

def __getitem__(self, key: str | int) -> int:
    """Returns an item from the storage"""
    key = Storage.parse_key_value(key)
    if key not in self.data:
        raise KeyError()
    return self.data[key]

__setitem__(key, value)

Sets an item in the storage

Source code in src/ethereum_test_tools/common/types.py

def __setitem__(self, key: str | int, value: str | int):  # noqa: SC200
    """Sets an item in the storage"""
    self.data[Storage.parse_key_value(key)] = Storage.parse_key_value(value)

__delitem__(key)

Deletes an item from the storage

Source code in src/ethereum_test_tools/common/types.py

def __delitem__(self, key: str | int):
    """Deletes an item from the storage"""
    del self.data[Storage.parse_key_value(key)]

to_dict()

Converts the storage into a string dict with appropriate 32-byte hex string formatting.

Source code in src/ethereum_test_tools/common/types.py

def to_dict(self) -> Mapping[str, str]:
    """
    Converts the storage into a string dict with appropriate 32-byte
    hex string formatting.
    """
    res: Dict[str, str] = {}
    for key in self.data:
        key_repr = Storage.key_value_to_string(key)
        val_repr = Storage.key_value_to_string(self.data[key])
        if key_repr in res and val_repr != res[key_repr]:
            raise Storage.AmbiguousKeyValue(key_repr, res[key_repr], key, val_repr)
        res[key_repr] = val_repr
    return res

contains(other)

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

Source code in src/ethereum_test_tools/common/types.py

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

must_contain(other)

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

Source code in src/ethereum_test_tools/common/types.py

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

must_be_equal(other)

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

Source code in src/ethereum_test_tools/common/types.py

def must_be_equal(self, other: "Storage"):
    """
    Succeeds only if "self" is equal to "other" storage.
    """
    # Test keys contained in both storage objects
    for key in self.data.keys() & other.data.keys():
        if self.data[key] != other.data[key]:
            raise Storage.KeyValueMismatch(key, self.data[key], other.data[key])

    # Test keys contained in either one of the storage objects
    for key in self.data.keys() ^ other.data.keys():
        if key in self.data:
            if self.data[key] != 0:
                raise Storage.KeyValueMismatch(key, self.data[key], 0)

        elif other.data[key] != 0:
            raise Storage.KeyValueMismatch(key, 0, other.data[key])

eip_2028_transaction_data_cost(data)

Calculates the cost of a given data as part of a transaction, based on the costs specified in EIP-2028: https://eips.ethereum.org/EIPS/eip-2028

Source code in src/ethereum_test_tools/common/helpers.py

def eip_2028_transaction_data_cost(data: bytes | str) -> int:
    """
    Calculates the cost of a given data as part of a transaction, based on the
    costs specified in EIP-2028: https://eips.ethereum.org/EIPS/eip-2028
    """
    if type(data) is str:
        if data.startswith("0x"):
            data = data[2:]
        data = bytes.fromhex(data)
    cost = 0
    for b in data:
        if b == 0:
            cost += 4
        else:
            cost += 16
    return cost

to_address(input)

Converts an int or str into proper address 20-byte hex string.

Source code in src/ethereum_test_tools/common/helpers.py

def to_address(input: int | str) -> str:
    """
    Converts an int or str into proper address 20-byte hex string.
    """
    if type(input) is str:
        # Convert to int
        input = int(input, 0)
    if type(input) is int:
        return "0x" + input.to_bytes(20, "big").hex()
    raise Exception("invalid type to convert to account address")

CodeGasMeasure dataclass

Bases: Code

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

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

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

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

    def assemble(self) -> bytes:
        """
        Assemble the bytecode that measures gas usage.
        """
        res = bytes()
        res += bytes(
            [
                0x5A,  # GAS
            ]
        )
        res += code_to_bytes(self.code)  # Execute code to measure its gas cost
        res += bytes(
            [
                0x5A,  # GAS
            ]
        )
        # We need to swap and pop for each extra stack item that remained from
        # the execution of the code
        res += (
            bytes(
                [
                    0x90,  # SWAP1
                    0x50,  # POP
                ]
            )
            * self.extra_stack_items
        )
        res += bytes(
            [
                0x90,  # SWAP1
                0x03,  # SUB
                0x60,  # PUSH1
                self.overhead_cost + 2,  # Overhead cost + GAS opcode price
                0x90,  # SWAP1
                0x03,  # SUB
                0x60,  # PUSH1
                self.sstore_key,  # -> SSTORE key
                0x55,  # SSTORE
                0x00,  # STOP
            ]
        )
        return res

code: bytes | str | Code instance-attribute

Bytecode to be executed to measure the gas usage.

overhead_cost: int = 0 instance-attribute class-attribute

Extra gas cost to be subtracted from extra operations.

extra_stack_items: int = 0 instance-attribute class-attribute

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

sstore_key: int = 0 instance-attribute class-attribute

Storage key to save the gas used.

assemble()

Assemble the bytecode that measures gas usage.

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

def assemble(self) -> bytes:
    """
    Assemble the bytecode that measures gas usage.
    """
    res = bytes()
    res += bytes(
        [
            0x5A,  # GAS
        ]
    )
    res += code_to_bytes(self.code)  # Execute code to measure its gas cost
    res += bytes(
        [
            0x5A,  # GAS
        ]
    )
    # We need to swap and pop for each extra stack item that remained from
    # the execution of the code
    res += (
        bytes(
            [
                0x90,  # SWAP1
                0x50,  # POP
            ]
        )
        * self.extra_stack_items
    )
    res += bytes(
        [
            0x90,  # SWAP1
            0x03,  # SUB
            0x60,  # PUSH1
            self.overhead_cost + 2,  # Overhead cost + GAS opcode price
            0x90,  # SWAP1
            0x03,  # SUB
            0x60,  # PUSH1
            self.sstore_key,  # -> SSTORE key
            0x55,  # SSTORE
            0x00,  # STOP
        ]
    )
    return res

to_hash_bytes(input)

Converts an int or str into proper 32-byte hash.

Source code in src/ethereum_test_tools/common/helpers.py

def to_hash_bytes(input: int | str) -> bytes:
    """
    Converts an int or str into proper 32-byte hash.
    """
    if type(input) is str:
        # Convert to int
        input = int(input, 0)
    if type(input) is int:
        return input.to_bytes(32, "big")
    raise Exception("invalid type to convert to hash")

to_hash(input)

Converts an int or str into proper 32-byte hash hex string.

Source code in src/ethereum_test_tools/common/helpers.py

def to_hash(input: int | str) -> str:
    """
    Converts an int or str into proper 32-byte hash hex string.
    """
    return "0x" + to_hash_bytes(input).hex()

add_kzg_version(b_hashes, kzg_version)

Adds the Kzg Version to each blob hash.

Source code in src/ethereum_test_tools/common/helpers.py

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

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

Opcodes

Bases: Opcode, Enum

Enum containing all known opcodes.

Contains deprecated and not yet implemented opcodes.

This enum is !! NOT !! meant to be iterated over by the tests. Instead, create a list with cherry-picked opcodes from this Enum within the test if iteration is needed.

Do !! NOT !! remove or modify existing opcodes from this list.

Source code in src/ethereum_test_tools/vm/opcode.py

class Opcodes(Opcode, Enum):
    """
    Enum containing all known opcodes.

    Contains deprecated and not yet implemented opcodes.

    This enum is !! NOT !! meant to be iterated over by the tests. Instead,
    create a list with cherry-picked opcodes from this Enum within the test
    if iteration is needed.

    Do !! NOT !! remove or modify existing opcodes from this list.
    """

    STOP = Opcode(0x00)
    ADD = Opcode(0x01, popped_stack_items=2, pushed_stack_items=1)
    MUL = Opcode(0x02, popped_stack_items=2, pushed_stack_items=1)
    SUB = Opcode(0x03, popped_stack_items=2, pushed_stack_items=1)
    DIV = Opcode(0x04, popped_stack_items=2, pushed_stack_items=1)
    SDIV = Opcode(0x05, popped_stack_items=2, pushed_stack_items=1)
    MOD = Opcode(0x06, popped_stack_items=2, pushed_stack_items=1)
    SMOD = Opcode(0x07, popped_stack_items=2, pushed_stack_items=1)
    ADDMOD = Opcode(0x08, popped_stack_items=3, pushed_stack_items=1)
    MULMOD = Opcode(0x09, popped_stack_items=3, pushed_stack_items=1)
    EXP = Opcode(0x0A, popped_stack_items=2, pushed_stack_items=1)
    SIGNEXTEND = Opcode(0x0B, popped_stack_items=2, pushed_stack_items=1)

    LT = Opcode(0x10, popped_stack_items=2, pushed_stack_items=1)
    GT = Opcode(0x11, popped_stack_items=2, pushed_stack_items=1)
    SLT = Opcode(0x12, popped_stack_items=2, pushed_stack_items=1)
    SGT = Opcode(0x13, popped_stack_items=2, pushed_stack_items=1)
    EQ = Opcode(0x14, popped_stack_items=2, pushed_stack_items=1)
    ISZERO = Opcode(0x15, popped_stack_items=1, pushed_stack_items=1)
    AND = Opcode(0x16, popped_stack_items=2, pushed_stack_items=1)
    OR = Opcode(0x17, popped_stack_items=2, pushed_stack_items=1)
    XOR = Opcode(0x18, popped_stack_items=2, pushed_stack_items=1)
    NOT = Opcode(0x19, popped_stack_items=1, pushed_stack_items=1)
    BYTE = Opcode(0x1A, popped_stack_items=2, pushed_stack_items=1)
    SHL = Opcode(0x1B, popped_stack_items=2, pushed_stack_items=1)
    SHR = Opcode(0x1C, popped_stack_items=2, pushed_stack_items=1)
    SAR = Opcode(0x1D, popped_stack_items=2, pushed_stack_items=1)

    SHA3 = Opcode(0x20, popped_stack_items=2, pushed_stack_items=1)

    ADDRESS = Opcode(0x30, pushed_stack_items=1)
    BALANCE = Opcode(0x31, popped_stack_items=1, pushed_stack_items=1)
    ORIGIN = Opcode(0x32, pushed_stack_items=1)
    CALLER = Opcode(0x33, pushed_stack_items=1)
    CALLVALUE = Opcode(0x34, pushed_stack_items=1)
    CALLDATALOAD = Opcode(0x35, popped_stack_items=1, pushed_stack_items=1)
    CALLDATASIZE = Opcode(0x36, pushed_stack_items=1)
    CALLDATACOPY = Opcode(0x37, popped_stack_items=3)
    CODESIZE = Opcode(0x38, pushed_stack_items=1)
    CODECOPY = Opcode(0x39, popped_stack_items=3)
    GASPRICE = Opcode(0x3A, pushed_stack_items=1)
    EXTCODESIZE = Opcode(0x3B, popped_stack_items=1, pushed_stack_items=1)
    EXTCODECOPY = Opcode(0x3C, popped_stack_items=4)
    RETURNDATASIZE = Opcode(0x3D, pushed_stack_items=1)
    RETURNDATACOPY = Opcode(0x3E, popped_stack_items=3)
    EXTCODEHASH = Opcode(0x3F, popped_stack_items=1, pushed_stack_items=1)

    BLOCKHASH = Opcode(0x40, popped_stack_items=1, pushed_stack_items=1)
    COINBASE = Opcode(0x41, pushed_stack_items=1)
    TIMESTAMP = Opcode(0x42, pushed_stack_items=1)
    NUMBER = Opcode(0x43, pushed_stack_items=1)
    PREVRANDAO = Opcode(0x44, pushed_stack_items=1)
    GASLIMIT = Opcode(0x45, pushed_stack_items=1)
    CHAINID = Opcode(0x46, pushed_stack_items=1)
    SELFBALANCE = Opcode(0x47, pushed_stack_items=1)
    BASEFEE = Opcode(0x48, pushed_stack_items=1)
    BLOBHASH = Opcode(0x49, popped_stack_items=1, pushed_stack_items=1)

    POP = Opcode(0x50, popped_stack_items=1)
    MLOAD = Opcode(0x51, popped_stack_items=1, pushed_stack_items=1)
    MSTORE = Opcode(0x52, popped_stack_items=2)
    MSTORE8 = Opcode(0x53, popped_stack_items=2)
    SLOAD = Opcode(0x54, popped_stack_items=1, pushed_stack_items=1)
    SSTORE = Opcode(0x55, popped_stack_items=2)
    JUMP = Opcode(0x56, popped_stack_items=1)
    JUMPI = Opcode(0x57, popped_stack_items=2)
    PC = Opcode(0x58, pushed_stack_items=1)
    MSIZE = Opcode(0x59, pushed_stack_items=1)
    GAS = Opcode(0x5A, pushed_stack_items=1)
    JUMPDEST = Opcode(0x5B)
    RJUMP = Opcode(0x5C, data_portion_length=2)
    RJUMPI = Opcode(0x5D, popped_stack_items=1, data_portion_length=2)
    CALLF = Opcode(0x5E, data_portion_length=2)
    RETF = Opcode(0x49)

    PUSH0 = Opcode(0x5F, pushed_stack_items=1)
    PUSH1 = Opcode(0x60, pushed_stack_items=1, data_portion_length=1)
    PUSH2 = Opcode(0x61, pushed_stack_items=1, data_portion_length=2)
    PUSH3 = Opcode(0x62, pushed_stack_items=1, data_portion_length=3)
    PUSH4 = Opcode(0x63, pushed_stack_items=1, data_portion_length=4)
    PUSH5 = Opcode(0x64, pushed_stack_items=1, data_portion_length=5)
    PUSH6 = Opcode(0x65, pushed_stack_items=1, data_portion_length=6)
    PUSH7 = Opcode(0x66, pushed_stack_items=1, data_portion_length=7)
    PUSH8 = Opcode(0x67, pushed_stack_items=1, data_portion_length=8)
    PUSH9 = Opcode(0x68, pushed_stack_items=1, data_portion_length=9)
    PUSH10 = Opcode(0x69, pushed_stack_items=1, data_portion_length=10)
    PUSH11 = Opcode(0x6A, pushed_stack_items=1, data_portion_length=11)
    PUSH12 = Opcode(0x6B, pushed_stack_items=1, data_portion_length=12)
    PUSH13 = Opcode(0x6C, pushed_stack_items=1, data_portion_length=13)
    PUSH14 = Opcode(0x6D, pushed_stack_items=1, data_portion_length=14)
    PUSH15 = Opcode(0x6E, pushed_stack_items=1, data_portion_length=15)
    PUSH16 = Opcode(0x6F, pushed_stack_items=1, data_portion_length=16)
    PUSH17 = Opcode(0x70, pushed_stack_items=1, data_portion_length=17)
    PUSH18 = Opcode(0x71, pushed_stack_items=1, data_portion_length=18)
    PUSH19 = Opcode(0x72, pushed_stack_items=1, data_portion_length=19)
    PUSH20 = Opcode(0x73, pushed_stack_items=1, data_portion_length=20)
    PUSH21 = Opcode(0x74, pushed_stack_items=1, data_portion_length=21)
    PUSH22 = Opcode(0x75, pushed_stack_items=1, data_portion_length=22)
    PUSH23 = Opcode(0x76, pushed_stack_items=1, data_portion_length=23)
    PUSH24 = Opcode(0x77, pushed_stack_items=1, data_portion_length=24)
    PUSH25 = Opcode(0x78, pushed_stack_items=1, data_portion_length=25)
    PUSH26 = Opcode(0x79, pushed_stack_items=1, data_portion_length=26)
    PUSH27 = Opcode(0x7A, pushed_stack_items=1, data_portion_length=27)
    PUSH28 = Opcode(0x7B, pushed_stack_items=1, data_portion_length=28)
    PUSH29 = Opcode(0x7C, pushed_stack_items=1, data_portion_length=29)
    PUSH30 = Opcode(0x7D, pushed_stack_items=1, data_portion_length=30)
    PUSH31 = Opcode(0x7E, pushed_stack_items=1, data_portion_length=31)
    PUSH32 = Opcode(0x7F, pushed_stack_items=1, data_portion_length=32)

    DUP1 = Opcode(0x80, pushed_stack_items=1, min_stack_height=1)
    DUP2 = Opcode(0x81, pushed_stack_items=1, min_stack_height=2)
    DUP3 = Opcode(0x82, pushed_stack_items=1, min_stack_height=3)
    DUP4 = Opcode(0x83, pushed_stack_items=1, min_stack_height=4)
    DUP5 = Opcode(0x84, pushed_stack_items=1, min_stack_height=5)
    DUP6 = Opcode(0x85, pushed_stack_items=1, min_stack_height=6)
    DUP7 = Opcode(0x86, pushed_stack_items=1, min_stack_height=7)
    DUP8 = Opcode(0x87, pushed_stack_items=1, min_stack_height=8)
    DUP9 = Opcode(0x88, pushed_stack_items=1, min_stack_height=9)
    DUP10 = Opcode(0x89, pushed_stack_items=1, min_stack_height=10)
    DUP11 = Opcode(0x8A, pushed_stack_items=1, min_stack_height=11)
    DUP12 = Opcode(0x8B, pushed_stack_items=1, min_stack_height=12)
    DUP13 = Opcode(0x8C, pushed_stack_items=1, min_stack_height=13)
    DUP14 = Opcode(0x8D, pushed_stack_items=1, min_stack_height=14)
    DUP15 = Opcode(0x8E, pushed_stack_items=1, min_stack_height=15)
    DUP16 = Opcode(0x8F, pushed_stack_items=1, min_stack_height=16)

    SWAP1 = Opcode(0x90, min_stack_height=2)
    SWAP2 = Opcode(0x91, min_stack_height=3)
    SWAP3 = Opcode(0x92, min_stack_height=4)
    SWAP4 = Opcode(0x93, min_stack_height=5)
    SWAP5 = Opcode(0x94, min_stack_height=6)
    SWAP6 = Opcode(0x95, min_stack_height=7)
    SWAP7 = Opcode(0x96, min_stack_height=8)
    SWAP8 = Opcode(0x97, min_stack_height=9)
    SWAP9 = Opcode(0x98, min_stack_height=10)
    SWAP10 = Opcode(0x99, min_stack_height=11)
    SWAP11 = Opcode(0x9A, min_stack_height=12)
    SWAP12 = Opcode(0x9B, min_stack_height=13)
    SWAP13 = Opcode(0x9C, min_stack_height=14)
    SWAP14 = Opcode(0x9D, min_stack_height=15)
    SWAP15 = Opcode(0x9E, min_stack_height=16)
    SWAP16 = Opcode(0x9F, min_stack_height=17)

    LOG0 = Opcode(0xA0, popped_stack_items=2)
    LOG1 = Opcode(0xA1, popped_stack_items=3)
    LOG2 = Opcode(0xA2, popped_stack_items=4)
    LOG3 = Opcode(0xA3, popped_stack_items=5)
    LOG4 = Opcode(0xA4, popped_stack_items=6)

    TLOAD = Opcode(0xB3, popped_stack_items=1, pushed_stack_items=1)
    TSTORE = Opcode(0xB4, popped_stack_items=2)

    CREATE = Opcode(0xF0, popped_stack_items=3, pushed_stack_items=1)
    CALL = Opcode(0xF1, popped_stack_items=7, pushed_stack_items=1)
    CALLCODE = Opcode(0xF2, popped_stack_items=7, pushed_stack_items=1)
    RETURN = Opcode(0xF3, popped_stack_items=2)
    DELEGATECALL = Opcode(0xF4, popped_stack_items=6, pushed_stack_items=1)
    CREATE2 = Opcode(0xF5, popped_stack_items=4, pushed_stack_items=1)

    STATICCALL = Opcode(0xFA, popped_stack_items=6, pushed_stack_items=1)

    REVERT = Opcode(0xFD, popped_stack_items=2)
    INVALID = Opcode(0xFE)

    SELFDESTRUCT = Opcode(0xFF, popped_stack_items=1)
    SENDALL = Opcode(0xFF, popped_stack_items=1)

Account

State associated with an address.

Source code in src/ethereum_test_tools/common/types.py

class Account:
    """
    State associated with an address.
    """

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

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

    class NonceMismatch(Exception):
        """
        Test expected a certain nonce value for an account but a different
        value was found.
        """

        address: str
        want: int | None
        got: int | None

        def __init__(self, address: str, want: int | None, got: int | None, *args):
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string"""
            return (
                f"unexpected nonce for account {self.address}: "
                + f"want {self.want}, got {self.got}"
            )

    class BalanceMismatch(Exception):
        """
        Test expected a certain balance for an account but a different
        value was found.
        """

        address: str
        want: int | None
        got: int | None

        def __init__(self, address: str, want: int | None, got: int | None, *args):
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string"""
            return (
                f"unexpected balance for account {self.address}: "
                + f"want {self.want}, got {self.got}"
            )

    class CodeMismatch(Exception):
        """
        Test expected a certain bytecode for an account but a different
        one was found.
        """

        address: str
        want: str | None
        got: str | None

        def __init__(self, address: str, want: str | None, got: str | None, *args):
            super().__init__(args)
            self.address = address
            self.want = want
            self.got = got

        def __str__(self):
            """Print exception string"""
            return (
                f"unexpected code for account {self.address}: "
                + f"want {self.want}, got {self.got}"
            )

    def __init__(
        self,
        *,
        nonce: int | None = None,
        balance: int | None = None,
        code: str | bytes | Code | None = None,
        storage: Storage | Dict[str | int | bytes, str | int | bytes] | None = None,
    ) -> None:
        """Init account members"""
        self.nonce = nonce
        self.balance = balance
        self.code = code
        if storage is not None and type(storage) is dict:
            self.storage = Storage(storage)

    def check_alloc(self: "Account", address: str, alloc: dict):
        """
        Checks the returned alloc against an expected account in post state.
        Raises exception on failure.
        """
        if self.nonce is not None:
            actual_nonce = int_or_none(alloc.get("nonce"), 0)
            if self.nonce != actual_nonce:
                raise Account.NonceMismatch(
                    address=address,
                    want=self.nonce,
                    got=actual_nonce,
                )

        if self.balance is not None:
            actual_balance = int_or_none(alloc.get("balance"), 0)
            if self.balance != actual_balance:
                raise Account.BalanceMismatch(
                    address=address,
                    want=self.balance,
                    got=actual_balance,
                )

        if self.code is not None:
            expected_code = code_to_hex(self.code)
            actual_code = str_or_none(alloc.get("code"), "0x")
            if expected_code != actual_code:
                raise Account.CodeMismatch(
                    address=address,
                    want=expected_code,
                    got=actual_code,
                )

        if self.storage is not None:
            expected_storage = (
                self.storage if isinstance(self.storage, Storage) else Storage(self.storage)
            )
            actual_storage = Storage(alloc["storage"]) if "storage" in alloc else Storage({})
            expected_storage.must_be_equal(actual_storage)

    @classmethod
    def with_code(cls: Type, code: bytes | str | Code) -> "Account":
        """
        Create account with provided `code` and nonce of `1`.
        """
        return Account(nonce=1, code=code)

storage: Storage | None = None instance-attribute class-attribute

Storage within a contract.

NONEXISTENT: object = object() class-attribute

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

NonceMismatch

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class NonceMismatch(Exception):
    """
    Test expected a certain nonce value for an account but a different
    value was found.
    """

    address: str
    want: int | None
    got: int | None

    def __init__(self, address: str, want: int | None, got: int | None, *args):
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string"""
        return (
            f"unexpected nonce for account {self.address}: "
            + f"want {self.want}, got {self.got}"
        )

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

def __str__(self):
    """Print exception string"""
    return (
        f"unexpected nonce for account {self.address}: "
        + f"want {self.want}, got {self.got}"
    )

BalanceMismatch

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class BalanceMismatch(Exception):
    """
    Test expected a certain balance for an account but a different
    value was found.
    """

    address: str
    want: int | None
    got: int | None

    def __init__(self, address: str, want: int | None, got: int | None, *args):
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string"""
        return (
            f"unexpected balance for account {self.address}: "
            + f"want {self.want}, got {self.got}"
        )

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

def __str__(self):
    """Print exception string"""
    return (
        f"unexpected balance for account {self.address}: "
        + f"want {self.want}, got {self.got}"
    )

CodeMismatch

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

class CodeMismatch(Exception):
    """
    Test expected a certain bytecode for an account but a different
    one was found.
    """

    address: str
    want: str | None
    got: str | None

    def __init__(self, address: str, want: str | None, got: str | None, *args):
        super().__init__(args)
        self.address = address
        self.want = want
        self.got = got

    def __str__(self):
        """Print exception string"""
        return (
            f"unexpected code for account {self.address}: "
            + f"want {self.want}, got {self.got}"
        )

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

def __str__(self):
    """Print exception string"""
    return (
        f"unexpected code for account {self.address}: "
        + f"want {self.want}, got {self.got}"
    )

__init__(*, nonce=None, balance=None, code=None, storage=None)

Init account members

Source code in src/ethereum_test_tools/common/types.py

def __init__(
    self,
    *,
    nonce: int | None = None,
    balance: int | None = None,
    code: str | bytes | Code | None = None,
    storage: Storage | Dict[str | int | bytes, str | int | bytes] | None = None,
) -> None:
    """Init account members"""
    self.nonce = nonce
    self.balance = balance
    self.code = code
    if storage is not None and type(storage) is dict:
        self.storage = Storage(storage)

nonce: int | None = nonce instance-attribute class-attribute

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

balance: int | None = balance instance-attribute class-attribute

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

code: str | bytes | Code | None = code instance-attribute class-attribute

Bytecode contained by the account.

check_alloc(address, alloc)

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

Source code in src/ethereum_test_tools/common/types.py

def check_alloc(self: "Account", address: str, alloc: dict):
    """
    Checks the returned alloc against an expected account in post state.
    Raises exception on failure.
    """
    if self.nonce is not None:
        actual_nonce = int_or_none(alloc.get("nonce"), 0)
        if self.nonce != actual_nonce:
            raise Account.NonceMismatch(
                address=address,
                want=self.nonce,
                got=actual_nonce,
            )

    if self.balance is not None:
        actual_balance = int_or_none(alloc.get("balance"), 0)
        if self.balance != actual_balance:
            raise Account.BalanceMismatch(
                address=address,
                want=self.balance,
                got=actual_balance,
            )

    if self.code is not None:
        expected_code = code_to_hex(self.code)
        actual_code = str_or_none(alloc.get("code"), "0x")
        if expected_code != actual_code:
            raise Account.CodeMismatch(
                address=address,
                want=expected_code,
                got=actual_code,
            )

    if self.storage is not None:
        expected_storage = (
            self.storage if isinstance(self.storage, Storage) else Storage(self.storage)
        )
        actual_storage = Storage(alloc["storage"]) if "storage" in alloc else Storage({})
        expected_storage.must_be_equal(actual_storage)

with_code(code) classmethod

Create account with provided code and nonce of 1.

Source code in src/ethereum_test_tools/common/types.py

@classmethod
def with_code(cls: Type, code: bytes | str | Code) -> "Account":
    """
    Create account with provided `code` and nonce of `1`.
    """
    return Account(nonce=1, code=code)

Withdrawal dataclass

Structure to represent a single withdrawal of a validator's balance from the beacon chain.

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Withdrawal:
    """
    Structure to represent a single withdrawal of a validator's balance from
    the beacon chain.
    """

    index: int
    validator: int
    address: str
    amount: int

    def to_serializable_list(self) -> List[Any]:
        """
        Returns a list of the withdrawal's attributes in the order they should
        be serialized.
        """
        return [
            Uint(self.index),
            Uint(self.validator),
            bytes.fromhex(self.address[2:]),
            Uint(self.amount),
        ]

to_serializable_list()

Returns a list of the withdrawal's attributes in the order they should be serialized.

Source code in src/ethereum_test_tools/common/types.py

def to_serializable_list(self) -> List[Any]:
    """
    Returns a list of the withdrawal's attributes in the order they should
    be serialized.
    """
    return [
        Uint(self.index),
        Uint(self.validator),
        bytes.fromhex(self.address[2:]),
        Uint(self.amount),
    ]

Environment dataclass

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

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Environment:
    """
    Structure used to keep track of the context in which a block
    must be executed.
    """

    coinbase: str | bytes = "0x2adc25665018aa1fe0e6bc666dac8fc2697ff9ba"
    gas_limit: int = 100000000000000000
    number: int = 1
    timestamp: int = 1000
    difficulty: Optional[int] = None
    prev_randao: Optional[int] = None
    block_hashes: Dict[int, bytes] = field(default_factory=dict)
    base_fee: Optional[int] = None
    parent_difficulty: Optional[int] = None
    parent_timestamp: Optional[int] = None
    parent_base_fee: Optional[int] = None
    parent_gas_used: Optional[int] = None
    parent_gas_limit: Optional[int] = None
    parent_ommers_hash: Optional[str | bytes] = None
    withdrawals: Optional[List[Withdrawal]] = None
    parent_data_gas_used: Optional[int] = None
    parent_excess_data_gas: Optional[int] = None
    excess_data_gas: Optional[int] = None
    data_gas_used: Optional[int] = None

    @staticmethod
    def from_parent_header(parent: "FixtureHeader") -> "Environment":
        """
        Instantiates a new environment with the provided header as parent.
        """
        return Environment(
            parent_difficulty=parent.difficulty,
            parent_timestamp=parent.timestamp,
            parent_base_fee=parent.base_fee,
            parent_data_gas_used=parent.data_gas_used,
            parent_excess_data_gas=parent.excess_data_gas,
            parent_gas_used=parent.gas_used,
            parent_gas_limit=parent.gas_limit,
            parent_ommers_hash=parent.ommers_hash,
            block_hashes={
                parent.number: parent.hash if parent.hash is not None else bytes([0] * 32)
            },
        )

    def parent_hash(self) -> bytes:
        """
        Obtjains the latest hash according to the highest block number in
        `block_hashes`.
        """
        if len(self.block_hashes) == 0:
            return bytes([0] * 32)

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

    def apply_new_parent(self, new_parent: "FixtureHeader") -> "Environment":
        """
        Applies a header as parent to a copy of this environment.
        """
        env = copy(self)
        env.parent_difficulty = new_parent.difficulty
        env.parent_timestamp = new_parent.timestamp
        env.parent_base_fee = new_parent.base_fee
        env.parent_data_gas_used = new_parent.data_gas_used
        env.parent_excess_data_gas = new_parent.excess_data_gas
        env.parent_gas_used = new_parent.gas_used
        env.parent_gas_limit = new_parent.gas_limit
        env.parent_ommers_hash = new_parent.ommers_hash
        env.block_hashes[new_parent.number] = (
            new_parent.hash if new_parent.hash is not None else bytes([0] * 32)
        )
        return env

    def set_fork_requirements(self, fork: Fork) -> "Environment":
        """
        Fills the required fields in an environment depending on the fork.
        """
        res = copy(self)

        if (
            fork.header_prev_randao_required(self.number, self.timestamp)
            and res.prev_randao is None
        ):
            res.prev_randao = 0

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

        if (
            fork.header_base_fee_required(self.number, self.timestamp)
            and res.base_fee is None
            and res.parent_base_fee is None
        ):
            res.base_fee = DEFAULT_BASE_FEE

        if fork.header_zero_difficulty_required(self.number, self.timestamp):
            res.difficulty = 0

        if (
            fork.header_excess_data_gas_required(self.number, self.timestamp)
            and res.excess_data_gas is None
            and res.parent_excess_data_gas is None
        ):
            res.excess_data_gas = 0

        if (
            fork.header_data_gas_used_required(self.number, self.timestamp)
            and res.data_gas_used is None
            and res.parent_data_gas_used is None
        ):
            res.data_gas_used = 0

        return res

from_parent_header(parent) staticmethod

Instantiates a new environment with the provided header as parent.

Source code in src/ethereum_test_tools/common/types.py

@staticmethod
def from_parent_header(parent: "FixtureHeader") -> "Environment":
    """
    Instantiates a new environment with the provided header as parent.
    """
    return Environment(
        parent_difficulty=parent.difficulty,
        parent_timestamp=parent.timestamp,
        parent_base_fee=parent.base_fee,
        parent_data_gas_used=parent.data_gas_used,
        parent_excess_data_gas=parent.excess_data_gas,
        parent_gas_used=parent.gas_used,
        parent_gas_limit=parent.gas_limit,
        parent_ommers_hash=parent.ommers_hash,
        block_hashes={
            parent.number: parent.hash if parent.hash is not None else bytes([0] * 32)
        },
    )

parent_hash()

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

Source code in src/ethereum_test_tools/common/types.py

def parent_hash(self) -> bytes:
    """
    Obtjains the latest hash according to the highest block number in
    `block_hashes`.
    """
    if len(self.block_hashes) == 0:
        return bytes([0] * 32)

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

apply_new_parent(new_parent)

Applies a header as parent to a copy of this environment.

Source code in src/ethereum_test_tools/common/types.py

def apply_new_parent(self, new_parent: "FixtureHeader") -> "Environment":
    """
    Applies a header as parent to a copy of this environment.
    """
    env = copy(self)
    env.parent_difficulty = new_parent.difficulty
    env.parent_timestamp = new_parent.timestamp
    env.parent_base_fee = new_parent.base_fee
    env.parent_data_gas_used = new_parent.data_gas_used
    env.parent_excess_data_gas = new_parent.excess_data_gas
    env.parent_gas_used = new_parent.gas_used
    env.parent_gas_limit = new_parent.gas_limit
    env.parent_ommers_hash = new_parent.ommers_hash
    env.block_hashes[new_parent.number] = (
        new_parent.hash if new_parent.hash is not None else bytes([0] * 32)
    )
    return env

set_fork_requirements(fork)

Fills the required fields in an environment depending on the fork.

Source code in src/ethereum_test_tools/common/types.py

def set_fork_requirements(self, fork: Fork) -> "Environment":
    """
    Fills the required fields in an environment depending on the fork.
    """
    res = copy(self)

    if (
        fork.header_prev_randao_required(self.number, self.timestamp)
        and res.prev_randao is None
    ):
        res.prev_randao = 0

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

    if (
        fork.header_base_fee_required(self.number, self.timestamp)
        and res.base_fee is None
        and res.parent_base_fee is None
    ):
        res.base_fee = DEFAULT_BASE_FEE

    if fork.header_zero_difficulty_required(self.number, self.timestamp):
        res.difficulty = 0

    if (
        fork.header_excess_data_gas_required(self.number, self.timestamp)
        and res.excess_data_gas is None
        and res.parent_excess_data_gas is None
    ):
        res.excess_data_gas = 0

    if (
        fork.header_data_gas_used_required(self.number, self.timestamp)
        and res.data_gas_used is None
        and res.parent_data_gas_used is None
    ):
        res.data_gas_used = 0

    return res

AccessList

Access List for transactions.

Source code in src/ethereum_test_tools/common/types.py

class AccessList:
    """
    Access List for transactions.
    """

    address: bytes
    storage_keys: List[bytes]

    def __init__(
        self,
        *,
        address: str | int | bytes,
        storage_keys: List[str | int | bytes],
    ) -> None:
        """
        Ensures the access list has the correct byte length for each field.
        """
        self.address = address_to_bytes(address)
        self.storage_keys = [hash_to_bytes(key) for key in storage_keys]

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

__init__(*, address, storage_keys)

Ensures the access list has the correct byte length for each field.

Source code in src/ethereum_test_tools/common/types.py

def __init__(
    self,
    *,
    address: str | int | bytes,
    storage_keys: List[str | int | bytes],
) -> None:
    """
    Ensures the access list has the correct byte length for each field.
    """
    self.address = address_to_bytes(address)
    self.storage_keys = [hash_to_bytes(key) for key in storage_keys]

to_list()

Returns the access list as a list of serializable elements.

Source code in src/ethereum_test_tools/common/types.py

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

Transaction dataclass

Generic object that can represent all Ethereum transaction types.

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Transaction:
    """
    Generic object that can represent all Ethereum transaction types.
    """

    ty: Optional[int] = None
    """
    Transaction type value.
    """
    chain_id: int = 1
    nonce: int = 0
    to: Optional[str | int] = AddrAA
    value: int = 0
    data: bytes | str | Code = bytes()
    gas_limit: int = 21000
    access_list: Optional[List[AccessList]] = None

    gas_price: Optional[int] = None
    max_fee_per_gas: Optional[int] = None
    max_priority_fee_per_gas: Optional[int] = None

    max_fee_per_data_gas: Optional[int] = None
    blob_versioned_hashes: Optional[Sequence[str | bytes]] = None

    wrapped_blob_transaction: bool = False
    blobs: Optional[Sequence[bytes]] = None
    blob_kzg_commitments: Optional[Sequence[bytes]] = None
    blob_kzg_proofs: Optional[Sequence[bytes]] = None

    signature: Optional[Tuple[int, int, int]] = None
    secret_key: Optional[str] = None
    sender: Optional[str | bytes] = None
    protected: bool = True
    error: Optional[str] = None

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

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

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

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

    def __post_init__(self) -> None:
        """
        Ensures the transaction has no conflicting properties.
        """
        if (
            self.gas_price is not None
            and self.max_fee_per_gas is not None
            and self.max_priority_fee_per_gas is not None
        ):
            raise Transaction.InvalidFeePayment()

        if (
            self.gas_price is None
            and self.max_fee_per_gas is None
            and self.max_priority_fee_per_gas is None
        ):
            self.gas_price = 10

        if self.signature is not None and self.secret_key is not None:
            raise Transaction.InvalidSignaturePrivateKey()

        if self.signature is None and self.secret_key is None:
            self.secret_key = TestPrivateKey

        if self.ty is None:
            # Try to deduce transaction type from included fields
            if self.max_fee_per_data_gas is not None:
                self.ty = 3
            elif self.max_fee_per_gas is not None:
                self.ty = 2
            elif self.access_list is not None:
                self.ty = 1
            else:
                self.ty = 0

    def with_error(self, error: str) -> "Transaction":
        """
        Create a copy of the transaction with an added error.
        """
        tx = copy(self)
        tx.error = error
        return tx

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

    def with_fields(self, **kwargs) -> "Transaction":
        """
        Create a deepcopy of the transaction with modified fields.
        """
        tx = deepcopy(self)
        for key, value in kwargs.items():
            if hasattr(tx, key):
                setattr(tx, key, value)
            else:
                raise ValueError(f"Invalid field '{key}' for Transaction")
        return tx

    def payload_body(self) -> List[Any]:
        """
        Returns the list of values included in the transaction body.
        """
        if self.signature is None:
            raise ValueError("signature must be set before serializing any tx type")

        if self.gas_limit is None:
            raise ValueError("gas_limit must be set for all tx types")
        to = address_to_bytes(self.to)

        if self.ty == 3:
            # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
            if self.max_priority_fee_per_gas is None:
                raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_data_gas is None:
                raise ValueError("max_fee_per_data_gas must be set for type 3 tx")
            if self.blob_versioned_hashes is None:
                raise ValueError("blob_versioned_hashes must be set for type 3 tx")

            if self.wrapped_blob_transaction:
                if self.blobs is None:
                    raise ValueError("blobs must be set for network version of type 3 tx")
                if self.blob_kzg_commitments is None:
                    raise ValueError(
                        "blob_kzg_commitments must be set for network version of type 3 tx"
                    )
                if self.blob_kzg_proofs is None:
                    raise ValueError(
                        "blob_kzg_proofs must be set for network version of type 3 tx"
                    )

                return [
                    [
                        Uint(self.chain_id),
                        Uint(self.nonce),
                        Uint(self.max_priority_fee_per_gas),
                        Uint(self.max_fee_per_gas),
                        Uint(self.gas_limit),
                        to,
                        Uint(self.value),
                        code_to_bytes(self.data),
                        [a.to_list() for a in self.access_list]
                        if self.access_list is not None
                        else [],
                        Uint(self.max_fee_per_data_gas),
                        [hash_to_bytes(h) for h in self.blob_versioned_hashes],
                        Uint(self.signature[0]),
                        Uint(self.signature[1]),
                        Uint(self.signature[2]),
                    ],
                    self.blobs,
                    self.blob_kzg_commitments,
                    self.blob_kzg_proofs,
                ]
            else:
                return [
                    Uint(self.chain_id),
                    Uint(self.nonce),
                    Uint(self.max_priority_fee_per_gas),
                    Uint(self.max_fee_per_gas),
                    Uint(self.gas_limit),
                    to,
                    Uint(self.value),
                    code_to_bytes(self.data),
                    [a.to_list() for a in self.access_list]
                    if self.access_list is not None
                    else [],
                    Uint(self.max_fee_per_data_gas),
                    [hash_to_bytes(h) for h in self.blob_versioned_hashes],
                    Uint(self.signature[0]),
                    Uint(self.signature[1]),
                    Uint(self.signature[2]),
                ]
        elif self.ty == 2:
            # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
            if self.max_priority_fee_per_gas is None:
                raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 3 tx")
            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.max_priority_fee_per_gas),
                Uint(self.max_fee_per_gas),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list] if self.access_list is not None else [],
                Uint(self.signature[0]),
                Uint(self.signature[1]),
                Uint(self.signature[2]),
            ]
        elif self.ty == 1:
            # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
            if self.gas_price is None:
                raise ValueError("gas_price must be set for type 1 tx")

            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.gas_price),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list] if self.access_list is not None else [],
                Uint(self.signature[0]),
                Uint(self.signature[1]),
                Uint(self.signature[2]),
            ]
        elif self.ty == 0:
            if self.gas_price is None:
                raise ValueError("gas_price must be set for type 0 tx")
            # EIP-155: https://eips.ethereum.org/EIPS/eip-155
            return [
                Uint(self.nonce),
                Uint(self.gas_price),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                Uint(self.signature[0]),
                Uint(self.signature[1]),
                Uint(self.signature[2]),
            ]

        raise NotImplementedError(f"serialized_bytes not implemented for tx type {self.ty}")

    def serialized_bytes(self) -> bytes:
        """
        Returns bytes of the serialized representation of the transaction,
        which is almost always RLP encoding.
        """
        if self.ty is None:
            raise ValueError("ty must be set for all tx types")

        if self.ty > 0:
            return bytes([self.ty]) + eth_rlp.encode(self.payload_body())
        else:
            return eth_rlp.encode(self.payload_body())

    def signing_envelope(self) -> List[Any]:
        """
        Returns the list of values included in the envelope used for signing.
        """
        if self.gas_limit is None:
            raise ValueError("gas_limit must be set for all tx types")
        to = address_to_bytes(self.to)

        if self.ty == 3:
            # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
            if self.max_priority_fee_per_gas is None:
                raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_data_gas is None:
                raise ValueError("max_fee_per_data_gas must be set for type 3 tx")
            if self.blob_versioned_hashes is None:
                raise ValueError("blob_versioned_hashes must be set for type 3 tx")
            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.max_priority_fee_per_gas),
                Uint(self.max_fee_per_gas),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list] if self.access_list is not None else [],
                Uint(self.max_fee_per_data_gas),
                [hash_to_bytes(h) for h in self.blob_versioned_hashes],
            ]
        elif self.ty == 2:
            # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
            if self.max_priority_fee_per_gas is None:
                raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 3 tx")
            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.max_priority_fee_per_gas),
                Uint(self.max_fee_per_gas),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list] if self.access_list is not None else [],
            ]
        elif self.ty == 1:
            # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
            if self.gas_price is None:
                raise ValueError("gas_price must be set for type 1 tx")

            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.gas_price),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list] if self.access_list is not None else [],
            ]
        elif self.ty == 0:
            if self.gas_price is None:
                raise ValueError("gas_price must be set for type 0 tx")

            if self.protected:
                # EIP-155: https://eips.ethereum.org/EIPS/eip-155
                return [
                    Uint(self.nonce),
                    Uint(self.gas_price),
                    Uint(self.gas_limit),
                    to,
                    Uint(self.value),
                    code_to_bytes(self.data),
                    Uint(self.chain_id),
                    Uint(0),
                    Uint(0),
                ]
            else:
                return [
                    Uint(self.nonce),
                    Uint(self.gas_price),
                    Uint(self.gas_limit),
                    to,
                    Uint(self.value),
                    code_to_bytes(self.data),
                ]
        raise NotImplementedError("sigining for transaction type {self.ty} not implemented")

    def signing_bytes(self) -> bytes:
        """
        Returns the serialized bytes of the transaction used for signing.
        """
        if self.ty is None:
            raise ValueError("ty must be set for all tx types")

        if self.ty > 0:
            return bytes([self.ty]) + eth_rlp.encode(self.signing_envelope())
        else:
            return eth_rlp.encode(self.signing_envelope())

    def with_signature_and_sender(self) -> "Transaction":
        """
        Returns a signed version of the transaction using the private key.
        """
        tx = copy(self)

        if tx.signature is not None:
            # Transaction already signed
            if tx.sender is None:
                # TODO: We need to recover the sender from the signature
                raise NotImplementedError("recovering sender from signature not implemented")
            return tx

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

        # Get the signing bytes
        signing_hash = keccak256(tx.signing_bytes())

        # Sign the bytes
        private_key = PrivateKey.from_int(hash_to_int(tx.secret_key))
        signature_bytes = private_key.sign_recoverable(signing_hash, hasher=None)
        public_key = PublicKey.from_signature_and_message(
            signature_bytes, signing_hash, hasher=None
        )
        tx.sender = 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 tx.ty == 0:
            if tx.protected:
                v += 35 + (tx.chain_id * 2)
            else:  # not protected
                v += 27

        tx.signature = (v, r, s)

        # Remove the secret key because otherwise we might attempt to sign again (?)
        tx.secret_key = None
        return tx

ty: Optional[int] = None instance-attribute class-attribute

Transaction type value.

InvalidFeePayment

Bases: Exception

Transaction described more than one fee payment type.

Source code in src/ethereum_test_tools/common/types.py

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

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

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

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

InvalidSignaturePrivateKey

Bases: Exception

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

Source code in src/ethereum_test_tools/common/types.py

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

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

__str__()

Print exception string

Source code in src/ethereum_test_tools/common/types.py

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

__post_init__()

Ensures the transaction has no conflicting properties.

Source code in src/ethereum_test_tools/common/types.py

def __post_init__(self) -> None:
    """
    Ensures the transaction has no conflicting properties.
    """
    if (
        self.gas_price is not None
        and self.max_fee_per_gas is not None
        and self.max_priority_fee_per_gas is not None
    ):
        raise Transaction.InvalidFeePayment()

    if (
        self.gas_price is None
        and self.max_fee_per_gas is None
        and self.max_priority_fee_per_gas is None
    ):
        self.gas_price = 10

    if self.signature is not None and self.secret_key is not None:
        raise Transaction.InvalidSignaturePrivateKey()

    if self.signature is None and self.secret_key is None:
        self.secret_key = TestPrivateKey

    if self.ty is None:
        # Try to deduce transaction type from included fields
        if self.max_fee_per_data_gas is not None:
            self.ty = 3
        elif self.max_fee_per_gas is not None:
            self.ty = 2
        elif self.access_list is not None:
            self.ty = 1
        else:
            self.ty = 0

with_error(error)

Create a copy of the transaction with an added error.

Source code in src/ethereum_test_tools/common/types.py

def with_error(self, error: str) -> "Transaction":
    """
    Create a copy of the transaction with an added error.
    """
    tx = copy(self)
    tx.error = error
    return tx

with_nonce(nonce)

Create a copy of the transaction with a modified nonce.

Source code in src/ethereum_test_tools/common/types.py

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

with_fields(**kwargs)

Create a deepcopy of the transaction with modified fields.

Source code in src/ethereum_test_tools/common/types.py

def with_fields(self, **kwargs) -> "Transaction":
    """
    Create a deepcopy of the transaction with modified fields.
    """
    tx = deepcopy(self)
    for key, value in kwargs.items():
        if hasattr(tx, key):
            setattr(tx, key, value)
        else:
            raise ValueError(f"Invalid field '{key}' for Transaction")
    return tx

payload_body()

Returns the list of values included in the transaction body.

Source code in src/ethereum_test_tools/common/types.py

def payload_body(self) -> List[Any]:
    """
    Returns the list of values included in the transaction body.
    """
    if self.signature is None:
        raise ValueError("signature must be set before serializing any tx type")

    if self.gas_limit is None:
        raise ValueError("gas_limit must be set for all tx types")
    to = address_to_bytes(self.to)

    if self.ty == 3:
        # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
        if self.max_priority_fee_per_gas is None:
            raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_data_gas is None:
            raise ValueError("max_fee_per_data_gas must be set for type 3 tx")
        if self.blob_versioned_hashes is None:
            raise ValueError("blob_versioned_hashes must be set for type 3 tx")

        if self.wrapped_blob_transaction:
            if self.blobs is None:
                raise ValueError("blobs must be set for network version of type 3 tx")
            if self.blob_kzg_commitments is None:
                raise ValueError(
                    "blob_kzg_commitments must be set for network version of type 3 tx"
                )
            if self.blob_kzg_proofs is None:
                raise ValueError(
                    "blob_kzg_proofs must be set for network version of type 3 tx"
                )

            return [
                [
                    Uint(self.chain_id),
                    Uint(self.nonce),
                    Uint(self.max_priority_fee_per_gas),
                    Uint(self.max_fee_per_gas),
                    Uint(self.gas_limit),
                    to,
                    Uint(self.value),
                    code_to_bytes(self.data),
                    [a.to_list() for a in self.access_list]
                    if self.access_list is not None
                    else [],
                    Uint(self.max_fee_per_data_gas),
                    [hash_to_bytes(h) for h in self.blob_versioned_hashes],
                    Uint(self.signature[0]),
                    Uint(self.signature[1]),
                    Uint(self.signature[2]),
                ],
                self.blobs,
                self.blob_kzg_commitments,
                self.blob_kzg_proofs,
            ]
        else:
            return [
                Uint(self.chain_id),
                Uint(self.nonce),
                Uint(self.max_priority_fee_per_gas),
                Uint(self.max_fee_per_gas),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                [a.to_list() for a in self.access_list]
                if self.access_list is not None
                else [],
                Uint(self.max_fee_per_data_gas),
                [hash_to_bytes(h) for h in self.blob_versioned_hashes],
                Uint(self.signature[0]),
                Uint(self.signature[1]),
                Uint(self.signature[2]),
            ]
    elif self.ty == 2:
        # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
        if self.max_priority_fee_per_gas is None:
            raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 3 tx")
        return [
            Uint(self.chain_id),
            Uint(self.nonce),
            Uint(self.max_priority_fee_per_gas),
            Uint(self.max_fee_per_gas),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            [a.to_list() for a in self.access_list] if self.access_list is not None else [],
            Uint(self.signature[0]),
            Uint(self.signature[1]),
            Uint(self.signature[2]),
        ]
    elif self.ty == 1:
        # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
        if self.gas_price is None:
            raise ValueError("gas_price must be set for type 1 tx")

        return [
            Uint(self.chain_id),
            Uint(self.nonce),
            Uint(self.gas_price),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            [a.to_list() for a in self.access_list] if self.access_list is not None else [],
            Uint(self.signature[0]),
            Uint(self.signature[1]),
            Uint(self.signature[2]),
        ]
    elif self.ty == 0:
        if self.gas_price is None:
            raise ValueError("gas_price must be set for type 0 tx")
        # EIP-155: https://eips.ethereum.org/EIPS/eip-155
        return [
            Uint(self.nonce),
            Uint(self.gas_price),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            Uint(self.signature[0]),
            Uint(self.signature[1]),
            Uint(self.signature[2]),
        ]

    raise NotImplementedError(f"serialized_bytes not implemented for tx type {self.ty}")

serialized_bytes()

Returns bytes of the serialized representation of the transaction, which is almost always RLP encoding.

Source code in src/ethereum_test_tools/common/types.py

def serialized_bytes(self) -> bytes:
    """
    Returns bytes of the serialized representation of the transaction,
    which is almost always RLP encoding.
    """
    if self.ty is None:
        raise ValueError("ty must be set for all tx types")

    if self.ty > 0:
        return bytes([self.ty]) + eth_rlp.encode(self.payload_body())
    else:
        return eth_rlp.encode(self.payload_body())

signing_envelope()

Returns the list of values included in the envelope used for signing.

Source code in src/ethereum_test_tools/common/types.py

def signing_envelope(self) -> List[Any]:
    """
    Returns the list of values included in the envelope used for signing.
    """
    if self.gas_limit is None:
        raise ValueError("gas_limit must be set for all tx types")
    to = address_to_bytes(self.to)

    if self.ty == 3:
        # EIP-4844: https://eips.ethereum.org/EIPS/eip-4844
        if self.max_priority_fee_per_gas is None:
            raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_data_gas is None:
            raise ValueError("max_fee_per_data_gas must be set for type 3 tx")
        if self.blob_versioned_hashes is None:
            raise ValueError("blob_versioned_hashes must be set for type 3 tx")
        return [
            Uint(self.chain_id),
            Uint(self.nonce),
            Uint(self.max_priority_fee_per_gas),
            Uint(self.max_fee_per_gas),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            [a.to_list() for a in self.access_list] if self.access_list is not None else [],
            Uint(self.max_fee_per_data_gas),
            [hash_to_bytes(h) for h in self.blob_versioned_hashes],
        ]
    elif self.ty == 2:
        # EIP-1559: https://eips.ethereum.org/EIPS/eip-1559
        if self.max_priority_fee_per_gas is None:
            raise ValueError("max_priority_fee_per_gas must be set for type 3 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 3 tx")
        return [
            Uint(self.chain_id),
            Uint(self.nonce),
            Uint(self.max_priority_fee_per_gas),
            Uint(self.max_fee_per_gas),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            [a.to_list() for a in self.access_list] if self.access_list is not None else [],
        ]
    elif self.ty == 1:
        # EIP-2930: https://eips.ethereum.org/EIPS/eip-2930
        if self.gas_price is None:
            raise ValueError("gas_price must be set for type 1 tx")

        return [
            Uint(self.chain_id),
            Uint(self.nonce),
            Uint(self.gas_price),
            Uint(self.gas_limit),
            to,
            Uint(self.value),
            code_to_bytes(self.data),
            [a.to_list() for a in self.access_list] if self.access_list is not None else [],
        ]
    elif self.ty == 0:
        if self.gas_price is None:
            raise ValueError("gas_price must be set for type 0 tx")

        if self.protected:
            # EIP-155: https://eips.ethereum.org/EIPS/eip-155
            return [
                Uint(self.nonce),
                Uint(self.gas_price),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
                Uint(self.chain_id),
                Uint(0),
                Uint(0),
            ]
        else:
            return [
                Uint(self.nonce),
                Uint(self.gas_price),
                Uint(self.gas_limit),
                to,
                Uint(self.value),
                code_to_bytes(self.data),
            ]
    raise NotImplementedError("sigining for transaction type {self.ty} not implemented")

signing_bytes()

Returns the serialized bytes of the transaction used for signing.

Source code in src/ethereum_test_tools/common/types.py

def signing_bytes(self) -> bytes:
    """
    Returns the serialized bytes of the transaction used for signing.
    """
    if self.ty is None:
        raise ValueError("ty must be set for all tx types")

    if self.ty > 0:
        return bytes([self.ty]) + eth_rlp.encode(self.signing_envelope())
    else:
        return eth_rlp.encode(self.signing_envelope())

with_signature_and_sender()

Returns a signed version of the transaction using the private key.

Source code in src/ethereum_test_tools/common/types.py

def with_signature_and_sender(self) -> "Transaction":
    """
    Returns a signed version of the transaction using the private key.
    """
    tx = copy(self)

    if tx.signature is not None:
        # Transaction already signed
        if tx.sender is None:
            # TODO: We need to recover the sender from the signature
            raise NotImplementedError("recovering sender from signature not implemented")
        return tx

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

    # Get the signing bytes
    signing_hash = keccak256(tx.signing_bytes())

    # Sign the bytes
    private_key = PrivateKey.from_int(hash_to_int(tx.secret_key))
    signature_bytes = private_key.sign_recoverable(signing_hash, hasher=None)
    public_key = PublicKey.from_signature_and_message(
        signature_bytes, signing_hash, hasher=None
    )
    tx.sender = 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 tx.ty == 0:
        if tx.protected:
            v += 35 + (tx.chain_id * 2)
        else:  # not protected
            v += 27

    tx.signature = (v, r, s)

    # Remove the secret key because otherwise we might attempt to sign again (?)
    tx.secret_key = None
    return tx

Header dataclass

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

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Header:
    """
    Header type used to describe block header properties in test specs.
    """

    parent_hash: Optional[bytes] = None
    ommers_hash: Optional[bytes] = None
    coinbase: Optional[bytes | str] = None
    state_root: Optional[bytes] = None
    transactions_root: Optional[bytes] = None
    receipt_root: Optional[bytes] = None
    bloom: Optional[bytes] = None
    difficulty: Optional[int] = None
    number: Optional[int] = None
    gas_limit: Optional[int] = None
    gas_used: Optional[int] = None
    timestamp: Optional[int] = None
    extra_data: Optional[bytes] = None
    mix_digest: Optional[bytes] = None
    nonce: Optional[bytes] = None
    base_fee: Optional[int | Removable] = None
    withdrawals_root: Optional[bytes | Removable] = None
    data_gas_used: Optional[int | Removable] = None
    excess_data_gas: Optional[int | Removable] = None
    hash: Optional[bytes] = None

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

REMOVE_FIELD: Removable = Removable() class-attribute

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

Block dataclass

Bases: Header

Block type used to describe block properties in test specs

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Block(Header):
    """
    Block type used to describe block properties in test specs
    """

    rlp: Optional[bytes] = None
    """
    If set, blockchain test will skip generating the block and will pass this value directly to
    the Fixture.

    Only meant to be used to simulate blocks with bad formats, and therefore
    requires the block to produce an exception.
    """
    rlp_modifier: Optional[Header] = None
    """
    An RLP modifying header which values would be used to override the ones
    returned by the  `evm_transition_tool`.
    """
    exception: Optional[str] = None
    """
    If set, the block is expected to be rejected by the client.
    """
    engine_api_error_code: Optional[EngineAPIError] = None
    """
    If set, the block is expected to produce an error response from the Engine API.
    """
    txs: Optional[List[Transaction]] = None
    """
    List of transactions included in the block.
    """
    ommers: Optional[List[Header]] = None
    """
    List of ommer headers included in the block.
    """
    withdrawals: Optional[List[Withdrawal]] = None
    """
    List of withdrawals to perform for this block.
    """

    def set_environment(self, env: Environment) -> Environment:
        """
        Creates a copy of the environment with the characteristics of this
        specific block.
        """
        new_env = copy(env)

        """
        Values that need to be set in the environment and are `None` for
        this block need to be set to their defaults.
        """
        environment_default = Environment()
        new_env.difficulty = self.difficulty
        new_env.coinbase = (
            self.coinbase if self.coinbase is not None else environment_default.coinbase
        )
        new_env.gas_limit = (
            self.gas_limit if self.gas_limit is not None else environment_default.gas_limit
        )
        if not isinstance(self.base_fee, Removable):
            new_env.base_fee = self.base_fee
        new_env.withdrawals = self.withdrawals
        if not isinstance(self.excess_data_gas, Removable):
            new_env.excess_data_gas = self.excess_data_gas
        if not isinstance(self.data_gas_used, Removable):
            new_env.data_gas_used = self.data_gas_used
        """
        These values are required, but they depend on the previous environment,
        so they can be calculated here.
        """
        if self.number is not None:
            new_env.number = self.number
        else:
            # calculate the next block number for the environment
            if len(new_env.block_hashes) == 0:
                new_env.number = 0
            else:
                new_env.number = max(new_env.block_hashes.keys()) + 1

        if self.timestamp is not None:
            new_env.timestamp = self.timestamp
        else:
            assert new_env.parent_timestamp is not None
            new_env.timestamp = new_env.parent_timestamp + 12

        return new_env

    def copy_with_rlp(self, rlp: bytes) -> "Block":
        """
        Creates a copy of the block and adds the specified RLP.
        """
        new_block = deepcopy(self)
        new_block.rlp = rlp
        return new_block

rlp: Optional[bytes] = None instance-attribute class-attribute

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

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

rlp_modifier: Optional[Header] = None instance-attribute class-attribute

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

exception: Optional[str] = None instance-attribute class-attribute

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

engine_api_error_code: Optional[EngineAPIError] = None instance-attribute class-attribute

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

txs: Optional[List[Transaction]] = None instance-attribute class-attribute

List of transactions included in the block.

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

List of ommer headers included in the block.

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

List of withdrawals to perform for this block.

set_environment(env)

Creates a copy of the environment with the characteristics of this specific block.

Source code in src/ethereum_test_tools/common/types.py

def set_environment(self, env: Environment) -> Environment:
    """
    Creates a copy of the environment with the characteristics of this
    specific block.
    """
    new_env = copy(env)

    """
    Values that need to be set in the environment and are `None` for
    this block need to be set to their defaults.
    """
    environment_default = Environment()
    new_env.difficulty = self.difficulty
    new_env.coinbase = (
        self.coinbase if self.coinbase is not None else environment_default.coinbase
    )
    new_env.gas_limit = (
        self.gas_limit if self.gas_limit is not None else environment_default.gas_limit
    )
    if not isinstance(self.base_fee, Removable):
        new_env.base_fee = self.base_fee
    new_env.withdrawals = self.withdrawals
    if not isinstance(self.excess_data_gas, Removable):
        new_env.excess_data_gas = self.excess_data_gas
    if not isinstance(self.data_gas_used, Removable):
        new_env.data_gas_used = self.data_gas_used
    """
    These values are required, but they depend on the previous environment,
    so they can be calculated here.
    """
    if self.number is not None:
        new_env.number = self.number
    else:
        # calculate the next block number for the environment
        if len(new_env.block_hashes) == 0:
            new_env.number = 0
        else:
            new_env.number = max(new_env.block_hashes.keys()) + 1

    if self.timestamp is not None:
        new_env.timestamp = self.timestamp
    else:
        assert new_env.parent_timestamp is not None
        new_env.timestamp = new_env.parent_timestamp + 12

    return new_env

copy_with_rlp(rlp)

Creates a copy of the block and adds the specified RLP.

Source code in src/ethereum_test_tools/common/types.py

def copy_with_rlp(self, rlp: bytes) -> "Block":
    """
    Creates a copy of the block and adds the specified RLP.
    """
    new_block = deepcopy(self)
    new_block.rlp = rlp
    return new_block

FixtureEngineNewPayload dataclass

Representation of the engine_newPayloadVX information to be sent using the block information.

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class FixtureEngineNewPayload:
    """
    Representation of the `engine_newPayloadVX` information to be
    sent using the block information.
    """

    payload: FixtureExecutionPayload
    version: int
    blob_versioned_hashes: Optional[List[bytes]] = None
    error_code: Optional[EngineAPIError] = None

    @classmethod
    def from_fixture_header(
        cls,
        fork: Fork,
        header: FixtureHeader,
        transactions: List[Transaction],
        withdrawals: Optional[List[Withdrawal]],
        error_code: Optional[EngineAPIError],
    ) -> Optional["FixtureEngineNewPayload"]:
        """
        Creates a `FixtureEngineNewPayload` from a `FixtureHeader`.
        """
        new_payload_version = fork.engine_new_payload_version(header.number, header.timestamp)

        if new_payload_version is None:
            return None

        new_payload = cls(
            payload=FixtureExecutionPayload(
                header=header, transactions=transactions, withdrawals=withdrawals
            ),
            version=new_payload_version,
            error_code=error_code,
        )

        if fork.engine_new_payload_blob_hashes(header.number, header.timestamp):
            new_payload.blob_versioned_hashes = blob_versioned_hashes_from_transactions(
                transactions
            )

        return new_payload

from_fixture_header(fork, header, transactions, withdrawals, error_code) classmethod

Creates a FixtureEngineNewPayload from a FixtureHeader.

Source code in src/ethereum_test_tools/common/types.py

@classmethod
def from_fixture_header(
    cls,
    fork: Fork,
    header: FixtureHeader,
    transactions: List[Transaction],
    withdrawals: Optional[List[Withdrawal]],
    error_code: Optional[EngineAPIError],
) -> Optional["FixtureEngineNewPayload"]:
    """
    Creates a `FixtureEngineNewPayload` from a `FixtureHeader`.
    """
    new_payload_version = fork.engine_new_payload_version(header.number, header.timestamp)

    if new_payload_version is None:
        return None

    new_payload = cls(
        payload=FixtureExecutionPayload(
            header=header, transactions=transactions, withdrawals=withdrawals
        ),
        version=new_payload_version,
        error_code=error_code,
    )

    if fork.engine_new_payload_blob_hashes(header.number, header.timestamp):
        new_payload.blob_versioned_hashes = blob_versioned_hashes_from_transactions(
            transactions
        )

    return new_payload

Fixture dataclass

Cross-client compatible Ethereum test fixture.

Source code in src/ethereum_test_tools/common/types.py

@dataclass(kw_only=True)
class Fixture:
    """
    Cross-client compatible Ethereum test fixture.
    """

    blocks: List[FixtureBlock]
    genesis: FixtureHeader
    genesis_rlp: bytes
    head: bytes
    fork: str
    pre_state: Mapping[str, Account]
    post_state: Optional[Mapping[str, Account]]
    seal_engine: str
    info: Dict[str, str] = field(default_factory=dict)
    name: str = ""

    _json: Dict[str, Any] | None = None

    def __post_init__(self):
        """
        Post init hook to convert to JSON after instantiation.
        """
        self._json = to_json(self)

    def fill_info(
        self,
        t8n: TransitionTool,
        ref_spec: ReferenceSpec | None,
    ):
        """
        Fill the info field for this fixture
        """
        self.info["filling-transition-tool"] = t8n.version()
        if ref_spec is not None:
            ref_spec.write_info(self.info)

__post_init__()

Post init hook to convert to JSON after instantiation.

Source code in src/ethereum_test_tools/common/types.py

def __post_init__(self):
    """
    Post init hook to convert to JSON after instantiation.
    """
    self._json = to_json(self)

fill_info(t8n, ref_spec)

Fill the info field for this fixture

Source code in src/ethereum_test_tools/common/types.py

def fill_info(
    self,
    t8n: TransitionTool,
    ref_spec: ReferenceSpec | None,
):
    """
    Fill the info field for this fixture
    """
    self.info["filling-transition-tool"] = t8n.version()
    if ref_spec is not None:
        ref_spec.write_info(self.info)

JSONEncoder

Bases: json.JSONEncoder

Custom JSON encoder for ethereum_test types.

Source code in src/ethereum_test_tools/common/types.py

class JSONEncoder(json.JSONEncoder):
    """
    Custom JSON encoder for `ethereum_test` types.
    """

    def default(self, obj: JSONEncoderSupportedType) -> Any:
        """
        Enocdes types defined in this module using basic python facilities.
        """
        if isinstance(obj, Storage):
            return obj.to_dict()
        elif isinstance(obj, Account):
            account = {
                "nonce": hex_or_none(obj.nonce, hex(0)),
                "balance": hex_or_none(obj.balance, hex(0)),
                "code": code_or_none(obj.code, "0x"),
                "storage": to_json_or_none(obj.storage, {}),
            }
            return even_padding(account, excluded=["storage"])
        elif isinstance(obj, AccessList):
            access_list: Dict[str, Any] = {
                "address": address_or_none(obj.address, "0x" + ZeroAddress.hex())
            }
            if obj.storage_keys is not None:
                access_list["storageKeys"] = [hex_or_none(k) for k in obj.storage_keys]
            return access_list
        elif isinstance(obj, Transaction):
            assert obj.ty is not None, "Transaction type must be set"
            tx: Dict[str, Any] = {
                "type": hex(obj.ty),
                "chainId": hex(obj.chain_id),
                "nonce": hex(obj.nonce),
                "gasPrice": hex_or_none(obj.gas_price),
                "maxPriorityFeePerGas": hex_or_none(obj.max_priority_fee_per_gas),
                "maxFeePerGas": hex_or_none(obj.max_fee_per_gas),
                "gas": hex(obj.gas_limit),
                "value": hex(obj.value),
                "input": code_to_hex(obj.data),
                "to": address_or_none(obj.to),
                "accessList": obj.access_list,
                "secretKey": obj.secret_key,
                "maxFeePerDataGas": hex_or_none(obj.max_fee_per_data_gas),
                "sender": address_or_none(obj.sender),
            }

            if obj.blob_versioned_hashes is not None:
                tx["blobVersionedHashes"] = [hash_string(h) for h in obj.blob_versioned_hashes]

            if obj.secret_key is None:
                assert obj.signature is not None
                assert len(obj.signature) == 3
                tx["v"] = hex(obj.signature[0])
                tx["r"] = hex(obj.signature[1])
                tx["s"] = hex(obj.signature[2])
            else:
                tx["v"] = ""
                tx["r"] = ""
                tx["s"] = ""

            return {k: v for (k, v) in tx.items() if v is not None}
        elif isinstance(obj, Withdrawal):
            withdrawal = {
                "index": hex(obj.index),
                "validatorIndex": hex(obj.validator),
                "address": obj.address,
                "amount": hex(obj.amount),
            }
            return withdrawal
        elif isinstance(obj, Environment):
            env: Dict[str, Any] = {
                "currentCoinbase": address_or_none(obj.coinbase, "0x" + ZeroAddress.hex()),
                "currentGasLimit": str_or_none(obj.gas_limit),
                "currentNumber": str_or_none(obj.number),
                "currentTimestamp": str_or_none(obj.timestamp),
                "currentRandom": str_or_none(obj.prev_randao),
                "currentDifficulty": str_or_none(obj.difficulty),
                "parentDifficulty": str_or_none(obj.parent_difficulty),
                "parentBaseFee": str_or_none(obj.parent_base_fee),
                "parentGasUsed": str_or_none(obj.parent_gas_used),
                "parentGasLimit": str_or_none(obj.parent_gas_limit),
                "parentTimestamp": str_or_none(obj.parent_timestamp),
                "blockHashes": {str(k): hex_or_none(v) for (k, v) in obj.block_hashes.items()},
                "ommers": [],
                "withdrawals": to_json_or_none(obj.withdrawals),
                "parentUncleHash": hash_string(obj.parent_ommers_hash),
                "currentBaseFee": str_or_none(obj.base_fee),
                "parentDataGasUsed": str_or_none(obj.parent_data_gas_used),
                "parentExcessDataGas": str_or_none(obj.parent_excess_data_gas),
                "currentExcessDataGas": str_or_none(obj.excess_data_gas),
                "currentDataGasUsed": str_or_none(obj.data_gas_used),
            }

            return {k: v for (k, v) in env.items() if v is not None}
        elif isinstance(obj, FixtureHeader):
            header = {
                "parentHash": hex_or_none(obj.parent_hash),
                "uncleHash": hex_or_none(obj.ommers_hash),
                "coinbase": hex_or_none(obj.coinbase),
                "stateRoot": hex_or_none(obj.state_root),
                "transactionsTrie": hex_or_none(obj.transactions_root),
                "receiptTrie": hex_or_none(obj.receipt_root),
                "bloom": hex_or_none(obj.bloom),
                "difficulty": hex(obj.difficulty),
                "number": hex(obj.number),
                "gasLimit": hex(obj.gas_limit),
                "gasUsed": hex(obj.gas_used),
                "timestamp": hex(obj.timestamp),
                "extraData": hex_or_none(obj.extra_data),
                "mixHash": hex_or_none(obj.mix_digest),
                "nonce": hex_or_none(obj.nonce),
            }
            if obj.base_fee is not None:
                header["baseFeePerGas"] = hex(obj.base_fee)
            if obj.hash is not None:
                header["hash"] = "0x" + obj.hash.hex()
            if obj.withdrawals_root is not None:
                header["withdrawalsRoot"] = hex_or_none(obj.withdrawals_root)
            if obj.data_gas_used is not None:
                header["dataGasUsed"] = hex(obj.data_gas_used)
            if obj.excess_data_gas is not None:
                header["excessDataGas"] = hex(obj.excess_data_gas)
            return even_padding(
                header,
                excluded=[
                    "parentHash",
                    "uncleHash",
                    "stateRoot",
                    "coinbase",
                    "transactionsTrie",
                    "receiptTrie",
                    "bloom",
                    "nonce",
                    "mixHash",
                    "hash",
                    "withdrawalsRoot",
                    "extraData",
                ],
            )
        elif isinstance(obj, FixtureTransaction):
            json_tx = to_json(obj.tx)
            if json_tx["v"] == "":
                del json_tx["v"]
                del json_tx["r"]
                del json_tx["s"]
            if "input" in json_tx:
                json_tx["data"] = json_tx["input"]
                del json_tx["input"]
            if "gas" in json_tx:
                json_tx["gasLimit"] = json_tx["gas"]
                del json_tx["gas"]
            if "to" not in json_tx:
                json_tx["to"] = ""
            return even_padding(
                json_tx,
                excluded=["data", "to", "accessList"],
            )
        elif isinstance(obj, FixtureExecutionPayload):
            payload: Dict[str, Any] = {
                "parentHash": hex_or_none(obj.header.parent_hash),
                "feeRecipient": hex_or_none(obj.header.coinbase),
                "stateRoot": hex_or_none(obj.header.state_root),
                "receiptsRoot": hex_or_none(obj.header.receipt_root),
                "logsBloom": hex_or_none(obj.header.bloom),
                "prevRandao": hex_or_none(obj.header.mix_digest),
                "blockNumber": hex(obj.header.number),
                "gasLimit": hex(obj.header.gas_limit),
                "gasUsed": hex(obj.header.gas_used),
                "timestamp": hex(obj.header.timestamp),
                "extraData": hex_or_none(obj.header.extra_data),
            }
            if obj.header.base_fee is not None:
                payload["baseFeePerGas"] = hex(obj.header.base_fee)
            if obj.header.hash is not None:
                payload["blockHash"] = "0x" + obj.header.hash.hex()

            if obj.transactions is not None:
                payload["transactions"] = [
                    hex_or_none(tx.serialized_bytes()) for tx in obj.transactions
                ]
            if obj.withdrawals is not None:
                payload["withdrawals"] = obj.withdrawals

            if obj.header.data_gas_used is not None:
                payload["dataGasUsed"] = hex(obj.header.data_gas_used)
            if obj.header.excess_data_gas is not None:
                payload["excessDataGas"] = hex(obj.header.excess_data_gas)

            return payload
        elif isinstance(obj, FixtureEngineNewPayload):
            new_payload: Dict[str, Any] = {
                "payload": to_json(obj.payload),
                "version": str_or_none(obj.version),
            }
            if obj.blob_versioned_hashes is not None:
                new_payload["blobVersionedHashes"] = [
                    "0x" + hash_to_bytes(h).hex() for h in obj.blob_versioned_hashes
                ]
            if obj.error_code is not None:
                new_payload["errorCode"] = str(int(obj.error_code))
            return new_payload

        elif isinstance(obj, FixtureBlock):
            b: Dict[str, Any] = {"rlp": hex_or_none(obj.rlp)}
            if obj.block_header is not None:
                b["blockHeader"] = json.loads(json.dumps(obj.block_header, cls=JSONEncoder))
            if obj.new_payload is not None:
                b["engineNewPayload"] = to_json_or_none(obj.new_payload)
            if obj.expected_exception is not None:
                b["expectException"] = obj.expected_exception
            if obj.block_number is not None:
                b["blocknumber"] = str(obj.block_number)
            if obj.txs is not None:
                b["transactions"] = [FixtureTransaction(tx=tx) for tx in obj.txs]
            if obj.ommers is not None:
                b["uncleHeaders"] = obj.ommers
            if obj.withdrawals is not None:
                b["withdrawals"] = [
                    even_padding(to_json(wd), excluded=["address"]) for wd in obj.withdrawals
                ]
            return b

        elif isinstance(obj, Fixture):
            if obj._json is not None:
                obj._json["_info"] = obj.info
                return obj._json

            f = {
                "_info": obj.info,
                "blocks": [json.loads(json.dumps(b, cls=JSONEncoder)) for b in obj.blocks],
                "genesisBlockHeader": self.default(obj.genesis),
                "genesisRLP": hex_or_none(obj.genesis_rlp),
                "lastblockhash": hex_or_none(obj.head),
                "network": obj.fork,
                "pre": json.loads(json.dumps(obj.pre_state, cls=JSONEncoder)),
                "postState": json.loads(json.dumps(obj.post_state, cls=JSONEncoder)),
                "sealEngine": obj.seal_engine,
            }
            if f["postState"] is None:
                del f["postState"]
            return f
        else:
            return super().default(obj)

default(obj)

Enocdes types defined in this module using basic python facilities.

Source code in src/ethereum_test_tools/common/types.py

def default(self, obj: JSONEncoderSupportedType) -> Any:
    """
    Enocdes types defined in this module using basic python facilities.
    """
    if isinstance(obj, Storage):
        return obj.to_dict()
    elif isinstance(obj, Account):
        account = {
            "nonce": hex_or_none(obj.nonce, hex(0)),
            "balance": hex_or_none(obj.balance, hex(0)),
            "code": code_or_none(obj.code, "0x"),
            "storage": to_json_or_none(obj.storage, {}),
        }
        return even_padding(account, excluded=["storage"])
    elif isinstance(obj, AccessList):
        access_list: Dict[str, Any] = {
            "address": address_or_none(obj.address, "0x" + ZeroAddress.hex())
        }
        if obj.storage_keys is not None:
            access_list["storageKeys"] = [hex_or_none(k) for k in obj.storage_keys]
        return access_list
    elif isinstance(obj, Transaction):
        assert obj.ty is not None, "Transaction type must be set"
        tx: Dict[str, Any] = {
            "type": hex(obj.ty),
            "chainId": hex(obj.chain_id),
            "nonce": hex(obj.nonce),
            "gasPrice": hex_or_none(obj.gas_price),
            "maxPriorityFeePerGas": hex_or_none(obj.max_priority_fee_per_gas),
            "maxFeePerGas": hex_or_none(obj.max_fee_per_gas),
            "gas": hex(obj.gas_limit),
            "value": hex(obj.value),
            "input": code_to_hex(obj.data),
            "to": address_or_none(obj.to),
            "accessList": obj.access_list,
            "secretKey": obj.secret_key,
            "maxFeePerDataGas": hex_or_none(obj.max_fee_per_data_gas),
            "sender": address_or_none(obj.sender),
        }

        if obj.blob_versioned_hashes is not None:
            tx["blobVersionedHashes"] = [hash_string(h) for h in obj.blob_versioned_hashes]

        if obj.secret_key is None:
            assert obj.signature is not None
            assert len(obj.signature) == 3
            tx["v"] = hex(obj.signature[0])
            tx["r"] = hex(obj.signature[1])
            tx["s"] = hex(obj.signature[2])
        else:
            tx["v"] = ""
            tx["r"] = ""
            tx["s"] = ""

        return {k: v for (k, v) in tx.items() if v is not None}
    elif isinstance(obj, Withdrawal):
        withdrawal = {
            "index": hex(obj.index),
            "validatorIndex": hex(obj.validator),
            "address": obj.address,
            "amount": hex(obj.amount),
        }
        return withdrawal
    elif isinstance(obj, Environment):
        env: Dict[str, Any] = {
            "currentCoinbase": address_or_none(obj.coinbase, "0x" + ZeroAddress.hex()),
            "currentGasLimit": str_or_none(obj.gas_limit),
            "currentNumber": str_or_none(obj.number),
            "currentTimestamp": str_or_none(obj.timestamp),
            "currentRandom": str_or_none(obj.prev_randao),
            "currentDifficulty": str_or_none(obj.difficulty),
            "parentDifficulty": str_or_none(obj.parent_difficulty),
            "parentBaseFee": str_or_none(obj.parent_base_fee),
            "parentGasUsed": str_or_none(obj.parent_gas_used),
            "parentGasLimit": str_or_none(obj.parent_gas_limit),
            "parentTimestamp": str_or_none(obj.parent_timestamp),
            "blockHashes": {str(k): hex_or_none(v) for (k, v) in obj.block_hashes.items()},
            "ommers": [],
            "withdrawals": to_json_or_none(obj.withdrawals),
            "parentUncleHash": hash_string(obj.parent_ommers_hash),
            "currentBaseFee": str_or_none(obj.base_fee),
            "parentDataGasUsed": str_or_none(obj.parent_data_gas_used),
            "parentExcessDataGas": str_or_none(obj.parent_excess_data_gas),
            "currentExcessDataGas": str_or_none(obj.excess_data_gas),
            "currentDataGasUsed": str_or_none(obj.data_gas_used),
        }

        return {k: v for (k, v) in env.items() if v is not None}
    elif isinstance(obj, FixtureHeader):
        header = {
            "parentHash": hex_or_none(obj.parent_hash),
            "uncleHash": hex_or_none(obj.ommers_hash),
            "coinbase": hex_or_none(obj.coinbase),
            "stateRoot": hex_or_none(obj.state_root),
            "transactionsTrie": hex_or_none(obj.transactions_root),
            "receiptTrie": hex_or_none(obj.receipt_root),
            "bloom": hex_or_none(obj.bloom),
            "difficulty": hex(obj.difficulty),
            "number": hex(obj.number),
            "gasLimit": hex(obj.gas_limit),
            "gasUsed": hex(obj.gas_used),
            "timestamp": hex(obj.timestamp),
            "extraData": hex_or_none(obj.extra_data),
            "mixHash": hex_or_none(obj.mix_digest),
            "nonce": hex_or_none(obj.nonce),
        }
        if obj.base_fee is not None:
            header["baseFeePerGas"] = hex(obj.base_fee)
        if obj.hash is not None:
            header["hash"] = "0x" + obj.hash.hex()
        if obj.withdrawals_root is not None:
            header["withdrawalsRoot"] = hex_or_none(obj.withdrawals_root)
        if obj.data_gas_used is not None:
            header["dataGasUsed"] = hex(obj.data_gas_used)
        if obj.excess_data_gas is not None:
            header["excessDataGas"] = hex(obj.excess_data_gas)
        return even_padding(
            header,
            excluded=[
                "parentHash",
                "uncleHash",
                "stateRoot",
                "coinbase",
                "transactionsTrie",
                "receiptTrie",
                "bloom",
                "nonce",
                "mixHash",
                "hash",
                "withdrawalsRoot",
                "extraData",
            ],
        )
    elif isinstance(obj, FixtureTransaction):
        json_tx = to_json(obj.tx)
        if json_tx["v"] == "":
            del json_tx["v"]
            del json_tx["r"]
            del json_tx["s"]
        if "input" in json_tx:
            json_tx["data"] = json_tx["input"]
            del json_tx["input"]
        if "gas" in json_tx:
            json_tx["gasLimit"] = json_tx["gas"]
            del json_tx["gas"]
        if "to" not in json_tx:
            json_tx["to"] = ""
        return even_padding(
            json_tx,
            excluded=["data", "to", "accessList"],
        )
    elif isinstance(obj, FixtureExecutionPayload):
        payload: Dict[str, Any] = {
            "parentHash": hex_or_none(obj.header.parent_hash),
            "feeRecipient": hex_or_none(obj.header.coinbase),
            "stateRoot": hex_or_none(obj.header.state_root),
            "receiptsRoot": hex_or_none(obj.header.receipt_root),
            "logsBloom": hex_or_none(obj.header.bloom),
            "prevRandao": hex_or_none(obj.header.mix_digest),
            "blockNumber": hex(obj.header.number),
            "gasLimit": hex(obj.header.gas_limit),
            "gasUsed": hex(obj.header.gas_used),
            "timestamp": hex(obj.header.timestamp),
            "extraData": hex_or_none(obj.header.extra_data),
        }
        if obj.header.base_fee is not None:
            payload["baseFeePerGas"] = hex(obj.header.base_fee)
        if obj.header.hash is not None:
            payload["blockHash"] = "0x" + obj.header.hash.hex()

        if obj.transactions is not None:
            payload["transactions"] = [
                hex_or_none(tx.serialized_bytes()) for tx in obj.transactions
            ]
        if obj.withdrawals is not None:
            payload["withdrawals"] = obj.withdrawals

        if obj.header.data_gas_used is not None:
            payload["dataGasUsed"] = hex(obj.header.data_gas_used)
        if obj.header.excess_data_gas is not None:
            payload["excessDataGas"] = hex(obj.header.excess_data_gas)

        return payload
    elif isinstance(obj, FixtureEngineNewPayload):
        new_payload: Dict[str, Any] = {
            "payload": to_json(obj.payload),
            "version": str_or_none(obj.version),
        }
        if obj.blob_versioned_hashes is not None:
            new_payload["blobVersionedHashes"] = [
                "0x" + hash_to_bytes(h).hex() for h in obj.blob_versioned_hashes
            ]
        if obj.error_code is not None:
            new_payload["errorCode"] = str(int(obj.error_code))
        return new_payload

    elif isinstance(obj, FixtureBlock):
        b: Dict[str, Any] = {"rlp": hex_or_none(obj.rlp)}
        if obj.block_header is not None:
            b["blockHeader"] = json.loads(json.dumps(obj.block_header, cls=JSONEncoder))
        if obj.new_payload is not None:
            b["engineNewPayload"] = to_json_or_none(obj.new_payload)
        if obj.expected_exception is not None:
            b["expectException"] = obj.expected_exception
        if obj.block_number is not None:
            b["blocknumber"] = str(obj.block_number)
        if obj.txs is not None:
            b["transactions"] = [FixtureTransaction(tx=tx) for tx in obj.txs]
        if obj.ommers is not None:
            b["uncleHeaders"] = obj.ommers
        if obj.withdrawals is not None:
            b["withdrawals"] = [
                even_padding(to_json(wd), excluded=["address"]) for wd in obj.withdrawals
            ]
        return b

    elif isinstance(obj, Fixture):
        if obj._json is not None:
            obj._json["_info"] = obj.info
            return obj._json

        f = {
            "_info": obj.info,
            "blocks": [json.loads(json.dumps(b, cls=JSONEncoder)) for b in obj.blocks],
            "genesisBlockHeader": self.default(obj.genesis),
            "genesisRLP": hex_or_none(obj.genesis_rlp),
            "lastblockhash": hex_or_none(obj.head),
            "network": obj.fork,
            "pre": json.loads(json.dumps(obj.pre_state, cls=JSONEncoder)),
            "postState": json.loads(json.dumps(obj.post_state, cls=JSONEncoder)),
            "sealEngine": obj.seal_engine,
        }
        if f["postState"] is None:
            del f["postState"]
        return f
    else:
        return super().default(obj)