Test Filler Plugin

A pytest plugin that provides fixtures that fill tests and generate fixtures.

Top-level pytest configuration file providing: - Command-line options, - Test-fixtures that can be used by all test cases, and that modifies pytest hooks in order to fill test specs for all tests and writes the generated fixtures to file.

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

Fills fixtures for the specified fork.

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

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

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

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

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

    return fixture

BlockchainTest dataclass

Bases: BaseTest

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

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

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

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

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

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

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

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

        return genesis_rlp, genesis

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

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

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

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

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

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

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

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

            assert len(header.state_root) == 32

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

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

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

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

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

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

        return (blocks, head, alloc)

pytest_parameter_name() classmethod

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

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

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

make_genesis(t8n, fork)

Create a genesis block from the state test definition.

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

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

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

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

    return genesis_rlp, genesis

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

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

Returns

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

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

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

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

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

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

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

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

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

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

        assert len(header.state_root) == 32

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

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

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

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

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

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

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

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

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

    return (blocks, head, alloc)

TransitionTool

Transition tool abstract base class which should be inherited by all transition tool implementations.

Source code in src/evm_transition_tool/transition_tool.py

class TransitionTool:
    """
    Transition tool abstract base class which should be inherited by all transition tool
    implementations.
    """

    traces: List[List[List[Dict]]] | None = None

    registered_tools: List[Type["TransitionTool"]] = []
    default_tool: Optional[Type["TransitionTool"]] = None
    default_binary: Path
    detect_binary_pattern: Pattern
    version_flag: str = "-v"

    # Abstract methods that each tool must implement

    @abstractmethod
    def __init__(
        self,
        *,
        binary: Optional[Path] = None,
        trace: bool = False,
    ):
        """
        Abstract initialization method that all subclasses must implement.
        """
        if binary is None:
            binary = self.default_binary
        else:
            # improve behavior of which by resolving the path: ~/relative paths don't work
            resolved_path = Path(os.path.expanduser(binary)).resolve()
            if resolved_path.exists():
                binary = resolved_path
        binary = shutil.which(binary)  # type: ignore
        if not binary:
            raise TransitionToolNotFoundInPath(binary=binary)
        self.binary = Path(binary)
        self.trace = trace

    def __init_subclass__(cls):
        """
        Registers all subclasses of TransitionTool as possible tools.
        """
        TransitionTool.register_tool(cls)

    @classmethod
    def register_tool(cls, tool_subclass: Type["TransitionTool"]):
        """
        Registers a given subclass as tool option.
        """
        cls.registered_tools.append(tool_subclass)

    @classmethod
    def set_default_tool(cls, tool_subclass: Type["TransitionTool"]):
        """
        Registers the default tool subclass.
        """
        cls.default_tool = tool_subclass

    @classmethod
    def from_binary_path(cls, *, binary_path: Optional[Path], **kwargs) -> "TransitionTool":
        """
        Instantiates the appropriate TransitionTool subclass derived from the
        tool's binary path.
        """
        assert cls.default_tool is not None, "default transition tool was never set"

        if binary_path is None:
            return cls.default_tool(binary=binary_path, **kwargs)

        resolved_path = Path(os.path.expanduser(binary_path)).resolve()
        if resolved_path.exists():
            binary_path = resolved_path
        binary = shutil.which(binary_path)  # type: ignore

        if not binary:
            raise TransitionToolNotFoundInPath(binary=binary)

        binary = Path(binary)

        # Group the tools by version flag, so we only have to call the tool once for all the
        # classes that share the same version flag
        for version_flag, subclasses in groupby(
            cls.registered_tools, key=lambda x: x.version_flag
        ):
            try:
                with os.popen(f"{binary} {version_flag}") as f:
                    binary_output = f.read()
            except Exception:
                # If the tool doesn't support the version flag,
                # we'll get an non-zero exit code.
                continue
            for subclass in subclasses:
                if subclass.detect_binary(binary_output):
                    return subclass(binary=binary, **kwargs)

        raise UnknownTransitionTool(f"Unknown transition tool binary: {binary_path}")

    @classmethod
    def detect_binary(cls, binary_output: str) -> bool:
        """
        Returns True if the binary matches the tool
        """
        assert cls.detect_binary_pattern is not None

        return cls.detect_binary_pattern.match(binary_output) is not None

    @abstractmethod
    def evaluate(
        self,
        alloc: Any,
        txs: Any,
        env: Any,
        fork: Fork,
        chain_id: int = 1,
        reward: int = 0,
        eips: Optional[List[int]] = None,
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Simulate a state transition with specified parameters
        """
        pass

    @abstractmethod
    def version(self) -> str:
        """
        Return name and version of tool used to state transition
        """
        pass

    @abstractmethod
    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool
        """
        pass

    def reset_traces(self):
        """
        Resets the internal trace storage for a new test to begin
        """
        self.traces = None

    def append_traces(self, new_traces: List[List[Dict]]):
        """
        Appends a list of traces of a state transition to the current list
        """
        if self.traces is None:
            self.traces = []
        self.traces.append(new_traces)

    def get_traces(self) -> List[List[List[Dict]]] | None:
        """
        Returns the accumulated traces
        """
        return self.traces

    def calc_state_root(self, alloc: Any, fork: Fork) -> bytes:
        """
        Calculate the state root for the given `alloc`.
        """
        env: Dict[str, Any] = {
            "currentCoinbase": "0x0000000000000000000000000000000000000000",
            "currentDifficulty": "0x0",
            "currentGasLimit": "0x0",
            "currentNumber": "0",
            "currentTimestamp": "0",
        }

        if fork.header_base_fee_required(0, 0):
            env["currentBaseFee"] = "7"

        if fork.header_prev_randao_required(0, 0):
            env["currentRandom"] = "0"

        if fork.header_withdrawals_required(0, 0):
            env["withdrawals"] = []

        _, result = self.evaluate(alloc, [], env, fork)
        state_root = result.get("stateRoot")
        if state_root is None or not isinstance(state_root, str):
            raise Exception("Unable to calculate state root")
        return bytes.fromhex(state_root[2:])

    def calc_withdrawals_root(self, withdrawals: Any, fork: Fork) -> bytes:
        """
        Calculate the state root for the given `alloc`.
        """
        if type(withdrawals) is list and len(withdrawals) == 0:
            # Optimize returning the empty root immediately
            return bytes.fromhex(
                "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
            )

        env: Dict[str, Any] = {
            "currentCoinbase": "0x0000000000000000000000000000000000000000",
            "currentDifficulty": "0x0",
            "currentGasLimit": "0x0",
            "currentNumber": "0",
            "currentTimestamp": "0",
            "withdrawals": withdrawals,
        }

        if fork.header_base_fee_required(0, 0):
            env["currentBaseFee"] = "7"

        if fork.header_prev_randao_required(0, 0):
            env["currentRandom"] = "0"

        if fork.header_excess_data_gas_required(0, 0):
            env["currentExcessDataGas"] = "0"

        _, result = self.evaluate({}, [], env, fork)
        withdrawals_root = result.get("withdrawalsRoot")
        if withdrawals_root is None:
            raise Exception(
                "Unable to calculate withdrawals root: no value returned from transition tool"
            )
        if type(withdrawals_root) is not str:
            raise Exception(
                "Unable to calculate withdrawals root: "
                + "incorrect type returned from transition tool: "
                + f"{withdrawals_root}"
            )
        return bytes.fromhex(withdrawals_root[2:])

__init__(*, binary=None, trace=False) abstractmethod

Abstract initialization method that all subclasses must implement.

Source code in src/evm_transition_tool/transition_tool.py

@abstractmethod
def __init__(
    self,
    *,
    binary: Optional[Path] = None,
    trace: bool = False,
):
    """
    Abstract initialization method that all subclasses must implement.
    """
    if binary is None:
        binary = self.default_binary
    else:
        # improve behavior of which by resolving the path: ~/relative paths don't work
        resolved_path = Path(os.path.expanduser(binary)).resolve()
        if resolved_path.exists():
            binary = resolved_path
    binary = shutil.which(binary)  # type: ignore
    if not binary:
        raise TransitionToolNotFoundInPath(binary=binary)
    self.binary = Path(binary)
    self.trace = trace

__init_subclass__()

Registers all subclasses of TransitionTool as possible tools.

Source code in src/evm_transition_tool/transition_tool.py

def __init_subclass__(cls):
    """
    Registers all subclasses of TransitionTool as possible tools.
    """
    TransitionTool.register_tool(cls)

register_tool(tool_subclass) classmethod

Registers a given subclass as tool option.

Source code in src/evm_transition_tool/transition_tool.py

@classmethod
def register_tool(cls, tool_subclass: Type["TransitionTool"]):
    """
    Registers a given subclass as tool option.
    """
    cls.registered_tools.append(tool_subclass)

set_default_tool(tool_subclass) classmethod

Registers the default tool subclass.

Source code in src/evm_transition_tool/transition_tool.py

@classmethod
def set_default_tool(cls, tool_subclass: Type["TransitionTool"]):
    """
    Registers the default tool subclass.
    """
    cls.default_tool = tool_subclass

from_binary_path(*, binary_path, **kwargs) classmethod

Instantiates the appropriate TransitionTool subclass derived from the tool's binary path.

Source code in src/evm_transition_tool/transition_tool.py

@classmethod
def from_binary_path(cls, *, binary_path: Optional[Path], **kwargs) -> "TransitionTool":
    """
    Instantiates the appropriate TransitionTool subclass derived from the
    tool's binary path.
    """
    assert cls.default_tool is not None, "default transition tool was never set"

    if binary_path is None:
        return cls.default_tool(binary=binary_path, **kwargs)

    resolved_path = Path(os.path.expanduser(binary_path)).resolve()
    if resolved_path.exists():
        binary_path = resolved_path
    binary = shutil.which(binary_path)  # type: ignore

    if not binary:
        raise TransitionToolNotFoundInPath(binary=binary)

    binary = Path(binary)

    # Group the tools by version flag, so we only have to call the tool once for all the
    # classes that share the same version flag
    for version_flag, subclasses in groupby(
        cls.registered_tools, key=lambda x: x.version_flag
    ):
        try:
            with os.popen(f"{binary} {version_flag}") as f:
                binary_output = f.read()
        except Exception:
            # If the tool doesn't support the version flag,
            # we'll get an non-zero exit code.
            continue
        for subclass in subclasses:
            if subclass.detect_binary(binary_output):
                return subclass(binary=binary, **kwargs)

    raise UnknownTransitionTool(f"Unknown transition tool binary: {binary_path}")

detect_binary(binary_output) classmethod

Returns True if the binary matches the tool

Source code in src/evm_transition_tool/transition_tool.py

@classmethod
def detect_binary(cls, binary_output: str) -> bool:
    """
    Returns True if the binary matches the tool
    """
    assert cls.detect_binary_pattern is not None

    return cls.detect_binary_pattern.match(binary_output) is not None

evaluate(alloc, txs, env, fork, chain_id=1, reward=0, eips=None) abstractmethod

Simulate a state transition with specified parameters

Source code in src/evm_transition_tool/transition_tool.py

@abstractmethod
def evaluate(
    self,
    alloc: Any,
    txs: Any,
    env: Any,
    fork: Fork,
    chain_id: int = 1,
    reward: int = 0,
    eips: Optional[List[int]] = None,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """
    Simulate a state transition with specified parameters
    """
    pass

version() abstractmethod

Return name and version of tool used to state transition

Source code in src/evm_transition_tool/transition_tool.py

@abstractmethod
def version(self) -> str:
    """
    Return name and version of tool used to state transition
    """
    pass

is_fork_supported(fork) abstractmethod

Returns True if the fork is supported by the tool

Source code in src/evm_transition_tool/transition_tool.py

@abstractmethod
def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    pass

reset_traces()

Resets the internal trace storage for a new test to begin

Source code in src/evm_transition_tool/transition_tool.py

def reset_traces(self):
    """
    Resets the internal trace storage for a new test to begin
    """
    self.traces = None

append_traces(new_traces)

Appends a list of traces of a state transition to the current list

Source code in src/evm_transition_tool/transition_tool.py

def append_traces(self, new_traces: List[List[Dict]]):
    """
    Appends a list of traces of a state transition to the current list
    """
    if self.traces is None:
        self.traces = []
    self.traces.append(new_traces)

get_traces()

Returns the accumulated traces

Source code in src/evm_transition_tool/transition_tool.py

def get_traces(self) -> List[List[List[Dict]]] | None:
    """
    Returns the accumulated traces
    """
    return self.traces

calc_state_root(alloc, fork)

Calculate the state root for the given alloc.

Source code in src/evm_transition_tool/transition_tool.py

def calc_state_root(self, alloc: Any, fork: Fork) -> bytes:
    """
    Calculate the state root for the given `alloc`.
    """
    env: Dict[str, Any] = {
        "currentCoinbase": "0x0000000000000000000000000000000000000000",
        "currentDifficulty": "0x0",
        "currentGasLimit": "0x0",
        "currentNumber": "0",
        "currentTimestamp": "0",
    }

    if fork.header_base_fee_required(0, 0):
        env["currentBaseFee"] = "7"

    if fork.header_prev_randao_required(0, 0):
        env["currentRandom"] = "0"

    if fork.header_withdrawals_required(0, 0):
        env["withdrawals"] = []

    _, result = self.evaluate(alloc, [], env, fork)
    state_root = result.get("stateRoot")
    if state_root is None or not isinstance(state_root, str):
        raise Exception("Unable to calculate state root")
    return bytes.fromhex(state_root[2:])

calc_withdrawals_root(withdrawals, fork)

Calculate the state root for the given alloc.

Source code in src/evm_transition_tool/transition_tool.py

def calc_withdrawals_root(self, withdrawals: Any, fork: Fork) -> bytes:
    """
    Calculate the state root for the given `alloc`.
    """
    if type(withdrawals) is list and len(withdrawals) == 0:
        # Optimize returning the empty root immediately
        return bytes.fromhex(
            "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
        )

    env: Dict[str, Any] = {
        "currentCoinbase": "0x0000000000000000000000000000000000000000",
        "currentDifficulty": "0x0",
        "currentGasLimit": "0x0",
        "currentNumber": "0",
        "currentTimestamp": "0",
        "withdrawals": withdrawals,
    }

    if fork.header_base_fee_required(0, 0):
        env["currentBaseFee"] = "7"

    if fork.header_prev_randao_required(0, 0):
        env["currentRandom"] = "0"

    if fork.header_excess_data_gas_required(0, 0):
        env["currentExcessDataGas"] = "0"

    _, result = self.evaluate({}, [], env, fork)
    withdrawals_root = result.get("withdrawalsRoot")
    if withdrawals_root is None:
        raise Exception(
            "Unable to calculate withdrawals root: no value returned from transition tool"
        )
    if type(withdrawals_root) is not str:
        raise Exception(
            "Unable to calculate withdrawals root: "
            + "incorrect type returned from transition tool: "
            + f"{withdrawals_root}"
        )
    return bytes.fromhex(withdrawals_root[2:])

pytest_addoption(parser)

Adds command-line options to pytest.

Source code in src/pytest_plugins/test_filler/test_filler.py

def pytest_addoption(parser):
    """
    Adds command-line options to pytest.
    """
    evm_group = parser.getgroup("evm", "Arguments defining evm executable behavior")
    evm_group.addoption(
        "--evm-bin",
        action="store",
        dest="evm_bin",
        type=Path,
        default=None,
        help=(
            "Path to an evm executable that provides `t8n`. " "Default: First 'evm' entry in PATH"
        ),
    )
    evm_group.addoption(
        "--traces",
        action="store_true",
        dest="evm_collect_traces",
        default=None,
        help="Collect traces of the execution information from the " + "transition tool",
    )

    solc_group = parser.getgroup("solc", "Arguments defining the solc executable")
    solc_group.addoption(
        "--solc-bin",
        action="store",
        dest="solc_bin",
        default=None,
        help=(
            "Path to a solc executable (for Yul source compilation). "
            "Default: First 'solc' entry in PATH"
        ),
    )

    test_group = parser.getgroup("tests", "Arguments defining filler location and output")
    test_group.addoption(
        "--filler-path",
        action="store",
        dest="filler_path",
        default="./tests/",
        help="Path to filler directives",
    )
    test_group.addoption(
        "--output",
        action="store",
        dest="output",
        default="./fixtures/",
        help="Directory to store the generated test fixtures. Can be deleted.",
    )
    test_group.addoption(
        "--flat-output",
        action="store_true",
        dest="flat_output",
        default=False,
        help="Output each test case in the directory without the folder structure.",
    )

StateTest dataclass

Bases: BaseTest

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

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

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

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

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

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

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

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

        return genesis_rlp, genesis

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

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

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

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

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

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

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

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

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

pytest_parameter_name() classmethod

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

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

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

make_genesis(t8n, fork)

Create a genesis block from the state test definition.

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

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

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

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

    return genesis_rlp, genesis

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

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

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

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

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

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

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

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

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

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

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

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

Yul

Bases: Code

Yul compiler. Compiles Yul source code into bytecode.

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

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

    source: str
    compiled: Optional[bytes] = None

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

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

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

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

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

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

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

assemble()

Assembles using solc --assemble.

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

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

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

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

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

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

version()

Return solc's version string

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

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

BaseTest

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

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

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

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

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

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

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

make_genesis(t8n, fork) abstractmethod

Create a genesis block from the test definition.

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

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

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

Generate the blockchain that must be executed sequentially during test.

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

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

pytest_parameter_name() classmethod abstractmethod

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

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

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

pytest_configure(config)

Register the plugin's custom markers and process command-line options.

Custom marker registration: https://docs.pytest.org/en/7.1.x/how-to/writing_plugins.html#registering-custom-markers

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
    """
    Register the plugin's custom markers and process command-line options.

    Custom marker registration:
    https://docs.pytest.org/en/7.1.x/how-to/writing_plugins.html#registering-custom-markers
    """
    config.addinivalue_line(
        "markers",
        "state_test: a test case that implement a single state transition test.",
    )
    config.addinivalue_line(
        "markers",
        "blockchain_test: a test case that implements a block transition test.",
    )
    config.addinivalue_line(
        "markers",
        "yul_test: a test case that compiles Yul code.",
    )
    config.addinivalue_line(
        "markers",
        "compile_yul_with(fork): Always compile Yul source using the corresponding evm version.",
    )
    if config.option.collectonly:
        return
    # Instantiate the transition tool here to check that the binary path/trace option is valid.
    # This ensures we only raise an error once, if appropriate, instead of for every test.
    TransitionTool.from_binary_path(
        binary_path=config.getoption("evm_bin"), trace=config.getoption("evm_collect_traces")
    )

EIPSpecTestItem

Bases: Item

Custom pytest test item to test EIP spec versions.

Source code in src/pytest_plugins/spec_version_checker/spec_version_checker.py

class EIPSpecTestItem(Item):
    """
    Custom pytest test item to test EIP spec versions.
    """

    def __init__(self, name, parent, module):
        super().__init__(name, parent)
        self.module = module

    @classmethod
    def from_parent(cls, parent, module):
        """
        Public constructor to define new tests.
        https://docs.pytest.org/en/latest/reference/reference.html#pytest.nodes.Node.from_parent
        """
        return super().from_parent(parent=parent, name="test_eip_spec_version", module=module)

    def runtest(self):
        """
        Define the test to execute for this item.
        """
        test_eip_spec_version(self.module)

    def reportinfo(self):
        """
        Get location information for this test item to use test reports.
        """
        return "spec_version_checker", 0, f"{self.name}"

from_parent(parent, module) classmethod

Public constructor to define new tests. https://docs.pytest.org/en/latest/reference/reference.html#pytest.nodes.Node.from_parent

Source code in src/pytest_plugins/spec_version_checker/spec_version_checker.py

@classmethod
def from_parent(cls, parent, module):
    """
    Public constructor to define new tests.
    https://docs.pytest.org/en/latest/reference/reference.html#pytest.nodes.Node.from_parent
    """
    return super().from_parent(parent=parent, name="test_eip_spec_version", module=module)

runtest()

Define the test to execute for this item.

Source code in src/pytest_plugins/spec_version_checker/spec_version_checker.py

def runtest(self):
    """
    Define the test to execute for this item.
    """
    test_eip_spec_version(self.module)

reportinfo()

Get location information for this test item to use test reports.

Source code in src/pytest_plugins/spec_version_checker/spec_version_checker.py

def reportinfo(self):
    """
    Get location information for this test item to use test reports.
    """
    return "spec_version_checker", 0, f"{self.name}"

pytest_report_header(config, start_path)

Add lines to pytest's console output header

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.hookimpl(trylast=True)
def pytest_report_header(config, start_path):
    """Add lines to pytest's console output header"""
    if config.option.collectonly:
        return
    binary_path = config.getoption("evm_bin")
    t8n = TransitionTool.from_binary_path(binary_path=binary_path)
    solc_version_string = Yul("", binary=config.getoption("solc_bin")).version()
    return [f"{t8n.version()}, solc version {solc_version_string}"]

evm_bin(request)

Returns the configured evm tool binary path.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def evm_bin(request) -> Path:
    """
    Returns the configured evm tool binary path.
    """
    return request.config.getoption("evm_bin")

solc_bin(request)

Returns the configured solc binary path.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def solc_bin(request):
    """
    Returns the configured solc binary path.
    """
    return request.config.getoption("solc_bin")

t8n(request, evm_bin)

Returns the configured transition tool.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def t8n(request, evm_bin: Path) -> TransitionTool:
    """
    Returns the configured transition tool.
    """
    return TransitionTool.from_binary_path(
        binary_path=evm_bin, trace=request.config.getoption("evm_collect_traces")
    )

strip_test_prefix(name)

Removes the test prefix from a test case name.

Source code in src/pytest_plugins/test_filler/test_filler.py

def strip_test_prefix(name: str) -> str:
    """
    Removes the test prefix from a test case name.
    """
    TEST_PREFIX = "test_"
    if name.startswith(TEST_PREFIX):
        return name[len(TEST_PREFIX) :]
    return name

FixtureCollector

Collects all fixtures generated by the test cases.

Source code in src/pytest_plugins/test_filler/test_filler.py

class FixtureCollector:
    """
    Collects all fixtures generated by the test cases.
    """

    all_fixtures: Dict[str, List[Tuple[str, Any]]]
    output_dir: str
    flat_output: bool

    def __init__(self, output_dir: str, flat_output: bool) -> None:
        self.all_fixtures = {}
        self.output_dir = output_dir
        self.flat_output = flat_output

    def add_fixture(self, item, fixture: Fixture) -> None:
        """
        Adds a fixture to the list of fixtures of a given test case.
        """

        def get_module_dir(item) -> str:
            """
            Returns the directory of the test case module.
            """
            dirname = os.path.dirname(item.path)
            basename, _ = os.path.splitext(item.path)
            basename = strip_test_prefix(os.path.basename(basename))
            module_path_no_ext = os.path.join(dirname, basename)
            module_dir = os.path.relpath(
                module_path_no_ext,
                item.funcargs["filler_path"],
            )
            return module_dir

        module_dir = (
            strip_test_prefix(item.originalname)
            if self.flat_output
            else os.path.join(
                get_module_dir(item),
                strip_test_prefix(item.originalname),
            )
        )
        if module_dir not in self.all_fixtures:
            self.all_fixtures[module_dir] = []
        m = re.match(r".*?\[(.*)\]", item.name)
        if not m:
            raise Exception("Could not parse test name: " + item.name)
        name = m.group(1)
        if fixture.name:
            name += "-" + fixture.name
        jsonFixture = json.loads(json.dumps(fixture, cls=JSONEncoder))
        self.all_fixtures[module_dir].append((name, jsonFixture))

    def dump_fixtures(self) -> None:
        """
        Dumps all collected fixtures to their respective files.
        """
        os.makedirs(self.output_dir, exist_ok=True)
        for module_file, fixtures in self.all_fixtures.items():
            output_json = {}
            for index, name_fixture in enumerate(fixtures):
                name, fixture = name_fixture
                name = str(index).zfill(3) + "-" + name
                output_json[name] = fixture
            file_path = os.path.join(self.output_dir, module_file + ".json")
            if not self.flat_output:
                os.makedirs(os.path.dirname(file_path), exist_ok=True)
            with open(file_path, "w") as f:
                json.dump(output_json, f, indent=4)

add_fixture(item, fixture)

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

Source code in src/pytest_plugins/test_filler/test_filler.py

def add_fixture(self, item, fixture: Fixture) -> None:
    """
    Adds a fixture to the list of fixtures of a given test case.
    """

    def get_module_dir(item) -> str:
        """
        Returns the directory of the test case module.
        """
        dirname = os.path.dirname(item.path)
        basename, _ = os.path.splitext(item.path)
        basename = strip_test_prefix(os.path.basename(basename))
        module_path_no_ext = os.path.join(dirname, basename)
        module_dir = os.path.relpath(
            module_path_no_ext,
            item.funcargs["filler_path"],
        )
        return module_dir

    module_dir = (
        strip_test_prefix(item.originalname)
        if self.flat_output
        else os.path.join(
            get_module_dir(item),
            strip_test_prefix(item.originalname),
        )
    )
    if module_dir not in self.all_fixtures:
        self.all_fixtures[module_dir] = []
    m = re.match(r".*?\[(.*)\]", item.name)
    if not m:
        raise Exception("Could not parse test name: " + item.name)
    name = m.group(1)
    if fixture.name:
        name += "-" + fixture.name
    jsonFixture = json.loads(json.dumps(fixture, cls=JSONEncoder))
    self.all_fixtures[module_dir].append((name, jsonFixture))

dump_fixtures()

Dumps all collected fixtures to their respective files.

Source code in src/pytest_plugins/test_filler/test_filler.py

def dump_fixtures(self) -> None:
    """
    Dumps all collected fixtures to their respective files.
    """
    os.makedirs(self.output_dir, exist_ok=True)
    for module_file, fixtures in self.all_fixtures.items():
        output_json = {}
        for index, name_fixture in enumerate(fixtures):
            name, fixture = name_fixture
            name = str(index).zfill(3) + "-" + name
            output_json[name] = fixture
        file_path = os.path.join(self.output_dir, module_file + ".json")
        if not self.flat_output:
            os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "w") as f:
            json.dump(output_json, f, indent=4)

fixture_collector(request)

Returns the configured fixture collector instance used for all tests in one test module.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(scope="module")
def fixture_collector(request):
    """
    Returns the configured fixture collector instance used for all tests
    in one test module.
    """
    fixture_collector = FixtureCollector(
        output_dir=request.config.getoption("output"),
        flat_output=request.config.getoption("flat_output"),
    )
    yield fixture_collector
    fixture_collector.dump_fixtures()

engine()

Returns the sealEngine used in the generated test fixtures.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def engine():
    """
    Returns the sealEngine used in the generated test fixtures.
    """
    return "NoProof"

filler_path(request)

Returns the directory containing the tests to execute.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def filler_path(request):
    """
    Returns the directory containing the tests to execute.
    """
    return request.config.getoption("filler_path")

eips()

A fixture specifying that, by default, no EIPs should be activated for tests.

This fixture (function) may be redefined in test filler modules in order to overwrite this default and return a list of integers specifying which EIPs should be activated for the tests in scope.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True)
def eips():
    """
    A fixture specifying that, by default, no EIPs should be activated for
    tests.

    This fixture (function) may be redefined in test filler modules in order
    to overwrite this default and return a list of integers specifying which
    EIPs should be activated for the tests in scope.
    """
    return []

