Ethereum Test Tools Package

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

CalldataCase dataclass

Small helper class to represent a single case whose condition depends on the value of the contract's calldata in a Switch case statement.

By default the calldata is read from position zero, but this can be overridden using position.

The condition is generated automatically based on the value (and optionally position) and may not be set directly.

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

@dataclass
class CalldataCase:
    """
    Small helper class to represent a single case whose condition depends
    on the value of the contract's calldata in a Switch case statement.

    By default the calldata is read from position zero, but this can be
    overridden using `position`.

    The `condition` is generated automatically based on the `value` (and
    optionally `position`) and may not be set directly.
    """

    action: str | bytes | SupportsBytes
    value: int | str | bytes | SupportsBytes
    position: int = 0
    condition: bytes = field(init=False)

    def __post_init__(self):
        """
        Generate the condition base on `value` and `position`.
        """
        value_as_bytes = self.value
        if not isinstance(self.value, int):
            value_as_bytes = Op.PUSH32(to_bytes(self.value))
        self.condition = Op.EQ(Op.CALLDATALOAD(self.position), value_as_bytes)
        self.action = to_bytes(self.action)

__post_init__()

Generate the condition base on value and position.

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

def __post_init__(self):
    """
    Generate the condition base on `value` and `position`.
    """
    value_as_bytes = self.value
    if not isinstance(self.value, int):
        value_as_bytes = Op.PUSH32(to_bytes(self.value))
    self.condition = Op.EQ(Op.CALLDATALOAD(self.position), value_as_bytes)
    self.action = to_bytes(self.action)

Case dataclass

Small helper class to represent a single, generic case in a Switch cases list.

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

@dataclass
class Case:
    """
    Small helper class to represent a single, generic case in a `Switch` cases
    list.
    """

    condition: str | bytes | SupportsBytes
    action: str | bytes | SupportsBytes

    def __post_init__(self):
        """
        Ensure that the condition and action are of type bytes.
        """
        self.condition = to_bytes(self.condition)
        self.action = to_bytes(self.action)

__post_init__()

Ensure that the condition and action are of type bytes.

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

def __post_init__(self):
    """
    Ensure that the condition and action are of type bytes.
    """
    self.condition = to_bytes(self.condition)
    self.action = to_bytes(self.action)

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.

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

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

__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

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

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

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 jump and
        jumpdest opcodes 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 jump and jumpdest opcodes 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 jump and
    jumpdest opcodes 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

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,
        initcode_prefix: str | bytes | SupportsBytes = b"",
        initcode_prefix_execution_gas: int = 0,
        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 = initcode_prefix_execution_gas
        self.deploy_code = deploy_code
        deploy_code_bytes = to_bytes(self.deploy_code)
        code_length = len(deploy_code_bytes)

        initcode_prefix_bytes = to_bytes(initcode_prefix)
        initcode = bytearray(initcode_prefix_bytes)

        # 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 + len(initcode_prefix_bytes) (constant)
        no_prefix_length = 0x0B
        assert no_prefix_length + len(initcode_prefix_bytes) <= 0xFF, "initcode prefix too long"
        initcode.append(0x60)
        initcode.append(no_prefix_length + len(initcode_prefix_bytes))
        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

        initcode_plus_deploy_code = bytes(initcode) + deploy_code_bytes
        padding_bytes = bytes()

        if initcode_length is not None:
            assert initcode_length >= len(
                initcode_plus_deploy_code
            ), "specified invalid length for initcode"

            padding_bytes = bytes(
                [padding_byte] * (initcode_length - len(initcode_plus_deploy_code))
            )

        self.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(deploy_code_bytes)

        super().__init__(initcode_plus_deploy_code + padding_bytes, name=name)

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

Bytecode to be deployed by the initcode.

execution_gas: int = initcode_prefix_execution_gas instance-attribute

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

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

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

