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, b11r, test_spec, fork, engine, spec, eips=None)

Fills fixtures for the specified fork.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/filling/fill.py

def fill_test(
    t8n: TransitionTool,
    b11r: BlockBuilder,
    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(b11r, t8n, fork)

    (blocks, head, alloc) = test_spec.make_blocks(
        b11r,
        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, b11r, spec)

    return fixture

StateTest dataclass

Bases: BaseTest

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

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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]
    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,
        b11r: BlockBuilder,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[str, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = self.env.set_fork_requirements(fork)

        genesis = FixtureHeader(
            parent_hash="0x0000000000000000000000000000000000000000000000000000000000000000",
            ommers_hash="0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
            coinbase="0x0000000000000000000000000000000000000000",
            state_root=t8n.calc_state_root(
                to_json(self.pre),
                fork,
            ),
            transactions_root=EmptyTrieRoot,
            receipt_root=EmptyTrieRoot,
            bloom="0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
            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="0x00",
            mix_digest="0x0000000000000000000000000000000000000000000000000000000000000000",
            nonce="0x0000000000000000",
            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) = b11r.build(
            header=genesis.to_geth_dict(),
            txs="",
            ommers=[],
            withdrawals=env.withdrawals,
        )
        return genesis_rlp, genesis

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

        (alloc, result, txs_rlp) = t8n.evaluate(
            alloc=to_json(self.pre),
            txs=to_json(self.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(self.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, head = b11r.build(
            header=header.to_geth_dict(),
            txs=txs_rlp,
            ommers=[],
            withdrawals=to_json_or_none(env.withdrawals),
        )
        header.hash = head

        return (
            [
                FixtureBlock(
                    rlp=block,
                    block_header=header,
                    txs=self.txs if self.txs is not None else [],
                    ommers=[],
                    withdrawals=env.withdrawals,
                )
            ],
            head,
            alloc,
        )

pytest_parameter_name() classmethod

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

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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(b11r, t8n, fork)

Create a genesis block from the state test definition.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/state_test.py

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

    genesis = FixtureHeader(
        parent_hash="0x0000000000000000000000000000000000000000000000000000000000000000",
        ommers_hash="0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
        coinbase="0x0000000000000000000000000000000000000000",
        state_root=t8n.calc_state_root(
            to_json(self.pre),
            fork,
        ),
        transactions_root=EmptyTrieRoot,
        receipt_root=EmptyTrieRoot,
        bloom="0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
        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="0x00",
        mix_digest="0x0000000000000000000000000000000000000000000000000000000000000000",
        nonce="0x0000000000000000",
        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) = b11r.build(
        header=genesis.to_geth_dict(),
        txs="",
        ommers=[],
        withdrawals=env.withdrawals,
    )
    return genesis_rlp, genesis

make_blocks(b11r, 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/state_test.py

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

    (alloc, result, txs_rlp) = t8n.evaluate(
        alloc=to_json(self.pre),
        txs=to_json(self.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(self.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, head = b11r.build(
        header=header.to_geth_dict(),
        txs=txs_rlp,
        ommers=[],
        withdrawals=to_json_or_none(env.withdrawals),
    )
    header.hash = head

    return (
        [
            FixtureBlock(
                rlp=block,
                block_header=header,
                txs=self.txs if self.txs is not None else [],
                ommers=[],
                withdrawals=env.withdrawals,
            )
        ],
        head,
        alloc,
    )

BlockchainTest dataclass

Bases: BaseTest

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

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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 = 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,
        b11r: BlockBuilder,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[str, FixtureHeader]:
        """
        Create a genesis block from the state test definition.
        """
        env = self.genesis_environment.set_fork_requirements(fork)

        genesis = FixtureHeader(
            parent_hash="0x0000000000000000000000000000000000000000000000000000000000000000",
            ommers_hash="0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
            coinbase="0x0000000000000000000000000000000000000000",
            state_root=t8n.calc_state_root(
                to_json(self.pre),
                fork,
            ),
            transactions_root=EmptyTrieRoot,
            receipt_root=EmptyTrieRoot,
            bloom="0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=0,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data="0x00",
            mix_digest="0x0000000000000000000000000000000000000000000000000000000000000000",
            nonce="0x0000000000000000",
            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) = b11r.build(
            header=genesis.to_geth_dict(),
            txs="",
            ommers=[],
            withdrawals=env.withdrawals,
        )
        return genesis_rlp, genesis

    def make_block(
        self,
        b11r: BlockBuilder,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Dict[str, Any],
        previous_head: str,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[FixtureBlock, Environment, Dict[str, Any], str]:
        """
        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)

            (next_alloc, result, txs_rlp) = t8n.evaluate(
                alloc=previous_alloc,
                txs=to_json_or_none(block.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(block.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) == 66

            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 = b11r.build(
                header=header.to_geth_dict(),
                txs=txs_rlp,
                ommers=[],
                withdrawals=to_json_or_none(env.withdrawals),
            )

            if block.exception is None:
                # Return environment and allocation of the following block
                return (
                    FixtureBlock(
                        rlp=rlp,
                        block_header=header,
                        block_number=header.number,
                        txs=block.txs if block.txs is not None else [],
                        ommers=[],
                        withdrawals=env.withdrawals,
                    ),
                    env.apply_new_parent(header),
                    next_alloc,
                    header.hash,
                )
            else:
                return (
                    FixtureBlock(
                        rlp=rlp,
                        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,
        b11r: BlockBuilder,
        t8n: TransitionTool,
        genesis: FixtureHeader,
        fork: Fork,
        chain_id=1,
        eips: Optional[List[int]] = None,
    ) -> Tuple[List[FixtureBlock], str, 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 "0x0000000000000000000000000000000000000000000000000000000000000000"
        )
        for block in self.blocks:
            fixture_block, env, alloc, head = self.make_block(
                b11r=b11r,
                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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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(b11r, t8n, fork)

Create a genesis block from the state test definition.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/blockchain_test.py

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

    genesis = FixtureHeader(
        parent_hash="0x0000000000000000000000000000000000000000000000000000000000000000",
        ommers_hash="0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",
        coinbase="0x0000000000000000000000000000000000000000",
        state_root=t8n.calc_state_root(
            to_json(self.pre),
            fork,
        ),
        transactions_root=EmptyTrieRoot,
        receipt_root=EmptyTrieRoot,
        bloom="0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",  # noqa: E501
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=0,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data="0x00",
        mix_digest="0x0000000000000000000000000000000000000000000000000000000000000000",
        nonce="0x0000000000000000",
        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) = b11r.build(
        header=genesis.to_geth_dict(),
        txs="",
        ommers=[],
        withdrawals=env.withdrawals,
    )
    return genesis_rlp, genesis

make_block(b11r, 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/blockchain_test.py

def make_block(
    self,
    b11r: BlockBuilder,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Dict[str, Any],
    previous_head: str,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[FixtureBlock, Environment, Dict[str, Any], str]:
    """
    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)

        (next_alloc, result, txs_rlp) = t8n.evaluate(
            alloc=previous_alloc,
            txs=to_json_or_none(block.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(block.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) == 66

        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 = b11r.build(
            header=header.to_geth_dict(),
            txs=txs_rlp,
            ommers=[],
            withdrawals=to_json_or_none(env.withdrawals),
        )

        if block.exception is None:
            # Return environment and allocation of the following block
            return (
                FixtureBlock(
                    rlp=rlp,
                    block_header=header,
                    block_number=header.number,
                    txs=block.txs if block.txs is not None else [],
                    ommers=[],
                    withdrawals=env.withdrawals,
                ),
                env.apply_new_parent(header),
                next_alloc,
                header.hash,
            )
        else:
            return (
                FixtureBlock(
                    rlp=rlp,
                    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(b11r, 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/blockchain_test.py

def make_blocks(
    self,
    b11r: BlockBuilder,
    t8n: TransitionTool,
    genesis: FixtureHeader,
    fork: Fork,
    chain_id=1,
    eips: Optional[List[int]] = None,
) -> Tuple[List[FixtureBlock], str, 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 "0x0000000000000000000000000000000000000000000000000000000000000000"
    )
    for block in self.blocks:
        fixture_block, env, alloc, head = self.make_block(
            b11r=b11r,
            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_addoption(parser)

Adds command-line options to pytest.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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",
        default=None,
        help=(
            "Path to an evm executable that provides `t8n` and `b11r.` "
            "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.",
    )

Yul

Bases: Code

Yul compiler. Compiles Yul source code into bytecode.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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

EvmBlockBuilder

Bases: BlockBuilder

Go-ethereum evm Block builder frontend.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_block_builder/__init__.py

class EvmBlockBuilder(BlockBuilder):
    """
    Go-ethereum `evm` Block builder frontend.
    """

    binary: Path
    cached_version: Optional[str] = None

    def __init__(self, binary: Optional[Path | str] = None):
        if binary is None:
            which_path = which("evm")
            if which_path is not None:
                binary = Path(which_path)
        if binary is None or not Path(binary).exists():
            raise Exception(
                """`evm` binary executable is not accessible, please refer to
                https://github.com/ethereum/go-ethereum on how to compile and
                install the full suite of utilities including the `evm` tool"""
            )
        self.binary = Path(binary)

    def build(
        self,
        header: Any,
        txs: Any,
        ommers: Any,
        withdrawals: Optional[Any] = None,
        clique: Optional[Any] = None,
        ethash: bool = False,
        ethash_mode: str = "normal",
    ) -> Tuple[str, str]:
        """
        Executes `evm b11r` with the specified arguments.
        """
        args = [
            str(self.binary),
            "b11r",
            "--input.header=stdin",
            "--input.txs=stdin",
            "--input.ommers=stdin",
            "--seal.clique=stdin",
            "--output.block=stdout",
            "--input.withdrawals=stdin" if withdrawals is not None else "",
        ]

        if ethash:
            args.append("--seal.ethash")
            args.append("--seal.ethash.mode=" + ethash_mode)

        stdin = {
            "header": header,
            "txs": txs,
            "uncles": ommers,
            "clique": clique,
        }
        if withdrawals is not None:
            stdin["withdrawals"] = withdrawals

        result = subprocess.run(
            args,
            input=str.encode(json.dumps(stdin)),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to build")

        output = json.loads(result.stdout)

        if "rlp" not in output or "hash" not in output:
            Exception("malformed result")

        return (output["rlp"], output["hash"])

    def version(self) -> str:
        """
        Gets `evm` binary version.
        """
        if self.cached_version is None:
            result = subprocess.run(
                [str(self.binary), "-v"],
                stdout=subprocess.PIPE,
            )

            if result.returncode != 0:
                raise Exception("failed to evaluate: " + result.stderr.decode())

            self.cached_version = result.stdout.decode().strip()

        return self.cached_version

build(header, txs, ommers, withdrawals=None, clique=None, ethash=False, ethash_mode='normal')

Executes evm b11r with the specified arguments.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_block_builder/__init__.py

def build(
    self,
    header: Any,
    txs: Any,
    ommers: Any,
    withdrawals: Optional[Any] = None,
    clique: Optional[Any] = None,
    ethash: bool = False,
    ethash_mode: str = "normal",
) -> Tuple[str, str]:
    """
    Executes `evm b11r` with the specified arguments.
    """
    args = [
        str(self.binary),
        "b11r",
        "--input.header=stdin",
        "--input.txs=stdin",
        "--input.ommers=stdin",
        "--seal.clique=stdin",
        "--output.block=stdout",
        "--input.withdrawals=stdin" if withdrawals is not None else "",
    ]

    if ethash:
        args.append("--seal.ethash")
        args.append("--seal.ethash.mode=" + ethash_mode)

    stdin = {
        "header": header,
        "txs": txs,
        "uncles": ommers,
        "clique": clique,
    }
    if withdrawals is not None:
        stdin["withdrawals"] = withdrawals

    result = subprocess.run(
        args,
        input=str.encode(json.dumps(stdin)),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception("failed to build")

    output = json.loads(result.stdout)

    if "rlp" not in output or "hash" not in output:
        Exception("malformed result")

    return (output["rlp"], output["hash"])

version()

Gets evm binary version.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_block_builder/__init__.py

def version(self) -> str:
    """
    Gets `evm` binary version.
    """
    if self.cached_version is None:
        result = subprocess.run(
            [str(self.binary), "-v"],
            stdout=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        self.cached_version = result.stdout.decode().strip()

    return self.cached_version

BaseTest

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

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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,
        b11r: BlockBuilder,
        t8n: TransitionTool,
        fork: Fork,
    ) -> Tuple[str, FixtureHeader]:
        """
        Create a genesis block from the test definition.
        """
        pass

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

Create a genesis block from the test definition.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/base_test.py

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

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

Generate the blockchain that must be executed sequentially during test.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/spec/base_test.py

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

pytest_parameter_name() abstractmethod classmethod

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

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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.",
    )

EIPSpecTestItem

Bases: Item

Custom pytest test item to test EIP spec versions.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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"""
    t8n = EvmTransitionTool(
        binary=config.getoption("evm_bin"),
        trace=config.getoption("evm_collect_traces"),
    )
    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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def evm_bin(request):
    """
    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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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")

EvmTransitionTool

Bases: TransitionTool

Go-ethereum evm Transition tool frontend.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_transition_tool/__init__.py

class EvmTransitionTool(TransitionTool):
    """
    Go-ethereum `evm` Transition tool frontend.
    """

    binary: Path
    cached_version: Optional[str] = None
    trace: bool

    def __init__(
        self,
        binary: Optional[Path | str] = None,
        trace: bool = False,
    ):
        if binary is None:
            which_path = which("evm")
            if which_path is not None:
                binary = Path(which_path)
        if binary is None or not Path(binary).exists():
            raise Exception(
                """`evm` binary executable is not accessible, please refer to
                https://github.com/ethereum/go-ethereum on how to compile and
                install the full suite of utilities including the `evm` tool"""
            )
        self.binary = Path(binary)
        self.trace = trace
        args = [str(self.binary), "t8n", "--help"]
        try:
            result = subprocess.run(args, capture_output=True, text=True)
        except subprocess.CalledProcessError as e:
            raise Exception("evm process unexpectedly returned a non-zero status code: " f"{e}.")
        except Exception as e:
            raise Exception(f"Unexpected exception calling evm tool: {e}.")
        self.help_string = result.stdout

    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], str]:
        """
        Executes `evm t8n` with the specified arguments.
        """
        fork_name = fork.name()
        if eips is not None:
            fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

        temp_dir = tempfile.TemporaryDirectory()

        args = [
            str(self.binary),
            "t8n",
            "--input.alloc=stdin",
            "--input.txs=stdin",
            "--input.env=stdin",
            "--output.result=stdout",
            "--output.alloc=stdout",
            "--output.body=txs.rlp",
            f"--output.basedir={temp_dir.name}",
            f"--state.fork={fork_name}",
            f"--state.chainid={chain_id}",
            f"--state.reward={reward}",
        ]

        if self.trace:
            args.append("--trace")

        stdin = {
            "alloc": alloc,
            "txs": txs,
            "env": env,
        }

        encoded_input = str.encode(json.dumps(stdin))
        result = subprocess.run(
            args,
            input=encoded_input,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        output = json.loads(result.stdout)

        if "alloc" not in output or "result" not in output:
            raise Exception("malformed result")

        with open(os.path.join(temp_dir.name, "txs.rlp"), "r") as txs_rlp_file:
            txs_rlp = txs_rlp_file.read().strip('"')

        if self.trace:
            receipts: List[Any] = output["result"]["receipts"]
            traces: List[List[Dict]] = []
            for i, r in enumerate(receipts):
                h = r["transactionHash"]
                trace_file_name = f"trace-{i}-{h}.jsonl"
                with open(os.path.join(temp_dir.name, trace_file_name), "r") as trace_file:
                    tx_traces: List[Dict] = []
                    for trace_line in trace_file.readlines():
                        tx_traces.append(json.loads(trace_line))
                    traces.append(tx_traces)
            self.append_traces(traces)

        temp_dir.cleanup()

        return (output["alloc"], output["result"], txs_rlp)

    def version(self) -> str:
        """
        Gets `evm` binary version.
        """
        if self.cached_version is None:
            result = subprocess.run(
                [str(self.binary), "-v"],
                stdout=subprocess.PIPE,
            )

            if result.returncode != 0:
                raise Exception("failed to evaluate: " + result.stderr.decode())

            self.cached_version = result.stdout.decode().strip()

        return self.cached_version

    def is_fork_supported(self, fork: Fork) -> bool:
        """
        Returns True if the fork is supported by the tool
        """
        return fork().name() in self.help_string

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

Executes evm t8n with the specified arguments.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_transition_tool/__init__.py

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], str]:
    """
    Executes `evm t8n` with the specified arguments.
    """
    fork_name = fork.name()
    if eips is not None:
        fork_name = "+".join([fork_name] + [str(eip) for eip in eips])

    temp_dir = tempfile.TemporaryDirectory()

    args = [
        str(self.binary),
        "t8n",
        "--input.alloc=stdin",
        "--input.txs=stdin",
        "--input.env=stdin",
        "--output.result=stdout",
        "--output.alloc=stdout",
        "--output.body=txs.rlp",
        f"--output.basedir={temp_dir.name}",
        f"--state.fork={fork_name}",
        f"--state.chainid={chain_id}",
        f"--state.reward={reward}",
    ]

    if self.trace:
        args.append("--trace")

    stdin = {
        "alloc": alloc,
        "txs": txs,
        "env": env,
    }

    encoded_input = str.encode(json.dumps(stdin))
    result = subprocess.run(
        args,
        input=encoded_input,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    if result.returncode != 0:
        raise Exception("failed to evaluate: " + result.stderr.decode())

    output = json.loads(result.stdout)

    if "alloc" not in output or "result" not in output:
        raise Exception("malformed result")

    with open(os.path.join(temp_dir.name, "txs.rlp"), "r") as txs_rlp_file:
        txs_rlp = txs_rlp_file.read().strip('"')

    if self.trace:
        receipts: List[Any] = output["result"]["receipts"]
        traces: List[List[Dict]] = []
        for i, r in enumerate(receipts):
            h = r["transactionHash"]
            trace_file_name = f"trace-{i}-{h}.jsonl"
            with open(os.path.join(temp_dir.name, trace_file_name), "r") as trace_file:
                tx_traces: List[Dict] = []
                for trace_line in trace_file.readlines():
                    tx_traces.append(json.loads(trace_line))
                traces.append(tx_traces)
        self.append_traces(traces)

    temp_dir.cleanup()

    return (output["alloc"], output["result"], txs_rlp)

version()

Gets evm binary version.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_transition_tool/__init__.py

def version(self) -> str:
    """
    Gets `evm` binary version.
    """
    if self.cached_version is None:
        result = subprocess.run(
            [str(self.binary), "-v"],
            stdout=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise Exception("failed to evaluate: " + result.stderr.decode())

        self.cached_version = result.stdout.decode().strip()

    return self.cached_version

is_fork_supported(fork)

Returns True if the fork is supported by the tool

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/evm_transition_tool/__init__.py

def is_fork_supported(self, fork: Fork) -> bool:
    """
    Returns True if the fork is supported by the tool
    """
    return fork().name() in self.help_string

t8n(request, evm_bin)

Returns the configured transition tool.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def t8n(request, evm_bin):
    """
    Returns the configured transition tool.
    """
    t8n = EvmTransitionTool(
        binary=evm_bin,
        trace=request.config.getoption("evm_collect_traces"),
    )
    return t8n

b11r(request, evm_bin)

Returns the configured block builder tool.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(autouse=True, scope="session")
def b11r(request, evm_bin):
    """
    Returns the configured block builder tool.
    """
    b11r = EvmBlockBuilder(binary=evm_bin)
    return b11r

strip_test_prefix(name)

Removes the test prefix from a test case name.

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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, b11r, 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(scope="function")
def state_test(
    request, t8n, b11r, 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,
                    b11r,
                    self,
                    fork,
                    engine,
                    reference_spec,
                    eips=eips,
                ),
            )

    return StateTestWrapper

blockchain_test(request, t8n, b11r, 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/pytest_plugins/test_filler/test_filler.py

@pytest.fixture(scope="function")
def blockchain_test(
    request, t8n, b11r, 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,
                    b11r,
                    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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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: str
    head: str
    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 = ""
    index: int = 0

    _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,
        b11r: BlockBuilder,
        ref_spec: ReferenceSpec | None,
    ):
        """
        Fill the info field for this fixture
        """
        self.info["filling-transition-tool"] = t8n.version()
        self.info["filling-block-build-tool"] = b11r.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 /home/dtopz/code/github/danceratopz/execution-spec-tests/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, b11r, ref_spec)

Fill the info field for this fixture

Source code in /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/common/types.py

def fill_info(
    self,
    t8n: TransitionTool,
    b11r: BlockBuilder,
    ref_spec: ReferenceSpec | None,
):
    """
    Fill the info field for this fixture
    """
    self.info["filling-transition-tool"] = t8n.version()
    self.info["filling-block-build-tool"] = b11r.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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/common/types.py

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

    def default(self, obj):
        """
        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(ACCOUNT_DEFAULTS.nonce)),
                "balance": hex_or_none(obj.balance, hex(ACCOUNT_DEFAULTS.balance)),
                "code": code_or_none(obj.code, "0x"),
                "storage": storage_padding(to_json_or_none(obj.storage, {})),
            }
            return even_padding(account, excluded=["storage"])
        elif isinstance(obj, AccessList):
            access_list = {"address": obj.address}
            if obj.storage_keys is not None:
                access_list["storageKeys"] = obj.storage_keys
            return access_list
        elif isinstance(obj, Transaction):
            tx = {
                "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": "0x" + int.to_bytes(obj.to, length=20, byteorder="big").hex()
                if obj.to is int
                else obj.to,
                "accessList": obj.access_list,
                "protected": obj.protected,
                "secretKey": obj.secret_key,
                "maxFeePerDataGas": hex_or_none(obj.max_fee_per_data_gas),
            }

            if obj.blob_versioned_hashes is not None:
                hashes: List[str] = []
                for h in obj.blob_versioned_hashes:
                    if type(h) is str:
                        hashes.append(h)
                    elif type(h) is bytes:
                        if len(h) != 32:
                            raise TypeError("improper byte size for blob_versioned_hashes")
                        hashes.append("0x" + h.hex())
                    else:
                        raise TypeError("improper type for blob_versioned_hashes")
                tx["blobVersionedHashes"] = hashes

            if obj.secret_key is None:
                assert obj.signature is not None
                assert len(obj.signature) == 3
                tx["v"] = obj.signature[0]
                tx["r"] = obj.signature[1]
                tx["s"] = 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 = {
                "currentCoinbase": obj.coinbase,
                "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): v for (k, v) in obj.block_hashes.items()},
                "ommers": [],
                "withdrawals": to_json_or_none(obj.withdrawals),
                "parentUncleHash": 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": obj.parent_hash,
                "uncleHash": obj.ommers_hash,
                "coinbase": obj.coinbase,
                "stateRoot": obj.state_root,
                "transactionsTrie": obj.transactions_root,
                "receiptTrie": obj.receipt_root,
                "bloom": 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": obj.extra_data if len(obj.extra_data) != 0 else "0x",
                "mixHash": obj.mix_digest,
                "nonce": obj.nonce,
            }
            if obj.base_fee is not None:
                header["baseFeePerGas"] = hex(obj.base_fee)
            if obj.hash is not None:
                header["hash"] = obj.hash
            if obj.withdrawals_root is not None:
                header["withdrawalsRoot"] = 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, FixtureBlock):
            b = {"rlp": obj.rlp}
            if obj.block_header is not None:
                b["blockHeader"] = json.loads(json.dumps(obj.block_header, cls=JSONEncoder))
            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": obj.genesis_rlp,
                "lastblockhash": 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 /home/dtopz/code/github/danceratopz/execution-spec-tests/src/ethereum_test_tools/common/types.py

def default(self, obj):
    """
    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(ACCOUNT_DEFAULTS.nonce)),
            "balance": hex_or_none(obj.balance, hex(ACCOUNT_DEFAULTS.balance)),
            "code": code_or_none(obj.code, "0x"),
            "storage": storage_padding(to_json_or_none(obj.storage, {})),
        }
        return even_padding(account, excluded=["storage"])
    elif isinstance(obj, AccessList):
        access_list = {"address": obj.address}
        if obj.storage_keys is not None:
            access_list["storageKeys"] = obj.storage_keys
        return access_list
    elif isinstance(obj, Transaction):
        tx = {
            "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": "0x" + int.to_bytes(obj.to, length=20, byteorder="big").hex()
            if obj.to is int
            else obj.to,
            "accessList": obj.access_list,
            "protected": obj.protected,
            "secretKey": obj.secret_key,
            "maxFeePerDataGas": hex_or_none(obj.max_fee_per_data_gas),
        }

        if obj.blob_versioned_hashes is not None:
            hashes: List[str] = []
            for h in obj.blob_versioned_hashes:
                if type(h) is str:
                    hashes.append(h)
                elif type(h) is bytes:
                    if len(h) != 32:
                        raise TypeError("improper byte size for blob_versioned_hashes")
                    hashes.append("0x" + h.hex())
                else:
                    raise TypeError("improper type for blob_versioned_hashes")
            tx["blobVersionedHashes"] = hashes

        if obj.secret_key is None:
            assert obj.signature is not None
            assert len(obj.signature) == 3
            tx["v"] = obj.signature[0]
            tx["r"] = obj.signature[1]
            tx["s"] = 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 = {
            "currentCoinbase": obj.coinbase,
            "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): v for (k, v) in obj.block_hashes.items()},
            "ommers": [],
            "withdrawals": to_json_or_none(obj.withdrawals),
            "parentUncleHash": 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": obj.parent_hash,
            "uncleHash": obj.ommers_hash,
            "coinbase": obj.coinbase,
            "stateRoot": obj.state_root,
            "transactionsTrie": obj.transactions_root,
            "receiptTrie": obj.receipt_root,
            "bloom": 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": obj.extra_data if len(obj.extra_data) != 0 else "0x",
            "mixHash": obj.mix_digest,
            "nonce": obj.nonce,
        }
        if obj.base_fee is not None:
            header["baseFeePerGas"] = hex(obj.base_fee)
        if obj.hash is not None:
            header["hash"] = obj.hash
        if obj.withdrawals_root is not None:
            header["withdrawalsRoot"] = 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, FixtureBlock):
        b = {"rlp": obj.rlp}
        if obj.block_header is not None:
            b["blockHeader"] = json.loads(json.dumps(obj.block_header, cls=JSONEncoder))
        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": obj.genesis_rlp,
            "lastblockhash": 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)