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]] = []
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = []

    @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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """
        Generate the list of test fixtures.
        """
        raise Exception(f"Unsupported execute format: {execute_format}")

    @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

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """
    Generate the list of test fixtures.
    """
    raise Exception(f"Unsupported execute format: {execute_format}")

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,
    ]
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = [
        TransactionPost,
    ]

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

        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(
                t8n.exception_mapper, 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)
            header.fork = fork  # Deleted during `apply` because `exclude=True`

        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
                    ),
                    fork=fork,
                ).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,
        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}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """
        Generate the list of test fixtures.
        """
        if execute_format == TransactionPost:
            txs: List[Transaction] = []
            for block in self.blocks:
                txs += block.txs
            return TransactionPost(
                transactions=txs,
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_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
        ),
        fork=fork,
    )

    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(
            t8n.exception_mapper, 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)
        header.fork = fork  # Deleted during `apply` because `exclude=True`

    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
                ),
                fork=fork,
            ).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(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}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/blockchain.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """
    Generate the list of test fixtures.
    """
    if execute_format == TransactionPost:
        txs: List[Transaction] = []
        for block in self.blocks:
            txs += block.txs
        return TransactionPost(
            transactions=txs,
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

BlockchainTestEngine

Bases: BlockchainTest

Filler type that tests multiple blocks (valid or invalid) in a chain, only for the Engine API.

Source code in src/ethereum_test_specs/blockchain.py

class BlockchainTestEngine(BlockchainTest):
    """
    Filler type that tests multiple blocks (valid or invalid) in a chain, only for the Engine API.
    """

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

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """
        Generate the list of test fixtures.
        """
        if execute_format == TransactionPost:
            return self.generate_state_test().execute(
                fork=fork, execute_format=execute_format, eips=eips
            )
        raise Exception(f"Unsupported execute format: {execute_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}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/eof.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """
    Generate the list of test fixtures.
    """
    if execute_format == TransactionPost:
        return self.generate_state_test().execute(
            fork=fork, execute_format=execute_format, eips=eips
        )
    raise Exception(f"Unsupported execute format: {execute_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,
    ]
    supported_execute_formats: ClassVar[List[ExecuteFormat]] = [
        TransactionPost,
    ]

    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

        try:
            verify_transactions(t8n.exception_mapper, [tx], transition_tool_output.result)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(transition_tool_output.alloc)
            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:
            return self.make_state_test_fixture(t8n, fork, eips)

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """
        Generate the list of test fixtures.
        """
        if execute_format == TransactionPost:
            return TransactionPost(
                transactions=[self.tx],
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_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

    try:
        verify_transactions(t8n.exception_mapper, [tx], transition_tool_output.result)
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(transition_tool_output.alloc)
        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:
        return self.make_state_test_fixture(t8n, fork, eips)

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

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/state.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """
    Generate the list of test fixtures.
    """
    if execute_format == TransactionPost:
        return TransactionPost(
            transactions=[self.tx],
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_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]