Ethereum Test Specs package

Test spec definitions and utilities.

BaseTest

Bases: BaseModel

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

Source code in src/ethereum_test_specs/base.py

class BaseTest(BaseModel):
    """
    Represents a base Ethereum test which must return a single test fixture.
    """

    tag: str = ""

    # Transition tool specific fields
    t8n_dump_dir: Path | None = Field(None, exclude=True)
    _t8n_call_counter: Iterator[int] = count(0)

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = []

    @abstractmethod
    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the list of test fixtures.
        """
        pass

    @classmethod
    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.

        By default, it returns the underscore separated name of the class.
        """
        return reduce(lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__).lower()

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

generate(*, request, t8n, fork, fixture_format, eips=None) abstractmethod

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py

@abstractmethod
def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the list of test fixtures.
    """
    pass

pytest_parameter_name() classmethod

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

By default, it returns the underscore separated name of the class.

Source code in src/ethereum_test_specs/base.py

@classmethod
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.

    By default, it returns the underscore separated name of the class.
    """
    return reduce(lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__).lower()

get_next_transition_tool_output_path()

Returns the path to the next transition tool output file.

Source code in src/ethereum_test_specs/base.py

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

BlockchainTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/blockchain.py

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

    pre: Alloc
    post: Alloc
    blocks: List[Block]
    genesis_environment: Environment = Field(default_factory=Environment)
    verify_sync: bool = False
    chain_id: int = 1

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
    ]

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

        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation_blockchain()),
            self.pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")

        state_root: bytes
        # TODO: refine, currently uses `evm verkle state-root` to get this.
        if fork < Verkle or fork is EIP6800Transition:
            state_root = pre_alloc.state_root()
        else:
            state_root = t8n.get_verkle_state_root(mpt_alloc=pre_alloc)

        genesis = FixtureHeader(
            parent_hash=0,
            ommers_hash=EmptyOmmersRoot,
            fee_recipient=0,
            state_root=state_root,
            transactions_trie=EmptyTrieRoot,
            receipts_root=EmptyTrieRoot,
            logs_bloom=0,
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=0,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data=b"\x00",
            prev_randao=0,
            nonce=0,
            base_fee_per_gas=env.base_fee_per_gas,
            blob_gas_used=env.blob_gas_used,
            excess_blob_gas=env.excess_blob_gas,
            withdrawals_root=(
                Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
            ),
            parent_beacon_block_root=env.parent_beacon_block_root,
            requests_root=(
                Requests(root=[]).trie_root if fork.header_requests_required(0, 0) else None
            ),
        )

        return (
            pre_alloc,
            FixtureBlockBase(
                header=genesis,
                withdrawals=None if env.withdrawals is None else [],
                deposit_requests=[] if fork.header_requests_required(0, 0) else None,
                withdrawal_requests=[] if fork.header_requests_required(0, 0) else None,
                consolidation_requests=[] if fork.header_requests_required(0, 0) else None,
            ).with_rlp(
                txs=[], requests=Requests() if fork.header_requests_required(0, 0) else None
            ),
        )

    def generate_block_data(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Alloc,
        previous_vkt: Optional[VerkleTree] = None,
        eips: Optional[List[int]] = None,
    ) -> Tuple[
        Environment,
        FixtureHeader,
        List[Transaction],
        Alloc,
        Optional[Requests],
        Optional[VerkleTree],
        Optional[Witness],
    ]:
        """
        Generate common block data for both make_fixture and make_hive_fixture.
        """
        if block.rlp and block.exception is not None:
            raise Exception(
                "test correctness: post-state cannot be verified if the "
                + "block's rlp is supplied and the block is not supposed "
                + "to produce an exception"
            )

        env = block.set_environment(previous_env)
        env = env.set_fork_requirements(fork)

        txs = [tx.with_signature_and_sender() for tx in block.txs]

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

        transition_tool_output = t8n.evaluate(
            alloc=previous_alloc,
            txs=txs,
            env=env,
            fork=fork,
            vkt=previous_vkt,
            chain_id=self.chain_id,
            reward=fork.get_reward(env.number, env.timestamp),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )

        try:  # General checks for the transition tool output
            rejected_txs = verify_transactions(txs, transition_tool_output.result)
            verify_result(transition_tool_output.result, env)
            if block.witness_check:
                if transition_tool_output.result.state_diff is None:
                    raise Exception(
                        "no state diff in transition tool output, cannot verify witness"
                    )
                # TODO: hack for now, temp addition to check hist. storage contract
                block.witness_check.add_storage_slot(
                    address=Address(0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFE),
                    storage_slot=env.number - 1,
                    # value=env.parent_hash,
                    value=None,
                )
                self.verify_witness(
                    t8n=t8n,
                    state_diff=transition_tool_output.result.state_diff,
                    witness_check=block.witness_check,
                )
        except Exception as e:
            print_traces(t8n.get_traces())
            print(
                "\nTransition tool output result:\n"
                f"{pformat(transition_tool_output.result.model_dump_json())}"
            )
            print(
                "\nPrevious transition tool alloc:\n"
                f"{pformat(previous_alloc.model_dump_json())}"
            )
            if transition_tool_output.alloc is not None:
                print(
                    "\nTransition tool output alloc:\n"
                    f"{pformat(transition_tool_output.alloc.model_dump_json())}"
                )
            if transition_tool_output.vkt is not None:
                print(
                    "\nTransition tools output verkle tree:\n"
                    f"{pformat(transition_tool_output.vkt.model_dump_json())}"
                )
            # TODO: t8n has the witness state diff from the result for now
            # if transition_tool_output.witness is not None:
            # print(
            # "\nTransition tools output witness:\n"
            # f"{pformat(transition_tool_output.witness.model_dump_json())}"
            # )
            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`"
            )

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

        header = FixtureHeader(
            **(
                transition_tool_output.result.model_dump(
                    exclude_none=True, exclude={"blob_gas_used", "transactions_trie"}
                )
                | env.model_dump(exclude_none=True, exclude={"blob_gas_used"})
            ),
            blob_gas_used=blob_gas_used,
            transactions_trie=Transaction.list_root(txs),
            extra_data=block.extra_data if block.extra_data is not None else b"",
            fork=fork,
        )

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

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

        requests = None
        if fork.header_requests_required(header.number, header.timestamp):
            requests_list: List[DepositRequest | WithdrawalRequest | ConsolidationRequest] = []
            if transition_tool_output.result.deposit_requests is not None:
                requests_list += transition_tool_output.result.deposit_requests
            if transition_tool_output.result.withdrawal_requests is not None:
                requests_list += transition_tool_output.result.withdrawal_requests
            if transition_tool_output.result.consolidation_requests is not None:
                requests_list += transition_tool_output.result.consolidation_requests
            requests = Requests(root=requests_list)

        if requests is not None and requests.trie_root != header.requests_root:
            raise Exception(
                f"Requests root in header does not match the requests root in the transition tool "
                "output: "
                f"{header.requests_root} != {requests.trie_root}"
            )

        if block.requests is not None:
            requests = Requests(root=block.requests)
            header.requests_root = requests.trie_root

        if fork.fork_at(env.number, env.timestamp) == Verkle:
            env = Environment(
                **(
                    env.model_dump(exclude_none=True)
                    | transition_tool_output.result.model_dump(exclude_none=True)
                )
            )
            transition_tool_output.alloc = previous_alloc
            # TODO: hack for now, replace with actual witness output once available from t8n
            transition_tool_output.witness = Witness(
                verkle_proof=transition_tool_output.result.verkle_proof,
                state_diff=transition_tool_output.result.state_diff,
            )

        return (
            env,
            header,
            txs,
            transition_tool_output.alloc,
            requests,
            transition_tool_output.vkt,
            transition_tool_output.witness,
        )

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

    def verify_post_state(
        self,
        env: Environment,
        t8n: TransitionTool,
        alloc: Alloc,
        vkt: Optional[VerkleTree] = None,
    ):
        """
        Verifies the post state after all block/s or payload/s are generated.
        """
        try:
            if env.verkle_conversion_started or env.verkle_conversion_ended:
                if vkt is not None:
                    pass  # TODO: skip exact account verify checks
                    # verify_post_vkt(t8n=t8n, expected_post=self.post, got_vkt=vkt)
                else:
                    raise Exception("vkt conversion started but no vkt was created.")
            else:
                self.post.verify_post_alloc(got_alloc=alloc)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

    def verify_witness(
        self,
        t8n: TransitionTool,
        state_diff: StateDiff,
        witness_check: WitnessCheck,
    ) -> None:
        """
        Compares the expected witness check allocation account against the values updated
        in the block execution witness state diff.
        """
        witness_check_state_diff, witness_check_address_mapping = t8n.get_witness_check_mapping(
            witness_check
        )
        print("\nExpected witness check state diff:")
        print(witness_check_state_diff.model_dump_json(indent=4))

        for stem_state_diff in state_diff.root:
            actual_stem = stem_state_diff.stem
            address = witness_check_address_mapping.get(actual_stem, None)
            print(f"\nChecking witness for stem: {actual_stem} at address: {address}")
            # check for stem in the expected witness check
            expected_stem_state_diff = next(
                (sd for sd in witness_check_state_diff.root if sd.stem == actual_stem), None
            )
            if not expected_stem_state_diff:
                raise ValueError(
                    "Witness check failed - missing stem not found in expected witness check.\n\n"
                    + pformat(
                        {
                            "test_account_address": str(address),
                            "stem": str(actual_stem),
                        },
                        indent=4,
                    )
                )
            for suffix_diff in stem_state_diff.suffix_diffs:
                actual_suffix = suffix_diff.suffix
                actual_current_value = suffix_diff.current_value
                # check for suffix in the expected witness check
                expected_suffix_state_diff = next(
                    (
                        sd
                        for sd in expected_stem_state_diff.suffix_diffs
                        if sd.suffix == actual_suffix
                    ),
                    None,
                )
                if not expected_suffix_state_diff:
                    raise ValueError(
                        "Witness check failed - actual suffix not found in expected witness"
                        " check.\n\n"
                        + pformat(
                            {
                                "test_account_address": str(address),
                                "stem": str(actual_stem),
                                "suffix": actual_suffix,
                                "value_actual": str(actual_current_value),
                                "value_expected": "value not found",
                            },
                            indent=4,
                        )
                    )
                # check the current value of the actual suffix state diff matches the expected
                if str(actual_current_value) != str(
                    expected_suffix_state_diff.current_value
                ):  # TODO: temp fix str casting
                    raise ValueError(
                        "Witness check failed - current value mismatch. The stem and suffix"
                        " exist.\n\n"
                        + pformat(
                            {
                                "test_account_address": str(address),
                                "stem": str(actual_stem),
                                "suffix": actual_suffix,
                                "value_actual": str(actual_current_value),
                                "value_expected": str(expected_suffix_state_diff.current_value),
                            },
                            indent=4,
                        )
                    )

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

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

        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head = genesis.header.block_hash
        vkt: Optional[VerkleTree] = None

        # Filling for verkle genesis tests
        if fork is Verkle:
            env.verkle_conversion_ended = True
            # convert alloc to vkt
            vkt = t8n.from_mpt_to_vkt(alloc)

        # Hack for filling naive verkle transition tests
        if fork is EIP6800Transition:
            # Add a dummy block before the fork transition
            self.blocks.insert(0, Block(timestamp=HexNumber(fork.transition_at() - 1)))
            # Set timestamp for the next block to verkle transition time
            self.blocks[1].timestamp = HexNumber(fork.transition_at())
            # Increment all other block numbers
            for i, block in enumerate(self.blocks[1:]):
                block.number = HexNumber(i + 2)
            # Add a dummy block at the end of the test blocks
            self.blocks.append(Block())

        for block in self.blocks:
            if block.rlp is None:
                # This is the most common case, the RLP needs to be constructed
                # based on the transactions to be included in the block.
                # Set the environment according to the block to execute.
                new_env, header, txs, new_alloc, requests, new_vkt, witness = (
                    self.generate_block_data(
                        t8n=t8n,
                        fork=fork,
                        block=block,
                        previous_env=env,
                        previous_alloc=alloc,
                        previous_vkt=vkt,
                        eips=eips,
                    )
                )
                fixture_block = FixtureBlockBase(
                    header=header,
                    txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                    ommers=[],
                    withdrawals=(
                        [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                        if new_env.withdrawals is not None
                        else None
                    ),
                    deposit_requests=(
                        [
                            FixtureDepositRequest.from_deposit_request(d)
                            for d in requests.deposit_requests()
                        ]
                        if requests is not None
                        else None
                    ),
                    withdrawal_requests=(
                        [
                            FixtureWithdrawalRequest.from_withdrawal_request(w)
                            for w in requests.withdrawal_requests()
                        ]
                        if requests is not None
                        else None
                    ),
                    consolidation_requests=(
                        [
                            FixtureConsolidationRequest.from_consolidation_request(c)
                            for c in requests.consolidation_requests()
                        ]
                        if requests is not None
                        else None
                    ),
                    witness=FixtureWitness.from_witness(witness) if witness is not None else None,
                ).with_rlp(txs=txs, requests=requests)
                if block.exception is None:
                    fixture_blocks.append(fixture_block)
                    # Update env, alloc, vkt, and last block hash for the next block.
                    alloc = new_alloc
                    env = apply_new_parent(new_env, header)
                    head = header.block_hash
                    vkt = new_vkt
                else:
                    fixture_blocks.append(
                        InvalidFixtureBlock(
                            rlp=fixture_block.rlp,
                            expect_exception=block.exception,
                            rlp_decoded=(
                                None
                                if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                                else fixture_block.without_rlp()
                            ),
                        ),
                    )
            else:
                assert block.exception is not None, (
                    "test correctness: if the block's rlp is hard-coded, "
                    + "the block is expected to produce an exception"
                )
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=block.rlp,
                        expect_exception=block.exception,
                    ),
                )

        self.verify_post_state(env=env, t8n=t8n, alloc=alloc, vkt=vkt)
        return Fixture(
            fork=self.network_info(fork, eips),
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            # TODO: post_state=alloc
        )

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

        pre, genesis = self.make_genesis(fork, t8n)
        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head_hash = genesis.header.block_hash
        vkt: Optional[VerkleTree] = None

        # Hack for filling naive verkle transition tests
        if fork is EIP6800Transition:
            # Add a dummy block before the fork transition
            self.blocks.insert(0, Block(timestamp=HexNumber(fork.transition_at() - 1)))
            # Set timestamp for the next block to verkle transition time
            self.blocks[1].timestamp = HexNumber(fork.transition_at())
            # Increment all other block numbers
            for i, block in enumerate(self.blocks[1:]):
                block.number = HexNumber(i + 2)
            # Add a dummy block at the end of the test blocks
            self.blocks.append(Block())

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

        self.verify_post_state(env=env, t8n=t8n, alloc=alloc, vkt=vkt)

        sync_payload: Optional[FixtureEngineNewPayload] = None
        if self.verify_sync:
            # Test is marked for syncing verification.
            assert (
                genesis.header.block_hash != head_hash
            ), "Invalid payload tests negative test via sync is not supported yet."

            # Most clients require the header to start the sync process, so we create an empty
            # block on top of the last block of the test to send it as new payload and trigger the
            # sync process.
            _, sync_header, _, _, requests, _, _ = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=Block(),
                previous_env=env,
                previous_alloc=alloc,
                previous_vkt=vkt,
                eips=eips,
            )
            sync_payload = FixtureEngineNewPayload.from_fixture_header(
                fork=fork,
                header=sync_header,
                transactions=[],
                withdrawals=[],
                requests=requests,
                validation_error=None,
                error_code=None,
            )

        return EngineFixture(
            fork=self.network_info(fork, eips),
            genesis=genesis.header,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre=pre,
            # TODO: post_state=alloc
            sync_payload=sync_payload,
            last_block_hash=head_hash,
        )

    def generate(
        self,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        t8n.reset_traces()
        if fixture_format == BlockchainEngineFixture:
            return self.make_hive_fixture(t8n, fork, eips)
        elif fixture_format == BlockchainFixture:
            return self.make_fixture(t8n, fork, eips)

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

make_genesis(fork, t8n)

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

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

    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation_blockchain()),
        self.pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")

    state_root: bytes
    # TODO: refine, currently uses `evm verkle state-root` to get this.
    if fork < Verkle or fork is EIP6800Transition:
        state_root = pre_alloc.state_root()
    else:
        state_root = t8n.get_verkle_state_root(mpt_alloc=pre_alloc)

    genesis = FixtureHeader(
        parent_hash=0,
        ommers_hash=EmptyOmmersRoot,
        fee_recipient=0,
        state_root=state_root,
        transactions_trie=EmptyTrieRoot,
        receipts_root=EmptyTrieRoot,
        logs_bloom=0,
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=0,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data=b"\x00",
        prev_randao=0,
        nonce=0,
        base_fee_per_gas=env.base_fee_per_gas,
        blob_gas_used=env.blob_gas_used,
        excess_blob_gas=env.excess_blob_gas,
        withdrawals_root=(
            Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
        ),
        parent_beacon_block_root=env.parent_beacon_block_root,
        requests_root=(
            Requests(root=[]).trie_root if fork.header_requests_required(0, 0) else None
        ),
    )

    return (
        pre_alloc,
        FixtureBlockBase(
            header=genesis,
            withdrawals=None if env.withdrawals is None else [],
            deposit_requests=[] if fork.header_requests_required(0, 0) else None,
            withdrawal_requests=[] if fork.header_requests_required(0, 0) else None,
            consolidation_requests=[] if fork.header_requests_required(0, 0) else None,
        ).with_rlp(
            txs=[], requests=Requests() if fork.header_requests_required(0, 0) else None
        ),
    )

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

Generate common block data for both make_fixture and make_hive_fixture.

Source code in src/ethereum_test_specs/blockchain.py

def generate_block_data(
    self,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Alloc,
    previous_vkt: Optional[VerkleTree] = None,
    eips: Optional[List[int]] = None,
) -> Tuple[
    Environment,
    FixtureHeader,
    List[Transaction],
    Alloc,
    Optional[Requests],
    Optional[VerkleTree],
    Optional[Witness],
]:
    """
    Generate common block data for both make_fixture and make_hive_fixture.
    """
    if block.rlp and block.exception is not None:
        raise Exception(
            "test correctness: post-state cannot be verified if the "
            + "block's rlp is supplied and the block is not supposed "
            + "to produce an exception"
        )

    env = block.set_environment(previous_env)
    env = env.set_fork_requirements(fork)

    txs = [tx.with_signature_and_sender() for tx in block.txs]

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

    transition_tool_output = t8n.evaluate(
        alloc=previous_alloc,
        txs=txs,
        env=env,
        fork=fork,
        vkt=previous_vkt,
        chain_id=self.chain_id,
        reward=fork.get_reward(env.number, env.timestamp),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )

    try:  # General checks for the transition tool output
        rejected_txs = verify_transactions(txs, transition_tool_output.result)
        verify_result(transition_tool_output.result, env)
        if block.witness_check:
            if transition_tool_output.result.state_diff is None:
                raise Exception(
                    "no state diff in transition tool output, cannot verify witness"
                )
            # TODO: hack for now, temp addition to check hist. storage contract
            block.witness_check.add_storage_slot(
                address=Address(0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFE),
                storage_slot=env.number - 1,
                # value=env.parent_hash,
                value=None,
            )
            self.verify_witness(
                t8n=t8n,
                state_diff=transition_tool_output.result.state_diff,
                witness_check=block.witness_check,
            )
    except Exception as e:
        print_traces(t8n.get_traces())
        print(
            "\nTransition tool output result:\n"
            f"{pformat(transition_tool_output.result.model_dump_json())}"
        )
        print(
            "\nPrevious transition tool alloc:\n"
            f"{pformat(previous_alloc.model_dump_json())}"
        )
        if transition_tool_output.alloc is not None:
            print(
                "\nTransition tool output alloc:\n"
                f"{pformat(transition_tool_output.alloc.model_dump_json())}"
            )
        if transition_tool_output.vkt is not None:
            print(
                "\nTransition tools output verkle tree:\n"
                f"{pformat(transition_tool_output.vkt.model_dump_json())}"
            )
        # TODO: t8n has the witness state diff from the result for now
        # if transition_tool_output.witness is not None:
        # print(
        # "\nTransition tools output witness:\n"
        # f"{pformat(transition_tool_output.witness.model_dump_json())}"
        # )
        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`"
        )

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

    header = FixtureHeader(
        **(
            transition_tool_output.result.model_dump(
                exclude_none=True, exclude={"blob_gas_used", "transactions_trie"}
            )
            | env.model_dump(exclude_none=True, exclude={"blob_gas_used"})
        ),
        blob_gas_used=blob_gas_used,
        transactions_trie=Transaction.list_root(txs),
        extra_data=block.extra_data if block.extra_data is not None else b"",
        fork=fork,
    )

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

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

    requests = None
    if fork.header_requests_required(header.number, header.timestamp):
        requests_list: List[DepositRequest | WithdrawalRequest | ConsolidationRequest] = []
        if transition_tool_output.result.deposit_requests is not None:
            requests_list += transition_tool_output.result.deposit_requests
        if transition_tool_output.result.withdrawal_requests is not None:
            requests_list += transition_tool_output.result.withdrawal_requests
        if transition_tool_output.result.consolidation_requests is not None:
            requests_list += transition_tool_output.result.consolidation_requests
        requests = Requests(root=requests_list)

    if requests is not None and requests.trie_root != header.requests_root:
        raise Exception(
            f"Requests root in header does not match the requests root in the transition tool "
            "output: "
            f"{header.requests_root} != {requests.trie_root}"
        )

    if block.requests is not None:
        requests = Requests(root=block.requests)
        header.requests_root = requests.trie_root

    if fork.fork_at(env.number, env.timestamp) == Verkle:
        env = Environment(
            **(
                env.model_dump(exclude_none=True)
                | transition_tool_output.result.model_dump(exclude_none=True)
            )
        )
        transition_tool_output.alloc = previous_alloc
        # TODO: hack for now, replace with actual witness output once available from t8n
        transition_tool_output.witness = Witness(
            verkle_proof=transition_tool_output.result.verkle_proof,
            state_diff=transition_tool_output.result.state_diff,
        )

    return (
        env,
        header,
        txs,
        transition_tool_output.alloc,
        requests,
        transition_tool_output.vkt,
        transition_tool_output.witness,
    )

