feat: implement phase manager

Author: LouisTsai-Csie

src/ethereum_test_execution/transaction_post.py

@@ -53,9 +53,10 @@ 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 tx.test_phase == "execution" else "setup"
                 tx.metadata = TransactionTestMetadata(
                     test_id=request.node.nodeid,
-                    phase="testing",
+                    phase=phase,
                     target=label,
                     tx_index=tx_index,
                 )

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,65 @@
+"""Test phase management for Ethereum tests."""
+
+from contextlib import contextmanager
+from enum import Enum
+from typing import ClassVar, Iterator, Optional
+
+
+class TestPhase(Enum):
+    """Test phase for state and blockchain tests."""
+
+    SETUP = "setup"
+    EXECUTION = "execution"
+
+
+class TestPhaseManager:
+    """
+    Manages test phases for transactions and blocks.
+
+    This singleton class provides context managers for "setup" and
+    "execution" phases. Transactions automatically detect and tag
+    themselves with the current phase.
+
+    Usage:
+        with TestPhaseManager.setup():
+            # Transactions created here have test_phase = "setup"
+            setup_tx = Transaction(...)
+
+        with TestPhaseManager.execution():
+            # Transactions created here have test_phase = "execution"
+            benchmark_tx = Transaction(...)
+    """
+
+    _current_phase: ClassVar[Optional[TestPhase]] = None
+
+    @classmethod
+    @contextmanager
+    def setup(cls) -> Iterator[None]:
+        """Context manager for the setup phase of a benchmark test."""
+        old_phase = cls._current_phase
+        cls._current_phase = TestPhase.SETUP
+        try:
+            yield
+        finally:
+            cls._current_phase = old_phase
+
+    @classmethod
+    @contextmanager
+    def execution(cls) -> Iterator[None]:
+        """Context manager for the execution phase of a test."""
+        old_phase = cls._current_phase
+        cls._current_phase = TestPhase.EXECUTION
+        try:
+            yield
+        finally:
+            cls._current_phase = old_phase
+
+    @classmethod
+    def get_current_phase(cls) -> Optional[TestPhase]:
+        """Get the current test phase."""
+        return cls._current_phase
+
+    @classmethod
+    def reset(cls) -> None:
+        """Reset the phase state to None (primarily for testing)."""
+        cls._current_phase = None

src/ethereum_test_types/tests/test_phase_manager.py

