feat(fill, execute): track execution & setup testing phase

src/ethereum_test_execution/transaction_post.py

@@ -8,7 +8,7 @@
 from ethereum_test_base_types import Address, Alloc, Hash
 from ethereum_test_forks import Fork
 from ethereum_test_rpc import EngineRPC, EthRPC, SendTransactionExceptionError
-from ethereum_test_types import Transaction, TransactionTestMetadata
+from ethereum_test_types import TestPhase, Transaction, TransactionTestMetadata
 
 from .base import BaseExecute
 
@@ -40,9 +40,15 @@ def execute(
                 tx = tx.with_signature_and_sender()
                 to_address = tx.to
                 label = to_address.label if isinstance(to_address, Address) else None
+                phase = (
+                    "testing"
+                    if getattr(tx, "test_phase", None) == TestPhase.EXECUTION
+                    or getattr(block, "test_phase", None) == TestPhase.EXECUTION
+                    else "setup"
+                )
                 tx.metadata = TransactionTestMetadata(
                     test_id=request.node.nodeid,
-                    phase="testing",
+                    phase=phase,
                     target=label,
                     tx_index=tx_index,
                 )

src/ethereum_test_specs/blockchain.py

@@ -51,7 +51,16 @@
 )
 from ethereum_test_fixtures.common import FixtureBlobSchedule
 from ethereum_test_forks import Fork
-from ethereum_test_types import Alloc, Environment, Removable, Requests, Transaction, Withdrawal
+from ethereum_test_types import (
+    Alloc,
+    Environment,
+    Removable,
+    Requests,
+    TestPhase,
+    TestPhaseManager,
+    Transaction,
+    Withdrawal,
+)
 from ethereum_test_types.block_access_list import BlockAccessList, BlockAccessListExpectation
 
 from .base import BaseTest, OpMode, verify_result
@@ -232,7 +241,13 @@ class Block(Header):
     expected_post_state: Alloc | None = None
     """Post state for verification after block execution in BlockchainTest"""
     block_access_list: Bytes | None = Field(None)
-    """EIP-7928: Block-level access lists (serialized)."""
+    """
+        EIP-7928: Block-level access lists (serialized).
+    """
+    test_phase: TestPhase | None = None
+    """
+    Test phase for this block.
+    """
 
     def set_environment(self, env: Environment) -> Environment:
         """
@@ -404,10 +419,11 @@ class BlockchainTest(BaseTest):
 
     pre: Alloc
     post: Alloc
-    blocks: List[Block]
+    blocks: List[Block] = Field(default_factory=list)
     genesis_environment: Environment = Field(default_factory=Environment)
     chain_id: int = 1
     exclude_full_post_state_in_output: bool = False
+    test_phase_manager: TestPhaseManager | None = Field(default=None, exclude=True)
     """
     Exclude the post state from the fixture output. In this case, the state
     verification is only performed based on the state root.
