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[Sequence[FixtureFormat | LabeledFixtureFormat]] = []
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = []

    supported_markers: ClassVar[Dict[str, str]] = {}

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        return False

    @classmethod
    def discard_execute_format_by_marks(
        cls,
        execute_format: ExecuteFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard an execute format from executing if the appropriate marker is used."""
        return False

    @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:
        """Return 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)),
        )

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/base.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    return False

discard_execute_format_by_marks(execute_format, fork, markers) classmethod

Discard an execute format from executing if the appropriate marker is used.

Source code in src/ethereum_test_specs/base.py

@classmethod
def discard_execute_format_by_marks(
    cls,
    execute_format: ExecuteFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard an execute format from executing if the appropriate marker is used."""
    return False

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()

Return 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:
    """Return 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[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "blockchain_test_engine_only": "Only generate a blockchain test engine fixture",
        "blockchain_test_only": "Only generate a blockchain test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "blockchain_test_only" in [m.name for m in markers]:
            return fixture_format != BlockchainFixture
        if "blockchain_test_engine_only" in [m.name for m in markers]:
            return fixture_format != BlockchainEngineFixture
        return False

    @staticmethod
    def make_genesis(
        genesis_environment: Environment,
        pre: Alloc,
        fork: Fork,
    ) -> Tuple[Alloc, FixtureBlock]:
        """Create a genesis block from the blockchain test definition."""
        env = 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()),
            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_hash=Requests() 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 [],
            ).with_rlp(txs=[]),
        )

    def generate_block_data(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Alloc,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | 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),
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            slow_request=slow,
        )

        try:
            rejected_txs = verify_transactions(
                txs=txs,
                exception_mapper=t8n.exception_mapper,
                result=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)

        requests_list: List[Bytes] | None = None
        if fork.header_requests_required(header.number, header.timestamp):
            assert transition_tool_output.result.requests is not None, (
                "Requests are required for this block"
            )
            requests = Requests(requests_lists=list(transition_tool_output.result.requests))

            if Hash(requests) != header.requests_hash:
                raise Exception(
                    "Requests root in header does not match the requests root in the transition "
                    "tool output: "
                    f"{header.requests_hash} != {Hash(requests)}"
                )

            requests_list = requests.requests_list

        if block.requests is not None:
            header.requests_hash = Hash(Requests(requests_lists=list(block.requests)))
            requests_list = block.requests

        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`

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

    @staticmethod
    def network_info(fork: Fork, eips: Optional[List[int]] = None):
        """Return 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, t8n_state: Alloc, expected_state: Alloc | None = None):
        """Verify post alloc after all block/s or payload/s are generated."""
        try:
            if expected_state:
                expected_state.verify_post_alloc(t8n_state)
            else:
                self.post.verify_post_alloc(t8n_state)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

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

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, 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, _, new_alloc, new_env = self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    eips=eips,
                    slow=slow,
                )
                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
                    ),
                    fork=fork,
                ).with_rlp(txs=txs)
                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,
                    ),
                )

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )

        self.verify_post_state(t8n, t8n_state=alloc)
        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainFixture(
            fork=network_info,
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            post_state=alloc,
            config=FixtureConfig(
                fork=network_info,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

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

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, 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,
                slow=slow,
            )
            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

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )

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

        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainEngineFixture(
            fork=network_info,
            genesis=genesis.header,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre=pre,
            post_state=alloc,
            sync_payload=sync_payload,
            last_block_hash=head_hash,
            config=FixtureConfig(
                fork=network_info,
                chain_id=self.chain_id,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            ),
        )

    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, slow=is_slow_test(request))
        elif fixture_format == BlockchainFixture:
            return self.make_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/blockchain.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "blockchain_test_only" in [m.name for m in markers]:
        return fixture_format != BlockchainFixture
    if "blockchain_test_engine_only" in [m.name for m in markers]:
        return fixture_format != BlockchainEngineFixture
    return False

make_genesis(genesis_environment, pre, fork) staticmethod

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py

@staticmethod
def make_genesis(
    genesis_environment: Environment,
    pre: Alloc,
    fork: Fork,
) -> Tuple[Alloc, FixtureBlock]:
    """Create a genesis block from the blockchain test definition."""
    env = 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()),
        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_hash=Requests() 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 [],
        ).with_rlp(txs=[]),
    )

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

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,
    slow: bool = False,
) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | 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),
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        slow_request=slow,
    )

    try:
        rejected_txs = verify_transactions(
            txs=txs,
            exception_mapper=t8n.exception_mapper,
            result=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)

    requests_list: List[Bytes] | None = None
    if fork.header_requests_required(header.number, header.timestamp):
        assert transition_tool_output.result.requests is not None, (
            "Requests are required for this block"
        )
        requests = Requests(requests_lists=list(transition_tool_output.result.requests))

        if Hash(requests) != header.requests_hash:
            raise Exception(
                "Requests root in header does not match the requests root in the transition "
                "tool output: "
                f"{header.requests_hash} != {Hash(requests)}"
            )

        requests_list = requests.requests_list

    if block.requests is not None:
        header.requests_hash = Hash(Requests(requests_lists=list(block.requests)))
        requests_list = block.requests

    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`

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

