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

    @abstractmethod
    def generate(
        self,
        *,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormats,
        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(*, 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,
    *,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormats,
    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[FixtureFormats]] = [
        FixtureFormats.BLOCKCHAIN_TEST,
        FixtureFormats.BLOCKCHAIN_TEST_ENGINE,
    ]

    def make_genesis(
        self,
        fork: Fork,
    ) -> 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 = pre_alloc.state_root()
        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,
        eips: Optional[List[int]] = None,
    ) -> Tuple[FixtureHeader, List[Transaction], Requests | None, Alloc, Environment]:
        """
        Generate common block data for both make_fixture and make_hive_fixture.
        """
        if block.rlp and block.exception is not None:
            raise Exception(
                "test correctness: post-state cannot be verified if the "
                + "block's rlp is supplied and the block is not supposed "
                + "to produce an exception"
            )

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

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

        if 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,
            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:
            rejected_txs = verify_transactions(txs, transition_tool_output.result)
            verify_result(transition_tool_output.result, env)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(previous_alloc)
            pprint(transition_tool_output.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`"
            )

        # 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

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

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

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

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

        pre, genesis = self.make_genesis(fork)

        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head = genesis.header.block_hash

        for block in self.blocks:
            if block.rlp is None:
                # This is the most common case, the RLP needs to be constructed
                # based on the transactions to be included in the block.
                # Set the environment according to the block to execute.
                header, txs, requests, new_alloc, new_env = self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    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,
                ).with_rlp(txs=txs, requests=requests)
                if block.exception is None:
                    fixture_blocks.append(fixture_block)
                    # Update env, alloc and last block hash for the next block.
                    alloc = new_alloc
                    env = apply_new_parent(new_env, header)
                    head = header.block_hash
                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(t8n, alloc)
        return Fixture(
            fork=self.network_info(fork, eips),
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            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)
        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head_hash = genesis.header.block_hash

        for block in self.blocks:
            header, txs, requests, new_alloc, new_env = self.generate_block_data(
                t8n=t8n, fork=fork, block=block, previous_env=env, previous_alloc=alloc, eips=eips
            )
            if block.rlp is None:
                fixture_payloads.append(
                    FixtureEngineNewPayload.from_fixture_header(
                        fork=fork,
                        header=header,
                        transactions=txs,
                        withdrawals=new_env.withdrawals,
                        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(env, header)
                    head_hash = header.block_hash
        fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
        assert (
            fcu_version is not None
        ), "A hive fixture was requested but no forkchoice update is defined. The framework should"
        " never try to execute this test case."

        self.verify_post_state(t8n, alloc)

        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,
                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,
            post_state=alloc,
            sync_payload=sync_payload,
            last_block_hash=head_hash,
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormats,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """
        Generate the BlockchainTest fixture.
        """
        t8n.reset_traces()
        if fixture_format == FixtureFormats.BLOCKCHAIN_TEST_ENGINE:
            return self.make_hive_fixture(t8n, fork, eips)
        elif fixture_format == FixtureFormats.BLOCKCHAIN_TEST:
            return self.make_fixture(t8n, fork, eips)

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

`make_genesis(fork)` ¶

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

def make_genesis(
    self,
    fork: Fork,
) -> 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 = pre_alloc.state_root()
    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, 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,
    eips: Optional[List[int]] = None,
) -> Tuple[FixtureHeader, List[Transaction], Requests | None, Alloc, Environment]:
    """
    Generate common block data for both make_fixture and make_hive_fixture.
    """
    if block.rlp and block.exception is not None:
        raise Exception(
            "test correctness: post-state cannot be verified if the "
            + "block's rlp is supplied and the block is not supposed "
            + "to produce an exception"
        )

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

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

    if 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,
        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:
        rejected_txs = verify_transactions(txs, transition_tool_output.result)
        verify_result(transition_tool_output.result, env)
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(previous_alloc)
        pprint(transition_tool_output.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`"
        )

    # 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

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

`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(t8n, alloc)` ¶

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

Source code in src/ethereum_test_specs/blockchain.py

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

`make_fixture(t8n, fork, eips=None)` ¶

Create a fixture from the blockchain test definition.

Source code in src/ethereum_test_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)

    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head = genesis.header.block_hash

    for block in self.blocks:
        if block.rlp is None:
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            header, txs, requests, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                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,
            ).with_rlp(txs=txs, requests=requests)
            if block.exception is None:
                fixture_blocks.append(fixture_block)
                # Update env, alloc and last block hash for the next block.
                alloc = new_alloc
                env = apply_new_parent(new_env, header)
                head = header.block_hash
            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(t8n, alloc)
    return Fixture(
        fork=self.network_info(fork, eips),
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        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)
    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head_hash = genesis.header.block_hash

    for block in self.blocks:
        header, txs, requests, new_alloc, new_env = self.generate_block_data(
            t8n=t8n, fork=fork, block=block, previous_env=env, previous_alloc=alloc, eips=eips
        )
        if block.rlp is None:
            fixture_payloads.append(
                FixtureEngineNewPayload.from_fixture_header(
                    fork=fork,
                    header=header,
                    transactions=txs,
                    withdrawals=new_env.withdrawals,
                    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(env, header)
                head_hash = header.block_hash
    fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
    assert (
        fcu_version is not None
    ), "A hive fixture was requested but no forkchoice update is defined. The framework should"
    " never try to execute this test case."

    self.verify_post_state(t8n, alloc)

    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,
            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,
        post_state=alloc,
        sync_payload=sync_payload,
        last_block_hash=head_hash,
    )

`generate(t8n, fork, fixture_format, eips=None)` ¶

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/blockchain.py

def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormats,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """
    Generate the BlockchainTest fixture.
    """
    t8n.reset_traces()
    if fixture_format == FixtureFormats.BLOCKCHAIN_TEST_ENGINE:
        return self.make_hive_fixture(t8n, fork, eips)
    elif fixture_format == FixtureFormats.BLOCKCHAIN_TEST:
        return self.make_fixture(t8n, fork, eips)

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