Ethereum Test Tools Package

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

Code

Bases: SupportsBytes, Sized

Generic code object.

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

class Code(SupportsBytes, Sized):
    """
    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 __init__(
        self,
        code: Optional[str | bytes | SupportsBytes] = None,
        *,
        name: Optional[str] = None,
    ):
        """
        Create a new Code object.
        """
        if code is not None:
            self.bytecode = to_bytes(code)
        self.name = name

    def __bytes__(self) -> bytes:
        """
        Transform the Code object into bytes.
        """
        if self.bytecode is None:
            return bytes()
        else:
            return self.bytecode

    def __len__(self) -> int:
        """
        Get the length of the Code object.
        """
        if self.bytecode is None:
            return 0
        else:
            return len(self.bytecode)

    def __add__(self, other: str | bytes | SupportsBytes) -> "Code":
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(to_bytes(self) + to_bytes(other))

    def __radd__(self, other: str | bytes | SupportsBytes) -> "Code":
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(to_bytes(other) + to_bytes(self))

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

bytes array that represents the bytecode of this object.

__init__(code=None, *, name=None)

Create a new Code object.

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

def __init__(
    self,
    code: Optional[str | bytes | SupportsBytes] = None,
    *,
    name: Optional[str] = None,
):
    """
    Create a new Code object.
    """
    if code is not None:
        self.bytecode = to_bytes(code)
    self.name = name

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

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

__bytes__()

Transform the Code object into bytes.

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

def __bytes__(self) -> bytes:
    """
    Transform the Code object into bytes.
    """
    if self.bytecode is None:
        return bytes()
    else:
        return self.bytecode

__len__()

Get the length of the Code object.

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

def __len__(self) -> int:
    """
    Get the length of the Code object.
    """
    if self.bytecode is None:
        return 0
    else:
        return len(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: str | bytes | SupportsBytes) -> "Code":
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(to_bytes(self) + 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: str | bytes | SupportsBytes) -> "Code":
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(to_bytes(other) + to_bytes(self))

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

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

    (blocks, head, alloc) = test_spec.make_blocks(
        t8n,
        genesis,
        pre,
        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=pre,
        post_state=alloc_to_accounts(alloc),
        seal_engine=engine,
        name=test_spec.tag,
    )
    fixture.fill_info(t8n, spec)

    return fixture

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: str | bytes | SupportsBytes
    """
    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 | SupportsBytes,
        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 = 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__(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 | SupportsBytes,
    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 = 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__(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: str | bytes | SupportsBytes = 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)

JSONEncoder

Bases: json.JSONEncoder

Custom JSON encoder for ethereum_test types.

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

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

    @dataclass(kw_only=True)
    class Field:
        """
        Settings for a field in a JSON object.
        """

        name: Optional[str] = None
        """
        The name of the field in the JSON object.
        """
        cast_type: Optional[Callable] = None
        """
        The type or function to use to cast the field to before serializing.
        """
        skip_string_convert: bool = False
        """
        By default, the fields are converted to string after serializing.
        """
        to_json: bool = False
        """
        Whether the field should be converted to JSON by itself.
        This option and `JSON_SKIP_STRING_CONVERT` are mutually exclusive.
        """
        default_value: Optional[Any] = None
        """
        Value to use if the field is not set before type casting.
        """
        default_value_skip_cast: Optional[Any] = None
        """
        Value to use if the field is not set and also skip type casting.
        """
        keep_none: bool = False
        """
        Whether to keep the field if it is `None`.
        """
        skip: bool = False
        """
        Whether to skip the field when serializing.
        """

        def apply(
            self, encoder: json.JSONEncoder, target: Dict[str, Any], field_name: str, value: Any
        ) -> None:
            """
            Applies the settings to the target dictionary.
            """
            if self.skip:
                return

            if self.name:
                field_name = self.name

            if value is None:
                if self.default_value is not None:
                    value = self.default_value
                elif self.default_value_skip_cast is not None:
                    target[field_name] = self.default_value_skip_cast
                    return

                if not self.keep_none and value is None:
                    return

            if value is not None:
                if self.cast_type is not None:
                    value = self.cast_type(value)

                if self.to_json:
                    value = encoder.default(value)
                elif not self.skip_string_convert:
                    value = str(value)

            target[field_name] = value

    def default(self, obj: Any) -> Any:
        """
        Encodes types defined in this module using basic python facilities.
        """
        if callable(getattr(obj, "__json__", False)):
            return obj.__json__(encoder=self)

        elif is_dataclass(obj):
            result: Dict[str, Any] = {}
            for object_field in fields(obj):
                field_name = object_field.name
                metadata = object_field.metadata
                value = getattr(obj, field_name)
                assert metadata is not None, f"Field {field_name} has no metadata"
                field_settings = metadata.get("json_encoder")
                assert isinstance(field_settings, self.Field), (
                    f"Field {field_name} has invalid json_encoder " f"metadata: {field_settings}"
                )
                field_settings.apply(self, result, field_name, value)
            return result

        elif isinstance(obj, dict):
            return {self.default(k): self.default(v) for k, v in obj.items()}

        elif isinstance(obj, list) or isinstance(obj, tuple):
            return [self.default(item) for item in obj]

        elif isinstance(obj, str) or isinstance(obj, int) or isinstance(obj, bool) or obj is None:
            return obj

        else:
            return super().default(obj)

Field dataclass

Settings for a field in a JSON object.

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

@dataclass(kw_only=True)
class Field:
    """
    Settings for a field in a JSON object.
    """

    name: Optional[str] = None
    """
    The name of the field in the JSON object.
    """
    cast_type: Optional[Callable] = None
    """
    The type or function to use to cast the field to before serializing.
    """
    skip_string_convert: bool = False
    """
    By default, the fields are converted to string after serializing.
    """
    to_json: bool = False
    """
    Whether the field should be converted to JSON by itself.
    This option and `JSON_SKIP_STRING_CONVERT` are mutually exclusive.
    """
    default_value: Optional[Any] = None
    """
    Value to use if the field is not set before type casting.
    """
    default_value_skip_cast: Optional[Any] = None
    """
    Value to use if the field is not set and also skip type casting.
    """
    keep_none: bool = False
    """
    Whether to keep the field if it is `None`.
    """
    skip: bool = False
    """
    Whether to skip the field when serializing.
    """

    def apply(
        self, encoder: json.JSONEncoder, target: Dict[str, Any], field_name: str, value: Any
    ) -> None:
        """
        Applies the settings to the target dictionary.
        """
        if self.skip:
            return

        if self.name:
            field_name = self.name

        if value is None:
            if self.default_value is not None:
                value = self.default_value
            elif self.default_value_skip_cast is not None:
                target[field_name] = self.default_value_skip_cast
                return

            if not self.keep_none and value is None:
                return

        if value is not None:
            if self.cast_type is not None:
                value = self.cast_type(value)

            if self.to_json:
                value = encoder.default(value)
            elif not self.skip_string_convert:
                value = str(value)

        target[field_name] = value

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

The name of the field in the JSON object.

cast_type: Optional[Callable] = None class-attribute instance-attribute

The type or function to use to cast the field to before serializing.

skip_string_convert: bool = False class-attribute instance-attribute

By default, the fields are converted to string after serializing.

to_json: bool = False class-attribute instance-attribute

Whether the field should be converted to JSON by itself. This option and JSON_SKIP_STRING_CONVERT are mutually exclusive.

default_value: Optional[Any] = None class-attribute instance-attribute

Value to use if the field is not set before type casting.

default_value_skip_cast: Optional[Any] = None class-attribute instance-attribute

Value to use if the field is not set and also skip type casting.

keep_none: bool = False class-attribute instance-attribute

Whether to keep the field if it is None.

skip: bool = False class-attribute instance-attribute

Whether to skip the field when serializing.

apply(encoder, target, field_name, value)

Applies the settings to the target dictionary.

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

def apply(
    self, encoder: json.JSONEncoder, target: Dict[str, Any], field_name: str, value: Any
) -> None:
    """
    Applies the settings to the target dictionary.
    """
    if self.skip:
        return

    if self.name:
        field_name = self.name

    if value is None:
        if self.default_value is not None:
            value = self.default_value
        elif self.default_value_skip_cast is not None:
            target[field_name] = self.default_value_skip_cast
            return

        if not self.keep_none and value is None:
            return

    if value is not None:
        if self.cast_type is not None:
            value = self.cast_type(value)

        if self.to_json:
            value = encoder.default(value)
        elif not self.skip_string_convert:
            value = str(value)

    target[field_name] = value

default(obj)

Encodes types defined in this module using basic python facilities.

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

def default(self, obj: Any) -> Any:
    """
    Encodes types defined in this module using basic python facilities.
    """
    if callable(getattr(obj, "__json__", False)):
        return obj.__json__(encoder=self)

    elif is_dataclass(obj):
        result: Dict[str, Any] = {}
        for object_field in fields(obj):
            field_name = object_field.name
            metadata = object_field.metadata
            value = getattr(obj, field_name)
            assert metadata is not None, f"Field {field_name} has no metadata"
            field_settings = metadata.get("json_encoder")
            assert isinstance(field_settings, self.Field), (
                f"Field {field_name} has invalid json_encoder " f"metadata: {field_settings}"
            )
            field_settings.apply(self, result, field_name, value)
        return result

    elif isinstance(obj, dict):
        return {self.default(k): self.default(v) for k, v in obj.items()}

    elif isinstance(obj, list) or isinstance(obj, tuple):
        return [self.default(item) for item in obj]

    elif isinstance(obj, str) or isinstance(obj, int) or isinstance(obj, bool) or obj is None:
        return obj

    else:
        return super().default(obj)

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: FixedSizeBytesConvertible, nonce: int) -> str:
    """
    Compute address of the resulting contract created using a transaction
    or the `CREATE` opcode.
    """
    nonce_bytes = bytes() if nonce == 0 else nonce.to_bytes(length=1, byteorder="big")
    hash = keccak256(encode([Address(address), nonce_bytes]))
    return "0x" + hash[-20:].hex()

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

        Hex-strings will automatically be converted to bytes.

        """
        args: List[Union[int, bytes, str, "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) or isinstance(data, str):
                if isinstance(data, str):
                    if data.startswith("0x"):
                        data = data[2:]
                    data = bytes.fromhex(data)
                assert len(data) <= self.data_portion_length
                data_portion = data.rjust(self.data_portion_length, b"\x00")
            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 bytes/hex string")

        # The rest of the arguments conform the stack.
        while len(args) > 0:
            data = args.pop()
            if isinstance(data, bytes) or isinstance(data, str):
                if isinstance(data, str):
                    if data.startswith("0x"):
                        data = data[2:]
                    data = bytes.fromhex(data)
                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/hex string")

        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.

Hex-strings will automatically be converted to bytes.

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

def __call__(self, *args_t: Union[int, bytes, str, "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.

    Hex-strings will automatically be converted to bytes.

    """
    args: List[Union[int, bytes, str, "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) or isinstance(data, str):
            if isinstance(data, str):
                if data.startswith("0x"):
                    data = data[2:]
                data = bytes.fromhex(data)
            assert len(data) <= self.data_portion_length
            data_portion = data.rjust(self.data_portion_length, b"\x00")
        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 bytes/hex string")

    # The rest of the arguments conform the stack.
    while len(args) > 0:
        data = args.pop()
        if isinstance(data, bytes) or isinstance(data, str):
            if isinstance(data, str):
                if data.startswith("0x"):
                    data = data[2:]
                data = bytes.fromhex(data)
            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/hex string")

    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_

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

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

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

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
    post: Mapping
    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[Alloc, Bytes, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = copy(self.env)

        # Remove fields that should not be present in the genesis block.
        env.withdrawals = None
        env.beacon_root = None

        env = env.set_fork_requirements(fork)

        new_alloc, state_root = t8n.calc_state_root(
            alloc=to_json(Alloc(self.pre)),
            fork=fork,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )

        genesis = FixtureHeader(
            parent_hash=Hash(0),
            ommers_hash=Hash(EmptyOmmersRoot),
            coinbase=Address(0),
            state_root=Hash(state_root),
            transactions_root=Hash(EmptyTrieRoot),
            receipt_root=Hash(EmptyTrieRoot),
            bloom=Bloom(0),
            difficulty=ZeroPaddedHexNumber(0x20000 if env.difficulty is None else env.difficulty),
            number=ZeroPaddedHexNumber(Number(env.number) - 1),
            gas_limit=ZeroPaddedHexNumber(env.gas_limit),
            gas_used=0,
            timestamp=0,
            extra_data=Bytes([0]),
            mix_digest=Hash(0),
            nonce=HeaderNonce(0),
            base_fee=ZeroPaddedHexNumber.or_none(env.base_fee),
            blob_gas_used=ZeroPaddedHexNumber.or_none(env.blob_gas_used),
            excess_blob_gas=ZeroPaddedHexNumber.or_none(env.excess_blob_gas),
            withdrawals_root=Hash.or_none(
                t8n.calc_withdrawals_root(
                    withdrawals=env.withdrawals,
                    fork=fork,
                    debug_output_path=self.get_next_transition_tool_output_path(),
                )
                if env.withdrawals is not None
                else None
            ),
            beacon_root=Hash.or_none(env.beacon_root),
        )

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

        return Alloc(new_alloc), genesis_rlp, genesis

    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        pre: Alloc,
        fork: Fork,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], Hash, 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(pre),
            txs=to_json(txs),
            env=to_json(env),
            fork_name=fork.fork(block_number=Number(env.number), timestamp=Number(env.timestamp)),
            chain_id=chain_id,
            reward=fork.get_reward(Number(env.number), Number(env.timestamp)),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )

        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

        env.extra_data = b"\x00"
        header = FixtureHeader.collect(
            fork=fork,
            transition_tool_result=result,
            environment=env,
        )

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

        new_payload: FixtureEngineNewPayload | None = None
        if not self.base_test_config.disable_hive:
            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[Alloc, Bytes, FixtureHeader]:
    """
    Create a genesis block from the state test definition.
    """
    env = copy(self.env)

    # Remove fields that should not be present in the genesis block.
    env.withdrawals = None
    env.beacon_root = None

    env = env.set_fork_requirements(fork)

    new_alloc, state_root = t8n.calc_state_root(
        alloc=to_json(Alloc(self.pre)),
        fork=fork,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )

    genesis = FixtureHeader(
        parent_hash=Hash(0),
        ommers_hash=Hash(EmptyOmmersRoot),
        coinbase=Address(0),
        state_root=Hash(state_root),
        transactions_root=Hash(EmptyTrieRoot),
        receipt_root=Hash(EmptyTrieRoot),
        bloom=Bloom(0),
        difficulty=ZeroPaddedHexNumber(0x20000 if env.difficulty is None else env.difficulty),
        number=ZeroPaddedHexNumber(Number(env.number) - 1),
        gas_limit=ZeroPaddedHexNumber(env.gas_limit),
        gas_used=0,
        timestamp=0,
        extra_data=Bytes([0]),
        mix_digest=Hash(0),
        nonce=HeaderNonce(0),
        base_fee=ZeroPaddedHexNumber.or_none(env.base_fee),
        blob_gas_used=ZeroPaddedHexNumber.or_none(env.blob_gas_used),
        excess_blob_gas=ZeroPaddedHexNumber.or_none(env.excess_blob_gas),
        withdrawals_root=Hash.or_none(
            t8n.calc_withdrawals_root(
                withdrawals=env.withdrawals,
                fork=fork,
                debug_output_path=self.get_next_transition_tool_output_path(),
            )
            if env.withdrawals is not None
            else None
        ),
        beacon_root=Hash.or_none(env.beacon_root),
    )

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

    return Alloc(new_alloc), genesis_rlp, genesis

make_blocks(t8n, genesis, pre, 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,
    pre: Alloc,
    fork: Fork,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], Hash, 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(pre),
        txs=to_json(txs),
        env=to_json(env),
        fork_name=fork.fork(block_number=Number(env.number), timestamp=Number(env.timestamp)),
        chain_id=chain_id,
        reward=fork.get_reward(Number(env.number), Number(env.timestamp)),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )

    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

    env.extra_data = b"\x00"
    header = FixtureHeader.collect(
        fork=fork,
        transition_tool_result=result,
        environment=env,
    )

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

    new_payload: FixtureEngineNewPayload | None = None
    if not self.base_test_config.disable_hive:
        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,
    )

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
    post: Mapping
    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[Alloc, Bytes, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = self.genesis_environment.set_fork_requirements(fork)

        new_alloc, state_root = t8n.calc_state_root(
            alloc=to_json(Alloc(self.pre)),
            fork=fork,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )
        genesis = FixtureHeader(
            parent_hash=Hash(0),
            ommers_hash=Hash(EmptyOmmersRoot),
            coinbase=Address(0),
            state_root=Hash(state_root),
            transactions_root=Hash(EmptyTrieRoot),
            receipt_root=Hash(EmptyTrieRoot),
            bloom=Bloom(0),
            difficulty=ZeroPaddedHexNumber(0x20000 if env.difficulty is None else env.difficulty),
            number=0,
            gas_limit=ZeroPaddedHexNumber(env.gas_limit),
            gas_used=0,
            timestamp=0,
            extra_data=Bytes([0]),
            mix_digest=Hash(0),
            nonce=HeaderNonce(0),
            base_fee=ZeroPaddedHexNumber.or_none(env.base_fee),
            blob_gas_used=ZeroPaddedHexNumber.or_none(env.blob_gas_used),
            excess_blob_gas=ZeroPaddedHexNumber.or_none(env.excess_blob_gas),
            withdrawals_root=Hash.or_none(
                t8n.calc_withdrawals_root(
                    withdrawals=env.withdrawals,
                    fork=fork,
                    debug_output_path=self.get_next_transition_tool_output_path(),
                )
                if env.withdrawals is not None
                else None
            ),
            beacon_root=Hash.or_none(env.beacon_root),
        )

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

        return Alloc(new_alloc), genesis_rlp, genesis

    def make_block(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Dict[str, Any],
        previous_head: Hash,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[FixtureBlock, Environment, Dict[str, Any], Hash]:
        """
        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(txs),
                env=to_json(env),
                fork_name=fork.fork(
                    block_number=Number(env.number), timestamp=Number(env.timestamp)
                ),
                chain_id=chain_id,
                reward=fork.get_reward(Number(env.number), Number(env.timestamp)),
                eips=eips,
                debug_output_path=self.get_next_transition_tool_output_path(),
            )
            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`"
                )
            env.extra_data = block.extra_data
            header = FixtureHeader.collect(
                fork=fork,
                transition_tool_result=result,
                environment=env,
            )

            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 | None = None
            if not self.base_test_config.disable_hive:
                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=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=Number(header.number),
                    ),
                    previous_env,
                    previous_alloc,
                    previous_head,
                )
        else:
            return (
                FixtureBlock(
                    rlp=Bytes(block.rlp),
                    expected_exception=block.exception,
                ),
                previous_env,
                previous_alloc,
                previous_head,
            )

    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        pre: Alloc,
        fork: Fork,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], Hash, 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(pre)
        env = Environment.from_parent_header(genesis)
        blocks: List[FixtureBlock] = []
        head = genesis.hash if genesis.hash is not None else Hash(0)
        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[Alloc, Bytes, FixtureHeader]:
    """
    Create a genesis block from the state test definition.
    """
    env = self.genesis_environment.set_fork_requirements(fork)

    new_alloc, state_root = t8n.calc_state_root(
        alloc=to_json(Alloc(self.pre)),
        fork=fork,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )
    genesis = FixtureHeader(
        parent_hash=Hash(0),
        ommers_hash=Hash(EmptyOmmersRoot),
        coinbase=Address(0),
        state_root=Hash(state_root),
        transactions_root=Hash(EmptyTrieRoot),
        receipt_root=Hash(EmptyTrieRoot),
        bloom=Bloom(0),
        difficulty=ZeroPaddedHexNumber(0x20000 if env.difficulty is None else env.difficulty),
        number=0,
        gas_limit=ZeroPaddedHexNumber(env.gas_limit),
        gas_used=0,
        timestamp=0,
        extra_data=Bytes([0]),
        mix_digest=Hash(0),
        nonce=HeaderNonce(0),
        base_fee=ZeroPaddedHexNumber.or_none(env.base_fee),
        blob_gas_used=ZeroPaddedHexNumber.or_none(env.blob_gas_used),
        excess_blob_gas=ZeroPaddedHexNumber.or_none(env.excess_blob_gas),
        withdrawals_root=Hash.or_none(
            t8n.calc_withdrawals_root(
                withdrawals=env.withdrawals,
                fork=fork,
                debug_output_path=self.get_next_transition_tool_output_path(),
            )
            if env.withdrawals is not None
            else None
        ),
        beacon_root=Hash.or_none(env.beacon_root),
    )

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

    return Alloc(new_alloc), 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: Hash,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[FixtureBlock, Environment, Dict[str, Any], Hash]:
    """
    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(txs),
            env=to_json(env),
            fork_name=fork.fork(
                block_number=Number(env.number), timestamp=Number(env.timestamp)
            ),
            chain_id=chain_id,
            reward=fork.get_reward(Number(env.number), Number(env.timestamp)),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )
        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`"
            )
        env.extra_data = block.extra_data
        header = FixtureHeader.collect(
            fork=fork,
            transition_tool_result=result,
            environment=env,
        )

        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 | None = None
        if not self.base_test_config.disable_hive:
            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=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=Number(header.number),
                ),
                previous_env,
                previous_alloc,
                previous_head,
            )
    else:
        return (
            FixtureBlock(
                rlp=Bytes(block.rlp),
                expected_exception=block.exception,
            ),
            previous_env,
            previous_alloc,
            previous_head,
        )

make_blocks(t8n, genesis, pre, 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,
    pre: Alloc,
    fork: Fork,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], Hash, 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(pre)
    env = Environment.from_parent_header(genesis)
    blocks: List[FixtureBlock] = []
    head = genesis.hash if genesis.hash is not None else Hash(0)
    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)

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: FixedSizeBytesConvertible, salt: FixedSizeBytesConvertible, initcode: BytesConvertible
) -> str:
    """
    Compute address of the resulting contract created using the `CREATE2`
    opcode.
    """
    hash = keccak256(b"\xff" + Address(address) + Hash(salt) + keccak256(Bytes(initcode)))
    return "0x" + hash[-20:].hex()

Yul

Bases: SupportsBytes, Sized

Yul compiler. Compiles Yul source code into bytecode.

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

class Yul(SupportsBytes, Sized):
    """
    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 __bytes__(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 __len__(self) -> int:
        """
        Get the length of the Yul bytecode.
        """
        return len(bytes(self))

    def __add__(self, other: str | bytes | SupportsBytes) -> Code:
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(bytes(self) + to_bytes(other))

    def __radd__(self, other: str | bytes | SupportsBytes) -> Code:
        """
        Adds two code objects together, by converting both to bytes first.
        """
        return Code(to_bytes(other) + bytes(self))

    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

__bytes__()

Assembles using solc --assemble.

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

def __bytes__(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

__len__()

Get the length of the Yul bytecode.

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

def __len__(self) -> int:
    """
    Get the length of the Yul bytecode.
    """
    return len(bytes(self))

__add__(other)

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

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

def __add__(self, other: str | bytes | SupportsBytes) -> Code:
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(bytes(self) + to_bytes(other))

__radd__(other)

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

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

def __radd__(self, other: str | bytes | SupportsBytes) -> Code:
    """
    Adds two code objects together, by converting both to bytes first.
    """
    return Code(to_bytes(other) + bytes(self))

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

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"

BaseTestConfig dataclass

General configuration that all tests must support.

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

@dataclass(kw_only=True)
class BaseTestConfig:
    """
    General configuration that all tests must support.
    """

    disable_hive: bool = False
    """
    Disable any hive-related properties that the output could contain.
    """

disable_hive: bool = False class-attribute instance-attribute

Disable any hive-related properties that the output could contain.

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)

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: BytesConvertible) -> 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
    """
    cost = 0
    for b in Bytes(data):
        if b == 0:
            cost += 4
        else:
            cost += 16
    return cost

BaseTest dataclass

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

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

@dataclass(kw_only=True)
class BaseTest:
    """
    Represents a base Ethereum test which must return a genesis and a
    blockchain.
    """

    pre: Mapping
    tag: str = ""
    base_test_config: BaseTestConfig = field(default_factory=BaseTestConfig)

    # Transition tool specific fields
    t8n_dump_dir: Optional[str] = ""
    t8n_call_counter: Iterator[int] = field(init=False, default_factory=count)

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

    @abstractmethod
    def make_blocks(
        self,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        pre: Alloc,
        fork: Fork,
        chain_id: int = 1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], Hash, 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

    def get_next_transition_tool_output_path(self) -> str:
        """
        Returns the path to the next transition tool output file.
        """
        if not self.t8n_dump_dir:
            return ""
        return path.join(
            self.t8n_dump_dir,
            str(next(self.t8n_call_counter)),
        )

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[Alloc, Bytes, FixtureHeader]:
    """
    Create a genesis block from the test definition.
    """
    pass

make_blocks(t8n, genesis, pre, 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,
    pre: Alloc,
    fork: Fork,
    chain_id: int = 1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], Hash, Dict[str, Any]]:
    """
    Generate the blockchain that must be executed sequentially during test.
    """
    pass

pytest_parameter_name() abstractmethod classmethod

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

get_next_transition_tool_output_path()

Returns the path to the next transition tool output file.

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

def get_next_transition_tool_output_path(self) -> str:
    """
    Returns the path to the next transition tool output file.
    """
    if not self.t8n_dump_dir:
        return ""
    return path.join(
        self.t8n_dump_dir,
        str(next(self.t8n_call_counter)),
    )

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: FixedSizeBytesConvertible) -> str:
    """
    Converts an int or str into proper address 20-byte hex string.
    """
    return str(Address(input))

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: FixedSizeBytesConvertible) -> bytes:
    """
    Converts an int or str into proper 32-byte hash.
    """
    return bytes(Hash(input))

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: FixedSizeBytesConvertible) -> str:
    """
    Converts an int or str into proper 32-byte hash hex string.
    """
    return str(Hash(input))

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 | SupportsBytes | 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) or isinstance(hash, SupportsBytes):
            if isinstance(hash, SupportsBytes):
                hash = bytes(hash)
            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

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: str | bytes | SupportsBytes
    """
    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 __post_init__(self):
        """
        Assemble the bytecode that measures gas usage.
        """
        res = bytes()
        res += bytes(
            [
                0x5A,  # GAS
            ]
        )
        res += 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
            ]
        )
        self.bytecode = res

code: str | bytes | SupportsBytes instance-attribute

Bytecode to be executed to measure the gas usage.

overhead_cost: int = 0 class-attribute instance-attribute

Extra gas cost to be subtracted from extra operations.

extra_stack_items: int = 0 class-attribute instance-attribute

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

sstore_key: int = 0 class-attribute instance-attribute

Storage key to save the gas used.

__post_init__()

Assemble the bytecode that measures gas usage.

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

def __post_init__(self):
    """
    Assemble the bytecode that measures gas usage.
    """
    res = bytes()
    res += bytes(
        [
            0x5A,  # GAS
        ]
    )
    res += 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
        ]
    )
    self.bytecode = res

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)
    MCOPY = Opcode(0x5E, popped_stack_items=3)
    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)

Conditional dataclass

Bases: Code

Helper class used to generate conditional bytecode.

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

@dataclass(kw_only=True)
class Conditional(Code):
    """
    Helper class used to generate conditional bytecode.
    """

    condition: str | bytes | SupportsBytes
    """
    Condition bytecode which must return the true or false condition of the conditional statement.
    """

    if_true: str | bytes | SupportsBytes
    """
    Bytecode to execute if the condition is true.
    """

    if_false: str | bytes | SupportsBytes
    """
    Bytecode to execute if the condition is false.
    """

    def __post_init__(self):
        """
        Assemble the conditional bytecode by generating the necessary jumps and
        jumpdests surrounding the condition and the two possible execution
        paths.

        In the future, PC usage should be replaced by using RJUMP and RJUMPI
        """
        condition_bytes = to_bytes(self.condition)
        if_true_bytes = to_bytes(self.if_true)
        if_false_bytes = to_bytes(self.if_false)

        # First we append a jumpdest to the start of the true branch
        if_true_bytes = Op.JUMPDEST + if_true_bytes

        # Then we append the unconditional jump to the end of the false branch, used to skip the
        # true branch
        if_false_bytes += Op.JUMP(Op.ADD(Op.PC, len(if_true_bytes) + 3))

        # Then we need to do the conditional jump by skipping the false branch
        condition_bytes = Op.JUMPI(Op.ADD(Op.PC, len(if_false_bytes) + 3), condition_bytes)

        # Finally we append the true and false branches, and the condition, plus the jumpdest at
        # the very end
        self.bytecode = condition_bytes + if_false_bytes + if_true_bytes + Op.JUMPDEST

condition: str | bytes | SupportsBytes instance-attribute

Condition bytecode which must return the true or false condition of the conditional statement.

if_true: str | bytes | SupportsBytes instance-attribute

Bytecode to execute if the condition is true.

if_false: str | bytes | SupportsBytes instance-attribute

Bytecode to execute if the condition is false.

__post_init__()

Assemble the conditional bytecode by generating the necessary jumps and jumpdests surrounding the condition and the two possible execution paths.

In the future, PC usage should be replaced by using RJUMP and RJUMPI

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

def __post_init__(self):
    """
    Assemble the conditional bytecode by generating the necessary jumps and
    jumpdests surrounding the condition and the two possible execution
    paths.

    In the future, PC usage should be replaced by using RJUMP and RJUMPI
    """
    condition_bytes = to_bytes(self.condition)
    if_true_bytes = to_bytes(self.if_true)
    if_false_bytes = to_bytes(self.if_false)

    # First we append a jumpdest to the start of the true branch
    if_true_bytes = Op.JUMPDEST + if_true_bytes

    # Then we append the unconditional jump to the end of the false branch, used to skip the
    # true branch
    if_false_bytes += Op.JUMP(Op.ADD(Op.PC, len(if_true_bytes) + 3))

    # Then we need to do the conditional jump by skipping the false branch
    condition_bytes = Op.JUMPI(Op.ADD(Op.PC, len(if_false_bytes) + 3), condition_bytes)

    # Finally we append the true and false branches, and the condition, plus the jumpdest at
    # the very end
    self.bytecode = condition_bytes + if_false_bytes + if_true_bytes + Op.JUMPDEST

Storage

Bases: SupportsJSON

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

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

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

    data: Dict[int, int]

    current_slot: Iterator[int]

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

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

        address: str
        key: int
        want: int
        got: int

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

        def __str__(self):
            """Print exception string"""
            return (
                f"incorrect value in address {self.address} for "
                + f"key {Storage.key_value_to_string(self.key)}:"
                + f" want {Storage.key_value_to_string(self.want)} (dec:{self.want}),"
                + f" got {Storage.key_value_to_string(self.got)} (dec:{self.got})"
            )

    @staticmethod
    def parse_key_value(input: str | int | bytes | SupportsBytes) -> int:
        """
        Parses a key or value to a valid int key for storage.
        """
        if isinstance(input, str):
            input = int(input, 0)
        elif isinstance(input, int):
            pass
        elif isinstance(input, bytes) or isinstance(input, SupportsBytes):
            input = int.from_bytes(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 = {}, start_slot: int = 0):
        """
        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
        self.current_slot = count(start_slot)

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

    def __contains__(self, key: str | int | bytes) -> 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 | bytes) -> 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 | bytes, value: str | int | bytes):  # 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 | bytes):
        """Deletes an item from the storage"""
        del self.data[Storage.parse_key_value(key)]

    def store_next(self, value: str | int | bytes) -> int:
        """
        Stores a value in the storage and returns the key where the value is stored.

        Increments the key counter so the next time this function is called,
        the next key is used.
        """
        self[slot := next(self.current_slot)] = value
        return slot

    def __json__(self, encoder: JSONEncoder) -> 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, address: str, 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(address, key, self.data[key], other.data[key])

    def must_be_equal(self, address: str, 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(address, 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(address, key, self.data[key], 0)

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

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

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

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

    address: str
    key: int
    want: int
    got: int

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

    def __str__(self):
        """Print exception string"""
        return (
            f"incorrect value in address {self.address} for "
            + f"key {Storage.key_value_to_string(self.key)}:"
            + f" want {Storage.key_value_to_string(self.want)} (dec:{self.want}),"
            + f" got {Storage.key_value_to_string(self.got)} (dec:{self.got})"
        )

__str__()

Print exception string

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

def __str__(self):
    """Print exception string"""
    return (
        f"incorrect value in address {self.address} for "
        + f"key {Storage.key_value_to_string(self.key)}:"
        + f" want {Storage.key_value_to_string(self.want)} (dec:{self.want}),"
        + f" got {Storage.key_value_to_string(self.got)} (dec:{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 | SupportsBytes) -> int:
    """
    Parses a key or value to a valid int key for storage.
    """
    if isinstance(input, str):
        input = int(input, 0)
    elif isinstance(input, int):
        pass
    elif isinstance(input, bytes) or isinstance(input, SupportsBytes):
        input = int.from_bytes(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={}, start_slot=0)

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 = {}, start_slot: int = 0):
    """
    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
    self.current_slot = count(start_slot)

__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 | bytes) -> 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 | bytes) -> 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 | bytes, value: str | int | bytes):  # 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 | bytes):
    """Deletes an item from the storage"""
    del self.data[Storage.parse_key_value(key)]

store_next(value)

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

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

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

def store_next(self, value: str | int | bytes) -> int:
    """
    Stores a value in the storage and returns the key where the value is stored.

    Increments the key counter so the next time this function is called,
    the next key is used.
    """
    self[slot := next(self.current_slot)] = value
    return slot

__json__(encoder)

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 __json__(self, encoder: JSONEncoder) -> 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(address, other)

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

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

def must_contain(self, address: str, 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(address, key, self.data[key], other.data[key])

must_be_equal(address, 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, address: str, 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(address, 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(address, key, self.data[key], 0)

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

Account dataclass

State associated with an address.

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

@dataclass(kw_only=True)
class Account:
    """
    State associated with an address.
    """

    nonce: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="nonce",
            cast_type=ZeroPaddedHexNumber,
            default_value=0,
        ),
    )
    """
    The scalar value equal to a) the number of transactions sent by
    an Externally Owned Account, b) the amount of contracts created by a
    contract.
    """
    balance: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="balance",
            cast_type=ZeroPaddedHexNumber,
            default_value=0,
        ),
    )
    """
    The amount of Wei (10<sup>-18</sup> Eth) the account has.
    """
    code: Optional[BytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="code",
            cast_type=Bytes,
            default_value=bytes(),
        ),
    )
    """
    Bytecode contained by the account.
    """
    storage: Optional[Storage | Storage.StorageDictType] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="storage",
            cast_type=Storage,
            to_json=True,
            default_value={},
        ),
    )
    """
    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 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)
            nonce = int(Number(self.nonce))
            if nonce != actual_nonce:
                raise Account.NonceMismatch(
                    address=address,
                    want=nonce,
                    got=actual_nonce,
                )

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

        if self.code is not None:
            expected_code = Bytes(self.code).hex()
            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(address=address, other=actual_storage)

    @classmethod
    def from_dict(cls: Type, data: "Dict | Account") -> "Account":
        """
        Create account from dictionary.
        """
        if isinstance(data, cls):
            return data
        return cls(**data)

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

nonce: Optional[NumberConvertible] = field(default=None, json_encoder=JSONEncoder.Field(name='nonce', cast_type=ZeroPaddedHexNumber, default_value=0)) class-attribute instance-attribute

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

balance: Optional[NumberConvertible] = field(default=None, json_encoder=JSONEncoder.Field(name='balance', cast_type=ZeroPaddedHexNumber, default_value=0)) class-attribute instance-attribute

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

code: Optional[BytesConvertible] = field(default=None, json_encoder=JSONEncoder.Field(name='code', cast_type=Bytes, default_value=bytes())) class-attribute instance-attribute

Bytecode contained by the account.

storage: Optional[Storage | Storage.StorageDictType] = field(default=None, json_encoder=JSONEncoder.Field(name='storage', cast_type=Storage, to_json=True, default_value={})) class-attribute instance-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}"
    )

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)
        nonce = int(Number(self.nonce))
        if nonce != actual_nonce:
            raise Account.NonceMismatch(
                address=address,
                want=nonce,
                got=actual_nonce,
            )

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

    if self.code is not None:
        expected_code = Bytes(self.code).hex()
        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(address=address, other=actual_storage)

from_dict(data) classmethod

Create account from dictionary.

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

@classmethod
def from_dict(cls: Type, data: "Dict | Account") -> "Account":
    """
    Create account from dictionary.
    """
    if isinstance(data, cls):
        return data
    return cls(**data)

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: BytesConvertible) -> "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: NumberConvertible = field(
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    validator: NumberConvertible = field(
        json_encoder=JSONEncoder.Field(
            name="validatorIndex",
            cast_type=HexNumber,
        ),
    )
    address: FixedSizeBytesConvertible = field(
        json_encoder=JSONEncoder.Field(
            cast_type=Address,
        ),
    )
    amount: NumberConvertible = field(
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )

    def to_serializable_list(self) -> List[Any]:
        """
        Returns a list of the withdrawal's attributes in the order they should
        be serialized.
        """
        return [
            Uint(Number(self.index)),
            Uint(Number(self.validator)),
            Address(self.address),
            Uint(Number(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(Number(self.index)),
        Uint(Number(self.validator)),
        Address(self.address),
        Uint(Number(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: FixedSizeBytesConvertible = field(
        default="0x2adc25665018aa1fe0e6bc666dac8fc2697ff9ba",
        json_encoder=JSONEncoder.Field(
            name="currentCoinbase",
            cast_type=Address,
        ),
    )
    gas_limit: NumberConvertible = field(
        default=100000000000000000,
        json_encoder=JSONEncoder.Field(
            name="currentGasLimit",
            cast_type=Number,
        ),
    )
    number: NumberConvertible = field(
        default=1,
        json_encoder=JSONEncoder.Field(
            name="currentNumber",
            cast_type=Number,
        ),
    )
    timestamp: NumberConvertible = field(
        default=1000,
        json_encoder=JSONEncoder.Field(
            name="currentTimestamp",
            cast_type=Number,
        ),
    )
    prev_randao: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="currentRandom",
            cast_type=Number,
        ),
    )
    difficulty: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="currentDifficulty",
            cast_type=Number,
        ),
    )
    block_hashes: Dict[NumberConvertible, FixedSizeBytesConvertible] = field(
        default_factory=dict,
        json_encoder=JSONEncoder.Field(
            name="blockHashes",
            cast_type=lambda x: {str(Number(k)): str(Hash(v)) for k, v in x.items()},
            skip_string_convert=True,
        ),
    )
    ommers: List[FixedSizeBytesConvertible] = field(
        default_factory=list,
        json_encoder=JSONEncoder.Field(
            name="ommers",
            cast_type=lambda x: [str(Hash(y)) for y in x],
            skip_string_convert=True,
        ),
    )
    withdrawals: Optional[List[Withdrawal]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="withdrawals",
            to_json=True,
        ),
    )

    base_fee: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="currentBaseFee",
            cast_type=Number,
        ),
    )
    parent_difficulty: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentDifficulty",
            cast_type=Number,
        ),
    )

    parent_timestamp: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentTimestamp",
            cast_type=Number,
        ),
    )
    parent_base_fee: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentBaseFee",
            cast_type=Number,
        ),
    )
    parent_gas_used: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentGasUsed",
            cast_type=Number,
        ),
    )
    parent_gas_limit: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentGasLimit",
            cast_type=Number,
        ),
    )
    parent_ommers_hash: FixedSizeBytesConvertible = field(
        default=0,
        json_encoder=JSONEncoder.Field(
            name="parentUncleHash",
            cast_type=Hash,
        ),
    )
    parent_blob_gas_used: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentBlobGasUsed",
            cast_type=Number,
        ),
    )
    parent_excess_blob_gas: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentExcessBlobGas",
            cast_type=Number,
        ),
    )
    blob_gas_used: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="currentBlobGasUsed",
            cast_type=Number,
        ),
    )
    excess_blob_gas: Optional[NumberConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="currentExcessBlobGas",
            cast_type=Number,
        ),
    )
    beacon_root: Optional[FixedSizeBytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="beaconRoot",
            cast_type=Hash,
        ),
    )
    extra_data: Optional[BytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )

    @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_blob_gas_used=parent.blob_gas_used,
            parent_excess_blob_gas=parent.excess_blob_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 0},
        )

    def parent_hash(self) -> bytes:
        """
        Obtains 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([Number(k) for k in self.block_hashes.keys()])
        return Hash(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_blob_gas_used = new_parent.blob_gas_used
        env.parent_excess_blob_gas = new_parent.excess_blob_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 0
        return env

    def set_fork_requirements(self, fork: Fork) -> "Environment":
        """
        Fills the required fields in an environment depending on the fork.
        """
        res = copy(self)
        number = Number(self.number)
        timestamp = Number(self.timestamp)
        if fork.header_prev_randao_required(number, timestamp) and res.prev_randao is None:
            res.prev_randao = 0

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

        if (
            fork.header_base_fee_required(number, 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(number, timestamp):
            res.difficulty = 0

        if (
            fork.header_excess_blob_gas_required(number, timestamp)
            and res.excess_blob_gas is None
            and res.parent_excess_blob_gas is None
        ):
            res.excess_blob_gas = 0

        if (
            fork.header_blob_gas_used_required(number, timestamp)
            and res.blob_gas_used is None
            and res.parent_blob_gas_used is None
        ):
            res.blob_gas_used = 0

        if fork.header_beacon_root_required(number, timestamp) and res.beacon_root is None:
            res.beacon_root = 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_blob_gas_used=parent.blob_gas_used,
        parent_excess_blob_gas=parent.excess_blob_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 0},
    )

parent_hash()

Obtains 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:
    """
    Obtains 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([Number(k) for k in self.block_hashes.keys()])
    return Hash(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_blob_gas_used = new_parent.blob_gas_used
    env.parent_excess_blob_gas = new_parent.excess_blob_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 0
    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)
    number = Number(self.number)
    timestamp = Number(self.timestamp)
    if fork.header_prev_randao_required(number, timestamp) and res.prev_randao is None:
        res.prev_randao = 0

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

    if (
        fork.header_base_fee_required(number, 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(number, timestamp):
        res.difficulty = 0

    if (
        fork.header_excess_blob_gas_required(number, timestamp)
        and res.excess_blob_gas is None
        and res.parent_excess_blob_gas is None
    ):
        res.excess_blob_gas = 0

    if (
        fork.header_blob_gas_used_required(number, timestamp)
        and res.blob_gas_used is None
        and res.parent_blob_gas_used is None
    ):
        res.blob_gas_used = 0

    if fork.header_beacon_root_required(number, timestamp) and res.beacon_root is None:
        res.beacon_root = 0

    return res

AccessList dataclass

Access List for transactions.

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

@dataclass(kw_only=True)
class AccessList:
    """
    Access List for transactions.
    """

    address: FixedSizeBytesConvertible = field(
        json_encoder=JSONEncoder.Field(
            cast_type=Address,
        ),
    )
    storage_keys: List[FixedSizeBytesConvertible] = field(
        default_factory=list,
        json_encoder=JSONEncoder.Field(
            name="storageKeys",
            cast_type=lambda x: [str(Hash(k)) for k in x],
            skip_string_convert=True,
        ),
    )

    def to_list(self) -> List[bytes | List[bytes]]:
        """
        Returns the access list as a list of serializable elements.
        """
        return [Address(self.address), [Hash(k) for k in self.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 [Address(self.address), [Hash(k) for k in 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] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="type",
            cast_type=HexNumber,
        ),
    )
    """
    Transaction type value.
    """
    chain_id: int = field(
        default=1,
        json_encoder=JSONEncoder.Field(
            name="chainId",
            cast_type=HexNumber,
        ),
    )
    nonce: int = field(
        default=0,
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    gas_price: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="gasPrice",
            cast_type=HexNumber,
        ),
    )
    max_priority_fee_per_gas: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="maxPriorityFeePerGas",
            cast_type=HexNumber,
        ),
    )
    max_fee_per_gas: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="maxFeePerGas",
            cast_type=HexNumber,
        ),
    )
    gas_limit: int = field(
        default=21000,
        json_encoder=JSONEncoder.Field(
            name="gas",
            cast_type=HexNumber,
        ),
    )
    to: Optional[FixedSizeBytesConvertible] = field(
        default=AddrAA,
        json_encoder=JSONEncoder.Field(
            cast_type=Address,
        ),
    )
    value: int = field(
        default=0,
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    data: BytesConvertible = field(
        default_factory=bytes,
        json_encoder=JSONEncoder.Field(
            name="input",
            cast_type=Bytes,
        ),
    )
    access_list: Optional[List[AccessList]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="accessList",
            to_json=True,
        ),
    )
    max_fee_per_blob_gas: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="maxFeePerBlobGas",
            cast_type=HexNumber,
        ),
    )
    blob_versioned_hashes: Optional[Sequence[FixedSizeBytesConvertible]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="blobVersionedHashes",
            cast_type=lambda x: [Hash(k) for k in x],
            to_json=True,
        ),
    )
    v: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    r: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    s: Optional[int] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            cast_type=HexNumber,
        ),
    )
    wrapped_blob_transaction: bool = field(
        default=False,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    blobs: Optional[Sequence[bytes]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    blob_kzg_commitments: Optional[Sequence[bytes]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    blob_kzg_proofs: Optional[Sequence[bytes]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    sender: Optional[FixedSizeBytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            cast_type=Address,
        ),
    )
    secret_key: Optional[FixedSizeBytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="secretKey",
            cast_type=Hash,
        ),
    )
    protected: bool = field(
        default=True,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    error: Optional[str] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )

    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
            or self.max_priority_fee_per_gas is not None
            or self.max_fee_per_blob_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
            and self.max_fee_per_blob_gas is None
        ):
            self.gas_price = 10

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

        if self.v 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_blob_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

        # Set default values for fields that are required for certain tx types
        if self.ty >= 1 and self.access_list is None:
            self.access_list = []

        if self.ty >= 2 and self.max_priority_fee_per_gas is None:
            self.max_priority_fee_per_gas = 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.v is None or self.r is None or self.s 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(self.to) if self.to is not None else bytes()

        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_blob_gas is None:
                raise ValueError("max_fee_per_blob_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.access_list is None:
                raise ValueError("access_list 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),
                        Bytes(self.data),
                        [a.to_list() for a in self.access_list],
                        Uint(self.max_fee_per_blob_gas),
                        [Hash(h) for h in self.blob_versioned_hashes],
                        Uint(self.v),
                        Uint(self.r),
                        Uint(self.s),
                    ],
                    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),
                    Bytes(self.data),
                    [a.to_list() for a in self.access_list],
                    Uint(self.max_fee_per_blob_gas),
                    [Hash(h) for h in self.blob_versioned_hashes],
                    Uint(self.v),
                    Uint(self.r),
                    Uint(self.s),
                ]
        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 2 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 2 tx")
            if self.access_list is None:
                raise ValueError("access_list must be set for type 2 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),
                Bytes(self.data),
                [a.to_list() for a in self.access_list],
                Uint(self.v),
                Uint(self.r),
                Uint(self.s),
            ]
        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")
            if self.access_list is None:
                raise ValueError("access_list 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),
                Bytes(self.data),
                [a.to_list() for a in self.access_list],
                Uint(self.v),
                Uint(self.r),
                Uint(self.s),
            ]
        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),
                Bytes(self.data),
                Uint(self.v),
                Uint(self.r),
                Uint(self.s),
            ]

        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(self.to) if self.to is not None else bytes()

        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_blob_gas is None:
                raise ValueError("max_fee_per_blob_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),
                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_blob_gas),
                [Hash(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 2 tx")
            if self.max_fee_per_gas is None:
                raise ValueError("max_fee_per_gas must be set for type 2 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),
                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),
                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),
                    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),
                    Bytes(self.data),
                ]
        raise NotImplementedError("signing 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.v 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(
            secret=Hash(tx.secret_key) if tx.secret_key is not None else bytes(32)
        )
        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 :]

        tx.v, tx.r, tx.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:
                tx.v += 35 + (tx.chain_id * 2)
            else:  # not protected
                tx.v += 27

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

ty: Optional[int] = field(default=None, json_encoder=JSONEncoder.Field(name='type', cast_type=HexNumber)) class-attribute instance-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
        or self.max_priority_fee_per_gas is not None
        or self.max_fee_per_blob_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
        and self.max_fee_per_blob_gas is None
    ):
        self.gas_price = 10

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

    if self.v 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_blob_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

    # Set default values for fields that are required for certain tx types
    if self.ty >= 1 and self.access_list is None:
        self.access_list = []

    if self.ty >= 2 and self.max_priority_fee_per_gas is None:
        self.max_priority_fee_per_gas = 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.v is None or self.r is None or self.s 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(self.to) if self.to is not None else bytes()

    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_blob_gas is None:
            raise ValueError("max_fee_per_blob_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.access_list is None:
            raise ValueError("access_list 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),
                    Bytes(self.data),
                    [a.to_list() for a in self.access_list],
                    Uint(self.max_fee_per_blob_gas),
                    [Hash(h) for h in self.blob_versioned_hashes],
                    Uint(self.v),
                    Uint(self.r),
                    Uint(self.s),
                ],
                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),
                Bytes(self.data),
                [a.to_list() for a in self.access_list],
                Uint(self.max_fee_per_blob_gas),
                [Hash(h) for h in self.blob_versioned_hashes],
                Uint(self.v),
                Uint(self.r),
                Uint(self.s),
            ]
    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 2 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 2 tx")
        if self.access_list is None:
            raise ValueError("access_list must be set for type 2 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),
            Bytes(self.data),
            [a.to_list() for a in self.access_list],
            Uint(self.v),
            Uint(self.r),
            Uint(self.s),
        ]
    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")
        if self.access_list is None:
            raise ValueError("access_list 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),
            Bytes(self.data),
            [a.to_list() for a in self.access_list],
            Uint(self.v),
            Uint(self.r),
            Uint(self.s),
        ]
    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),
            Bytes(self.data),
            Uint(self.v),
            Uint(self.r),
            Uint(self.s),
        ]

    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(self.to) if self.to is not None else bytes()

    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_blob_gas is None:
            raise ValueError("max_fee_per_blob_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),
            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_blob_gas),
            [Hash(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 2 tx")
        if self.max_fee_per_gas is None:
            raise ValueError("max_fee_per_gas must be set for type 2 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),
            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),
            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),
                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),
                Bytes(self.data),
            ]
    raise NotImplementedError("signing 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.v 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(
        secret=Hash(tx.secret_key) if tx.secret_key is not None else bytes(32)
    )
    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 :]

    tx.v, tx.r, tx.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:
            tx.v += 35 + (tx.chain_id * 2)
        else:  # not protected
            tx.v += 27

    # 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[FixedSizeBytesConvertible] = None
    ommers_hash: Optional[FixedSizeBytesConvertible] = None
    coinbase: Optional[FixedSizeBytesConvertible] = None
    state_root: Optional[FixedSizeBytesConvertible] = None
    transactions_root: Optional[FixedSizeBytesConvertible] = None
    receipt_root: Optional[FixedSizeBytesConvertible] = None
    bloom: Optional[FixedSizeBytesConvertible] = None
    difficulty: Optional[NumberConvertible] = None
    number: Optional[NumberConvertible] = None
    gas_limit: Optional[NumberConvertible] = None
    gas_used: Optional[NumberConvertible] = None
    timestamp: Optional[NumberConvertible] = None
    extra_data: Optional[BytesConvertible] = None
    mix_digest: Optional[FixedSizeBytesConvertible] = None
    nonce: Optional[FixedSizeBytesConvertible] = None
    base_fee: Optional[NumberConvertible | Removable] = None
    withdrawals_root: Optional[FixedSizeBytesConvertible | Removable] = None
    blob_gas_used: Optional[NumberConvertible | Removable] = None
    excess_blob_gas: Optional[NumberConvertible | Removable] = None
    beacon_root: Optional[FixedSizeBytesConvertible | Removable] = None
    hash: Optional[FixedSizeBytesConvertible] = 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[BytesConvertible] = 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_blob_gas, Removable):
            new_env.excess_blob_gas = self.excess_blob_gas
        if not isinstance(self.blob_gas_used, Removable):
            new_env.blob_gas_used = self.blob_gas_used
        if not isinstance(self.beacon_root, Removable):
            new_env.beacon_root = self.beacon_root
        """
        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([Number(n) for n in 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 = int(Number(new_env.parent_timestamp) + 12)

        return new_env

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

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

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

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

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

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

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

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

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

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

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

List of transactions included in the block.

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

List of ommer headers included in the block.

withdrawals: Optional[List[Withdrawal]] = None class-attribute instance-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_blob_gas, Removable):
        new_env.excess_blob_gas = self.excess_blob_gas
    if not isinstance(self.blob_gas_used, Removable):
        new_env.blob_gas_used = self.blob_gas_used
    if not isinstance(self.beacon_root, Removable):
        new_env.beacon_root = self.beacon_root
    """
    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([Number(n) for n in 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 = int(Number(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 | BytesConvertible | None) -> "Block":
    """
    Creates a copy of the block and adds the specified RLP.
    """
    new_block = deepcopy(self)
    new_block.rlp = Bytes.or_none(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 = field(
        json_encoder=JSONEncoder.Field(
            name="executionPayload",
            to_json=True,
        )
    )
    blob_versioned_hashes: Optional[List[FixedSizeBytesConvertible]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="expectedBlobVersionedHashes",
            cast_type=lambda hashes: [Hash(hash) for hash in hashes],
            to_json=True,
        ),
    )
    version: int = field(
        json_encoder=JSONEncoder.Field(),
    )
    beacon_root: Optional[FixedSizeBytesConvertible] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="parentBeaconBlockRoot",
            cast_type=Hash,
        ),
    )
    error_code: Optional[EngineAPIError] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="errorCode",
            cast_type=int,
        ),
    )

    @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.from_fixture_header(
                header=replace(header, beacon_root=None),
                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
            )

        if fork.engine_new_payload_beacon_root(header.number, header.timestamp):
            new_payload.beacon_root = header.beacon_root

        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.from_fixture_header(
            header=replace(header, beacon_root=None),
            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
        )

    if fork.engine_new_payload_beacon_root(header.number, header.timestamp):
        new_payload.beacon_root = header.beacon_root

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

    info: Dict[str, str] = field(
        default_factory=dict,
        json_encoder=JSONEncoder.Field(
            name="_info",
            to_json=True,
        ),
    )
    blocks: List[FixtureBlock] = field(
        json_encoder=JSONEncoder.Field(
            name="blocks",
            to_json=True,
        ),
    )
    genesis: FixtureHeader = field(
        json_encoder=JSONEncoder.Field(
            name="genesisBlockHeader",
            to_json=True,
        ),
    )
    genesis_rlp: Bytes = field(
        json_encoder=JSONEncoder.Field(
            name="genesisRLP",
        ),
    )
    head: Hash = field(
        json_encoder=JSONEncoder.Field(
            name="lastblockhash",
        ),
    )
    fork: str = field(
        json_encoder=JSONEncoder.Field(
            name="network",
        ),
    )
    pre_state: Mapping[str, Account] = field(
        json_encoder=JSONEncoder.Field(
            name="pre",
            cast_type=Alloc,
            to_json=True,
        ),
    )
    post_state: Optional[Mapping[str, Account]] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            name="postState",
            cast_type=Alloc,
            to_json=True,
        ),
    )
    seal_engine: str = field(
        json_encoder=JSONEncoder.Field(
            name="sealEngine",
        ),
    )
    name: str = field(
        default="",
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )

    _json: Dict[str, Any] | None = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )

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

    def to_json(self) -> Dict[str, Any]:
        """
        Convert to JSON.
        """
        assert self._json is not None, "Fixture not initialized"
        self._json["_info"] = self.info
        return self._json

    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)

to_json()

Convert to JSON.

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

def to_json(self) -> Dict[str, Any]:
    """
    Convert to JSON.
    """
    assert self._json is not None, "Fixture not initialized"
    self._json["_info"] = self.info
    return self._json

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)