network_info(fork, eips=None) staticmethod

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

Source code in src/ethereum_test_specs/blockchain.py

@staticmethod
def network_info(fork: Fork, eips: Optional[List[int]] = None):
    """Return 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, t8n_state, expected_state=None)

Verify 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, t8n_state: Alloc, expected_state: Alloc | None = None):
    """Verify post alloc after all block/s or payload/s are generated."""
    try:
        if expected_state:
            expected_state.verify_post_alloc(t8n_state)
        else:
            self.post.verify_post_alloc(t8n_state)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

make_fixture(t8n, fork, eips=None, slow=False)

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,
    slow: bool = False,
) -> BlockchainFixture:
    """Create a fixture from the blockchain test definition."""
    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, 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, _, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
                slow=slow,
            )
            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
                ),
                fork=fork,
            ).with_rlp(txs=txs)
            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,
                ),
            )

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )

    self.verify_post_state(t8n, t8n_state=alloc)
    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainFixture(
        fork=network_info,
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        post_state=alloc,
        config=FixtureConfig(
            fork=network_info,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

make_hive_fixture(t8n, fork, eips=None, slow=False)

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,
    slow: bool = False,
) -> BlockchainEngineFixture:
    """Create a hive fixture from the blocktest definition."""
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, 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,
            slow=slow,
        )
        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

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )

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

    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainEngineFixture(
        fork=network_info,
        genesis=genesis.header,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre=pre,
        post_state=alloc,
        sync_payload=sync_payload,
        last_block_hash=head_hash,
        config=FixtureConfig(
            fork=network_info,
            chain_id=self.chain_id,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
        ),
    )

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, slow=is_slow_test(request))
    elif fixture_format == BlockchainFixture:
        return self.make_fixture(t8n, fork, eips, slow=is_slow_test(request))

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

EOFStateTest

Bases: EOFTest, Transaction

Filler type that generates an EOF test for container validation, and also tests the container during runtime using a state test (and blockchain test).

In the state or blockchain test, the container is first deployed to the pre-allocation and then a transaction is sent to the deployed container.

Container deployment/validation is not tested like in the EOFTest unless the container under test is an initcode container.

All fields from ethereum_test_types.Transaction are available for use in the test.

Source code in src/ethereum_test_specs/eof.py

class EOFStateTest(EOFTest, Transaction):
    """
    Filler type that generates an EOF test for container validation, and also tests the container
    during runtime using a state test (and blockchain test).

    In the state or blockchain test, the container is first deployed to the pre-allocation and
    then a transaction is sent to the deployed container.

    Container deployment/validation is **not** tested like in the `EOFTest` unless the container
    under test is an initcode container.

    All fields from `ethereum_test_types.Transaction` are available for use in the test.
    """

    gas_limit: HexNumber = Field(HexNumber(10_000_000), serialization_alias="gas")
    """
    Gas limit for the transaction that deploys the container.
    """
    tx_sender_funding_amount: int = 1_000_000_000_000_000_000_000
    """
    Amount of funds to send to the sender EOA before the transaction.
    """
    env: Environment = Field(default_factory=Environment)
    """
    Environment object that is used during State Test generation.
    """
    container_post: Account = Field(default_factory=Account)
    """
    Account object used to verify the container post state.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"eof_{fixture_format.format_name}",
            f"Tests that generate an EOF {fixture_format.format_name}.",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"eof_{execute_format.format_name}",
            f"Tests that generate an EOF {execute_format.format_name}.",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

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

    def model_post_init(self, __context):
        """Prepare the transaction parameters required to fill the test."""
        assert self.pre is not None, "pre must be set to generate a StateTest."

        EOFTest.model_post_init(self, __context)

        self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
        if self.post is None:
            self.post = Alloc()

        if self.expect_exception is not None:  # Invalid EOF
            self.to = None  # Make EIP-7698 create transaction
            self.data = Bytes(
                bytes(self.container) + self.data
            )  # by concatenating container and tx data.

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.created_contract] = None  # Expect failure.
        elif self.container_kind == ContainerKind.INITCODE:
            self.to = None  # Make EIP-7698 create transaction
            self.data = Bytes(
                bytes(self.container) + self.data
            )  # by concatenating container and tx data.

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.created_contract] = self.container_post  # Successful.
        else:
            self.to = self.pre.deploy_contract(code=self.container)

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.to] = self.container_post

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        assert self.pre is not None, "pre must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."

        return StateTest(
            pre=self.pre,
            tx=self,
            env=self.env,
            post=self.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 Bytes(self.container) 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 StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).generate(
                request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )

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