__init__(*, deploy_code, initcode_length=None, initcode_prefix=b'', initcode_prefix_execution_gas=0, 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,
    initcode_prefix: str | bytes | SupportsBytes = b"",
    initcode_prefix_execution_gas: int = 0,
    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 = initcode_prefix_execution_gas
    self.deploy_code = deploy_code
    deploy_code_bytes = to_bytes(self.deploy_code)
    code_length = len(deploy_code_bytes)

    initcode_prefix_bytes = to_bytes(initcode_prefix)
    initcode = bytearray(initcode_prefix_bytes)

    # 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 + len(initcode_prefix_bytes) (constant)
    no_prefix_length = 0x0B
    assert no_prefix_length + len(initcode_prefix_bytes) <= 0xFF, "initcode prefix too long"
    initcode.append(0x60)
    initcode.append(no_prefix_length + len(initcode_prefix_bytes))
    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

    initcode_plus_deploy_code = bytes(initcode) + deploy_code_bytes
    padding_bytes = bytes()

    if initcode_length is not None:
        assert initcode_length >= len(
            initcode_plus_deploy_code
        ), "specified invalid length for initcode"

        padding_bytes = bytes(
            [padding_byte] * (initcode_length - len(initcode_plus_deploy_code))
        )

    self.deployment_gas = GAS_PER_DEPLOYED_CODE_BYTE * len(deploy_code_bytes)

    super().__init__(initcode_plus_deploy_code + padding_bytes, name=name)

Switch dataclass

Bases: Code

Helper class used to generate switch-case expressions in EVM bytecode.

Switch-case behavior

• If no condition is met in the list of BytecodeCases conditions, the default_action bytecode is executed.
• If multiple conditions are met, the action from the first valid condition is the only one executed.
• There is no fall through; it is not possible to execute multiple actions.

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

@dataclass(kw_only=True)
class Switch(Code):
    """
    Helper class used to generate switch-case expressions in EVM bytecode.

    Switch-case behavior:
        - If no condition is met in the list of BytecodeCases conditions,
            the `default_action` bytecode is executed.
        - If multiple conditions are met, the action from the first valid
            condition is the only one executed.
        - There is no fall through; it is not possible to execute multiple
            actions.
    """

    default_action: str | bytes | SupportsBytes
    """
    The default bytecode to execute; if no condition is met, this bytecode is
    executed.
    """

    cases: List[Case | CalldataCase]
    """
    A list of Case or CalldataCase: The first element with a condition that
    evaluates to a non-zero value is the one that is executed.
    """

    def __post_init__(self):
        """
        Assemble the bytecode by looping over the list of cases and adding
        the necessary JUMPI and JUMPDEST opcodes in order to replicate
        switch-case behavior.

        In the future, PC usage should be replaced by using RJUMP and RJUMPI.
        """
        # The length required to jump over subsequent actions to the final JUMPDEST at the end
        # of the switch-case block:
        # - add 6 per case for the length of the JUMPDEST and JUMP(ADD(PC, action_jump_length))
        #   bytecode
        # - add 3 to the total to account for this action's JUMP; the PC within the call
        #   requires a "correction" of 3.
        action_jump_length = sum(len(case.action) + 6 for case in self.cases) + 3

        # All conditions get pre-pended to this bytecode; if none are met, we reach the default
        self.bytecode = to_bytes(self.default_action) + Op.JUMP(Op.ADD(Op.PC, action_jump_length))

        # The length required to jump over the default action and its JUMP bytecode
        condition_jump_length = len(self.bytecode) + 3

        # Reversed: first case in the list has priority; it will become the outer-most onion layer.
        # We build up layers around the default_action, after 1 iteration of the loop, a simplified
        # representation of the bytecode is:
        #
        #  JUMPI(case[n-1].condition)
        #  + default_action + JUMP()
        #  + JUMPDEST + case[n-1].action + JUMP()
        #
        # and after n=len(cases) iterations:
        #
        #  JUMPI(case[0].condition)
        #  + JUMPI(case[1].condition)
        #    ...
        #  + JUMPI(case[n-1].condition)
        #  + default_action + JUMP()
        #  + JUMPDEST + case[n-1].action + JUMP()
        #  + ...
        #  + JUMPDEST + case[1].action + JUMP()
        #  + JUMPDEST + case[0].action + JUMP()
        #
        for case in reversed(self.cases):
            action_jump_length -= len(case.action) + 6
            action = Op.JUMPDEST + case.action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
            condition = Op.JUMPI(Op.ADD(Op.PC, condition_jump_length), case.condition)
            # wrap the current case around the onion as its next layer
            self.bytecode = condition + self.bytecode + action
            condition_jump_length += len(condition) + len(action)

        self.bytecode += Op.JUMPDEST

default_action: str | bytes | SupportsBytes instance-attribute

The default bytecode to execute; if no condition is met, this bytecode is executed.

cases: List[Case | CalldataCase] instance-attribute

A list of Case or CalldataCase: The first element with a condition that evaluates to a non-zero value is the one that is executed.

__post_init__()

Assemble the bytecode by looping over the list of cases and adding the necessary JUMPI and JUMPDEST opcodes in order to replicate switch-case behavior.

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 bytecode by looping over the list of cases and adding
    the necessary JUMPI and JUMPDEST opcodes in order to replicate
    switch-case behavior.

    In the future, PC usage should be replaced by using RJUMP and RJUMPI.
    """
    # The length required to jump over subsequent actions to the final JUMPDEST at the end
    # of the switch-case block:
    # - add 6 per case for the length of the JUMPDEST and JUMP(ADD(PC, action_jump_length))
    #   bytecode
    # - add 3 to the total to account for this action's JUMP; the PC within the call
    #   requires a "correction" of 3.
    action_jump_length = sum(len(case.action) + 6 for case in self.cases) + 3

    # All conditions get pre-pended to this bytecode; if none are met, we reach the default
    self.bytecode = to_bytes(self.default_action) + Op.JUMP(Op.ADD(Op.PC, action_jump_length))

    # The length required to jump over the default action and its JUMP bytecode
    condition_jump_length = len(self.bytecode) + 3

    # Reversed: first case in the list has priority; it will become the outer-most onion layer.
    # We build up layers around the default_action, after 1 iteration of the loop, a simplified
    # representation of the bytecode is:
    #
    #  JUMPI(case[n-1].condition)
    #  + default_action + JUMP()
    #  + JUMPDEST + case[n-1].action + JUMP()
    #
    # and after n=len(cases) iterations:
    #
    #  JUMPI(case[0].condition)
    #  + JUMPI(case[1].condition)
    #    ...
    #  + JUMPI(case[n-1].condition)
    #  + default_action + JUMP()
    #  + JUMPDEST + case[n-1].action + JUMP()
    #  + ...
    #  + JUMPDEST + case[1].action + JUMP()
    #  + JUMPDEST + case[0].action + JUMP()
    #
    for case in reversed(self.cases):
        action_jump_length -= len(case.action) + 6
        action = Op.JUMPDEST + case.action + Op.JUMP(Op.ADD(Op.PC, action_jump_length))
        condition = Op.JUMPI(Op.ADD(Op.PC, condition_jump_length), case.condition)
        # wrap the current case around the onion as its next layer
        self.bytecode = condition + self.bytecode + action
        condition_jump_length += len(condition) + len(action)

    self.bytecode += Op.JUMPDEST

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) -> Version:
        """
        Return solc's version string
        """
        result = run(
            [self.binary, "--version"],
            stdout=PIPE,
            stderr=PIPE,
        )
        solc_output = result.stdout.decode().split("\n")
        version_pattern = r"Version: (.*)"
        solc_version_string = ""
        for line in solc_output:
            if match := re.search(version_pattern, line):
                solc_version_string = match.group(1)
                break
        if not solc_version_string:
            warnings.warn("Unable to determine solc version.")
            return Version(0)
        # Sanitize
        solc_version_string = solc_version_string.replace("g++", "gpp")
        return Version.parse(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) -> Version:
    """
    Return solc's version string
    """
    result = run(
        [self.binary, "--version"],
        stdout=PIPE,
        stderr=PIPE,
    )
    solc_output = result.stdout.decode().split("\n")
    version_pattern = r"Version: (.*)"
    solc_version_string = ""
    for line in solc_output:
        if match := re.search(version_pattern, line):
            solc_version_string = match.group(1)
            break
    if not solc_version_string:
        warnings.warn("Unable to determine solc version.")
        return Version(0)
    # Sanitize
    solc_version_string = solc_version_string.replace("g++", "gpp")
    return Version.parse(solc_version_string)

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

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

    @dataclass(kw_only=True)
    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}"
            )

    @dataclass(kw_only=True)
    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}"
            )

    @dataclass(kw_only=True)
    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)

    def is_empty(self: "Account") -> bool:
        """
        Returns true if an account deemed empty.
        """
        return (
            (self.nonce == 0 or self.nonce is None)
            and (self.balance == 0 or self.balance is None)
            and (not self.code and self.code is None)
            and (not self.storage or self.storage == {} or self.storage is None)
        )

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

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

        def to_kwargs_dict(account: "Dict | Account | None") -> Dict:
            if account is None:
                return {}
            if isinstance(account, dict):
                return account
            elif isinstance(account, cls):
                return {
                    f.name: v for f in fields(cls) if (v := getattr(account, f.name)) is not None
                }
            raise TypeError(f"Unexpected type for account merge: {type(account)}")

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

        return cls(**kwargs)

nonce: 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 dataclass

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

@dataclass(kw_only=True)
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 dataclass

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

@dataclass(kw_only=True)
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 dataclass

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

@dataclass(kw_only=True)
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)

is_empty()

Returns true if an account deemed empty.

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

def is_empty(self: "Account") -> bool:
    """
    Returns true if an account deemed empty.
    """
    return (
        (self.nonce == 0 or self.nonce is None)
        and (self.balance == 0 or self.balance is None)
        and (not self.code and self.code is None)
        and (not self.storage or self.storage == {} or self.storage is None)
    )

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)

merge(account_1, account_2) classmethod

Create a merged account from two sources.

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

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

    def to_kwargs_dict(account: "Dict | Account | None") -> Dict:
        if account is None:
            return {}
        if isinstance(account, dict):
            return account
        elif isinstance(account, cls):
            return {
                f.name: v for f in fields(cls) if (v := getattr(account, f.name)) is not None
            }
        raise TypeError(f"Unexpected type for account merge: {type(account)}")

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

    return cls(**kwargs)

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"

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

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="parentBeaconBlockRoot",
            cast_type=Hash,
        ),
    )
    extra_data: Optional[BytesConvertible] = field(
        default=b"\x00",
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )

    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 set_fork_requirements(self, fork: Fork, in_place: bool = False) -> "Environment":
        """
        Fills the required fields in an environment depending on the fork.
        """
        res = self if in_place else copy(self)
        number = Number(res.number)
        timestamp = Number(res.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
        elif res.difficulty is None and res.parent_difficulty is None:
            res.difficulty = 0x20000

        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

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

set_fork_requirements(fork, in_place=False)

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, in_place: bool = False) -> "Environment":
    """
    Fills the required fields in an environment depending on the fork.
    """
    res = self if in_place else copy(self)
    number = Number(res.number)
    timestamp = Number(res.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
    elif res.difficulty is None and res.parent_difficulty is None:
        res.difficulty = 0x20000

    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

JSONEncoder

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

Removable

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

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

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

    pass

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

    @dataclass(kw_only=True)
    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}"

    @dataclass(kw_only=True)
    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}"

    @dataclass(kw_only=True)
    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}
            """

    @dataclass(kw_only=True)
    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))

    @dataclass(kw_only=True)
    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(key_or_value=input)

        if input > MAX_STORAGE_KEY_VALUE or input < MIN_STORAGE_KEY_VALUE:
            raise Storage.InvalidValue(key_or_value=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 | "Storage" = {}, 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 __iter__(self) -> Iterator[int]:
        """Returns iterator of the storage"""
        return iter(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_1=key_repr, val_1=res[key_repr], key_2=key, val_2=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=key)
            elif self.data[key] != other.data[key]:
                raise Storage.KeyValueMismatch(
                    address=address, key=key, want=self.data[key], got=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=address, key=key, want=self.data[key], got=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=address, key=key, want=self.data[key], got=0
                    )

            elif other.data[key] != 0:
                raise Storage.KeyValueMismatch(
                    address=address, key=key, want=0, got=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 dataclass

Bases: Exception

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

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

@dataclass(kw_only=True)
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 dataclass

Bases: Exception

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

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

@dataclass(kw_only=True)
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 dataclass

Bases: Exception

Key is represented twice in the storage.

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

@dataclass(kw_only=True)
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 dataclass

Bases: Exception

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

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

@dataclass(kw_only=True)
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 dataclass

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

@dataclass(kw_only=True)
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(key_or_value=input)

    if input > MAX_STORAGE_KEY_VALUE or input < MIN_STORAGE_KEY_VALUE:
        raise Storage.InvalidValue(key_or_value=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 | "Storage" = {}, 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)

__iter__()

Returns iterator of the storage

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

def __iter__(self) -> Iterator[int]:
    """Returns iterator of the storage"""
    return iter(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_1=key_repr, val_1=res[key_repr], key_2=key, val_2=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=key)
        elif self.data[key] != other.data[key]:
            raise Storage.KeyValueMismatch(
                address=address, key=key, want=self.data[key], got=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=address, key=key, want=self.data[key], got=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=address, key=key, want=self.data[key], got=0
                )

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

TestParameterGroup dataclass

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

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

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

    __test__ = False  # explicitly prevent pytest collecting this class

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

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

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

__repr__()

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

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

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

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

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

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[TransactionException | ExceptionList] = field(
        default=None,
        json_encoder=JSONEncoder.Field(
            skip=True,
        ),
    )
    rlp: Optional[bytes] = 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: TransactionException | ExceptionList) -> "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.rlp is not None:
            return self.rlp

        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 signature_bytes(self) -> bytes:
        """
        Returns the serialized bytes of the transaction signature.
        """
        assert self.v is not None and self.r is not None and self.s is not None
        v = self.v
        if self.ty == 0:
            if self.protected:
                assert self.chain_id is not None
                v -= 35 + (self.chain_id * 2)
            else:
                v -= 27
        return (
            self.r.to_bytes(32, byteorder="big")
            + self.s.to_bytes(32, byteorder="big")
            + bytes([v])
        )

    def with_signature_and_sender(self, *, keep_secret_key: bool = False) -> "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:
                public_key = PublicKey.from_signature_and_message(
                    tx.signature_bytes(), keccak256(tx.signing_bytes()), hasher=None
                )
                tx.sender = keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
            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 if requested
        if not keep_secret_key:
            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: TransactionException | ExceptionList) -> "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.rlp is not None:
        return self.rlp

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