yul(fork, request)

A fixture that allows contract code to be defined with Yul code.

This fixture defines a class that wraps the ::ethereum_test_tools.Yul class so that upon instantiation within the test case, it provides the test case's current fork parameter. The forks is then available for use in solc's arguments for the Yul code compilation.

Test cases can override the default value by specifying a fixed version with the @pytest.mark.compile_yul_with(FORK) marker.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture
def yul(fork: Fork, request):
    """
    A fixture that allows contract code to be defined with Yul code.

    This fixture defines a class that wraps the ::ethereum_test_tools.Yul
    class so that upon instantiation within the test case, it provides the
    test case's current fork parameter. The forks is then available for use
    in solc's arguments for the Yul code compilation.

    Test cases can override the default value by specifying a fixed version
    with the @pytest.mark.compile_yul_with(FORK) marker.
    """
    marker = request.node.get_closest_marker("compile_yul_with")
    if marker:
        if not marker.args[0]:
            pytest.fail(
                f"{request.node.name}: Expected one argument in 'compile_yul_with' marker."
            )
        fork = request.config.fork_map[marker.args[0]]

    class YulWrapper(Yul):
        def __init__(self, *args, **kwargs):
            super(YulWrapper, self).__init__(*args, **kwargs, fork=fork)

    return YulWrapper