@@ -698,6 +714,33 @@ def verify_post_state(self, t8n, t8n_state: Alloc, expected_state: Alloc | None
             print_traces(t8n.get_traces())
             raise e
 
+    def _get_test_phase_blocks(self) -> List[Block]:
+        """
+        Get additional blocks from phase manager for setup
+        and execution phases.
+        """
+        assert self.test_phase_manager is not None, "Test phase manager is not set"
+
+        blocks = []
+        if self.test_phase_manager.setup_blocks:
+            for block in self.test_phase_manager.setup_blocks:
+                block.test_phase = TestPhase.SETUP
+            blocks.extend(self.test_phase_manager.setup_blocks)
+        elif self.test_phase_manager.setup_transactions:
+            for tx in self.test_phase_manager.setup_transactions:
+                tx.test_phase = TestPhase.SETUP
+            blocks.append(Block(txs=self.test_phase_manager.setup_transactions))
+
+        if self.test_phase_manager.execution_blocks:
+            for block in self.test_phase_manager.execution_blocks:
+                block.test_phase = TestPhase.EXECUTION
+            blocks.extend(self.test_phase_manager.execution_blocks)
+        elif self.test_phase_manager.execution_transactions:
+            for tx in self.test_phase_manager.execution_transactions:
+                tx.test_phase = TestPhase.EXECUTION
+            blocks.append(Block(txs=self.test_phase_manager.execution_transactions))
+        return blocks
+
     def make_fixture(
         self,
         t8n: TransitionTool,
@@ -712,6 +755,7 @@ def make_fixture(
         env = environment_from_parent_header(genesis.header)
         head = genesis.header.block_hash
         invalid_blocks = 0
+
         for i, block in enumerate(self.blocks):
             # This is the most common case, the RLP needs to be constructed
             # based on the transactions to be included in the block.
@@ -776,6 +820,8 @@ def make_hive_fixture(
         env = environment_from_parent_header(genesis.header)
         head_hash = genesis.header.block_hash
         invalid_blocks = 0
+
+        built_block = None
         for i, block in enumerate(self.blocks):
             built_block = self.generate_block_data(
                 t8n=t8n,
@@ -798,6 +844,10 @@ def make_hive_fixture(
                     t8n, t8n_state=alloc, expected_state=block.expected_post_state
                 )
         self.check_exception_test(exception=invalid_blocks > 0)
+
+        if built_block is None:
+            raise Exception("No blocks to process in the test")
+
         fcu_version = fork.engine_forkchoice_updated_version(
             built_block.header.number, built_block.header.timestamp
         )
@@ -877,6 +927,10 @@ def generate(
     ) -> BaseFixture:
         """Generate the BlockchainTest fixture."""
         t8n.reset_traces()
+
+        if self.test_phase_manager is not None:
+            self.blocks.extend(self._get_test_phase_blocks())
+
         if fixture_format in [
             BlockchainEngineFixture,
             BlockchainEngineXFixture,
@@ -897,6 +951,8 @@ def execute(
         """Generate the list of test fixtures."""
         if execute_format == TransactionPost:
             blocks: List[List[Transaction]] = []
+            if self.test_phase_manager is not None:
+                self.blocks.extend(self._get_test_phase_blocks())
             for block in self.blocks:
                 blocks += [block.txs]
             return TransactionPost(

src/ethereum_test_types/__init__.py

@@ -26,6 +26,7 @@
     compute_create_address,
     compute_eofcreate_address,
 )
+from .phase_manager import TestPhase, TestPhaseManager
 from .receipt_types import TransactionReceipt
 from .request_types import (
     ConsolidationRequest,
@@ -66,6 +67,8 @@
     "Removable",
     "Requests",
     "TestParameterGroup",
+    "TestPhase",
+    "TestPhaseManager",
     "Transaction",
     "TransactionDefaults",
     "TransactionReceipt",

src/ethereum_test_types/phase_manager.py

@@ -0,0 +1,95 @@
+"""Test phase management for Ethereum tests."""
+
+from contextlib import contextmanager
+from enum import Enum
+from typing import Any, Iterator, List, Optional
+
+from pydantic import GetCoreSchemaHandler
+from pydantic_core.core_schema import (
+    PlainValidatorFunctionSchema,
+    no_info_plain_validator_function,
+)
+
+
+class TestPhase(Enum):
+    """Test phase for state and blockchain tests."""
+
+    SETUP = "setup"
+    EXECUTION = "execution"
+
+
+class TestPhaseManager:
+    """
+    Manages test phases and collects transactions and blocks by phase.
+    This class provides a mechanism for "setup" and "execution" phases,
+    Only supports "setup" and "execution" phases now.
+    """
+
+    def __init__(self, *args, **kwargs):
+        """Initialize TestPhaseManager with empty transactions and blocks."""
+        self.setup_transactions: List = []
+        self.setup_blocks: List = []
+        self.execution_transactions: List = []
+        self.execution_blocks: List = []
+        self._current_phase: Optional[TestPhase] = TestPhase.EXECUTION
+
+    @contextmanager
+    def setup(self) -> Iterator["TestPhaseManager"]:
+        """Context manager for the setup phase of a benchmark test."""
+        old_phase = self._current_phase
+        self._current_phase = TestPhase.SETUP
+        try:
+            yield self
+        finally:
+            self._current_phase = old_phase
+
+    @contextmanager
+    def execution(self) -> Iterator["TestPhaseManager"]:
+        """Context manager for the execution phase of a test."""
+        old_phase = self._current_phase
+        self._current_phase = TestPhase.EXECUTION
+        try:
+            yield self
+        finally:
+            self._current_phase = old_phase
+
+    def add_transaction(self, tx) -> None:
+        """Add a transaction to the current phase."""
+        current_phase = self._current_phase
+        tx.test_phase = current_phase
+
+        if current_phase == TestPhase.EXECUTION:
+            self.execution_transactions.append(tx)
+        else:
+            self.setup_transactions.append(tx)
+
+    def add_block(self, block) -> None:
+        """Add a block to the current phase."""
+        current_phase = self._current_phase
+        for tx in block.txs:
+            tx.test_phase = current_phase
+
+        if current_phase == TestPhase.EXECUTION:
+            self.execution_blocks.append(block)
+        else:
+            self.setup_blocks.append(block)
+
+    def get_current_phase(self) -> Optional[TestPhase]:
+        """Get the current test phase."""
+        return self._current_phase
+
+    @staticmethod
+    def __get_pydantic_core_schema__(
+        source_type: Any, handler: GetCoreSchemaHandler
+    ) -> PlainValidatorFunctionSchema:
+        """Pydantic schema for TestPhaseManager."""
+
+        def validate_test_phase_manager(value):
+            """Return the TestPhaseManager instance as-is."""
+            if isinstance(value, source_type):
+                return value
+            return source_type()
+
+        return no_info_plain_validator_function(
+            validate_test_phase_manager,
+        )