refactor: update test phase manager instance model

Author: LouisTsai-Csie

src/ethereum_test_specs/blockchain.py

@@ -440,7 +440,7 @@ class BlockchainTest(BaseTest):
     genesis_environment: Environment = Field(default_factory=Environment)
     chain_id: int = 1
     exclude_full_post_state_in_output: bool = False
-    test_phase_manager: TestPhaseManager | None = None
+    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.
@@ -742,7 +742,6 @@ def _get_test_phase_blocks(self) -> List[Block]:
             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(
@@ -760,9 +759,6 @@ def make_fixture(
         head = genesis.header.block_hash
         invalid_blocks = 0
 
-        if self.test_phase_manager is not None:
-            self.blocks.extend(self._get_test_phase_blocks())
-
         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.
@@ -827,9 +823,7 @@ def make_hive_fixture(
         head_hash = genesis.header.block_hash
         invalid_blocks = 0
 
-        if self.test_phase_manager is not None:
-            self.blocks.extend(self._get_test_phase_blocks())
-
+        built_block = None
         for i, block in enumerate(self.blocks):
             built_block = self.generate_block_data(
                 t8n=t8n,
@@ -852,6 +846,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
         )
@@ -931,6 +929,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,

src/ethereum_test_types/phase_manager.py

@@ -1,11 +1,14 @@
 """Test phase management for Ethereum tests."""
 
 from contextlib import contextmanager
-from contextvars import ContextVar
 from enum import Enum
-from typing import Iterator, Optional
+from typing import Any, Iterator, List, Optional
 
-from pydantic import BaseModel
+from pydantic import GetCoreSchemaHandler
+from pydantic_core.core_schema import (
+    PlainValidatorFunctionSchema,
+    no_info_plain_validator_function,
+)
 
 
 class TestPhase(Enum):
@@ -15,52 +18,44 @@ class TestPhase(Enum):
     EXECUTION = "execution"
 
 
-_current_phase: ContextVar[Optional[TestPhase]] = ContextVar("test_phase", default=TestPhase.SETUP)
-
-
-class TestPhaseManager(BaseModel):
+class TestPhaseManager:
     """
     Manages test phases and collects transactions and blocks by phase.
     This class provides a mechanism for tracking "setup" and "execution" phases,
     Only supports "setup" and "execution" phases now.
     """
 
-    model_config = {"arbitrary_types_allowed": True}
-
-    setup_transactions: list = []
-    setup_blocks: list = []
-    execution_transactions: list = []
-    execution_blocks: list = []
-
-    def __init__(self, **data):
+    def __init__(self, *args, **kwargs):
         """Initialize the TestPhaseManager with empty transaction and block lists."""
-        super().__init__(**data)
-        self.setup_transactions = []
-        self.setup_blocks = []
-        self.execution_transactions = []
-        self.execution_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."""
-        token = _current_phase.set(TestPhase.SETUP)
+        old_phase = self._current_phase
+        self._current_phase = TestPhase.SETUP
         try:
             yield self
         finally:
-            _current_phase.reset(token)
+            self._current_phase = old_phase
 
     @contextmanager
     def execution(self) -> Iterator["TestPhaseManager"]:
         """Context manager for the execution phase of a test."""
-        token = _current_phase.set(TestPhase.EXECUTION)
+        old_phase = self._current_phase
+        self._current_phase = TestPhase.EXECUTION
         try:
             yield self
         finally:
-            _current_phase.reset(token)
+            self._current_phase = old_phase
 
     def add_transaction(self, tx) -> None:
         """Add a transaction to the current phase."""
-        current_phase = _current_phase.get()
+        current_phase = self._current_phase
         tx.test_phase = current_phase
 
         if current_phase == TestPhase.EXECUTION:
@@ -70,7 +65,7 @@ def add_transaction(self, tx) -> None:
 
     def add_block(self, block) -> None:
         """Add a block to the current phase."""
-        current_phase = _current_phase.get()
+        current_phase = self._current_phase
         for tx in block.txs:
             tx.test_phase = current_phase
 
@@ -81,4 +76,20 @@ def add_block(self, block) -> None:
 
     def get_current_phase(self) -> Optional[TestPhase]:
         """Get the current test phase."""
-        return _current_phase.get()
+        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,
+        )

src/pytest_plugins/shared/transaction_fixtures.py

@@ -162,13 +162,7 @@ def typed_transaction(request, fork):
         ) from e
 
 
-@pytest.fixture
-def phase_manager():
-    """Fixture providing access to a test phase manager."""
-    return TestPhaseManager()
-
-
-@pytest.fixture
+@pytest.fixture(scope="function")
 def test_phase_manager():
-    """Fixture providing access to a test phase manager for blockchain tests."""
+    """Fixture providing access to a test phase manager."""
     return TestPhaseManager()

tests/benchmark/test_worst_stateful_opcodes.py

@@ -454,13 +454,13 @@ def test_worst_storage_access_warm(
 def test_worst_blockhash(
     blockchain_test: BlockchainTestFiller,
     pre: Alloc,
-    phase_manager: TestPhaseManager,
+    test_phase_manager: TestPhaseManager,
     gas_benchmark_value: int,
 ):
     """Test running a block with as many blockhash accessing oldest allowed block as possible."""
-    with phase_manager.setup():
+    with test_phase_manager.setup():
         for _ in range(256):
-            phase_manager.add_block(Block())
+            test_phase_manager.add_block(Block())
 
     # Always ask for the oldest allowed BLOCKHASH block.
     execution_code = Op.PUSH1(1) + While(
@@ -473,13 +473,13 @@ def test_worst_blockhash(
         sender=pre.fund_eoa(),
     )
 
-    with phase_manager.execution():
-        phase_manager.add_transaction(op_tx)
+    with test_phase_manager.execution():
+        test_phase_manager.add_transaction(op_tx)
 
     blockchain_test(
         pre=pre,
         post={},
-        test_phase_manager=phase_manager,
+        test_phase_manager=test_phase_manager,
     )