feat(execute,tests): implement blob_transaction_test execute spec

docs/CHANGELOG.md

@@ -34,6 +34,10 @@ Users can select any of the artifacts depending on their testing needs for their
 
 #### `consume`
 
+#### `execute`
+
+- ✨ Add `blob_transaction_test` execute test spec, which allows tests that send blob transactions to a running client and verifying its `engine_getBlobsVX` endpoint behavior ([#1644](https://github.com/ethereum/execution-spec-tests/pull/1644)).
+
 ### 📋 Misc
 
 - ✨ Added the [EIP checklist template](https://eest.ethereum.org/main/writing_tests/checklist_templates/eip_testing_checklist_template/) that serves as a reference to achieve better coverage when implementing tests for new EIPs ([#1327](https://github.com/ethereum/execution-spec-tests/pull/1327)).
@@ -46,6 +50,7 @@ Users can select any of the artifacts depending on their testing needs for their
 
 - 🔀 Refactored `BLOBHASH` opcode context tests to use the `pre_alloc` plugin in order to avoid contract and EOA address collisions ([#1637](https://github.com/ethereum/execution-spec-tests/pull/1637)).
 - 🔀 Refactored `SELFDESTRUCT` opcode collision tests to use the `pre_alloc` plugin in order to avoid contract and EOA address collisions ([#1643](https://github.com/ethereum/execution-spec-tests/pull/1643)).
+- ✨ EIP-7594: Sanity test cases to send blob transactions and verify `engine_getBlobsVX` using the `execute` command ([#1644](https://github.com/ethereum/execution-spec-tests/pull/1644)).
 
 ## [v4.5.0](https://github.com/ethereum/execution-spec-tests/releases/tag/v4.5.0) - 2025-05-14
 

src/ethereum_test_execution/__init__.py

@@ -1,11 +1,13 @@
 """Ethereum test execution package."""
 
 from .base import BaseExecute, ExecuteFormat, LabeledExecuteFormat
+from .blob_transaction import BlobTransaction
 from .transaction_post import TransactionPost
 
 __all__ = [
     "BaseExecute",
     "ExecuteFormat",
+    "BlobTransaction",
     "LabeledExecuteFormat",
     "TransactionPost",
 ]

src/ethereum_test_execution/base.py

@@ -6,7 +6,8 @@
 from pydantic import PlainSerializer, PlainValidator
 
 from ethereum_test_base_types import CamelModel
-from ethereum_test_rpc import EthRPC
+from ethereum_test_forks import Fork
+from ethereum_test_rpc import EngineRPC, EthRPC
 
 
 class BaseExecute(CamelModel):
@@ -18,6 +19,7 @@ class BaseExecute(CamelModel):
     # Execute format properties
     format_name: ClassVar[str] = ""
     description: ClassVar[str] = "Unknown execute format; it has not been set."
+    requires_engine_rpc: ClassVar[bool] = False
 
     @classmethod
     def __pydantic_init_subclass__(cls, **kwargs):
@@ -30,7 +32,7 @@ def __pydantic_init_subclass__(cls, **kwargs):
             BaseExecute.formats[cls.format_name] = cls
 
     @abstractmethod
-    def execute(self, eth_rpc: EthRPC):
+    def execute(self, fork: Fork, eth_rpc: EthRPC, engine_rpc: EngineRPC | None):
         """Execute the format."""
         pass
 
@@ -71,6 +73,11 @@ def format_name(self) -> str:
         """Get the execute format name."""
         return self.format.format_name
 
+    @property
+    def requires_engine_rpc(self) -> bool:
+        """Get the requires engine RPC flag."""
+        return self.format.requires_engine_rpc
+
     def __eq__(self, other: Any) -> bool:
         """
         Check if two labeled execute formats are equal.

src/ethereum_test_execution/blob_transaction.py

@@ -0,0 +1,103 @@
+"""Test execution format to get blobs from the execution client."""
+
+from typing import ClassVar, Dict, List
+
+from ethereum_test_base_types import Hash
+from ethereum_test_forks import Fork
+from ethereum_test_rpc import BlobAndProofV1, BlobAndProofV2, EngineRPC, EthRPC
+from ethereum_test_types import NetworkWrappedTransaction, Transaction
+
+from .base import BaseExecute
+
+
+def versioned_hashes_with_blobs_and_proofs(
+    tx: NetworkWrappedTransaction,
+) -> Dict[Hash, BlobAndProofV1 | BlobAndProofV2]:
+    """
+    Return a dictionary of versioned hashes with their corresponding blobs and
+    proofs.
+    """
+    versioned_hashes: Dict[Hash, BlobAndProofV1 | BlobAndProofV2] = {}
+    for blob in tx.blobs:
+        versioned_hash = blob.versioned_hash()
+        if blob.kzg_proof is not None:
+            versioned_hashes[versioned_hash] = BlobAndProofV1(blob=blob.data, proof=blob.kzg_proof)
+        elif blob.kzg_cell_proofs is not None:
+            versioned_hashes[versioned_hash] = BlobAndProofV2(
+                blob=blob.data, proofs=blob.kzg_cell_proofs
+            )
+        else:
+            raise ValueError(
+                f"Blob with versioned hash {versioned_hash.hex()} requires either kzg_proof "
+                "or kzg_cell_proofs, but both are None"
+            )
+
+    return versioned_hashes
+
+
+class BlobTransaction(BaseExecute):
+    """
+    Represents a test execution format to send blob transactions to the client and then
+    use `engine_getBlobsV*` end points to validate the proofs generated by the execution client.
+    """
+
+    format_name: ClassVar[str] = "blob_transaction_test"
+    description: ClassVar[str] = (
+        "Send blob transactions to the execution client and validate their availability via "
+        "`engine_getBlobsV*`"
+    )
+    requires_engine_rpc: ClassVar[bool] = True
+
+    txs: List[NetworkWrappedTransaction | Transaction]
+
+    def execute(self, fork: Fork, eth_rpc: EthRPC, engine_rpc: EngineRPC | None):
+        """Execute the format."""
+        assert engine_rpc is not None, "Engine RPC is required for this format."
+        versioned_hashes: Dict[Hash, BlobAndProofV1 | BlobAndProofV2] = {}
+        sent_txs: List[Transaction] = []
+        for tx in self.txs:
+            if isinstance(tx, NetworkWrappedTransaction):
+                tx.tx = tx.tx.with_signature_and_sender()
+                sent_txs.append(tx.tx)
+                expected_hash = tx.tx.hash
+                versioned_hashes.update(versioned_hashes_with_blobs_and_proofs(tx))
+            else:
+                tx = tx.with_signature_and_sender()
+                sent_txs.append(tx)
+                expected_hash = tx.hash
+            received_hash = eth_rpc.send_raw_transaction(tx.rlp())
+            assert expected_hash == received_hash, (
+                f"Expected hash {expected_hash} does not match received hash {received_hash}."
+            )
+        version = fork.engine_get_blobs_version()
+        assert version is not None, "Engine get blobs version is not supported by the fork."
+        blob_response = engine_rpc.get_blobs(list(versioned_hashes.keys()), version=version)
+        local_blobs_and_proofs = list(versioned_hashes.values())
+        if len(blob_response) != len(local_blobs_and_proofs):
+            raise ValueError(
+                f"Expected {len(local_blobs_and_proofs)} blobs and proofs, "
+                f"got {len(blob_response)}."
+            )
+        for expected_blob, received_blob in zip(
+            local_blobs_and_proofs, blob_response.root, strict=False
+        ):
+            if received_blob is None:
+                raise ValueError("Received blob is empty.")
+            if isinstance(expected_blob, BlobAndProofV1):
+                if not isinstance(received_blob, BlobAndProofV1):
+                    raise ValueError("Received blob is not a BlobAndProofV1.")
+                if expected_blob.blob != received_blob.blob:
+                    raise ValueError("Blob mismatch.")
+                if expected_blob.proof != received_blob.proof:
+                    raise ValueError("Proof mismatch.")
+            elif isinstance(expected_blob, BlobAndProofV2):
+                if not isinstance(received_blob, BlobAndProofV2):
+                    raise ValueError("Received blob is not a BlobAndProofV2.")
+                if expected_blob.blob != received_blob.blob:
+                    raise ValueError("Blob mismatch.")
+                if expected_blob.proofs != received_blob.proofs:
+                    raise ValueError("Proofs mismatch.")
+            else:
+                raise ValueError(f"Unexpected blob type: {type(expected_blob)}")
+
+        eth_rpc.wait_for_transactions(sent_txs)

src/ethereum_test_execution/transaction_post.py

@@ -5,7 +5,8 @@
 import pytest
 
 from ethereum_test_base_types import Alloc, Hash
-from ethereum_test_rpc import EthRPC, SendTransactionExceptionError
+from ethereum_test_forks import Fork
+from ethereum_test_rpc import EngineRPC, EthRPC, SendTransactionExceptionError
 from ethereum_test_types import Transaction
 
 from .base import BaseExecute
@@ -17,12 +18,12 @@ class TransactionPost(BaseExecute):
     blocks: List[List[Transaction]]
     post: Alloc
 
-    format_name: ClassVar[str] = "transaction_post"
+    format_name: ClassVar[str] = "transaction_post_test"
     description: ClassVar[str] = (
         "Simple transaction sending, then post-check after all transactions are included"
     )
 
-    def execute(self, eth_rpc: EthRPC):
+    def execute(self, fork: Fork, eth_rpc: EthRPC, engine_rpc: EngineRPC | None):
         """Execute the format."""
         assert not any(tx.ty == 3 for block in self.blocks for tx in block), (
             "Transaction type 3 is not supported in execute mode."

src/ethereum_test_forks/base_fork.py

@@ -452,7 +452,16 @@ def engine_forkchoice_updated_version(
     def engine_get_payload_version(
         cls, block_number: int = 0, timestamp: int = 0
     ) -> Optional[int]:
-        """Return `None` if the forks canonical chain cannot be set using the forkchoice method."""
+        """
+        Return `None` if the forks canonical chain cannot build a payload using the engine
+        API.
+        """
+        pass
+
+    @classmethod
+    @abstractmethod
+    def engine_get_blobs_version(cls, block_number: int = 0, timestamp: int = 0) -> Optional[int]:
+        """Return `None` if the fork does not support the engine get blobs version."""
         pass
 
     # EVM information abstract methods

src/ethereum_test_forks/forks/forks.py

@@ -331,6 +331,11 @@ def engine_get_payload_version(
         """At genesis, payloads cannot be retrieved through the engine API."""
         return cls.engine_new_payload_version(block_number, timestamp)
 
+    @classmethod
+    def engine_get_blobs_version(cls, block_number: int = 0, timestamp: int = 0) -> Optional[int]:
+        """At genesis, blobs cannot be retrieved through the engine API."""
+        return None
+
     @classmethod
     def get_reward(cls, block_number: int = 0, timestamp: int = 0) -> int:
         """
@@ -1017,6 +1022,11 @@ def engine_new_payload_version(
         """From Cancun, new payload calls must use version 3."""
         return 3
 
+    @classmethod
+    def engine_get_blobs_version(cls, block_number: int = 0, timestamp: int = 0) -> Optional[int]:
+        """At Cancun, the engine get blobs version is 1."""
+        return 1
+
     @classmethod
     def engine_new_payload_blob_hashes(cls, block_number: int = 0, timestamp: int = 0) -> bool:
         """From Cancun, payloads must have blob hashes."""
@@ -1302,6 +1312,11 @@ def engine_get_payload_version(
         """From Osaka, get payload calls must use version 5."""
         return 5
 
+    @classmethod
+    def engine_get_blobs_version(cls, block_number: int = 0, timestamp: int = 0) -> Optional[int]:
+        """At Osaka, the engine get blobs version is 2."""
+        return 2
+
     @classmethod
     def is_deployed(cls) -> bool:
         """

src/ethereum_test_rpc/__init__.py

@@ -1,8 +1,11 @@
 """JSON-RPC methods and helper functions for EEST consume based hive simulators."""
 
 from .rpc import BlockNumberType, DebugRPC, EngineRPC, EthRPC, SendTransactionExceptionError
+from .types import BlobAndProofV1, BlobAndProofV2
 
 __all__ = [
+    "BlobAndProofV1",
+    "BlobAndProofV2",
     "BlockNumberType",
     "DebugRPC",
     "EngineRPC",

src/ethereum_test_rpc/rpc.py

@@ -348,15 +348,15 @@ def get_payload(
 
     def get_blobs(
         self,
-        params: List[Hash],
+        versioned_hashes: List[Hash],
         *,
         version: int,
     ) -> GetBlobsResponse:
         """`engine_getBlobsVX`: Retrieves blobs from an execution layers tx pool."""
         return GetBlobsResponse.model_validate(
             self.post_request(
                 f"getBlobsV{version}",
-                *[to_json(param) for param in params],
+                [f"{h}" for h in versioned_hashes],
             ),
             context=self.response_validation_context,
         )

src/ethereum_test_rpc/types.py

@@ -1,11 +1,19 @@
 """Types used in the RPC module for `eth` and `engine` namespaces' requests."""
 
 from enum import Enum
+from hashlib import sha256
 from typing import Annotated, Any, List, Union
 
 from pydantic import AliasChoices, Field, model_validator
 
-from ethereum_test_base_types import Address, Bytes, CamelModel, Hash, HexNumber
+from ethereum_test_base_types import (
+    Address,
+    Bytes,
+    CamelModel,
+    EthereumTestRootModel,
+    Hash,
+    HexNumber,
+)
 from ethereum_test_exceptions import (
     BlockException,
     ExceptionMapperValidator,
@@ -131,18 +139,28 @@ class BlobsBundle(CamelModel):
     proofs: List[Bytes]
     blobs: List[Bytes]
 
-    def blob_versioned_hashes(self) -> List[Hash]:
+    def blob_versioned_hashes(self, versioned_hash_version: int = 1) -> List[Hash]:
         """Return versioned hashes of the blobs."""
-        return [Hash(b"\1" + commitment[1:]) for commitment in self.commitments]
+        versioned_hashes: List[Hash] = []
+        for commitment in self.commitments:
+            commitment_hash = sha256(commitment).digest()
+            versioned_hash = Hash(bytes([versioned_hash_version]) + commitment_hash[1:])
+            versioned_hashes.append(versioned_hash)
+        return versioned_hashes
 
 
-class BlobAndProof(CamelModel):
-    """Represents a blob and proof structure."""
+class BlobAndProofV1(CamelModel):
+    """Represents a blob and single-proof structure."""
 
     blob: Bytes
-    proofs: List[Bytes] | None = None  # >= Osaka (V2)
+    proof: Bytes
 
-    proof: Bytes | None = None  # <= Prague (V1)
+
+class BlobAndProofV2(CamelModel):
+    """Represents a blob and proof structure."""
+
+    blob: Bytes
+    proofs: List[Bytes]
 
 
 class GetPayloadResponse(CamelModel):
@@ -153,7 +171,15 @@ class GetPayloadResponse(CamelModel):
     execution_requests: List[Bytes] | None = None
 
 
-class GetBlobsResponse(CamelModel):
+class GetBlobsResponse(EthereumTestRootModel):
     """Represents the response of a get blobs request."""
 
-    result: List[BlobAndProof | None]
+    root: List[BlobAndProofV1 | BlobAndProofV2 | None]
+
+    def __len__(self) -> int:
+        """Return the number of blobs in the response."""
+        return len(self.root)
+
+    def __getitem__(self, index: int) -> BlobAndProofV1 | BlobAndProofV2 | None:
+        """Return the blob at the given index."""
+        return self.root[index]