network_info(fork, eips=None)

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

Source code in src/ethereum_test_specs/blockchain.py

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

verify_post_state(env, t8n, alloc, vkt=None)

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

Source code in src/ethereum_test_specs/blockchain.py

def verify_post_state(
    self,
    env: Environment,
    t8n: TransitionTool,
    alloc: Alloc,
    vkt: Optional[VerkleTree] = None,
):
    """
    Verifies the post state after all block/s or payload/s are generated.
    """
    try:
        if env.verkle_conversion_started or env.verkle_conversion_ended:
            if vkt is not None:
                pass  # TODO: skip exact account verify checks
                # verify_post_vkt(t8n=t8n, expected_post=self.post, got_vkt=vkt)
            else:
                raise Exception("vkt conversion started but no vkt was created.")
        else:
            self.post.verify_post_alloc(got_alloc=alloc)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

verify_witness(t8n, state_diff, witness_check)

Compares the expected witness check allocation account against the values updated in the block execution witness state diff.

Source code in src/ethereum_test_specs/blockchain.py

def verify_witness(
    self,
    t8n: TransitionTool,
    state_diff: StateDiff,
    witness_check: WitnessCheck,
) -> None:
    """
    Compares the expected witness check allocation account against the values updated
    in the block execution witness state diff.
    """
    witness_check_state_diff, witness_check_address_mapping = t8n.get_witness_check_mapping(
        witness_check
    )
    print("\nExpected witness check state diff:")
    print(witness_check_state_diff.model_dump_json(indent=4))

    for stem_state_diff in state_diff.root:
        actual_stem = stem_state_diff.stem
        address = witness_check_address_mapping.get(actual_stem, None)
        print(f"\nChecking witness for stem: {actual_stem} at address: {address}")
        # check for stem in the expected witness check
        expected_stem_state_diff = next(
            (sd for sd in witness_check_state_diff.root if sd.stem == actual_stem), None
        )
        if not expected_stem_state_diff:
            raise ValueError(
                "Witness check failed - missing stem not found in expected witness check.\n\n"
                + pformat(
                    {
                        "test_account_address": str(address),
                        "stem": str(actual_stem),
                    },
                    indent=4,
                )
            )
        for suffix_diff in stem_state_diff.suffix_diffs:
            actual_suffix = suffix_diff.suffix
            actual_current_value = suffix_diff.current_value
            # check for suffix in the expected witness check
            expected_suffix_state_diff = next(
                (
                    sd
                    for sd in expected_stem_state_diff.suffix_diffs
                    if sd.suffix == actual_suffix
                ),
                None,
            )
            if not expected_suffix_state_diff:
                raise ValueError(
                    "Witness check failed - actual suffix not found in expected witness"
                    " check.\n\n"
                    + pformat(
                        {
                            "test_account_address": str(address),
                            "stem": str(actual_stem),
                            "suffix": actual_suffix,
                            "value_actual": str(actual_current_value),
                            "value_expected": "value not found",
                        },
                        indent=4,
                    )
                )
            # check the current value of the actual suffix state diff matches the expected
            if str(actual_current_value) != str(
                expected_suffix_state_diff.current_value
            ):  # TODO: temp fix str casting
                raise ValueError(
                    "Witness check failed - current value mismatch. The stem and suffix"
                    " exist.\n\n"
                    + pformat(
                        {
                            "test_account_address": str(address),
                            "stem": str(actual_stem),
                            "suffix": actual_suffix,
                            "value_actual": str(actual_current_value),
                            "value_expected": str(expected_suffix_state_diff.current_value),
                        },
                        indent=4,
                    )
                )