state_test(request, t8n, fork, engine, reference_spec, eips, fixture_collector)

Fixture used to instantiate an auto-fillable StateTest object from within a test function.

Every test that defines a StateTest filler must explicitly specify this fixture in its function arguments and set the StateTestWrapper's spec property.

Implementation detail: It must be scoped on test function level to avoid leakage between tests.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(scope="function")
def state_test(
    request, t8n, fork, engine, reference_spec, eips, fixture_collector
) -> StateTestFiller:
    """
    Fixture used to instantiate an auto-fillable StateTest object from within
    a test function.

    Every test that defines a StateTest filler must explicitly specify this
    fixture in its function arguments and set the StateTestWrapper's spec
    property.

    Implementation detail: It must be scoped on test function level to avoid
    leakage between tests.
    """

    class StateTestWrapper(StateTest):
        def __init__(self, *args, **kwargs):
            super(StateTestWrapper, self).__init__(*args, **kwargs)
            fixture_collector.add_fixture(
                request.node,
                fill_test(
                    t8n,
                    self,
                    fork,
                    engine,
                    reference_spec,
                    eips=eips,
                ),
            )

    return StateTestWrapper

blockchain_test(request, t8n, fork, engine, reference_spec, eips, fixture_collector)

Fixture used to define an auto-fillable BlockchainTest analogous to the state_test fixture for StateTests. See the state_test fixture docstring for details.