signature_bytes()

Returns the serialized bytes of the transaction signature.

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

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

with_signature_and_sender(*, keep_secret_key=False)

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, *, keep_secret_key: bool = False) -> "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:
            public_key = PublicKey.from_signature_and_message(
                tx.signature_bytes(), keccak256(tx.signing_bytes()), hasher=None
            )
            tx.sender = keccak256(public_key.format(compressed=False)[1:])[32 - 20 :]
        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 if requested
    if not keep_secret_key:
        tx.secret_key = None
    return tx

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

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

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)

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

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

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)

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)

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

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

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

transaction_list_root(input_txs)

Returns the transactions root of a list of transactions.

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

def transaction_list_root(input_txs: List[Transaction] | None) -> Hash:
    """
    Returns the transactions root of a list of transactions.
    """
    t = HexaryTrie(db={})
    for i, tx in enumerate(input_txs or []):
        t.set(eth_rlp.encode(Uint(i)), tx.serialized_bytes())
    return Hash(t.root_hash)

BlockException

Bases: ExceptionBase

Exception raised when a block is invalid, but not due to a transaction.

E.g. all transactions in the block are valid, and can be applied to the state, but the block header contains an invalid field.

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

@unique
class BlockException(ExceptionBase):
    """
    Exception raised when a block is invalid, but not due to a transaction.

    E.g. all transactions in the block are valid, and can be applied to the state, but the
    block header contains an invalid field.
    """

    INCORRECT_BLOCK_FORMAT = auto()
    """
    Block's format is incorrect, contains invalid fields, is missing fields, or contains fields of
    a fork that is not active yet.
    """
    BLOB_GAS_USED_ABOVE_LIMIT = auto()
    """
    Block's blob gas used in header is above the limit.
    """
    INCORRECT_BLOB_GAS_USED = auto()
    """
    Block's blob gas used in header is incorrect.
    """
    INCORRECT_EXCESS_BLOB_GAS = auto()
    """
    Block's excess blob gas in header is incorrect.
    """

INCORRECT_BLOCK_FORMAT = auto() class-attribute instance-attribute

Block's format is incorrect, contains invalid fields, is missing fields, or contains fields of a fork that is not active yet.

BLOB_GAS_USED_ABOVE_LIMIT = auto() class-attribute instance-attribute

Block's blob gas used in header is above the limit.

INCORRECT_BLOB_GAS_USED = auto() class-attribute instance-attribute

Block's blob gas used in header is incorrect.

INCORRECT_EXCESS_BLOB_GAS = auto() class-attribute instance-attribute

Block's excess blob gas in header is incorrect.

ExceptionList

Bases: list

A list of exceptions.

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

class ExceptionList(list):
    """
    A list of exceptions.
    """

    def __init__(self, *exceptions: "ExceptionBase") -> None:
        """
        Create a new ExceptionList.
        """
        exceptions_set: List[ExceptionBase] = []
        for exception in exceptions:
            if not isinstance(exception, ExceptionBase):
                raise TypeError(f"Expected ExceptionBase, got {type(exception)}")
            if exception not in exceptions_set:
                exceptions_set.append(exception)
        super().__init__(exceptions_set)

    def __or__(self, other: Union["ExceptionBase", "ExceptionList"]) -> "ExceptionList":
        """
        Combine two ExceptionLists.
        """
        if isinstance(other, list):
            return ExceptionList(*(self + other))
        return ExceptionList(*(self + [other]))

    def __str__(self) -> str:
        """
        String representation of the ExceptionList.
        """
        return "|".join(str(exception) for exception in self)

__init__(*exceptions)

Create a new ExceptionList.

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

def __init__(self, *exceptions: "ExceptionBase") -> None:
    """
    Create a new ExceptionList.
    """
    exceptions_set: List[ExceptionBase] = []
    for exception in exceptions:
        if not isinstance(exception, ExceptionBase):
            raise TypeError(f"Expected ExceptionBase, got {type(exception)}")
        if exception not in exceptions_set:
            exceptions_set.append(exception)
    super().__init__(exceptions_set)

__or__(other)

Combine two ExceptionLists.

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

def __or__(self, other: Union["ExceptionBase", "ExceptionList"]) -> "ExceptionList":
    """
    Combine two ExceptionLists.
    """
    if isinstance(other, list):
        return ExceptionList(*(self + other))
    return ExceptionList(*(self + [other]))

__str__()

String representation of the ExceptionList.

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

def __str__(self) -> str:
    """
    String representation of the ExceptionList.
    """
    return "|".join(str(exception) for exception in self)

TransactionException

Bases: ExceptionBase

Exception raised when a transaction is invalid, and thus cannot be executed.

If a transaction with any of these exceptions is included in a block, the block is invalid.

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

@unique
class TransactionException(ExceptionBase):
    """
    Exception raised when a transaction is invalid, and thus cannot be executed.

    If a transaction with any of these exceptions is included in a block, the block is invalid.
    """

    INSUFFICIENT_ACCOUNT_FUNDS = auto()
    """
    Transaction's sender does not have enough funds to pay for the transaction.
    """
    INSUFFICIENT_MAX_FEE_PER_GAS = auto()
    """
    Transaction's max-fee-per-gas is lower than the block base-fee.
    """
    PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS = auto()
    """
    Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas.
    """
    INSUFFICIENT_MAX_FEE_PER_BLOB_GAS = auto()
    """
    Transaction's max-fee-per-blob-gas is lower than the block's blob-gas price.
    """
    INTRINSIC_GAS_TOO_LOW = auto()
    """
    Transaction's gas limit is too low.
    """
    INITCODE_SIZE_EXCEEDED = auto()
    """
    Transaction's initcode for a contract-creating transaction is too large.
    """
    TYPE_3_TX_PRE_FORK = auto()
    """
    Transaction type 3 included before activation fork.
    """
    TYPE_3_TX_ZERO_BLOBS_PRE_FORK = auto()
    """
    Transaction type 3, with zero blobs, included before activation fork.
    """
    TYPE_3_TX_INVALID_BLOB_VERSIONED_HASH = auto()
    """
    Transaction contains a blob versioned hash with an invalid version.
    """
    TYPE_3_TX_WITH_FULL_BLOBS = auto()
    """
    Transaction contains full blobs (network-version of the transaction).
    """
    TYPE_3_TX_BLOB_COUNT_EXCEEDED = auto()
    """
    Transaction contains too many blob versioned hashes.
    """
    TYPE_3_TX_CONTRACT_CREATION = auto()
    """
    Transaction is a type 3 transaction and has an empty `to`.
    """
    TYPE_3_TX_MAX_BLOB_GAS_ALLOWANCE_EXCEEDED = auto()
    """
    Transaction causes block to go over blob gas limit.
    """
    TYPE_3_TX_ZERO_BLOBS = auto()
    """
    Transaction is type 3, but has no blobs.
    """