make_fixture(t8n, fork, eips=None)

Create a fixture from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

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

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

    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head = genesis.header.block_hash
    vkt: Optional[VerkleTree] = None

    # Filling for verkle genesis tests
    if fork is Verkle:
        env.verkle_conversion_ended = True
        # convert alloc to vkt
        vkt = t8n.from_mpt_to_vkt(alloc)

    # Hack for filling naive verkle transition tests
    if fork is EIP6800Transition:
        # Add a dummy block before the fork transition
        self.blocks.insert(0, Block(timestamp=HexNumber(fork.transition_at() - 1)))
        # Set timestamp for the next block to verkle transition time
        self.blocks[1].timestamp = HexNumber(fork.transition_at())
        # Increment all other block numbers
        for i, block in enumerate(self.blocks[1:]):
            block.number = HexNumber(i + 2)
        # Add a dummy block at the end of the test blocks
        self.blocks.append(Block())

    for block in self.blocks:
        if block.rlp is None:
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            new_env, header, txs, new_alloc, requests, new_vkt, witness = (
                self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    previous_vkt=vkt,
                    eips=eips,
                )
            )
            fixture_block = FixtureBlockBase(
                header=header,
                txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                ommers=[],
                withdrawals=(
                    [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                    if new_env.withdrawals is not None
                    else None
                ),
                deposit_requests=(
                    [
                        FixtureDepositRequest.from_deposit_request(d)
                        for d in requests.deposit_requests()
                    ]
                    if requests is not None
                    else None
                ),
                withdrawal_requests=(
                    [
                        FixtureWithdrawalRequest.from_withdrawal_request(w)
                        for w in requests.withdrawal_requests()
                    ]
                    if requests is not None
                    else None
                ),
                consolidation_requests=(
                    [
                        FixtureConsolidationRequest.from_consolidation_request(c)
                        for c in requests.consolidation_requests()
                    ]
                    if requests is not None
                    else None
                ),
                witness=FixtureWitness.from_witness(witness) if witness is not None else None,
            ).with_rlp(txs=txs, requests=requests)
            if block.exception is None:
                fixture_blocks.append(fixture_block)
                # Update env, alloc, vkt, and last block hash for the next block.
                alloc = new_alloc
                env = apply_new_parent(new_env, header)
                head = header.block_hash
                vkt = new_vkt
            else:
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=fixture_block.rlp,
                        expect_exception=block.exception,
                        rlp_decoded=(
                            None
                            if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                            else fixture_block.without_rlp()
                        ),
                    ),
                )
        else:
            assert block.exception is not None, (
                "test correctness: if the block's rlp is hard-coded, "
                + "the block is expected to produce an exception"
            )
            fixture_blocks.append(
                InvalidFixtureBlock(
                    rlp=block.rlp,
                    expect_exception=block.exception,
                ),
            )

    self.verify_post_state(env=env, t8n=t8n, alloc=alloc, vkt=vkt)
    return Fixture(
        fork=self.network_info(fork, eips),
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        # TODO: post_state=alloc
    )