gas_limit: HexNumber = Field(HexNumber(10000000), serialization_alias='gas') class-attribute instance-attribute

Gas limit for the transaction that deploys the container.

tx_sender_funding_amount: int = 1000000000000000000000 class-attribute instance-attribute

Amount of funds to send to the sender EOA before the transaction.

env: Environment = Field(default_factory=Environment) class-attribute instance-attribute

Environment object that is used during State Test generation.

container_post: Account = Field(default_factory=Account) class-attribute instance-attribute

Account object used to verify the container post state.

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"

model_post_init(__context)

Prepare the transaction parameters required to fill the test.

Source code in src/ethereum_test_specs/eof.py

def model_post_init(self, __context):
    """Prepare the transaction parameters required to fill the test."""
    assert self.pre is not None, "pre must be set to generate a StateTest."

    EOFTest.model_post_init(self, __context)

    self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
    if self.post is None:
        self.post = Alloc()

    if self.expect_exception is not None:  # Invalid EOF
        self.to = None  # Make EIP-7698 create transaction
        self.data = Bytes(
            bytes(self.container) + self.data
        )  # by concatenating container and tx data.

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.created_contract] = None  # Expect failure.
    elif self.container_kind == ContainerKind.INITCODE:
        self.to = None  # Make EIP-7698 create transaction
        self.data = Bytes(
            bytes(self.container) + self.data
        )  # by concatenating container and tx data.

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.created_contract] = self.container_post  # Successful.
    else:
        self.to = self.pre.deploy_contract(code=self.container)

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.to] = self.container_post

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    assert self.pre is not None, "pre must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."

    return StateTest(
        pre=self.pre,
        tx=self,
        env=self.env,
        post=self.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 Bytes(self.container) 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 StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).generate(
            request=request, t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )

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

EOFTest

Bases: BaseTest

Filler type that generates a test for EOF container validation.

A state test is also automatically generated where the container is wrapped in a contract-creating transaction to test deployment/validation on the instantiated blockchain.

Source code in src/ethereum_test_specs/eof.py