Source code in src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(scope="function")
def blockchain_test(
    request, t8n, fork, engine, reference_spec, eips, fixture_collector
) -> BlockchainTestFiller:
    """
    Fixture used to define an auto-fillable BlockchainTest analogous to the
    state_test fixture for StateTests.
    See the state_test fixture docstring for details.
    """

    class BlockchainTestWrapper(BlockchainTest):
        def __init__(self, *args, **kwargs):
            super(BlockchainTestWrapper, self).__init__(*args, **kwargs)
            fixture_collector.add_fixture(
                request.node,
                fill_test(
                    t8n,
                    self,
                    fork,
                    engine,
                    reference_spec,
                    eips=eips,
                ),
            )

    return BlockchainTestWrapper

pytest_collection_modifyitems(items, config)

A pytest hook called during collection, after all items have been collected.

Here we dynamically apply "state_test" or "blockchain_test" markers to a test if the test function uses the corresponding fixture.

Source code in src/pytest_plugins/test_filler/test_filler.py

def pytest_collection_modifyitems(items, config):
    """
    A pytest hook called during collection, after all items have been
    collected.

    Here we dynamically apply "state_test" or "blockchain_test" markers
    to a test if the test function uses the corresponding fixture.
    """
    for item in items:
        if isinstance(item, EIPSpecTestItem):
            continue
        if "state_test" in item.fixturenames:
            marker = pytest.mark.state_test()
            item.add_marker(marker)
        elif "blockchain_test" in item.fixturenames:
            marker = pytest.mark.blockchain_test()
            item.add_marker(marker)
        if "yul" in item.fixturenames:
            marker = pytest.mark.yul_test()
            item.add_marker(marker)