INSUFFICIENT_ACCOUNT_FUNDS = auto() class-attribute instance-attribute

Transaction's sender does not have enough funds to pay for the transaction.

INSUFFICIENT_MAX_FEE_PER_GAS = auto() class-attribute instance-attribute

Transaction's max-fee-per-gas is lower than the block base-fee.

PRIORITY_GREATER_THAN_MAX_FEE_PER_GAS = auto() class-attribute instance-attribute

Transaction's max-priority-fee-per-gas is greater than the max-fee-per-gas.

INSUFFICIENT_MAX_FEE_PER_BLOB_GAS = auto() class-attribute instance-attribute

Transaction's max-fee-per-blob-gas is lower than the block's blob-gas price.

INTRINSIC_GAS_TOO_LOW = auto() class-attribute instance-attribute

Transaction's gas limit is too low.

INITCODE_SIZE_EXCEEDED = auto() class-attribute instance-attribute

Transaction's initcode for a contract-creating transaction is too large.

TYPE_3_TX_PRE_FORK = auto() class-attribute instance-attribute

Transaction type 3 included before activation fork.

TYPE_3_TX_ZERO_BLOBS_PRE_FORK = auto() class-attribute instance-attribute

Transaction type 3, with zero blobs, included before activation fork.

TYPE_3_TX_INVALID_BLOB_VERSIONED_HASH = auto() class-attribute instance-attribute

Transaction contains a blob versioned hash with an invalid version.

TYPE_3_TX_WITH_FULL_BLOBS = auto() class-attribute instance-attribute

Transaction contains full blobs (network-version of the transaction).

TYPE_3_TX_BLOB_COUNT_EXCEEDED = auto() class-attribute instance-attribute

Transaction contains too many blob versioned hashes.

TYPE_3_TX_CONTRACT_CREATION = auto() class-attribute instance-attribute

Transaction is a type 3 transaction and has an empty to.

TYPE_3_TX_MAX_BLOB_GAS_ALLOWANCE_EXCEEDED = auto() class-attribute instance-attribute

Transaction causes block to go over blob gas limit.

TYPE_3_TX_ZERO_BLOBS = auto() class-attribute instance-attribute

Transaction is type 3, but has no blobs.

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

BaseFixture dataclass

Represents a base Ethereum test fixture of any type.

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

@dataclass(kw_only=True)
class BaseFixture:
    """
    Represents a base Ethereum test fixture of any type.
    """

    info: Dict[str, str] = json_field(
        default_factory=dict,
        json_encoder=JSONEncoder.Field(
            name="_info",
            to_json=True,
        ),
    )

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

    @abstractmethod
    def to_json(self) -> Dict[str, Any]:
        """
        Convert to JSON.
        """
        pass

    @classmethod
    @abstractmethod
    def format(cls) -> FixtureFormats:
        """
        Returns the fixture format which the evm tool can use to determine how to verify the
        fixture.
        """
        pass

    @classmethod
    @abstractmethod
    def collect_into_file(cls, fd: TextIO, fixtures: Dict[str, "BaseFixture"]):
        """
        Returns the name of the subdirectory where this type of fixture should be dumped to.
        """
        pass

    @classmethod
    @abstractmethod
    def output_base_dir_name(cls) -> Path:
        """
        Returns the name of the subdirectory where this type of fixture should be dumped to.
        """
        pass

    @classmethod
    def output_file_extension(cls) -> str:
        """
        Returns the file extension for this type of fixture.

        By default, fixtures are dumped as JSON files.
        """
        return ".json"

fill_info(t8n, ref_spec)

Fill the info field for this fixture

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

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

to_json() abstractmethod

Convert to JSON.

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

@abstractmethod
def to_json(self) -> Dict[str, Any]:
    """
    Convert to JSON.
    """
    pass

format() abstractmethod classmethod

Returns the fixture format which the evm tool can use to determine how to verify the fixture.

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

@classmethod
@abstractmethod
def format(cls) -> FixtureFormats:
    """
    Returns the fixture format which the evm tool can use to determine how to verify the
    fixture.
    """
    pass

collect_into_file(fd, fixtures) abstractmethod classmethod

Returns the name of the subdirectory where this type of fixture should be dumped to.

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

@classmethod
@abstractmethod
def collect_into_file(cls, fd: TextIO, fixtures: Dict[str, "BaseFixture"]):
    """
    Returns the name of the subdirectory where this type of fixture should be dumped to.
    """
    pass

output_base_dir_name() abstractmethod classmethod

Returns the name of the subdirectory where this type of fixture should be dumped to.

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

@classmethod
@abstractmethod
def output_base_dir_name(cls) -> Path:
    """
    Returns the name of the subdirectory where this type of fixture should be dumped to.
    """
    pass

output_file_extension() classmethod

Returns the file extension for this type of fixture.

By default, fixtures are dumped as JSON files.

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

@classmethod
def output_file_extension(cls) -> str:
    """
    Returns the file extension for this type of fixture.

    By default, fixtures are dumped as JSON files.
    """
    return ".json"

BaseTest dataclass

Represents a base Ethereum test which must return a single test fixture.

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

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

    pre: Mapping
    tag: str = ""
    # Setting a default here is just for type checking, the correct value is automatically set
    # by pytest.
    fixture_format: FixtureFormats = FixtureFormats.UNSET_TEST_FORMAT

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

    @abstractmethod
    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the list of test fixtures.
        """
        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

    @classmethod
    @abstractmethod
    def fixture_formats(cls) -> List[FixtureFormats]:
        """
        Returns a list of fixture formats that can be output to the test spec.
        """
        pass

    def __post_init__(self) -> None:
        """
        Validate the fixture format.
        """
        if self.fixture_format not in self.fixture_formats():
            raise ValueError(
                f"Invalid fixture format {self.fixture_format} for {self.__class__.__name__}."
            )

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

generate(t8n, fork, eips=None) abstractmethod

Generate the list of test fixtures.

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

@abstractmethod
def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the list of test fixtures.
    """
    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/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

fixture_formats() abstractmethod classmethod

Returns a list of fixture formats that can be output to the test spec.

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

@classmethod
@abstractmethod
def fixture_formats(cls) -> List[FixtureFormats]:
    """
    Returns a list of fixture formats that can be output to the test spec.
    """
    pass

__post_init__()

Validate the fixture format.

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