class EOFTest(BaseTest):
    """
    Filler type that generates a test for EOF container validation.

    A state test is also automatically generated where the container is wrapped in a
    contract-creating transaction to test deployment/validation on the instantiated blockchain.
    """

    container: Container
    """
    EOF container that will be tested for validity.

    The only supported type at the moment is `ethereum_test_types.eof.v1.Container`.

    If an invalid container needs to be tested, and it cannot be generated using the
    Container class features, the `raw_bytes` field can be used to provide the raw
    container bytes.
    """
    expect_exception: EOFExceptionInstanceOrList | None = None
    """
    Expected exception that the container should raise when parsed by an EOF parser.

    Can be a single exception or a list of exceptions that the container is expected to raise,
    in which case the test will pass if any of the exceptions are raised.

    The list of supported exceptions can be found in the `ethereum_test_exceptions.EOFException`
    class.
    """
    container_kind: ContainerKind = ContainerKind.RUNTIME
    """
    Container kind type that the container should be treated as.

    The container kind can be one of the following:
    - `ContainerKind.INITCODE`: The container is an initcode container.
    - `ContainerKind.RUNTIME`: The container is a runtime container.

    The default value is `ContainerKind.RUNTIME`.
    """
    deployed_container: Container | None = None
    """
    To be used when the container is an initcode container and the expected deployed container is
    known.

    The value is only used when a State Test is generated from this EOF test to set the expected
    deployed container that should be found in the post state.

    If this field is not set, and the container is valid:
      - If the container kind is `ContainerKind.RUNTIME`, the deployed container is assumed to be
        the container itself, and an initcode container that wraps the container is generated
        automatically.
      - If the container kind is `ContainerKind.INITCODE`, `model_post_init` will attempt to infer
        the deployed container from the sections of the init-container, and the first
        container-type section will be used. An error will be raised if the deployed container
        cannot be inferred.

    If the value is set to `None`, it is assumed that the container is invalid and the test will
    expect that no contract is created.

    It is considered an error if:
      - The `deployed_container` field is set and the `container_kind` field is not set to
        `ContainerKind.INITCODE`.
      - The `deployed_container` field is set and the `expect_exception` is not `None`.

    The deployed container is **not** executed at any point during the EOF validation test nor
    the generated State Test. For container runtime testing use the `EOFStateTest` class.
    """
    pre: Alloc | None = None
    """
    Pre alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and should otherwise be left unset.
    """
    post: Alloc | None = None
    """
    Post alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and is normally not set by the user.
    """
    sender: EOA | None = None
    """
    Sender EOA object that is used during State Test generation.

    This field is automatically set by the `model_post_init` method and should otherwise be left
    unset.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_eof_test",
            f"A {fixture_format.format_name} generated from an eof_test.",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"{execute_format.format_name}_from_eof_test",
            f"A {execute_format.format_name} generated from an eof_test.",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "eof_test_only": "Only generate an EOF test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "eof_test_only" in [m.name for m in markers]:
            return fixture_format != EOFFixture
        return False

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

    def model_post_init(self, __context):
        """Prepare the test exception based on the container."""
        if self.container.validity_error is not None:
            if self.expect_exception is not None:
                assert self.expect_exception == self.container.validity_error, (
                    f"Container validity error {self.container.validity_error} "
                    f"does not match expected exception {self.expect_exception}."
                )
            self.expect_exception = self.container.validity_error
            assert self.deployed_container is None, (
                "deployed_container must be None for invalid containers."
            )
        if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
            if (
                "kind" in self.container.model_fields_set
                and "container_kind" in self.model_fields_set
            ):
                assert self.container.kind == self.container_kind, (
                    f"Container kind type {str(self.container.kind)} "
                    f"does not match test {self.container_kind}."
                )
            elif "kind" in self.container.model_fields_set:
                self.container_kind = self.container.kind
            elif "container_kind" in self.model_fields_set:
                self.container.kind = self.container_kind

        assert self.pre is not None, "pre must be set to generate a StateTest."
        self.sender = self.pre.fund_eoa()
        if self.post is None:
            self.post = Alloc()

    def make_eof_test_fixture(
        self,
        *,
        request: pytest.FixtureRequest,
        fork: Fork,
        eips: Optional[List[int]],
    ) -> EOFFixture:
        """Generate the EOF test fixture."""
        container_bytes = Bytes(self.container)
        if container_bytes in existing_tests:
            pytest.fail(
                f"Duplicate EOF test: {container_bytes}, "
                f"existing test: {existing_tests[container_bytes]}"
            )
        existing_tests[container_bytes] = request.node.nodeid
        vectors = [
            Vector(
                code=container_bytes,
                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!", stacklevel=2
            )
            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_value=str(vector.code))
            self.verify_result(result, expected_result, vector.code)

        return fixture

    def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
        """Check 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 UnexpectedEOFExceptionError(
                    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 ExpectedEOFExceptionError(
                    code=code,
                    expected=f"{expected_string}",
                )
            elif actual_exception in expected_result.exception:
                return
            else:
                raise EOFExceptionMismatchError(
                    code=code,
                    expected=f"{expected_string}",
                    got=f"{actual_exception} ({actual_message})",
                )

    def generate_eof_contract_create_transaction(self) -> Transaction:
        """Generate a transaction that creates a contract."""
        assert self.sender is not None, "sender must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."

        initcode: Container
        deployed_container: Container | Bytes | None = None
        if self.container_kind == ContainerKind.INITCODE:
            initcode = self.container
            if "deployed_container" in self.model_fields_set:
                # In the case of an initcontainer where we know the deployed container,
                # we can use the initcontainer as-is.
                deployed_container = self.deployed_container
            elif self.expect_exception is None:
                # We have a valid init-container, but we don't know the deployed container.
                # Try to infer the deployed container from the sections of the init-container.
                assert self.container.raw_bytes is None, (
                    "deployed_container must be set for initcode containers with raw_bytes."
                )
                for section in self.container.sections:
                    if section.kind == SectionKind.CONTAINER:
                        deployed_container = section.data
                        break

                assert deployed_container is not None, (
                    "Unable to infer deployed container for init-container. "
                    "Use field `deployed_container` to set the expected deployed container."
                )
        else:
            assert self.deployed_container is None, (
                "deployed_container must be None for runtime containers."
            )
            initcode = Container(
                sections=[
                    Section.Code(Op.RETURNCONTRACT[0](0, 0)),
                    Section.Container(self.container),
                ]
            )
            deployed_container = self.container

        tx = Transaction(
            sender=self.sender,
            to=None,
            gas_limit=10_000_000,
            data=initcode,
        )

        if self.expect_exception is not None or deployed_container is None:
            self.post[tx.created_contract] = None
        else:
            self.post[tx.created_contract] = Account(
                code=deployed_container,
            )
        return tx

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        return StateTest(
            pre=self.pre,
            tx=self.generate_eof_contract_create_transaction(),
            env=Environment(),
            post=self.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:
            return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
        elif fixture_format in StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).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(fork).execute(
                fork=fork, execute_format=execute_format, eips=eips
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

container: Container instance-attribute

EOF container that will be tested for validity.

The only supported type at the moment is ethereum_test_types.eof.v1.Container.

If an invalid container needs to be tested, and it cannot be generated using the Container class features, the raw_bytes field can be used to provide the raw container bytes.

expect_exception: EOFExceptionInstanceOrList | None = None class-attribute instance-attribute

Expected exception that the container should raise when parsed by an EOF parser.

Can be a single exception or a list of exceptions that the container is expected to raise, in which case the test will pass if any of the exceptions are raised.

The list of supported exceptions can be found in the ethereum_test_exceptions.EOFException class.

container_kind: ContainerKind = ContainerKind.RUNTIME class-attribute instance-attribute

Container kind type that the container should be treated as.

The container kind can be one of the following: - ContainerKind.INITCODE: The container is an initcode container. - ContainerKind.RUNTIME: The container is a runtime container.

The default value is ContainerKind.RUNTIME.

deployed_container: Container | None = None class-attribute instance-attribute

To be used when the container is an initcode container and the expected deployed container is known.

The value is only used when a State Test is generated from this EOF test to set the expected deployed container that should be found in the post state.

If this field is not set, and the container is valid: - If the container kind is ContainerKind.RUNTIME, the deployed container is assumed to be the container itself, and an initcode container that wraps the container is generated automatically. - If the container kind is ContainerKind.INITCODE, model_post_init will attempt to infer the deployed container from the sections of the init-container, and the first container-type section will be used. An error will be raised if the deployed container cannot be inferred.

If the value is set to None, it is assumed that the container is invalid and the test will expect that no contract is created.

It is considered an error if

• The deployed_container field is set and the container_kind field is not set to ContainerKind.INITCODE.
• The deployed_container field is set and the expect_exception is not None.

The deployed container is not executed at any point during the EOF validation test nor the generated State Test. For container runtime testing use the EOFStateTest class.

pre: Alloc | None = None class-attribute instance-attribute

Pre alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and should otherwise be left unset.

post: Alloc | None = None class-attribute instance-attribute

Post alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and is normally not set by the user.

sender: EOA | None = None class-attribute instance-attribute

Sender EOA object that is used during State Test generation.

This field is automatically set by the model_post_init method and should otherwise be left unset.

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/eof.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "eof_test_only" in [m.name for m in markers]:
        return fixture_format != EOFFixture
    return False

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"

model_post_init(__context)

Prepare the test exception based on the container.

Source code in src/ethereum_test_specs/eof.py

def model_post_init(self, __context):
    """Prepare the test exception based on the container."""
    if self.container.validity_error is not None:
        if self.expect_exception is not None:
            assert self.expect_exception == self.container.validity_error, (
                f"Container validity error {self.container.validity_error} "
                f"does not match expected exception {self.expect_exception}."
            )
        self.expect_exception = self.container.validity_error
        assert self.deployed_container is None, (
            "deployed_container must be None for invalid containers."
        )
    if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
        if (
            "kind" in self.container.model_fields_set
            and "container_kind" in self.model_fields_set
        ):
            assert self.container.kind == self.container_kind, (
                f"Container kind type {str(self.container.kind)} "
                f"does not match test {self.container_kind}."
            )
        elif "kind" in self.container.model_fields_set:
            self.container_kind = self.container.kind
        elif "container_kind" in self.model_fields_set:
            self.container.kind = self.container_kind

    assert self.pre is not None, "pre must be set to generate a StateTest."
    self.sender = self.pre.fund_eoa()
    if self.post is None:
        self.post = Alloc()

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."""
    container_bytes = Bytes(self.container)
    if container_bytes in existing_tests:
        pytest.fail(
            f"Duplicate EOF test: {container_bytes}, "
            f"existing test: {existing_tests[container_bytes]}"
        )
    existing_tests[container_bytes] = request.node.nodeid
    vectors = [
        Vector(
            code=container_bytes,
            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!", stacklevel=2
        )
        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_value=str(vector.code))
        self.verify_result(result, expected_result, vector.code)

    return fixture

verify_result(result, expected_result, code)

Check 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):
    """Check 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 UnexpectedEOFExceptionError(
                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 ExpectedEOFExceptionError(
                code=code,
                expected=f"{expected_string}",
            )
        elif actual_exception in expected_result.exception:
            return
        else:
            raise EOFExceptionMismatchError(
                code=code,
                expected=f"{expected_string}",
                got=f"{actual_exception} ({actual_message})",
            )

generate_eof_contract_create_transaction()

Generate a transaction that creates a contract.

Source code in src/ethereum_test_specs/eof.py

def generate_eof_contract_create_transaction(self) -> Transaction:
    """Generate a transaction that creates a contract."""
    assert self.sender is not None, "sender must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."

    initcode: Container
    deployed_container: Container | Bytes | None = None
    if self.container_kind == ContainerKind.INITCODE:
        initcode = self.container
        if "deployed_container" in self.model_fields_set:
            # In the case of an initcontainer where we know the deployed container,
            # we can use the initcontainer as-is.
            deployed_container = self.deployed_container
        elif self.expect_exception is None:
            # We have a valid init-container, but we don't know the deployed container.
            # Try to infer the deployed container from the sections of the init-container.
            assert self.container.raw_bytes is None, (
                "deployed_container must be set for initcode containers with raw_bytes."
            )
            for section in self.container.sections:
                if section.kind == SectionKind.CONTAINER:
                    deployed_container = section.data
                    break

            assert deployed_container is not None, (
                "Unable to infer deployed container for init-container. "
                "Use field `deployed_container` to set the expected deployed container."
            )
    else:
        assert self.deployed_container is None, (
            "deployed_container must be None for runtime containers."
        )
        initcode = Container(
            sections=[
                Section.Code(Op.RETURNCONTRACT[0](0, 0)),
                Section.Container(self.container),
            ]
        )
        deployed_container = self.container

    tx = Transaction(
        sender=self.sender,
        to=None,
        gas_limit=10_000_000,
        data=initcode,
    )

    if self.expect_exception is not None or deployed_container is None:
        self.post[tx.created_contract] = None
    else:
        self.post[tx.created_contract] = Account(
            code=deployed_container,
        )
    return tx

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py

def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    return StateTest(
        pre=self.pre,
        tx=self.generate_eof_contract_create_transaction(),
        env=Environment(),
        post=self.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:
        return self.make_eof_test_fixture(request=request, fork=fork, eips=eips)
    elif fixture_format in StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).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(fork).execute(
            fork=fork, execute_format=execute_format, eips=eips
        )
    raise Exception(f"Unsupported execute format: {execute_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 = Field(default_factory=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[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        StateFixture,
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_state_test",
            f"A {fixture_format.format_name} generated from a state_test",
        )
        for fixture_format in BlockchainTest.supported_fixture_formats
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "state_test_only": "Only generate a state test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "state_test_only" in [m.name for m in markers]:
            return fixture_format != StateFixture
        return False

    def _generate_blockchain_genesis_environment(self, *, fork: Fork) -> 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 + (
                fork.target_blobs_per_block() * fork.blob_gas_per_blob()
            )

        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, *, fork: Fork) -> BlockchainTest:
        """Generate a BlockchainTest fixture from this StateTest fixture."""
        return BlockchainTest(
            genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
            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,
        slow: bool = False,
    ) -> StateFixture:
        """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
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            state_test=True,
            slow_request=slow,
        )

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

        try:
            verify_transactions(
                txs=[tx],
                exception_mapper=t8n.exception_mapper,
                result=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 StateFixture(
            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,
                        state=transition_tool_output.alloc,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
            config=FixtureConfig(
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

    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(fork=fork).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, slow=is_slow_test(request))

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

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/state.py

@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "state_test_only" in [m.name for m in markers]:
        return fixture_format != StateFixture
    return False

generate_blockchain_test(*, fork)

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in src/ethereum_test_specs/state.py

def generate_blockchain_test(self, *, fork: Fork) -> BlockchainTest:
    """Generate a BlockchainTest fixture from this StateTest fixture."""
    return BlockchainTest(
        genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
        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, slow=False)

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,
    slow: bool = False,
) -> StateFixture:
    """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
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        state_test=True,
        slow_request=slow,
    )

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

    try:
        verify_transactions(
            txs=[tx],
            exception_mapper=t8n.exception_mapper,
            result=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 StateFixture(
        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,
                    state=transition_tool_output.alloc,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
        config=FixtureConfig(
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

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(fork=fork).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, slow=is_slow_test(request))

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

TransactionTest

Bases: BaseTest

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

Source code in src/ethereum_test_specs/transaction.py

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

    tx: Transaction
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        TransactionFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[ExecuteFormat | LabeledExecuteFormat]] = [
        TransactionPost,
    ]

    def make_transaction_test_fixture(
        self,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> TransactionFixture:
        """Create a fixture from the transaction test definition."""
        if self.tx.error is not None:
            result = FixtureResult(
                exception=self.tx.error,
                hash=None,
                intrinsic_gas=0,
                sender=None,
            )
        else:
            intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
            intrinsic_gas = intrinsic_gas_cost_calculator(
                calldata=self.tx.data,
                contract_creation=self.tx.to is None,
                access_list=self.tx.access_list,
                authorization_list_or_count=self.tx.authorization_list,
            )
            result = FixtureResult(
                exception=None,
                hash=self.tx.hash,
                intrinsic_gas=intrinsic_gas,
                sender=self.tx.sender,
            )

        return TransactionFixture(
            result={
                fork.blockchain_test_network_name(): result,
            },
            transaction=self.tx.with_signature_and_sender().rlp,
        )

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

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

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Execute the transaction test by sending it to the live network."""
        if execute_format == TransactionPost:
            return TransactionPost(
                transactions=[self.tx],
                post={},
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

make_transaction_test_fixture(fork, eips=None)

Create a fixture from the transaction test definition.

Source code in src/ethereum_test_specs/transaction.py

def make_transaction_test_fixture(
    self,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> TransactionFixture:
    """Create a fixture from the transaction test definition."""
    if self.tx.error is not None:
        result = FixtureResult(
            exception=self.tx.error,
            hash=None,
            intrinsic_gas=0,
            sender=None,
        )
    else:
        intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
        intrinsic_gas = intrinsic_gas_cost_calculator(
            calldata=self.tx.data,
            contract_creation=self.tx.to is None,
            access_list=self.tx.access_list,
            authorization_list_or_count=self.tx.authorization_list,
        )
        result = FixtureResult(
            exception=None,
            hash=self.tx.hash,
            intrinsic_gas=intrinsic_gas,
            sender=self.tx.sender,
        )

    return TransactionFixture(
        result={
            fork.blockchain_test_network_name(): result,
        },
        transaction=self.tx.with_signature_and_sender().rlp,
    )

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

Generate the TransactionTest fixture.

Source code in src/ethereum_test_specs/transaction.py

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

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

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

Execute the transaction test by sending it to the live network.

Source code in src/ethereum_test_specs/transaction.py

def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Execute the transaction test by sending it to the live network."""
    if execute_format == TransactionPost:
        return TransactionPost(
            transactions=[self.tx],
            post={},
        )
    raise Exception(f"Unsupported execute format: {execute_format}")