make_hive_fixture(t8n, fork, eips=None)

Create a hive fixture from the blocktest definition.

Source code in src/ethereum_test_specs/blockchain.py

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

    pre, genesis = self.make_genesis(fork, t8n)
    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head_hash = genesis.header.block_hash
    vkt: Optional[VerkleTree] = None

    # Hack for filling naive verkle transition tests
    if fork is EIP6800Transition:
        # Add a dummy block before the fork transition
        self.blocks.insert(0, Block(timestamp=HexNumber(fork.transition_at() - 1)))
        # Set timestamp for the next block to verkle transition time
        self.blocks[1].timestamp = HexNumber(fork.transition_at())
        # Increment all other block numbers
        for i, block in enumerate(self.blocks[1:]):
            block.number = HexNumber(i + 2)
        # Add a dummy block at the end of the test blocks
        self.blocks.append(Block())

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

    self.verify_post_state(env=env, t8n=t8n, alloc=alloc, vkt=vkt)

    sync_payload: Optional[FixtureEngineNewPayload] = None
    if self.verify_sync:
        # Test is marked for syncing verification.
        assert (
            genesis.header.block_hash != head_hash
        ), "Invalid payload tests negative test via sync is not supported yet."

        # Most clients require the header to start the sync process, so we create an empty
        # block on top of the last block of the test to send it as new payload and trigger the
        # sync process.
        _, sync_header, _, _, requests, _, _ = self.generate_block_data(
            t8n=t8n,
            fork=fork,
            block=Block(),
            previous_env=env,
            previous_alloc=alloc,
            previous_vkt=vkt,
            eips=eips,
        )
        sync_payload = FixtureEngineNewPayload.from_fixture_header(
            fork=fork,
            header=sync_header,
            transactions=[],
            withdrawals=[],
            requests=requests,
            validation_error=None,
            error_code=None,
        )

    return EngineFixture(
        fork=self.network_info(fork, eips),
        genesis=genesis.header,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre=pre,
        # TODO: post_state=alloc
        sync_payload=sync_payload,
        last_block_hash=head_hash,
    )

