feat(fw): pydantic rebase, post vkt verify boilerplate, vkt subcommand.

src/ethereum_test_forks/base_fork.py

@@ -3,7 +3,7 @@
 """
 
 from abc import ABC, ABCMeta, abstractmethod
-from typing import Any, ClassVar, Dict, List, Optional, Protocol, Type
+from typing import Any, ClassVar, List, Mapping, Optional, Protocol, Type
 
 from semver import Version
 
@@ -184,9 +184,7 @@ def precompiles(cls, block_number: int = 0, timestamp: int = 0) -> List[int]:
     @classmethod
     @prefer_transition_to_method
     @abstractmethod
-    def pre_allocation(
-        cls, block_number: int = 0, timestamp: int = 0
-    ) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation(cls) -> Mapping:
         """
         Returns required pre-allocation of accounts for any kind of test.
 
@@ -198,9 +196,7 @@ def pre_allocation(
     @classmethod
     @prefer_transition_to_method
     @abstractmethod
-    def pre_allocation_blockchain(
-        cls, block_number: int = 0, timestamp: int = 0
-    ) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation_blockchain(cls) -> Mapping:
         """
         Returns required pre-allocation of accounts for any blockchain tests.
 

src/ethereum_test_forks/forks/constants.py

@@ -2,7 +2,7 @@
 Constant values used by the forks.
 """
 
-from typing import Dict, Generator, Iterator, Tuple
+from typing import Dict, Generator, Iterator, Mapping, Tuple
 
 from Crypto.Hash import SHA256
 
@@ -47,7 +47,7 @@ def account_generator(
         }
 
 
-VERKLE_PRE_ALLOCATION: Dict[int, Dict[str, str | int | Dict[int, int]]] = {
+VERKLE_PRE_ALLOCATION: Mapping = {
     addr: account
     for addr, account in account_generator(seed=seed_generator(0), max_accounts=MAX_ACCOUNTS)
 }

src/ethereum_test_forks/forks/forks.py

@@ -2,7 +2,7 @@
 All Ethereum fork class definitions.
 """
 
-from typing import Dict, List, Optional
+from typing import List, Mapping, Optional
 
 from semver import Version
 
@@ -151,7 +151,7 @@ def precompiles(cls, block_number: int = 0, timestamp: int = 0) -> List[int]:
         return []
 
     @classmethod
-    def pre_allocation(cls) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation(cls) -> Mapping:
         """
         Returns whether the fork expects pre-allocation of accounts
 
@@ -160,7 +160,7 @@ def pre_allocation(cls) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
         return {}
 
     @classmethod
-    def pre_allocation_blockchain(cls) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation_blockchain(cls) -> Mapping:
         """
         Returns whether the fork expects pre-allocation of accounts
 
@@ -434,14 +434,12 @@ def precompiles(cls, block_number: int = 0, timestamp: int = 0) -> List[int]:
         return [0xA] + super(Cancun, cls).precompiles(block_number, timestamp)
 
     @classmethod
-    def pre_allocation_blockchain(
-        cls, block_number: int = 0, timestamp: int = 0
-    ) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation_blockchain(cls) -> Mapping:
         """
         Cancun requires pre-allocation of the beacon root contract for EIP-4788 on blockchain
         type tests
         """
-        new_allocation: Dict[int, Dict[str, str | int | Dict[int, int]]] = {
+        new_allocation = {
             0x000F3DF6D732807EF1319FB7B8BB8522D0BEAC02: {
                 "nonce": 1,
                 "code": "0x3373fffffffffffffffffffffffffffffffffffffffe14604d57602036146024575f5f"

src/ethereum_test_forks/forks/transition.py

@@ -2,7 +2,7 @@
 List of all transition fork definitions.
 """
 
-from typing import Dict
+from typing import Mapping
 
 from ..transition_base_fork import transition_fork
 from .constants import VERKLE_PRE_ALLOCATION
@@ -58,12 +58,10 @@ class ShanghaiToPragueVerkleTransition(Shanghai):
     """
 
     @classmethod
-    def pre_allocation(
-        cls, block_number: int = 0, timestamp: int = 0
-    ) -> Dict[int, Dict[str, str | int | Dict[int, int]]]:
+    def pre_allocation(cls) -> Mapping:
         """
         Pre-allocates a big state full of accounts and storage to test the MPT to Verkle tree
         conversion.
         """
 
-        return VERKLE_PRE_ALLOCATION | super(Shanghai, cls).pre_allocation(block_number, timestamp)
+        return VERKLE_PRE_ALLOCATION | super(Shanghai, cls).pre_allocation()

src/ethereum_test_tools/common/__init__.py

@@ -42,6 +42,7 @@
     Removable,
     Storage,
     Transaction,
+    VerkleTree,
     Withdrawal,
 )
 
@@ -70,6 +71,7 @@
     "TestPrivateKey",
     "TestPrivateKey2",
     "Transaction",
+    "VerkleTree",
     "Withdrawal",
     "ZeroPaddedHexNumber",
     "add_kzg_version",

src/ethereum_test_tools/common/types.py

@@ -789,7 +789,7 @@ def set_fork_requirements(self, fork: Fork) -> "Environment":
             updated_values["parent_beacon_block_root"] = 0
 
         if fork.environment_verkle_conversion_starts(number, timestamp):
-            if updated_values["verkle_conversion_ended"] is None:
+            if self.verkle_conversion_ended:
                 # Conversion is marked as completed if this is the genesis block, or we are
                 # past the conversion end fork.
                 updated_values["verkle_conversion_ended"] = (
@@ -798,29 +798,25 @@ def set_fork_requirements(self, fork: Fork) -> "Environment":
 
         return self.copy(**updated_values)
 
-    def update_from_result(self, transition_tool_result: Dict[str, Any]) -> "Environment":
+    def update_from_result(self, result: "Result") -> "Environment":
         """
         Updates the environment with the result of a transition tool execution.
         """
-        if "currentConversionAddress" in transition_tool_result:
-            self.verkle_conversion_address = transition_tool_result["currentConversionAddress"]
-        if "currentConversionSlotHash" in transition_tool_result:
-            self.verkle_conversion_slot_hash = transition_tool_result["currentConversionSlotHash"]
-        if "currentConversionStarted" in transition_tool_result:
-            conversion_started = transition_tool_result["currentConversionStarted"]
-            assert conversion_started is not None and isinstance(conversion_started, bool)
-            self.verkle_conversion_started = conversion_started
-        if "currentConversionEnded" in transition_tool_result:
-            conversion_ended = transition_tool_result["currentConversionEnded"]
-            assert conversion_ended is not None and isinstance(conversion_ended, bool)
-            self.verkle_conversion_ended = transition_tool_result["currentConversionEnded"]
-        if "currentConversionStorageProcessed" in transition_tool_result:
-            conversion_storage_processed = transition_tool_result[
-                "currentConversionStorageProcessed"
-            ]
-            assert conversion_storage_processed is not None and isinstance(
-                conversion_storage_processed, bool
-            )
+        if result.conversion_address:
+            self.verkle_conversion_address = result.conversion_address
+        if result.conversion_slot_hash:
+            self.verkle_conversion_slot_hash = result.conversion_slot_hash
+        if result.conversion_started:
+            conversion_started = result.conversion_started
+            assert isinstance(conversion_started, bool)
+            self.verkle_conversion_started = result.conversion_started
+        if result.conversion_ended:
+            conversion_ended = result.conversion_ended
+            assert isinstance(conversion_ended, bool)
+            self.verkle_conversion_ended = result.conversion_ended
+        if result.conversion_storage_processed:
+            conversion_storage_processed = result.conversion_storage_processed
+            assert isinstance(conversion_storage_processed, bool)
             self.verkle_conversion_storage_processed = conversion_storage_processed
         return self
 
@@ -1302,8 +1298,17 @@ class Result(CamelModel):
     excess_blob_gas: HexNumber | None = Field(None, alias="currentExcessBlobGas")
     blob_gas_used: HexNumber | None = None
 
+    # Verkle tree related: TODO
+    conversion_address: Address | None = Field(None, alias="currentConversionAddress")
+    conversion_slot_hash: Hash | None = Field(None, alias="currentConversionSlotHash")
+    conversion_started: bool | None = Field(None, alias="currentConversionStarted")
+    conversion_ended: bool | None = Field(None, alias="currentConversionEnded")
+    conversion_storage_processed: bool | None = Field(
+        None, alias="currentConversionStorageProcessed"
+    )
+
 
-class VerkleTree(RootModel):
+class VerkleTree(RootModel[Dict[str, str | None]]):
     # TODO: Implement VerkleTree model
     root: Dict[str, str | None] = Field(default_factory=dict, validate_default=True)
 

src/ethereum_test_tools/spec/blockchain/blockchain_test.py

@@ -10,7 +10,7 @@
 from ethereum_test_forks import Fork, Prague
 from evm_transition_tool import FixtureFormats, TransitionTool
 
-from ...common import Alloc, EmptyTrieRoot, Environment, Hash, Transaction, Withdrawal
+from ...common import Alloc, EmptyTrieRoot, Environment, Hash, Transaction, VerkleTree, Withdrawal
 from ...common.constants import EmptyOmmersRoot
 from ...common.json import to_json
 from ...common.types import TransitionToolOutput
@@ -155,9 +155,9 @@ def generate_block_data(
         block: Block,
         previous_env: Environment,
         previous_alloc: Alloc,
-        previous_vkt: Optional[Alloc] = None,
+        previous_vkt: Optional[VerkleTree] = None,
         eips: Optional[List[int]] = None,
-    ) -> Tuple[FixtureHeader, List[Transaction], Alloc, Optional[Alloc], Environment]:
+    ) -> Tuple[FixtureHeader, List[Transaction], Alloc, Optional[VerkleTree], Environment]:
         """
         Generate common block data for both make_fixture and make_hive_fixture.
         """
@@ -192,7 +192,7 @@ def generate_block_data(
                 fork_name=fork.transition_tool_name(
                     block_number=env.number, timestamp=env.timestamp
                 ),
-                vkt=previous_vkt,
+                vkt=to_json(previous_vkt) if previous_vkt is not None else None,
                 chain_id=self.chain_id,
                 reward=fork.get_reward(env.number, env.timestamp),
                 eips=eips,
@@ -253,21 +253,13 @@ def generate_block_data(
             header = header.join(block.rlp_modifier)
 
         env.update_from_result(transition_tool_output.result)
-        rlp, header.hash = header.build(
-            txs=txs,
-            ommers=[],
-            withdrawals=env.withdrawals,
-        )
-
-        env.update_from_result(transition_tool_output.result)
-
         if fork.fork_at(env.number, env.timestamp) >= Prague:
             if env.verkle_conversion_ended:
-                transition_tool_output.alloc = {}
+                transition_tool_output.alloc = Alloc()
             else:
                 transition_tool_output.alloc = previous_alloc
 
-        return header, rlp, txs, transition_tool_output.alloc, transition_tool_output.vkt, env
+        return header, txs, transition_tool_output.alloc, transition_tool_output.vkt, env
 
     def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
         """
@@ -281,10 +273,14 @@ def network_info(self, fork: Fork, eips: Optional[List[int]] = None):
 
     def verify_post_state(self, *, t8n, alloc: Alloc, vkt=None):
         """
-        Verifies the post alloc after all block/s or payload/s are generated.
+        Verifies the post state after all block/s or payload/s are generated.
         """
         try:
-            self.post.verify_post_alloc(alloc)
+            if vkt is not None:
+                # self.post.verify_post_vkt(vkt)  # TODO: implement this method
+                print("Skipping VKT verification for now.")
+            else:
+                self.post.verify_post_alloc(alloc)
         except Exception as e:
             print_traces(t8n.get_traces())
             raise e
@@ -312,7 +308,7 @@ def make_fixture(
                 # 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, rlp, txs, new_alloc, new_vkt, new_env = self.generate_block_data(
+                header, txs, new_alloc, new_vkt, new_env = self.generate_block_data(
                     t8n=t8n,
                     fork=fork,
                     block=block,
@@ -422,7 +418,7 @@ def make_hive_fixture(
         ), "A hive fixture was requested but no forkchoice update is defined. The framework should"
         " never try to execute this test case."
 
-        self.verify_post_state(t8n, alloc, vkt)
+        self.verify_post_state(t8n=t8n, alloc=alloc, vkt=vkt)
 
         sync_payload: Optional[FixtureEngineNewPayload] = None
         if self.verify_sync:
@@ -434,12 +430,13 @@ def make_hive_fixture(
             # 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, _, _, _ = self.generate_block_data(
+            sync_header, _, _, _, _ = self.generate_block_data(
                 t8n=t8n,
                 fork=fork,
                 block=Block(),
                 previous_env=env,
                 previous_alloc=alloc,
+                previous_vkt=vkt,
                 eips=eips,
             )
             sync_payload = FixtureEngineNewPayload.from_fixture_header(

src/evm_transition_tool/geth.py

@@ -27,6 +27,7 @@ class GethTransitionTool(TransitionTool):
     t8n_subcommand: Optional[str] = "t8n"
     statetest_subcommand: Optional[str] = "statetest"
     blocktest_subcommand: Optional[str] = "blocktest"
+    verkle_subcommand: Optional[str] = "verkle"
 
     binary: Path
     cached_version: Optional[str] = None
@@ -125,3 +126,30 @@ def verify_fixture(
                 f"Failed to verify fixture via: '{' '.join(command)}'. "
                 f"Error: '{result.stderr.decode()}'"
             )
+
+    def verkle_tree_key(self, account: str, storage_slot: Optional[str] = None) -> str:
+        """
+        Returns the verkle tree key for the input account using the verkle subcommand.
+        Optionally the key for the storage slot if specified.
+        """
+        command = [
+            str(self.binary),
+            str(self.t8n_subcommand),
+            str(self.verkle_subcommand),
+            str(account),
+        ]
+        if storage_slot:
+            command.append(storage_slot)
+
+        result = subprocess.run(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+
+        if result.returncode != 0:
+            raise Exception(
+                f"Failed to run verkle subcommand: '{' '.join(command)}'. "
+                f"Error: '{result.stderr.decode()}'"
+            )
+        return result.stdout.decode().strip()  # strip the newline character

src/evm_transition_tool/transition_tool.py

@@ -118,6 +118,7 @@ class TransitionTool:
     blocktest_subcommand: Optional[str] = None
     cached_version: Optional[str] = None
     t8n_use_stream: bool = True
+    verkle_subcommand: Optional[str] = None
 
     # Abstract methods that each tool must implement
 

whitelist.txt

@@ -177,6 +177,7 @@ metaclass
 Misspelled words:
 mkdocs
 mkdocstrings
+mpt
 mypy
 namespace
 nav