def __post_init__(self) -> None:
    """
    Validate the fixture format.
    """
    if self.fixture_format not in self.fixture_formats():
        raise ValueError(
            f"Invalid fixture format {self.fixture_format} for {self.__class__.__name__}."
        )

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

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/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 = ""
    chain_id: int = 1

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

    @classmethod
    def fixture_formats(cls) -> List[FixtureFormats]:
        """
        Returns a list of fixture formats that can be output to the test spec.
        """
        return [
            FixtureFormats.BLOCKCHAIN_TEST,
            FixtureFormats.BLOCKCHAIN_TEST_HIVE,
        ]

    def make_genesis(
        self,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[Alloc, Bytes, FixtureHeader]:
        """
        Create a genesis block from the blockchain test definition.
        """
        env = self.genesis_environment.set_fork_requirements(fork)
        if env.withdrawals is not None:
            assert len(env.withdrawals) == 0, "withdrawals must be empty at genesis"
        if env.beacon_root is not None:
            assert Hash(env.beacon_root) == Hash(0), "beacon_root must be empty at genesis"

        pre_alloc = Alloc(
            fork.pre_allocation(block_number=0, timestamp=Number(env.timestamp)),
        )

        new_alloc, state_root = t8n.calc_state_root(
            alloc=to_json(Alloc.merge(pre_alloc, 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(
                withdrawals_root(env.withdrawals) 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 generate_block_data(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Dict[str, Any],
        eips: Optional[List[int]] = None,
    ) -> Tuple[FixtureHeader, Bytes, List[Transaction], Dict[str, Any], Environment]:
        """
        Generate common block data for both make_fixture and make_hive_fixture.
        """
        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"
            )

        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 []

        if failing_tx_count := len([tx for tx in txs if tx.error]) > 0:
            if failing_tx_count > 1:
                raise Exception(
                    "test correctness: only one transaction can produce an exception in a block"
                )
            if not txs[-1].error:
                raise Exception(
                    "test correctness: the transaction that produces an exception "
                    + "must be the last transaction in the block"
                )

        next_alloc, result = t8n.evaluate(
            alloc=previous_alloc,
            txs=to_json(txs),
            env=to_json(env),
            fork_name=fork.transition_tool_name(
                block_number=Number(env.number), timestamp=Number(env.timestamp)
            ),
            chain_id=self.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)
            verify_result(result, env)
        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,
        )

        # Update the transactions root to the one calculated locally.
        header.transactions_root = transaction_list_root(txs)

        # One special case of the invalid transactions is the blob gas used, since this value
        # is not included in the transition tool result, but it is included in the block header,
        # and some clients check it before executing the block by simply counting the type-3 txs,
        # we need to set the correct value by default.
        if (
            blob_gas_per_blob := fork.blob_gas_per_blob(Number(env.number), Number(env.timestamp))
        ) > 0:
            header.blob_gas_used = blob_gas_per_blob * count_blobs(txs)

        if block.header_verify is not None:
            # Verify the header after transition tool processing.
            header.verify(block.header_verify)

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

        return header, rlp, txs, next_alloc, env

    def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
        """
        Returns fixture network information for the fork & EIP/s.
        """
        return (
            "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
            if eips
            else fork.blockchain_test_network_name()
        )

    def verify_post_state(self, t8n, alloc):
        """
        Verifies the post alloc after all block/s or payload/s are generated.
        """
        try:
            verify_post_alloc(self.post, alloc)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

    def make_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> Fixture:
        """
        Create a fixture from the blockchain test definition.
        """
        fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

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

        alloc = to_json(pre)
        env = environment_from_parent_header(genesis)
        head = genesis.hash if genesis.hash is not None else Hash(0)

        for block in self.blocks:
            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.
                header, rlp, txs, new_alloc, new_env = self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    eips=eips,
                )
                fixture_block = FixtureBlock(
                    rlp=rlp,
                    block_header=header,
                    block_number=Number(header.number),
                    txs=txs,
                    ommers=[],
                    withdrawals=new_env.withdrawals,
                )
                if block.exception is None:
                    fixture_blocks.append(fixture_block)
                    # Update env, alloc and last block hash for the next block.
                    alloc = new_alloc
                    env = apply_new_parent(new_env, header)
                    head = header.hash if header.hash is not None else Hash(0)
                else:
                    fixture_blocks.append(
                        InvalidFixtureBlock(
                            rlp=rlp,
                            expected_exception=block.exception,
                            rlp_decoded=replace(fixture_block, rlp=None),
                        ),
                    )
            else:
                assert block.exception is not None, (
                    "test correctness: if the block's rlp is hard-coded, "
                    + "the block is expected to produce an exception"
                )
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=Bytes(block.rlp),
                        expected_exception=block.exception,
                    ),
                )

        self.verify_post_state(t8n, alloc)
        return Fixture(
            fork=self.network_info(fork, eips),
            genesis=genesis,
            genesis_rlp=genesis_rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre_state=pre,
            post_state=alloc_to_accounts(alloc),
            name=self.tag,
        )

    def make_hive_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> HiveFixture:
        """
        Create a hive fixture from the blocktest definition.
        """
        fixture_payloads: List[FixtureEngineNewPayload] = []

        pre, _, genesis = self.make_genesis(t8n, fork)
        alloc = to_json(pre)
        env = environment_from_parent_header(genesis)

        for block in self.blocks:
            header, _, txs, new_alloc, new_env = self.generate_block_data(
                t8n=t8n, fork=fork, block=block, previous_env=env, previous_alloc=alloc, eips=eips
            )
            if block.rlp is None:
                fixture_payloads.append(
                    FixtureEngineNewPayload.from_fixture_header(
                        fork=fork,
                        header=header,
                        transactions=txs,
                        withdrawals=new_env.withdrawals,
                        validation_error=block.exception,
                        error_code=block.engine_api_error_code,
                    )
                )
                if block.exception is None:
                    alloc = new_alloc
                    env = apply_new_parent(env, header)
        fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
        assert (
            fcu_version is not None
        ), "A hive fixture was requested but no forkchoice update is defined. The framework should"
        " never try to execute this test case."

        self.verify_post_state(t8n, alloc)
        return HiveFixture(
            fork=self.network_info(fork, eips),
            genesis=genesis,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre_state=pre,
            post_state=alloc_to_accounts(alloc),
            name=self.tag,
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        t8n.reset_traces()
        if self.fixture_format == FixtureFormats.BLOCKCHAIN_TEST_HIVE:
            if fork.engine_forkchoice_updated_version() is None:
                raise Exception(
                    "A hive fixture was requested but no forkchoice update is defined. "
                    "The framework should never try to execute this test case."
                )
            return self.make_hive_fixture(t8n, fork, eips)
        elif self.fixture_format == FixtureFormats.BLOCKCHAIN_TEST:
            return self.make_fixture(t8n, fork, eips)

        raise Exception(f"Unknown fixture format: {self.fixture_format}")

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

fixture_formats() classmethod

Returns a list of fixture formats that can be output to the test spec.

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

@classmethod
def fixture_formats(cls) -> List[FixtureFormats]:
    """
    Returns a list of fixture formats that can be output to the test spec.
    """
    return [
        FixtureFormats.BLOCKCHAIN_TEST,
        FixtureFormats.BLOCKCHAIN_TEST_HIVE,
    ]

make_genesis(t8n, fork)

Create a genesis block from the blockchain test definition.

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

def make_genesis(
    self,
    t8n: TransitionTool,
    fork: Fork,
) -> Tuple[Alloc, Bytes, FixtureHeader]:
    """
    Create a genesis block from the blockchain test definition.
    """
    env = self.genesis_environment.set_fork_requirements(fork)
    if env.withdrawals is not None:
        assert len(env.withdrawals) == 0, "withdrawals must be empty at genesis"
    if env.beacon_root is not None:
        assert Hash(env.beacon_root) == Hash(0), "beacon_root must be empty at genesis"

    pre_alloc = Alloc(
        fork.pre_allocation(block_number=0, timestamp=Number(env.timestamp)),
    )

    new_alloc, state_root = t8n.calc_state_root(
        alloc=to_json(Alloc.merge(pre_alloc, 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(
            withdrawals_root(env.withdrawals) 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

generate_block_data(t8n, fork, block, previous_env, previous_alloc, eips=None)

Generate common block data for both make_fixture and make_hive_fixture.

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

def generate_block_data(
    self,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Dict[str, Any],
    eips: Optional[List[int]] = None,
) -> Tuple[FixtureHeader, Bytes, List[Transaction], Dict[str, Any], Environment]:
    """
    Generate common block data for both make_fixture and make_hive_fixture.
    """
    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"
        )

    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 []

    if failing_tx_count := len([tx for tx in txs if tx.error]) > 0:
        if failing_tx_count > 1:
            raise Exception(
                "test correctness: only one transaction can produce an exception in a block"
            )
        if not txs[-1].error:
            raise Exception(
                "test correctness: the transaction that produces an exception "
                + "must be the last transaction in the block"
            )

    next_alloc, result = t8n.evaluate(
        alloc=previous_alloc,
        txs=to_json(txs),
        env=to_json(env),
        fork_name=fork.transition_tool_name(
            block_number=Number(env.number), timestamp=Number(env.timestamp)
        ),
        chain_id=self.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)
        verify_result(result, env)
    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,
    )

    # Update the transactions root to the one calculated locally.
    header.transactions_root = transaction_list_root(txs)

    # One special case of the invalid transactions is the blob gas used, since this value
    # is not included in the transition tool result, but it is included in the block header,
    # and some clients check it before executing the block by simply counting the type-3 txs,
    # we need to set the correct value by default.
    if (
        blob_gas_per_blob := fork.blob_gas_per_blob(Number(env.number), Number(env.timestamp))
    ) > 0:
        header.blob_gas_used = blob_gas_per_blob * count_blobs(txs)

    if block.header_verify is not None:
        # Verify the header after transition tool processing.
        header.verify(block.header_verify)

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

    return header, rlp, txs, next_alloc, env

network_info(fork, eips=None)

Returns fixture network information for the fork & EIP/s.

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

def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
    """
    Returns fixture network information for the fork & EIP/s.
    """
    return (
        "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
        if eips
        else fork.blockchain_test_network_name()
    )

verify_post_state(t8n, alloc)

Verifies the post alloc after all block/s or payload/s are generated.

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

def verify_post_state(self, t8n, alloc):
    """
    Verifies the post alloc after all block/s or payload/s are generated.
    """
    try:
        verify_post_alloc(self.post, alloc)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

make_fixture(t8n, fork, eips=None)

Create a fixture from the blockchain test definition.

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

def make_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> Fixture:
    """
    Create a fixture from the blockchain test definition.
    """
    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

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

    alloc = to_json(pre)
    env = environment_from_parent_header(genesis)
    head = genesis.hash if genesis.hash is not None else Hash(0)

    for block in self.blocks:
        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.
            header, rlp, txs, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
            )
            fixture_block = FixtureBlock(
                rlp=rlp,
                block_header=header,
                block_number=Number(header.number),
                txs=txs,
                ommers=[],
                withdrawals=new_env.withdrawals,
            )
            if block.exception is None:
                fixture_blocks.append(fixture_block)
                # Update env, alloc and last block hash for the next block.
                alloc = new_alloc
                env = apply_new_parent(new_env, header)
                head = header.hash if header.hash is not None else Hash(0)
            else:
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=rlp,
                        expected_exception=block.exception,
                        rlp_decoded=replace(fixture_block, rlp=None),
                    ),
                )
        else:
            assert block.exception is not None, (
                "test correctness: if the block's rlp is hard-coded, "
                + "the block is expected to produce an exception"
            )
            fixture_blocks.append(
                InvalidFixtureBlock(
                    rlp=Bytes(block.rlp),
                    expected_exception=block.exception,
                ),
            )

    self.verify_post_state(t8n, alloc)
    return Fixture(
        fork=self.network_info(fork, eips),
        genesis=genesis,
        genesis_rlp=genesis_rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre_state=pre,
        post_state=alloc_to_accounts(alloc),
        name=self.tag,
    )