generate(request, t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/blockchain.py

def generate(
    self,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    t8n.reset_traces()
    if fixture_format == BlockchainEngineFixture:
        return self.make_hive_fixture(t8n, fork, eips)
    elif fixture_format == BlockchainFixture:
        return self.make_fixture(t8n, fork, eips)

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

EOFStateTest

Bases: EOFTest

Filler type that tests EOF containers and also generates a state/blockchain test.

Source code in src/ethereum_test_specs/eof.py

class EOFStateTest(EOFTest):
    """
    Filler type that tests EOF containers and also generates a state/blockchain test.
    """

    deploy_tx: bool = False
    tx_gas_limit: int = 10_000_000
    tx_data: Bytes = Bytes(b"")
    tx_sender_funding_amount: int = 1_000_000_000_000_000_000_000
    env: Environment = Field(default_factory=Environment)
    container_post: Account = Field(default_factory=Account)
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        EOFFixture,
        StateFixture,
        BlockchainFixture,
        BlockchainEngineFixture,
    ]

    @model_validator(mode="before")
    @classmethod
    def check_container_type(cls, data: Any) -> Any:
        """
        Check if the container exception matches the expected exception.
        """
        if isinstance(data, dict):
            container = data.get("data")
            deploy_tx = data.get("deploy_tx")
            container_kind = data.get("container_kind")
            if deploy_tx is None:
                if (
                    container is not None
                    and isinstance(container, Container)
                    and "kind" in container.model_fields_set
                    and container.kind == ContainerKind.INITCODE
                ) or (container_kind is not None and container_kind == ContainerKind.INITCODE):
                    data["deploy_tx"] = True
        return data

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Workaround for pytest parameter name.
        """
        return "eof_state_test"

    def generate_state_test(self) -> StateTest:
        """
        Generate the StateTest filler.
        """
        assert self.pre is not None, "pre must be set to generate a StateTest."
        tx = Transaction(
            sender=self.pre.fund_eoa(amount=self.tx_sender_funding_amount),
            gas_limit=self.tx_gas_limit,
        )
        post = Alloc()
        if self.expect_exception is not None:  # Invalid EOF
            tx.to = None  # Make EIP-7698 create transaction
            tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
            post[tx.created_contract] = None  # Expect failure.
        elif self.deploy_tx:
            tx.to = None  # Make EIP-7698 create transaction
            tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
            post[tx.created_contract] = self.container_post  # Successful.
        else:
            tx.to = self.pre.deploy_contract(code=self.data)
            tx.data = self.tx_data
            post[tx.to] = self.container_post
        return StateTest(
            pre=self.pre,
            tx=tx,
            env=self.env,
            post=post,
            t8n_dump_dir=self.t8n_dump_dir,
        )

    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        if fixture_format == EOFFixture:
            if self.data in existing_tests:
                # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
                # state fixtures with the same data.
                pytest.skip(f"Duplicate EOF container on EOFStateTest: {request.node.nodeid}")
            return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
        elif fixture_format in (
            StateFixture,
            BlockchainFixture,
            BlockchainEngineFixture,
        ):
            return self.generate_state_test().generate(
                request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )

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

check_container_type(data) classmethod

Check if the container exception matches the expected exception.

Source code in src/ethereum_test_specs/eof.py

@model_validator(mode="before")
@classmethod
def check_container_type(cls, data: Any) -> Any:
    """
    Check if the container exception matches the expected exception.
    """
    if isinstance(data, dict):
        container = data.get("data")
        deploy_tx = data.get("deploy_tx")
        container_kind = data.get("container_kind")
        if deploy_tx is None:
            if (
                container is not None
                and isinstance(container, Container)
                and "kind" in container.model_fields_set
                and container.kind == ContainerKind.INITCODE
            ) or (container_kind is not None and container_kind == ContainerKind.INITCODE):
                data["deploy_tx"] = True
    return data

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Workaround for pytest parameter name.
    """
    return "eof_state_test"

generate_state_test()

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self) -> StateTest:
    """
    Generate the StateTest filler.
    """
    assert self.pre is not None, "pre must be set to generate a StateTest."
    tx = Transaction(
        sender=self.pre.fund_eoa(amount=self.tx_sender_funding_amount),
        gas_limit=self.tx_gas_limit,
    )
    post = Alloc()
    if self.expect_exception is not None:  # Invalid EOF
        tx.to = None  # Make EIP-7698 create transaction
        tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
        post[tx.created_contract] = None  # Expect failure.
    elif self.deploy_tx:
        tx.to = None  # Make EIP-7698 create transaction
        tx.data = Bytes(self.data + self.tx_data)  # by concatenating container and tx data.
        post[tx.created_contract] = self.container_post  # Successful.
    else:
        tx.to = self.pre.deploy_contract(code=self.data)
        tx.data = self.tx_data
        post[tx.to] = self.container_post
    return StateTest(
        pre=self.pre,
        tx=tx,
        env=self.env,
        post=post,
        t8n_dump_dir=self.t8n_dump_dir,
    )

