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

Author: marioevz

src/ethereum_test_execution/__init__.py

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

src/ethereum_test_execution/get_blobs.py

@@ -0,0 +1,78 @@
+"""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 BlobAndProof, EngineRPC, EthRPC
+from ethereum_test_types import NetworkWrappedTransaction, Transaction
+
+from .base import BaseExecute
+
+
+def versioned_hashes_with_blobs_and_proofs(
+    tx: NetworkWrappedTransaction,
+) -> Dict[Hash, BlobAndProof]:
+    """
+    Return a dictionary of versioned hashes with their corresponding blobs and
+    proofs.
+    """
+    versioned_hashes: Dict[Hash, BlobAndProof] = {}
+    for blob in tx.blobs:
+        versioned_hash = blob.versioned_hash()
+        if blob.kzg_proof is not None:
+            versioned_hashes[versioned_hash] = BlobAndProof(blob=blob.data, proof=blob.kzg_proof)
+        elif blob.kzg_cell_proofs is not None:
+            versioned_hashes[versioned_hash] = BlobAndProof(
+                blob=blob.data, proofs=blob.kzg_cell_proofs
+            )
+        else:
+            raise ValueError("Invalid blob: missing kzg proof or cell proofs")
+    return versioned_hashes
+
+
+class GetBlobs(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] = "get_blobs"
+    description: ClassVar[str] = "Get blobs from the execution client"
+    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, BlobAndProof] = {}
+        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."
+        resp = engine_rpc.get_blobs(list(versioned_hashes.keys()), version=version)
+        local_blobs_and_proofs = list(versioned_hashes.values())
+        assert len(resp) == len(local_blobs_and_proofs), (
+            f"Expected {len(local_blobs_and_proofs)} blobs and proofs, got {len(resp)}."
+        )
+        for local, remote in zip(local_blobs_and_proofs, resp.root, strict=False):
+            if remote is None:
+                raise ValueError("Remote blob is None.")
+            assert local.blob == remote.blob, "Blob mismatch."
+            assert local.proofs == remote.proofs, "Proofs mismatch."
+
+        eth_rpc.wait_for_transactions(sent_txs)

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, GetBlobs
+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(
+            GetBlobs,
+            "get_blobs",
+            "A test that executes a get blobs request",
+        ),
+    ]
+
+    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 == GetBlobs:
+            return GetBlobs(
+                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",