pytest_make_parametrize_id(config, val, argname)

Pytest hook called when generating test ids. We use this to generate more readable test ids for the generated tests.

Source code in src/pytest_plugins/test_filler/test_filler.py

def pytest_make_parametrize_id(config, val, argname):
    """
    Pytest hook called when generating test ids. We use this to generate
    more readable test ids for the generated tests.
    """
    return f"{argname}={val}"

pytest_runtest_call(item)

Pytest hook called in the context of test execution.

Source code in src/pytest_plugins/test_filler/test_filler.py

def pytest_runtest_call(item):
    """
    Pytest hook called in the context of test execution.
    """
    if isinstance(item, EIPSpecTestItem):
        return

    class InvalidFiller(Exception):
        def __init__(self, message):
            super().__init__(message)

    if "state_test" in item.fixturenames and "blockchain_test" in item.fixturenames:
        raise InvalidFiller(
            "A filler should only implement either a state test or " "a blockchain test; not both."
        )

    # Check that the test defines either test type as parameter.
    if not any([i for i in item.funcargs if i in SPEC_TYPES_PARAMETERS]):
        pytest.fail(
            "Test must define either one of the following parameters to "
            + "properly generate a test: "
            + ", ".join(SPEC_TYPES_PARAMETERS)
        )

Fixture dataclass

Cross-client compatible Ethereum test fixture.

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

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

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

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

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

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

__post_init__()

Post init hook to convert to JSON after instantiation.

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

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

fill_info(t8n, ref_spec)

Fill the info field for this fixture

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

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

JSONEncoder

Bases: json.JSONEncoder

Custom JSON encoder for ethereum_test types.

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

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

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

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

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

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

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

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

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

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

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

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

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

default(obj)

Enocdes types defined in this module using basic python facilities.

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

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

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

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

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

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

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

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

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

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

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

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