generate(*, request, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py

def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    if fixture_format == EOFFixture:
        if self.data in existing_tests:
            # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
            # state fixtures with the same data.
            pytest.skip(f"Duplicate EOF container on EOFStateTest: {request.node.nodeid}")
        return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
    elif fixture_format in (
        StateFixture,
        BlockchainFixture,
        BlockchainEngineFixture,
    ):
        return self.generate_state_test().generate(
            request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )

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

EOFTest

Bases: BaseTest

Filler type that tests EOF containers.

Source code in src/ethereum_test_specs/eof.py

class EOFTest(BaseTest):
    """
    Filler type that tests EOF containers.
    """

    data: Bytes
    expect_exception: EOFExceptionInstanceOrList | None = None
    container_kind: ContainerKind | None = None

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        EOFFixture,
    ]

    @model_validator(mode="before")
    @classmethod
    def check_container_exception(cls, data: Any) -> Any:
        """
        Check if the container exception matches the expected exception.
        """
        if isinstance(data, dict):
            container = data.get("data")
            expect_exception = data.get("expect_exception")
            container_kind = data.get("container_kind")
            if container is not None and isinstance(container, Container):
                if (
                    "validity_error" in container.model_fields_set
                    and container.validity_error is not None
                ):
                    if expect_exception is not None:
                        assert container.validity_error == expect_exception, (
                            f"Container validity error {container.validity_error} "
                            f"does not match expected exception {expect_exception}."
                        )
                    if expect_exception is None:
                        data["expect_exception"] = container.validity_error
                if "kind" in container.model_fields_set:
                    if container_kind is not None:
                        assert container.kind == container_kind, (
                            f"Container kind type {str(container.kind)} "
                            f"does not match test {container_kind}."
                        )
                    if container.kind != ContainerKind.RUNTIME:
                        data["container_kind"] = container.kind
        return data

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Workaround for pytest parameter name.
        """
        return "eof_test"

    def make_eof_test_fixture(
        self,
        *,
        request: pytest.FixtureRequest,
        fork: Fork,
        eips: Optional[List[int]],
    ) -> EOFFixture:
        """
        Generate the EOF test fixture.
        """
        if self.data in existing_tests:
            pytest.fail(
                f"Duplicate EOF test: {self.data}, existing test: {existing_tests[self.data]}"
            )
        existing_tests[self.data] = request.node.nodeid
        vectors = [
            Vector(
                code=self.data,
                container_kind=self.container_kind,
                results={
                    fork.blockchain_test_network_name(): Result(
                        exception=self.expect_exception,
                        valid=self.expect_exception is None,
                    ),
                },
            )
        ]
        fixture = EOFFixture(vectors=dict(enumerate(vectors)))
        try:
            eof_parse = EOFParse()
        except FileNotFoundError as e:
            warnings.warn(f"{e} Skipping EOF fixture verification. Fixtures may be invalid!")
            return fixture

        for _, vector in fixture.vectors.items():
            expected_result = vector.results.get(fork.blockchain_test_network_name())
            if expected_result is None:
                raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
            args = []
            if vector.container_kind == ContainerKind.INITCODE:
                args.append("--initcode")
            result = eof_parse.run(*args, input=str(vector.code))
            self.verify_result(result, expected_result, vector.code)

        return fixture

    def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
        """
        Checks that the reported exception string matches the expected error.
        """
        parser = EvmoneExceptionMapper()
        actual_message = result.stdout.strip()
        actual_exception = parser.message_to_exception(actual_message)

        if expected_result.exception is None:
            if "OK" in actual_message:
                return
            else:
                raise UnexpectedEOFException(
                    code=code, got=f"{actual_exception} ({actual_message})"
                )
        else:
            expected_string = to_pipe_str(expected_result.exception)
            print(expected_string)
            print(actual_exception)
            if "OK" in actual_message:
                raise ExpectedEOFException(
                    code=code,
                    expected=f"{expected_string}",
                )
            elif actual_exception in expected_result.exception:
                return
            else:
                raise EOFExceptionMismatch(
                    code=code,
                    expected=f"{expected_string}",
                    got=f"{actual_exception} ({actual_message})",
                )

    def generate(
        self,
        *,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        if fixture_format == EOFFixture:
            return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)

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

check_container_exception(data) classmethod

Check if the container exception matches the expected exception.

Source code in src/ethereum_test_specs/eof.py

@model_validator(mode="before")
@classmethod
def check_container_exception(cls, data: Any) -> Any:
    """
    Check if the container exception matches the expected exception.
    """
    if isinstance(data, dict):
        container = data.get("data")
        expect_exception = data.get("expect_exception")
        container_kind = data.get("container_kind")
        if container is not None and isinstance(container, Container):
            if (
                "validity_error" in container.model_fields_set
                and container.validity_error is not None
            ):
                if expect_exception is not None:
                    assert container.validity_error == expect_exception, (
                        f"Container validity error {container.validity_error} "
                        f"does not match expected exception {expect_exception}."
                    )
                if expect_exception is None:
                    data["expect_exception"] = container.validity_error
            if "kind" in container.model_fields_set:
                if container_kind is not None:
                    assert container.kind == container_kind, (
                        f"Container kind type {str(container.kind)} "
                        f"does not match test {container_kind}."
                    )
                if container.kind != ContainerKind.RUNTIME:
                    data["container_kind"] = container.kind
    return data

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py

@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Workaround for pytest parameter name.
    """
    return "eof_test"

