unit test improvements for transition forks, wip

Author: felix314159

src/ethereum_test_types/blob_types.py

@@ -13,18 +13,30 @@
 
 from ethereum_test_base_types.base_types import Bytes, Hash
 from ethereum_test_base_types.pydantic import CamelModel
-from ethereum_test_forks import (
-    Cancun,
-    Fork,
-    Osaka,
-    Prague,
-)
+from ethereum_test_forks import Cancun, CancunToPragueAtTime15k, Fork, Osaka, Prague
+from ethereum_test_forks.forks.transition import PragueToOsakaAtTime15k
+from ethereum_test_forks.transition_base_fork import TransitionBaseClass
 
 CACHED_BLOBS_DIRECTORY: Path = (
     Path(platformdirs.user_cache_dir("ethereum-execution-spec-tests")) / "cached_blobs"
 )
 
 
+def clear_blob_cache(cached_blobs_folder_path: Path):
+    """Delete all cached blobs."""
+    if not cached_blobs_folder_path.is_dir():
+        return
+    for f in cached_blobs_folder_path.glob("*.json"):  # only delete .json files
+        try:
+            f.unlink()  # permanently delete this file
+        except OSError as e:
+            print(
+                f"Critical error while trying to delete file {f}:{e}.. "
+                "Aborting clearing of blob cache folder."
+            )
+            return
+
+
 def fork_string_to_object(fork_name: str) -> type[Cancun] | type[Prague] | type[Osaka]:
     """Take a fork string and return the respective fork as object."""
     fork_name = fork_name.lower()
@@ -201,21 +213,21 @@ def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
                 f"cell proofs {amount_cell_proofs} but expected 128."
             )
 
-        assert fork.supports_blobs(), f"Provided fork {fork.name()} does not support blobs!"
-
         # if this blob already exists then load from file
         blob_location: Path = Blob.get_filepath(fork, seed)
         if blob_location.exists():
             print(f"Blob exists already, reading it from file {blob_location}")
             return Blob.from_file(Blob.get_filename(fork, seed))
 
-        # loaded trusted setup if it is not already loaded
-        Blob.trusted_setup()
-
         # handle transition forks (not needed when default timestamp of 0 is used)
         if timestamp > 0:
             fork = fork.fork_at(timestamp=timestamp)
 
+        assert fork.supports_blobs(), f"Provided fork {fork.name()} does not support blobs!"
+
+        # loaded trusted setup if it is not already loaded
+        Blob.trusted_setup()
+
         # get data for blob parameters
         data: Bytes = generate_blob_data(seed)
         commitment: Bytes = get_commitment(data)
@@ -435,20 +447,3 @@ def corrupt_byte(b: bytes) -> Bytes:
             self.proof = Bytes(bytes(len(self.proof)))
         elif mode == self.ProofCorruptionMode.CORRUPT_ALL_BYTES:
             self.proof = Bytes(b"".join(corrupt_byte(bytes([byte])) for byte in self.proof))
-
-
-# test fork_at for when timestamp > 0
-# transitionfork: Fork = Cancun
-# new_timestamp = 15000  # is supposed to return prague, but still returns cancun.. TODO: ?
-# transitionforkblob: Blob = Blob.from_fork(transitionfork, timestamp=new_timestamp)
-# transitionforkblob.write_to_file()
-# restored_transitionforkblob: Blob = Blob.from_file("blob_0_cell_proofs_0")
-# assert transitionforkblob.data == restored_transitionforkblob.data
-# assert transitionforkblob.commitment == restored_transitionforkblob.commitment
-# assert transitionforkblob.proof == restored_transitionforkblob.proof
-# assert transitionforkblob.cells == restored_transitionforkblob.cells
-# assert transitionforkblob.versioned_hash == restored_transitionforkblob.versioned_hash
-# assert transitionforkblob.name == restored_transitionforkblob.name
-# assert transitionforkblob.fork == restored_transitionforkblob.fork
-# assert transitionforkblob.seed == restored_transitionforkblob.seed
-# assert transitionforkblob.timestamp == restored_transitionforkblob.timestamp

src/ethereum_test_types/tests/test_blob_types.py

@@ -1,16 +1,28 @@
 """Test suite for blobs."""
 
 import copy
+from pathlib import Path
 
+import platformdirs
 import pytest
 
 from ethereum_test_forks import (
     Cancun,
     Osaka,
     Prague,
 )
+from ethereum_test_forks.base_fork import Fork
+from ethereum_test_forks.forks.transition import (
+    CancunToPragueAtTime15k,
+    PragueToOsakaAtTime15k,
+    ShanghaiToCancunAtTime15k,
+)
+
+from ..blob_types import Blob, clear_blob_cache
 
-from ..blob_types import Blob
+CACHED_BLOBS_DIRECTORY: Path = (
+    Path(platformdirs.user_cache_dir("ethereum-execution-spec-tests")) / "cached_blobs"
+)
 
 
 @pytest.mark.parametrize("seed", [0, 10, 100])
@@ -45,8 +57,6 @@ def test_blob_creation_and_writing_and_reading(
     assert b.seed == restored.seed
     assert b.timestamp == restored.timestamp
 
-    print("SUCCESSFUL UNIT TEST")
-
 
 @pytest.mark.parametrize(
     "corruption_mode",
@@ -75,3 +85,45 @@ def test_blob_proof_corruption(
         f"Proof corruption mode {corruption_mode} for fork {fork.name()} failed, "
         "proof is unchanged!"
     )
+
+
+@pytest.mark.parametrize("timestamp", [14999, 15000])
+@pytest.mark.parametrize(
+    "fork", [ShanghaiToCancunAtTime15k, CancunToPragueAtTime15k, PragueToOsakaAtTime15k]
+)
+def test_transition_fork_blobs(
+    fork,
+    timestamp,
+):
+    """Generates blobs for transition forks (time 14999 is old fork, time 15000 is new fork)."""
+    clear_blob_cache(CACHED_BLOBS_DIRECTORY)
+
+    print(f"Original fork: {fork}, Timestamp: {timestamp}")
+    pre_transition_fork = fork.transitions_from()
+    post_transition_fork_at_15k = fork.transitions_to()  # only reached if timestamp >= 15000
+
+    if not pre_transition_fork.supports_blobs() and timestamp < 15000:
+        print(
+            f"Skipping blob creation because pre-transition fork is {pre_transition_fork} "
+            f"and timestamp is {timestamp}"
+        )
+        return
+
+    # b has already applied transition if requirements were met
+    b = Blob.from_fork(fork=fork, timestamp=timestamp)
+    print(f"Fork of created blob: {b.fork.name()}")
+
+    if timestamp == 14999:  # case: no transition yet
+        assert b.fork.name() == pre_transition_fork.name(), (
+            f"Transition fork failure! Fork {fork.name()} at timestamp: {timestamp} should have "
+            f"stayed at fork {pre_transition_fork.name()} but has unexpectedly transitioned "
+            f"to {b.fork.name()}"
+        )
+    elif timestamp == 15000:  # case: transition to next fork has happened
+        assert b.fork.name() == post_transition_fork_at_15k.name(), (
+            f"Transition fork failure! Fork {fork.name()} at timestamp: {timestamp} should have "
+            f"transitioned to {post_transition_fork_at_15k.name()} but is still at {b.fork.name()}"
+        )
+
+
+# TODO: fix issue with ShanghaiToCancunAtTime15k