make_hive_fixture(t8n, fork, eips=None)

Create a hive fixture from the blocktest definition.

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

def make_hive_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> HiveFixture:
    """
    Create a hive fixture from the blocktest definition.
    """
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, _, genesis = self.make_genesis(t8n, fork)
    alloc = to_json(pre)
    env = environment_from_parent_header(genesis)

    for block in self.blocks:
        header, _, txs, new_alloc, new_env = self.generate_block_data(
            t8n=t8n, fork=fork, block=block, previous_env=env, previous_alloc=alloc, eips=eips
        )
        if block.rlp is None:
            fixture_payloads.append(
                FixtureEngineNewPayload.from_fixture_header(
                    fork=fork,
                    header=header,
                    transactions=txs,
                    withdrawals=new_env.withdrawals,
                    validation_error=block.exception,
                    error_code=block.engine_api_error_code,
                )
            )
            if block.exception is None:
                alloc = new_alloc
                env = apply_new_parent(env, header)
    fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
    assert (
        fcu_version is not None
    ), "A hive fixture was requested but no forkchoice update is defined. The framework should"
    " never try to execute this test case."

    self.verify_post_state(t8n, alloc)
    return HiveFixture(
        fork=self.network_info(fork, eips),
        genesis=genesis,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre_state=pre,
        post_state=alloc_to_accounts(alloc),
        name=self.tag,
    )

generate(t8n, fork, eips=None)

Generate the BlockchainTest fixture.

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

def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    t8n.reset_traces()
    if self.fixture_format == FixtureFormats.BLOCKCHAIN_TEST_HIVE:
        if fork.engine_forkchoice_updated_version() is None:
            raise Exception(
                "A hive fixture was requested but no forkchoice update is defined. "
                "The framework should never try to execute this test case."
            )
        return self.make_hive_fixture(t8n, fork, eips)
    elif self.fixture_format == FixtureFormats.BLOCKCHAIN_TEST:
        return self.make_fixture(t8n, fork, eips)

    raise Exception(f"Unknown fixture format: {self.fixture_format}")

FixtureCollector dataclass

Collects all fixtures generated by the test cases.

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

@dataclass(kw_only=True)
class FixtureCollector:
    """
    Collects all fixtures generated by the test cases.
    """

    output_dir: str
    flat_output: bool
    single_fixture_per_file: bool
    filler_path: Path
    base_dump_dir: Optional[Path] = None

    # Internal state
    all_fixtures: Dict[Path, Dict[str, BaseFixture]] = field(default_factory=dict)
    json_path_to_fixture_type: Dict[Path, FixtureFormats] = field(default_factory=dict)
    json_path_to_test_item: Dict[Path, TestInfo] = field(default_factory=dict)

    def get_fixture_basename(self, info: TestInfo) -> Path:
        """
        Returns the basename of the fixture file for a given test case.
        """
        if self.flat_output:
            if self.single_fixture_per_file:
                return Path(strip_test_prefix(info.get_single_test_name()))
            return Path(strip_test_prefix(info.original_name))
        else:
            relative_fixture_output_dir = Path(info.path).parent / strip_test_prefix(
                Path(info.path).stem
            )
            module_relative_output_dir = get_module_relative_output_dir(
                relative_fixture_output_dir, self.filler_path
            )

            if self.single_fixture_per_file:
                return module_relative_output_dir / strip_test_prefix(info.get_single_test_name())
            return module_relative_output_dir / strip_test_prefix(info.original_name)

    def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> None:
        """
        Adds a fixture to the list of fixtures of a given test case.
        """
        fixture_basename = self.get_fixture_basename(info)

        fixture_path = (
            self.output_dir
            / fixture.output_base_dir_name()
            / fixture_basename.with_suffix(fixture.output_file_extension())
        )
        if fixture_path not in self.all_fixtures:  # relevant when we group by test function
            self.all_fixtures[fixture_path] = {}
            if fixture_path in self.json_path_to_fixture_type:
                if self.json_path_to_fixture_type[fixture_path] != fixture.format():
                    raise Exception(
                        f"Fixture {fixture_path} has two different types: "
                        f"{self.json_path_to_fixture_type[fixture_path]} "
                        f"and {fixture.format()}"
                    )
            else:
                self.json_path_to_fixture_type[fixture_path] = fixture.format()
            self.json_path_to_test_item[fixture_path] = info

        self.all_fixtures[fixture_path][info.id] = fixture

    def dump_fixtures(self) -> None:
        """
        Dumps all collected fixtures to their respective files.
        """
        os.makedirs(self.output_dir, exist_ok=True)
        for fixture_path, fixtures in self.all_fixtures.items():
            os.makedirs(fixture_path.parent, exist_ok=True)

            # Get the first fixture to dump to get its type
            fixture = next(iter(fixtures.values()))
            # Call class method to dump all the fixtures
            with open(fixture_path, "w") as fd:
                fixture.collect_into_file(fd, fixtures)

    def verify_fixture_files(self, evm_fixture_verification: TransitionTool) -> None:
        """
        Runs `evm [state|block]test` on each fixture.
        """
        for fixture_path, fixture_format in self.json_path_to_fixture_type.items():
            if FixtureFormats.is_verifiable(fixture_format):
                info = self.json_path_to_test_item[fixture_path]
                verify_fixtures_dump_dir = self._get_verify_fixtures_dump_dir(info)
                evm_fixture_verification.verify_fixture(
                    fixture_format, fixture_path, verify_fixtures_dump_dir
                )

    def _get_verify_fixtures_dump_dir(
        self,
        info: TestInfo,
    ):
        """
        The directory to dump the current test function's fixture.json and fixture
        verification debug output.
        """
        if not self.base_dump_dir:
            return None
        if self.single_fixture_per_file:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_parameter"
            )
        else:
            return info.get_dump_dir_path(
                self.base_dump_dir, self.filler_path, level="test_function"
            )