make_eof_test_fixture(*, request, fork, eips)

Generate the EOF test fixture.

Source code in src/ethereum_test_specs/eof.py

def make_eof_test_fixture(
    self,
    *,
    request: pytest.FixtureRequest,
    fork: Fork,
    eips: Optional[List[int]],
) -> EOFFixture:
    """
    Generate the EOF test fixture.
    """
    if self.data in existing_tests:
        pytest.fail(
            f"Duplicate EOF test: {self.data}, existing test: {existing_tests[self.data]}"
        )
    existing_tests[self.data] = request.node.nodeid
    vectors = [
        Vector(
            code=self.data,
            container_kind=self.container_kind,
            results={
                fork.blockchain_test_network_name(): Result(
                    exception=self.expect_exception,
                    valid=self.expect_exception is None,
                ),
            },
        )
    ]
    fixture = EOFFixture(vectors=dict(enumerate(vectors)))
    try:
        eof_parse = EOFParse()
    except FileNotFoundError as e:
        warnings.warn(f"{e} Skipping EOF fixture verification. Fixtures may be invalid!")
        return fixture

    for _, vector in fixture.vectors.items():
        expected_result = vector.results.get(fork.blockchain_test_network_name())
        if expected_result is None:
            raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
        args = []
        if vector.container_kind == ContainerKind.INITCODE:
            args.append("--initcode")
        result = eof_parse.run(*args, input=str(vector.code))
        self.verify_result(result, expected_result, vector.code)

    return fixture