@@ -0,0 +1,151 @@
+"""Test suite for TestPhaseManager functionality."""
+
+import pytest
+
+from ethereum_test_base_types import Address
+from ethereum_test_tools import Transaction
+
+from ..phase_manager import TestPhase, TestPhaseManager
+
+
+@pytest.fixture(autouse=True)
+def reset_phase_manager() -> None:
+    """Reset TestPhaseManager singleton state before each test."""
+    TestPhaseManager.reset()
+
+
+def test_test_phase_enum_values() -> None:
+    """Test that TestPhase enum has correct values."""
+    assert TestPhase.SETUP.value == "setup"
+    assert TestPhase.EXECUTION.value == "execution"
+
+
+def test_phase_manager_class_state() -> None:
+    """Test TestPhaseManager uses class-level state."""
+    # All access is through class methods, no instance needed
+    assert TestPhaseManager.get_current_phase() is None
+
+    # Setting phase through class method
+    with TestPhaseManager.setup():
+        assert TestPhaseManager.get_current_phase() == TestPhase.SETUP
+
+    # Phase persists at class level
+    assert TestPhaseManager.get_current_phase() is None
+
+
+def test_default_phase_is_none() -> None:
+    """Test that default phase is None (no context set)."""
+    assert TestPhaseManager.get_current_phase() is None
+
+
+def test_transaction_auto_detects_execution_phase() -> None:
+    """Test that transactions default to 'execution' when no phase set."""
+    tx = Transaction(to=Address(0x123), value=100, gas_limit=21000)
+    assert tx.test_phase == "execution"
+
+
+def test_transaction_auto_detects_setup_phase() -> None:
+    """Test that transactions created in setup context get SETUP phase."""
+    with TestPhaseManager.setup():
+        tx = Transaction(to=Address(0x456), value=50, gas_limit=21000)
+        assert tx.test_phase == "setup"
+
+
+def test_phase_context_switching() -> None:
+    """Test that phase switching works correctly."""
+    # Start with no phase set (defaults to execution)
+    tx1 = Transaction(to=Address(0x100), value=100, gas_limit=21000)
+    assert tx1.test_phase == "execution"
+
+    # Switch to SETUP
+    with TestPhaseManager.setup():
+        assert TestPhaseManager.get_current_phase() == TestPhase.SETUP
+        tx2 = Transaction(to=Address(0x200), value=200, gas_limit=21000)
+        assert tx2.test_phase == "setup"
+
+    # Back to None after context (transactions default to execution)
+    assert TestPhaseManager.get_current_phase() is None
+    tx3 = Transaction(to=Address(0x300), value=300, gas_limit=21000)
+    assert tx3.test_phase == "execution"
+
+
+def test_nested_phase_contexts() -> None:
+    """Test that nested phase contexts work correctly."""
+    with TestPhaseManager.setup():
+        tx1 = Transaction(to=Address(0x100), value=100, gas_limit=21000)
+        assert tx1.test_phase == "setup"
+
+        # Nested execution context
+        with TestPhaseManager.execution():
+            tx2 = Transaction(to=Address(0x200), value=200, gas_limit=21000)
+            assert tx2.test_phase == "execution"
+
+        # Back to setup after nested context
+        tx3 = Transaction(to=Address(0x300), value=300, gas_limit=21000)
+        assert tx3.test_phase == "setup"
+
+
+@pytest.mark.parametrize(
+    ["num_setup_txs", "num_exec_txs"],
+    [
+        pytest.param(0, 1, id="exec_only"),
+        pytest.param(1, 0, id="setup_only"),
+        pytest.param(3, 5, id="mixed"),
+        pytest.param(10, 10, id="many"),
+    ],
+)
+def test_multiple_transactions_phase_tagging(num_setup_txs: int, num_exec_txs: int) -> None:
+    """Test that multiple transactions are correctly tagged by phase."""
+    setup_txs = []
+    exec_txs = []
+
+    # Create setup transactions
+    with TestPhaseManager.setup():
+        for i in range(num_setup_txs):
+            tx = Transaction(to=Address(0x1000 + i), value=i * 10, gas_limit=21000)
+            setup_txs.append(tx)
+
+    # Create execution transactions
+    for i in range(num_exec_txs):
+        tx = Transaction(to=Address(0x2000 + i), value=i * 20, gas_limit=21000)
+        exec_txs.append(tx)
+
+    # Verify all setup transactions have SETUP phase
+    for tx in setup_txs:
+        assert tx.test_phase == "setup"
+
+    # Verify all execution transactions have EXECUTION phase
+    for tx in exec_txs:
+        assert tx.test_phase == "execution"
+
+
+def test_phase_reset() -> None:
+    """Test that reset() restores default phase."""
+    # Change phase
+    with TestPhaseManager.setup():
+        pass
+
+    # Manually set to SETUP
+    TestPhaseManager._current_phase = TestPhase.SETUP
+    assert TestPhaseManager.get_current_phase() == TestPhase.SETUP
+
+    # Reset should restore None
+    TestPhaseManager.reset()
+    assert TestPhaseManager.get_current_phase() is None
+
+
+def test_class_state_shared() -> None:
+    """Test that phase state is shared at class level."""
+    # Phase changes are visible globally since it's class-level state
+    assert TestPhaseManager.get_current_phase() is None
+
+    with TestPhaseManager.setup():
+        # All access to the class sees the same phase
+        assert TestPhaseManager.get_current_phase() == TestPhase.SETUP
+
+        # Transactions created during this context get "setup" phase
+        tx = Transaction(to=Address(0x789), value=75, gas_limit=21000)
+        assert tx.test_phase == "setup"
+
+    # After context, phase returns to None
+    assert TestPhaseManager.get_current_phase() is None

