feat(execution,specs): Implement BlobsTest execute-only spec type

Update exception message in src/ethereum_test_execution/get_blobs.py

Co-authored-by: danceratopz <[email protected]>

Update description src/ethereum_test_execution/get_blobs.py

Co-authored-by: danceratopz <[email protected]>

Apply suggestions from code review to src/ethereum_test_execution/get_blobs.py

Co-authored-by: danceratopz <[email protected]>

Update docs/CHANGELOG.md for ethereum_test_execution

Co-authored-by: danceratopz <[email protected]>

fix ethereum_test_execution comment

fix ethereum_test_execution tox

fix(execution): Use BlobAndProofV1 and BlobAndProofV2

fix(execution): Rename blob_transaction_test

fix(sepcs): Rename blob_transaction_test

Author: marioevz

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 execute test spec, which allows to write tests that send blob transactions to a running client and verify `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)).

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/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

@@ -18,7 +18,7 @@ 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"
     )

src/ethereum_test_specs/__init__.py

@@ -2,6 +2,7 @@
 
 from .base import BaseTest, TestSpec
 from .base_static import BaseStaticTest
+from .blobs import BlobsTest, BlobsTestFiller, BlobsTestSpec
 from .blockchain import (
     BlockchainTest,
     BlockchainTestFiller,
@@ -22,6 +23,9 @@
 __all__ = (
     "BaseStaticTest",
     "BaseTest",
+    "BlobsTest",
+    "BlobsTestFiller",
+    "BlobsTestSpec",
     "BlockchainTest",
     "BlockchainTestEngineFiller",
     "BlockchainTestEngineSpec",

src/ethereum_test_specs/blobs.py

@@ -0,0 +1,59 @@
+"""Test specification for blob tests."""
+
+from typing import Callable, ClassVar, Generator, List, Optional, Sequence, Type
+
+from ethereum_clis import TransitionTool
+from ethereum_test_base_types import Alloc
+from ethereum_test_execution import BaseExecute, BlobTransaction
+from ethereum_test_fixtures import (
+    BaseFixture,
+    FixtureFormat,
+)
+from ethereum_test_forks import Fork
+from ethereum_test_types import NetworkWrappedTransaction, Transaction
+
+from .base import BaseTest, ExecuteFormat, LabeledExecuteFormat
+
+
+class BlobsTest(BaseTest):
+    """Test specification for blob tests."""
+
+    pre: Alloc
+    txs: List[NetworkWrappedTransaction | Transaction]
+
+    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
+        LabeledExecuteFormat(
+            BlobTransaction,
+            "blob_transaction_test",
+            "A test that executes a blob transaction",
+        ),
+    ]
+
+    def generate(
+        self,
+        *,
+        t8n: TransitionTool,
+        fork: Fork,
+        fixture_format: FixtureFormat,
+        eips: Optional[List[int]] = None,
+    ) -> BaseFixture:
+        """Generate the list of test fixtures."""
+        raise Exception(f"Unknown fixture format: {fixture_format}")
+
+    def execute(
+        self,
+        *,
+        fork: Fork,
+        execute_format: ExecuteFormat,
+        eips: Optional[List[int]] = None,
+    ) -> BaseExecute:
+        """Generate the list of test fixtures."""
+        if execute_format == BlobTransaction:
+            return BlobTransaction(
+                txs=self.txs,
+            )
+        raise Exception(f"Unsupported execute format: {execute_format}")
+
+
+BlobsTestSpec = Callable[[str], Generator[BlobsTest, None, None]]
+BlobsTestFiller = Type[BlobsTest]

src/ethereum_test_tools/__init__.py

@@ -24,6 +24,8 @@
 from ethereum_test_fixtures import BaseFixture, FixtureCollector
 from ethereum_test_specs import (
     BaseTest,
+    BlobsTest,
+    BlobsTestFiller,
     BlockchainTest,
     BlockchainTestFiller,
     EOFStateTest,
@@ -99,6 +101,8 @@
     "BaseFixture",
     "BaseTest",
     "Blob",
+    "BlobsTest",
+    "BlobsTestFiller",
     "Block",
     "BlockchainTest",
     "BlockchainTestFiller",