verify_result(result, expected_result, code)

Checks that the reported exception string matches the expected error.

Source code in src/ethereum_test_specs/eof.py

def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
    """
    Checks that the reported exception string matches the expected error.
    """
    parser = EvmoneExceptionMapper()
    actual_message = result.stdout.strip()
    actual_exception = parser.message_to_exception(actual_message)

    if expected_result.exception is None:
        if "OK" in actual_message:
            return
        else:
            raise UnexpectedEOFException(
                code=code, got=f"{actual_exception} ({actual_message})"
            )
    else:
        expected_string = to_pipe_str(expected_result.exception)
        print(expected_string)
        print(actual_exception)
        if "OK" in actual_message:
            raise ExpectedEOFException(
                code=code,
                expected=f"{expected_string}",
            )
        elif actual_exception in expected_result.exception:
            return
        else:
            raise EOFExceptionMismatch(
                code=code,
                expected=f"{expected_string}",
                got=f"{actual_exception} ({actual_message})",
            )

generate(*, request, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py

def generate(
    self,
    *,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    if fixture_format == EOFFixture:
        return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)

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

StateTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/state.py

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

    env: Environment
    pre: Alloc
    post: Alloc
    tx: Transaction
    engine_api_error_code: Optional[EngineAPIError] = None
    blockchain_test_header_verify: Optional[Header] = None
    blockchain_test_rlp_modifier: Optional[Header] = None
    chain_id: int = 1

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
        StateFixture,
    ]

    def _generate_blockchain_genesis_environment(self) -> Environment:
        """
        Generate the genesis environment for the BlockchainTest formatted test.
        """
        assert (
            self.env.number >= 1
        ), "genesis block number cannot be negative, set state test env.number to 1"

        # Modify values to the proper values for the genesis block
        # TODO: All of this can be moved to a new method in `Fork`
        updated_values: Dict[str, Any] = {
            "withdrawals": None,
            "parent_beacon_block_root": None,
            "number": self.env.number - 1,
        }
        if self.env.excess_blob_gas:
            # The excess blob gas environment value means the value of the context (block header)
            # where the transaction is executed. In a blockchain test, we need to indirectly
            # set the excess blob gas by setting the excess blob gas of the genesis block
            # to the expected value plus the TARGET_BLOB_GAS_PER_BLOCK, which is the value
            # that will be subtracted from the excess blob gas when the first block is mined.
            updated_values["excess_blob_gas"] = (
                self.env.excess_blob_gas + TARGET_BLOB_GAS_PER_BLOCK
            )

        return self.env.copy(**updated_values)

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

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

    def make_state_test_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> Fixture:
        """
        Create a fixture from the state test definition.
        """
        # We can't generate a state test fixture that names a transition fork,
        # so we get the fork at the block number and timestamp of the state test
        fork = fork.fork_at(self.env.number, self.env.timestamp)

        env = self.env.set_fork_requirements(fork)
        tx = self.tx.with_signature_and_sender(keep_secret_key=True)
        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation()),
            self.pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")

        transition_tool_output = t8n.evaluate(
            alloc=pre_alloc,
            txs=[tx],
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
        )

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

        return Fixture(
            env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
            pre=pre_alloc,
            post={
                fork.blockchain_test_network_name(): [
                    FixtureForkPost(
                        state_root=transition_tool_output.result.state_root,
                        logs_hash=transition_tool_output.result.logs_hash,
                        tx_bytes=tx.rlp,
                        expect_exception=tx.error,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
        )

    def generate(
        self,
        request: pytest.FixtureRequest,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        if fixture_format in BlockchainTest.supported_fixture_formats:
            return self.generate_blockchain_test().generate(
                request=request,
                t8n=t8n,
                fork=fork,
                fixture_format=fixture_format,
                eips=eips,
            )
        elif fixture_format == StateFixture:
            if fork is not EIP6800Transition:
                return self.make_state_test_fixture(t8n, fork, eips)
            else:
                pytest.skip("State tests are not supported for EIP-6800 transition.")

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

generate_blockchain_test()

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in src/ethereum_test_specs/state.py

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

make_state_test_fixture(t8n, fork, eips=None)

Create a fixture from the state test definition.

Source code in src/ethereum_test_specs/state.py

def make_state_test_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> Fixture:
    """
    Create a fixture from the state test definition.
    """
    # We can't generate a state test fixture that names a transition fork,
    # so we get the fork at the block number and timestamp of the state test
    fork = fork.fork_at(self.env.number, self.env.timestamp)

    env = self.env.set_fork_requirements(fork)
    tx = self.tx.with_signature_and_sender(keep_secret_key=True)
    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation()),
        self.pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")

    transition_tool_output = t8n.evaluate(
        alloc=pre_alloc,
        txs=[tx],
        env=env,
        fork=fork,
        chain_id=self.chain_id,
        reward=0,  # Reward on state tests is always zero
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
    )

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

    return Fixture(
        env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
        pre=pre_alloc,
        post={
            fork.blockchain_test_network_name(): [
                FixtureForkPost(
                    state_root=transition_tool_output.result.state_root,
                    logs_hash=transition_tool_output.result.logs_hash,
                    tx_bytes=tx.rlp,
                    expect_exception=tx.error,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
    )

generate(request, t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/state.py

def generate(
    self,
    request: pytest.FixtureRequest,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    if fixture_format in BlockchainTest.supported_fixture_formats:
        return self.generate_blockchain_test().generate(
            request=request,
            t8n=t8n,
            fork=fork,
            fixture_format=fixture_format,
            eips=eips,
        )
    elif fixture_format == StateFixture:
        if fork is not EIP6800Transition:
            return self.make_state_test_fixture(t8n, fork, eips)
        else:
            pytest.skip("State tests are not supported for EIP-6800 transition.")

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

StateTestOnly

Bases: StateTest

StateTest filler that only generates a state test fixture.

Source code in src/ethereum_test_specs/state.py

class StateTestOnly(StateTest):
    """
    StateTest filler that only generates a state test fixture.
    """

    supported_fixture_formats: ClassVar[List[FixtureFormat]] = [StateFixture]