src/ethereum_test_types/transaction_types.py

@@ -38,6 +38,7 @@
 from .account_types import EOA
 from .blob_types import Blob
 from .chain_config_types import ChainConfigDefaults
+from .phase_manager import TestPhaseManager
 from .receipt_types import TransactionReceipt
 from .utils import int_to_bytes, keccak256
 
@@ -292,6 +293,13 @@ class Transaction(
     zero: ClassVar[Literal[0]] = 0
 
     metadata: TransactionTestMetadata | None = Field(None, exclude=True)
+    test_phase: str | None = Field(
+        default_factory=lambda: (
+            phase.value
+            if (phase := TestPhaseManager.get_current_phase()) is not None
+            else "execution"
+        )
+    )
 
     model_config = ConfigDict(validate_assignment=True)
 

tests/benchmark/test_worst_stateful_opcodes.py

@@ -22,6 +22,7 @@
     compute_create2_address,
     compute_create_address,
 )
+from ethereum_test_types import TestPhaseManager
 from ethereum_test_vm import Opcodes as Op
 
 from .helpers import code_loop_precompile_call
@@ -434,23 +435,26 @@ def test_worst_storage_access_warm(
         )
         + Op.RETURN(0, Op.MSIZE)
     )
-    sender_addr = pre.fund_eoa()
-    setup_tx = Transaction(
-        to=None,
-        gas_limit=env.gas_limit,
-        data=creation_code,
-        sender=sender_addr,
-    )
-    blocks.append(Block(txs=[setup_tx]))
+
+    with TestPhaseManager.setup():
+        sender_addr = pre.fund_eoa()
+        setup_tx = Transaction(
+            to=None,
+            gas_limit=env.gas_limit,
+            data=creation_code,
+            sender=sender_addr,
+        )
+        blocks.append(Block(txs=[setup_tx]))
 
     contract_address = compute_create_address(address=sender_addr, nonce=0)
 
-    op_tx = Transaction(
-        to=contract_address,
-        gas_limit=attack_gas_limit,
-        sender=pre.fund_eoa(),
-    )
-    blocks.append(Block(txs=[op_tx]))
+    with TestPhaseManager.execution():
+        op_tx = Transaction(
+            to=contract_address,
+            gas_limit=attack_gas_limit,
+            sender=pre.fund_eoa(),
+        )
+        blocks.append(Block(txs=[op_tx]))
 
     blockchain_test(
         pre=pre,
@@ -468,20 +472,26 @@ def test_worst_blockhash(
     Test running a block with as many blockhash accessing oldest allowed block
     as possible.
     """
-    # Create 256 dummy blocks to fill the blockhash window.
-    blocks = [Block()] * 256
+    blocks = []
 
-    # Always ask for the oldest allowed BLOCKHASH block.
-    execution_code = Op.PUSH1(1) + While(
-        body=Op.POP(Op.BLOCKHASH(Op.DUP1)),
-    )
-    execution_code_address = pre.deploy_contract(code=execution_code)
-    op_tx = Transaction(
-        to=execution_code_address,
-        gas_limit=gas_benchmark_value,
-        sender=pre.fund_eoa(),
-    )
-    blocks.append(Block(txs=[op_tx]))
+    # Setup phase: Create 256 empty blocks
+    with TestPhaseManager.setup():
+        for _ in range(256):
+            blocks.append(Block())
+
+    # Execution phase: Deploy contract and create benchmark transaction
+    with TestPhaseManager.execution():
+        # Always ask for the oldest allowed BLOCKHASH block.
+        execution_code = Op.PUSH1(1) + While(
+            body=Op.POP(Op.BLOCKHASH(Op.DUP1)),
+        )
+        execution_code_address = pre.deploy_contract(code=execution_code)
+        op_tx = Transaction(
+            to=execution_code_address,
+            gas_limit=gas_benchmark_value,
+            sender=pre.fund_eoa(),
+        )
+        blocks.append(Block(txs=[op_tx]))
 
     blockchain_test(
         pre=pre,