get_fixture_basename(info)

Returns the basename of the fixture file for a given test case.

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

def get_fixture_basename(self, info: TestInfo) -> Path:
    """
    Returns the basename of the fixture file for a given test case.
    """
    if self.flat_output:
        if self.single_fixture_per_file:
            return Path(strip_test_prefix(info.get_single_test_name()))
        return Path(strip_test_prefix(info.original_name))
    else:
        relative_fixture_output_dir = Path(info.path).parent / strip_test_prefix(
            Path(info.path).stem
        )
        module_relative_output_dir = get_module_relative_output_dir(
            relative_fixture_output_dir, self.filler_path
        )

        if self.single_fixture_per_file:
            return module_relative_output_dir / strip_test_prefix(info.get_single_test_name())
        return module_relative_output_dir / strip_test_prefix(info.original_name)

add_fixture(info, fixture)

Adds a fixture to the list of fixtures of a given test case.

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

def add_fixture(self, info: TestInfo, fixture: BaseFixture) -> None:
    """
    Adds a fixture to the list of fixtures of a given test case.
    """
    fixture_basename = self.get_fixture_basename(info)

    fixture_path = (
        self.output_dir
        / fixture.output_base_dir_name()
        / fixture_basename.with_suffix(fixture.output_file_extension())
    )
    if fixture_path not in self.all_fixtures:  # relevant when we group by test function
        self.all_fixtures[fixture_path] = {}
        if fixture_path in self.json_path_to_fixture_type:
            if self.json_path_to_fixture_type[fixture_path] != fixture.format():
                raise Exception(
                    f"Fixture {fixture_path} has two different types: "
                    f"{self.json_path_to_fixture_type[fixture_path]} "
                    f"and {fixture.format()}"
                )
        else:
            self.json_path_to_fixture_type[fixture_path] = fixture.format()
        self.json_path_to_test_item[fixture_path] = info

    self.all_fixtures[fixture_path][info.id] = fixture

dump_fixtures()

Dumps all collected fixtures to their respective files.

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

def dump_fixtures(self) -> None:
    """
    Dumps all collected fixtures to their respective files.
    """
    os.makedirs(self.output_dir, exist_ok=True)
    for fixture_path, fixtures in self.all_fixtures.items():
        os.makedirs(fixture_path.parent, exist_ok=True)

        # Get the first fixture to dump to get its type
        fixture = next(iter(fixtures.values()))
        # Call class method to dump all the fixtures
        with open(fixture_path, "w") as fd:
            fixture.collect_into_file(fd, fixtures)

verify_fixture_files(evm_fixture_verification)

Runs evm [state|block]test on each fixture.

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

def verify_fixture_files(self, evm_fixture_verification: TransitionTool) -> None:
    """
    Runs `evm [state|block]test` on each fixture.
    """
    for fixture_path, fixture_format in self.json_path_to_fixture_type.items():
        if FixtureFormats.is_verifiable(fixture_format):
            info = self.json_path_to_test_item[fixture_path]
            verify_fixtures_dump_dir = self._get_verify_fixtures_dump_dir(info)
            evm_fixture_verification.verify_fixture(
                fixture_format, fixture_path, verify_fixtures_dump_dir
            )

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/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
    tx: Transaction
    engine_api_error_code: Optional[EngineAPIError] = None
    blockchain_test_header_verify: Optional[Header] = None
    blockchain_test_rlp_modifier: Optional[Header] = None
    tag: str = ""
    chain_id: int = 1

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

    @classmethod
    def fixture_formats(cls) -> List[FixtureFormats]:
        """
        Returns a list of fixture formats that can be output to the test spec.
        """
        return [
            FixtureFormats.BLOCKCHAIN_TEST,
            FixtureFormats.BLOCKCHAIN_TEST_HIVE,
            FixtureFormats.STATE_TEST,
        ]

    def _generate_blockchain_genesis_environment(self) -> Environment:
        """
        Generate the genesis environment for the BlockchainTest formatted test.
        """
        genesis_env = copy(self.env)

        # Modify values to the proper values for the genesis block
        # TODO: All of this can be moved to a new method in `Fork`
        genesis_env.withdrawals = None
        genesis_env.beacon_root = None
        genesis_env.number = Number(genesis_env.number) - 1
        assert (
            genesis_env.number >= 0
        ), "genesis block number cannot be negative, set state test env.number to 1"
        if genesis_env.excess_blob_gas:
            # The excess blob gas environment value means the value of the context (block header)
            # where the transaction is executed. In a blockchain test, we need to indirectly
            # set the excess blob gas by setting the excess blob gas of the genesis block
            # to the expected value plus the TARGET_BLOB_GAS_PER_BLOCK, which is the value
            # that will be subtracted from the excess blob gas when the first block is mined.
            genesis_env.excess_blob_gas = (
                Number(genesis_env.excess_blob_gas) + TARGET_BLOB_GAS_PER_BLOCK
            )

        return genesis_env

    def _generate_blockchain_blocks(self) -> List[Block]:
        """
        Generate the single block that represents this state test in a BlockchainTest format.
        """
        return [
            Block(
                number=self.env.number,
                timestamp=self.env.timestamp,
                coinbase=self.env.coinbase,
                difficulty=self.env.difficulty,
                gas_limit=self.env.gas_limit,
                extra_data=self.env.extra_data,
                withdrawals=self.env.withdrawals,
                beacon_root=self.env.beacon_root,
                txs=[self.tx],
                ommers=[],
                exception=self.tx.error,
                header_verify=self.blockchain_test_header_verify,
                rlp_modifier=self.blockchain_test_rlp_modifier,
            )
        ]

    def generate_blockchain_test(self) -> BlockchainTest:
        """
        Generate a BlockchainTest fixture from this StateTest fixture.
        """
        return BlockchainTest(
            genesis_environment=self._generate_blockchain_genesis_environment(),
            pre=self.pre,
            post=self.post,
            blocks=self._generate_blockchain_blocks(),
            fixture_format=self.fixture_format,
            t8n_dump_dir=self.t8n_dump_dir,
        )

    def make_state_test_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> Fixture:
        """
        Create a fixture from the state test definition.
        """
        env = self.env.set_fork_requirements(fork)
        tx = self.tx.with_signature_and_sender(keep_secret_key=True)
        pre_alloc = Alloc.merge(
            Alloc(
                fork.pre_allocation(block_number=env.number, timestamp=Number(env.timestamp)),
            ),
            Alloc(self.pre),
        )
        transition_tool_name = fork.transition_tool_name(
            block_number=Number(self.env.number),
            timestamp=Number(self.env.timestamp),
        )
        fork_name = (
            "+".join([transition_tool_name] + [str(eip) for eip in eips])
            if eips
            else transition_tool_name
        )
        next_alloc, result = t8n.evaluate(
            alloc=to_json(pre_alloc),
            txs=to_json([tx]),
            env=to_json(env),
            fork_name=fork_name,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )

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

        # Perform post state processing required for some forks
        if fork >= Cancun:
            # StateTest does not execute any beacon root contract logic, but we still need to
            # set the beacon root to the correct value, because most tests assume this happens,
            # so we copy the beacon root contract storage from the post state into the pre state
            # and the transaction is executed in isolation properly.
            if beacon_roots_account := next_alloc.get(str(BEACON_ROOTS_ADDRESS)):
                if beacon_roots_storage := beacon_roots_account.get("storage"):
                    pre_alloc = Alloc.merge(
                        pre_alloc,
                        Alloc({BEACON_ROOTS_ADDRESS: Account(storage=beacon_roots_storage)}),
                    )

        return Fixture(
            env=env,
            pre_state=pre_alloc,
            post={
                fork.blockchain_test_network_name(): [
                    FixtureForkPost.collect(
                        transition_tool_result=result,
                        transaction=tx.with_signature_and_sender(),
                    )
                ]
            },
            transaction=tx,
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        if self.fixture_format in BlockchainTest.fixture_formats():
            return self.generate_blockchain_test().generate(t8n, fork, eips)
        elif self.fixture_format == FixtureFormats.STATE_TEST:
            return self.make_state_test_fixture(t8n, fork, eips)

        raise Exception(f"Unknown fixture format: {self.fixture_format}")

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

fixture_formats() classmethod

Returns a list of fixture formats that can be output to the test spec.

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

@classmethod
def fixture_formats(cls) -> List[FixtureFormats]:
    """
    Returns a list of fixture formats that can be output to the test spec.
    """
    return [
        FixtureFormats.BLOCKCHAIN_TEST,
        FixtureFormats.BLOCKCHAIN_TEST_HIVE,
        FixtureFormats.STATE_TEST,
    ]

generate_blockchain_test()

Generate a BlockchainTest fixture from this StateTest fixture.

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

def generate_blockchain_test(self) -> BlockchainTest:
    """
    Generate a BlockchainTest fixture from this StateTest fixture.
    """
    return BlockchainTest(
        genesis_environment=self._generate_blockchain_genesis_environment(),
        pre=self.pre,
        post=self.post,
        blocks=self._generate_blockchain_blocks(),
        fixture_format=self.fixture_format,
        t8n_dump_dir=self.t8n_dump_dir,
    )

make_state_test_fixture(t8n, fork, eips=None)

Create a fixture from the state test definition.

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

def make_state_test_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> Fixture:
    """
    Create a fixture from the state test definition.
    """
    env = self.env.set_fork_requirements(fork)
    tx = self.tx.with_signature_and_sender(keep_secret_key=True)
    pre_alloc = Alloc.merge(
        Alloc(
            fork.pre_allocation(block_number=env.number, timestamp=Number(env.timestamp)),
        ),
        Alloc(self.pre),
    )
    transition_tool_name = fork.transition_tool_name(
        block_number=Number(self.env.number),
        timestamp=Number(self.env.timestamp),
    )
    fork_name = (
        "+".join([transition_tool_name] + [str(eip) for eip in eips])
        if eips
        else transition_tool_name
    )
    next_alloc, result = t8n.evaluate(
        alloc=to_json(pre_alloc),
        txs=to_json([tx]),
        env=to_json(env),
        fork_name=fork_name,
        chain_id=self.chain_id,
        reward=0,  # Reward on state tests is always zero
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )

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

    # Perform post state processing required for some forks
    if fork >= Cancun:
        # StateTest does not execute any beacon root contract logic, but we still need to
        # set the beacon root to the correct value, because most tests assume this happens,
        # so we copy the beacon root contract storage from the post state into the pre state
        # and the transaction is executed in isolation properly.
        if beacon_roots_account := next_alloc.get(str(BEACON_ROOTS_ADDRESS)):
            if beacon_roots_storage := beacon_roots_account.get("storage"):
                pre_alloc = Alloc.merge(
                    pre_alloc,
                    Alloc({BEACON_ROOTS_ADDRESS: Account(storage=beacon_roots_storage)}),
                )

    return Fixture(
        env=env,
        pre_state=pre_alloc,
        post={
            fork.blockchain_test_network_name(): [
                FixtureForkPost.collect(
                    transition_tool_result=result,
                    transaction=tx.with_signature_and_sender(),
                )
            ]
        },
        transaction=tx,
    )

generate(t8n, fork, eips=None)

Generate the BlockchainTest fixture.

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

def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    if self.fixture_format in BlockchainTest.fixture_formats():
        return self.generate_blockchain_test().generate(t8n, fork, eips)
    elif self.fixture_format == FixtureFormats.STATE_TEST:
        return self.make_state_test_fixture(t8n, fork, eips)

    raise Exception(f"Unknown fixture format: {self.fixture_format}")

TestInfo dataclass

Contains test information from the current node.

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

@dataclass(kw_only=True)
class TestInfo:
    """
    Contains test information from the current node.
    """

    name: str  # pytest: Item.name
    id: str  # pytest: Item.nodeid
    original_name: str  # pytest: Item.originalname
    path: Path  # pytest: Item.path

    def get_name_and_parameters(self) -> Tuple[str, str]:
        """
        Converts a test name to a tuple containing the test name and test parameters.

        Example:
        test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai
        """
        test_name, parameters = self.name.split("[")
        return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

    def get_single_test_name(self) -> str:
        """
        Converts a test name to a single test name.
        """
        test_name, test_parameters = self.get_name_and_parameters()
        return f"{test_name}__{test_parameters}"

    def get_dump_dir_path(
        self,
        base_dump_dir: Optional[Path],
        filler_path: Path,
        level: Literal["test_module", "test_function", "test_parameter"] = "test_parameter",
    ) -> Optional[Path]:
        """
        The path to dump the debug output as defined by the level to dump at.
        """
        if not base_dump_dir:
            return None
        test_module_relative_dir = get_module_relative_output_dir(self.path, filler_path)
        if level == "test_module":
            return Path(base_dump_dir) / Path(str(test_module_relative_dir).replace(os.sep, "__"))
        test_name, test_parameter_string = self.get_name_and_parameters()
        flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{test_name}"
        if level == "test_function":
            return Path(base_dump_dir) / flat_path
        elif level == "test_parameter":
            return Path(base_dump_dir) / flat_path / test_parameter_string
        raise Exception("Unexpected level.")

get_name_and_parameters()

Converts a test name to a tuple containing the test name and test parameters.

Example: test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai

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

def get_name_and_parameters(self) -> Tuple[str, str]:
    """
    Converts a test name to a tuple containing the test name and test parameters.

    Example:
    test_push0_key_sstore[fork_Shanghai] -> test_push0_key_sstore, fork_Shanghai
    """
    test_name, parameters = self.name.split("[")
    return test_name, re.sub(r"[\[\-]", "_", parameters).replace("]", "")

get_single_test_name()

Converts a test name to a single test name.

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

def get_single_test_name(self) -> str:
    """
    Converts a test name to a single test name.
    """
    test_name, test_parameters = self.get_name_and_parameters()
    return f"{test_name}__{test_parameters}"

get_dump_dir_path(base_dump_dir, filler_path, level='test_parameter')

The path to dump the debug output as defined by the level to dump at.

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

def get_dump_dir_path(
    self,
    base_dump_dir: Optional[Path],
    filler_path: Path,
    level: Literal["test_module", "test_function", "test_parameter"] = "test_parameter",
) -> Optional[Path]:
    """
    The path to dump the debug output as defined by the level to dump at.
    """
    if not base_dump_dir:
        return None
    test_module_relative_dir = get_module_relative_output_dir(self.path, filler_path)
    if level == "test_module":
        return Path(base_dump_dir) / Path(str(test_module_relative_dir).replace(os.sep, "__"))
    test_name, test_parameter_string = self.get_name_and_parameters()
    flat_path = f"{str(test_module_relative_dir).replace(os.sep, '__')}__{test_name}"
    if level == "test_function":
        return Path(base_dump_dir) / flat_path
    elif level == "test_parameter":
        return Path(base_dump_dir) / flat_path / test_parameter_string
    raise Exception("Unexpected level.")

Block dataclass

Bases: Header

Block type used to describe block properties in test specs

Source code in src/ethereum_test_tools/spec/blockchain/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.
    """
    header_verify: Optional[Header] = None
    """
    If set, the block header will be verified against the specified values.
    """
    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[BlockException | TransactionException | ExceptionList] = 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.

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

If set, the block header will be verified against the specified values.

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[BlockException | TransactionException | ExceptionList] = 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/spec/blockchain/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/spec/blockchain/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

Header dataclass

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

Source code in src/ethereum_test_tools/spec/blockchain/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.
    """
    EMPTY_FIELD: ClassVar[Removable] = Removable()
    """
    Sentinel object used to specify that a header field must be empty during verification.
    """

REMOVE_FIELD: Removable = Removable() class-attribute

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

EMPTY_FIELD: Removable = Removable() class-attribute

Sentinel object used to specify that a header field must be empty during verification.

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_

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)
    BLOBBASEFEE = Opcode(0x4A, popped_stack_items=0, 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)
    TLOAD = Opcode(0x5C, popped_stack_items=1, pushed_stack_items=1)
    TSTORE = Opcode(0x5D, popped_stack_items=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)

    RJUMP = Opcode(0xE0, data_portion_length=2)
    RJUMPI = Opcode(0xE1, popped_stack_items=1, data_portion_length=2)
    RJUMPV = Opcode(0xE2)

    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)