removed static blobs, cleaned code

Author: felix314159

src/ethereum_test_forks/base_fork.py

@@ -1,7 +1,7 @@
 """Abstract base class for Ethereum forks."""
 
 from abc import ABC, ABCMeta, abstractmethod
-from typing import Any, ClassVar, List, Mapping, Optional, Protocol, Sized, Tuple, Type
+from typing import Any, ClassVar, List, Literal, Mapping, Optional, Protocol, Sized, Tuple, Type
 
 from semver import Version
 
@@ -155,6 +155,14 @@ class BaseFork(ABC, metaclass=BaseForkMeta):
     _solc_name: ClassVar[Optional[str]] = None
     _ignore: ClassVar[bool] = False
 
+    # make mypy happy
+    BLOB_CONSTANTS: ClassVar[dict[str, int | Literal["big"]]] = {}
+
+    @classmethod
+    def get_blob_constant(cls, name: str) -> int | Literal["big"]:
+        """Return value of requested blob constant."""
+        raise NotImplementedError
+
     def __init_subclass__(
         cls,
         *,

src/ethereum_test_types/blob.py

@@ -5,9 +5,9 @@
 from hashlib import sha256
 from os.path import realpath
 from pathlib import Path
-from typing import Any, ClassVar, List
+from typing import Any, ClassVar, List, Literal, cast
 
-import ckzg
+import ckzg  # type: ignore
 import platformdirs
 
 from ethereum_test_base_types.base_types import Bytes, Hash
@@ -20,15 +20,6 @@
 )
 
 
-def get_eest_root_folder(marker_files=("pyproject.toml", ".git", "tests", "src")) -> Path:
-    """Search for a folder where all files/folders listed above exist (root of project)."""
-    current = Path(__file__).resolve()
-    for parent in current.parents:
-        if all((parent / marker).exists() for marker in marker_files):
-            return parent
-    raise RuntimeError("Project root folder of execution-spec-tests was not found")
-
-
 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()
@@ -42,8 +33,6 @@ def fork_string_to_object(fork_name: str) -> type[Cancun] | type[Prague] | type[
     raise ValueError(f"Fork {fork_name} has not yet been implemented in this function.")
 
 
-eest_root = get_eest_root_folder()
-
 CACHED_BLOBS_DIRECTORY: Path = (
     Path(platformdirs.user_cache_dir("ethereum-execution-spec-tests")) / "cached_blobs"
 )
@@ -77,7 +66,7 @@ def trusted_setup(cls):
     @staticmethod
     def get_filename(fork: Fork, seed: int) -> str:
         """Return filename this blob would have as string (with .json extension)."""
-        amount_cell_proofs: int = fork.get_blob_constant("AMOUNT_CELL_PROOFS")
+        amount_cell_proofs: int = cast(int, fork.get_blob_constant("AMOUNT_CELL_PROOFS"))
         return "blob_" + str(seed) + "_cell_proofs_" + str(amount_cell_proofs) + ".json"
 
     @staticmethod
@@ -88,7 +77,6 @@ def get_filepath(fork: Fork, seed: int):
 
         # create cached blobs dir if necessary
         if not CACHED_BLOBS_DIRECTORY.exists():
-            # print(f"Blob cache directory does not exist already: {CACHED_BLOBS_DIRECTORY}. I create it now.")  # noqa: E501
             CACHED_BLOBS_DIRECTORY.mkdir(
                 parents=True, exist_ok=True
             )  # create all necessary dirs on the way
@@ -97,8 +85,8 @@ def get_filepath(fork: Fork, seed: int):
         return CACHED_BLOBS_DIRECTORY / would_be_filename
 
     @staticmethod
-    def from_fork(fork: Fork, seed: int = 0, timestamp: int = 0) -> "Blob":  # noqa: N802
-        """Construct Blob instances. Fork-specific logic is encapsulated within nested functions."""  # noqa: E501
+    def from_fork(fork: Fork, seed: int = 0, timestamp: int = 0) -> "Blob":
+        """Construct Blob instances. Fork logic is encapsulated within nested functions."""
 
         def generate_blob_data(rng_seed: int = 0) -> Bytes:
             """Calculate blob data deterministically via provided seed."""
@@ -107,14 +95,14 @@ def generate_blob_data(rng_seed: int = 0) -> Bytes:
 
             # generate blob
             ints: list[int] = [
-                rng.randrange(fork.get_blob_constant("BLS_MODULUS"))
-                for _ in range(fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB"))
+                rng.randrange(cast(int, fork.get_blob_constant("BLS_MODULUS")))
+                for _ in range(cast(int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB")))
             ]
 
             encoded: list[bytes] = [
                 i.to_bytes(
-                    fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"),
-                    fork.get_blob_constant("KZG_ENDIANNESS"),
+                    cast(int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")),
+                    cast(Literal["big"], fork.get_blob_constant("KZG_ENDIANNESS")),
                 )
                 for i in ints
             ]
@@ -127,19 +115,24 @@ def get_versioned_hash(commitment: Bytes, version: int = 1) -> Hash:
             return Hash(bytes([version]) + sha256(commitment).digest()[1:])
 
         def get_commitment(data: Bytes) -> Bytes:
-            """Take a blob and returns a cryptographic commitment to it. Note: Each cell seems to hold a copy of this commitment."""  # noqa: E501
+            """
+            Take a blob and returns a cryptographic commitment to it.
+
+            Note: Each cell seems to hold a copy of this commitment.
+            """
             # sanity check
-            assert len(data) == fork.get_blob_constant(
-                "FIELD_ELEMENTS_PER_BLOB"
-            ) * fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"), (
-                f"Expected blob of length {fork.get_blob_constant('FIELD_ELEMENTS_PER_BLOB') * fork.get_blob_constant('BYTES_PER_FIELD_ELEMENT')} but got blob of length {len(data)}"  # noqa: E501
+            field_elements: int = cast(int, fork.get_blob_constant("FIELD_ELEMENTS_PER_BLOB"))
+            bytes_per_field: int = cast(int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"))
+            assert len(data) == field_elements * bytes_per_field, (
+                f"Expected blob of length "
+                f"{field_elements * bytes_per_field} but got blob of length {len(data)}"
             )
 
             # calculate commitment
             commitment = ckzg.blob_to_kzg_commitment(data, Blob._trusted_setup)
-
             assert len(commitment) == fork.get_blob_constant("BYTES_PER_COMMITMENT"), (
-                f"Expected {fork.get_blob_constant('BYTES_PER_COMMITMENT')} resulting commitments but got {len(commitment)} commitments"  # noqa: E501
+                f"Expected {fork.get_blob_constant('BYTES_PER_COMMITMENT')} "
+                f"resulting commitments but got {len(commitment)} commitments"
             )
 
             return commitment
@@ -152,7 +145,7 @@ def get_proof(fork: Fork, data: Bytes) -> List[Bytes] | Bytes:
             if amount_cell_proofs == 0:
                 z = 2  # 2 is one of many possible valid field elements z (https://github.com/ethereum/consensus-specs/blob/ad884507f7a1d5962cd3dfb5f7b3e41aab728c55/tests/core/pyspec/eth2spec/test/utils/kzg_tests.py#L58-L66)
                 z_valid_size: bytes = z.to_bytes(
-                    fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT"), byteorder="big"
+                    cast(int, fork.get_blob_constant("BYTES_PER_FIELD_ELEMENT")), byteorder="big"
                 )
                 proof, _ = ckzg.compute_kzg_proof(data, z_valid_size, Blob._trusted_setup)
                 return proof
@@ -165,7 +158,8 @@ def get_proof(fork: Fork, data: Bytes) -> List[Bytes] | Bytes:
                 return proofs
 
             raise AssertionError(
-                f"get_proof() has not been implemented yet for fork: {fork_str}. Got amount of cell proofs {amount_cell_proofs} but expected 128."  # noqa: E501
+                f"get_proof() has not been implemented yet for fork: {fork_str}. "
+                f"Got amount of cell proofs {amount_cell_proofs} but expected 128."
             )
 
         def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
@@ -184,7 +178,8 @@ def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
                 return cells  # List[bytes]
 
             raise AssertionError(
-                f"get_cells() has not been implemented yet for fork: {fork_str}. Got amount of cell proofs {amount_cell_proofs} but expected 128."  # noqa: E501
+                f"get_cells() has not been implemented yet for fork: {fork_str}. Got amount of "
+                f"cell proofs {amount_cell_proofs} but expected 128."
             )
 
         assert fork.supports_blobs(), f"Provided fork {fork.name()} does not support blobs!"
@@ -221,11 +216,16 @@ def get_cells(fork: Fork, data: Bytes) -> List[Bytes] | None:
         )
 
     @staticmethod
-    def from_file(file_name: str) -> "Blob":  # noqa: N802
-        """Read a .json file and reconstruct object it represents. You can load a blob only via its filename (with or without .json extension)."""  # noqa: E501
+    def from_file(file_name: str) -> "Blob":
+        """
+        Read a .json file and reconstruct object it represents.
+
+        You can load a blob only via its filename (with or without .json extension).
+        """
         # ensure filename was passed
         assert file_name.startswith("blob_"), (
-            f"You provided an invalid blob filename. Expected it to start with 'blob_' but got: {file_name}"  # noqa: E501
+            f"You provided an invalid blob filename. Expected it to start with 'blob_' "
+            f"but got: {file_name}"
         )
 
         # loaded trusted setup if it is not already loaded
@@ -261,16 +261,17 @@ def write_to_file(self):
         with open(output_location, "w", encoding="utf-8") as f:  # overwrite existing
             f.write(json_str)
 
-    def verify_cell_kzg_proof_batch(self, cell_indices: list) -> bool:  # noqa: E501
+    def verify_cell_kzg_proof_batch(self, cell_indices: list) -> bool:
         """Check whether all cell proofs are valid and returns True only if that is the case."""
         assert self.fork in ["osaka"], (
             f"verify_cell_kzg_proof_batch() is not available for fork: {self.fork}"
         )
 
-        assert self.cells is not None, ""  # TODO: write message
+        assert self.cells is not None, "self.cells is None, critical error."
 
         assert len(cell_indices) == len(self.cells), (
-            f"Cell Indices list (detected length {len(cell_indices)}) and Cell list (detected length {len(self.cells)}) should have same length."  # noqa: E501
+            f"Cell Indices list (detected length {len(cell_indices)}) and Cell list "
+            f"(detected length {len(self.cells)}) should have same length."
         )
 
         # each cell refers to the same commitment
@@ -288,26 +289,30 @@ def delete_cells_then_recover_them(self, deletion_indices: list[int]):
 
         Note: Requirement for successful reconstruction is having at least N of the 2N cells.
 
-        Theoretical Usage: You pass a cell list with to 128 elements to this function along with a list of deletion indices.
-        These cells will be deleted and then the ckzg recovery mechanism is used to repair the missing cells.
+        Theoretical Usage: You pass a cell list with to 128 elements to this function
+        along with a list of deletion indices. These cells will be deleted and then
+        the ckzg recovery mechanism is used to repair the missing cells.
         If no assertion is triggered the reconstruction was successful.
-        """  # noqa: E501
+        """
         assert self.fork in ["osaka"], (
             f"delete_cells_then_recover_them() is not available for fork: {self.fork}"
         )
 
-        assert self.cells is not None, "..."  # TODO: write text
+        assert self.cells is not None, "self.cells is None, critical problem."
 
         assert isinstance(self.proof, list), (
-            "This function only works when self.proof is a list, but it seems to be of type bytes (not a list)"  # noqa: E501
+            "This function only works when self.proof is a list, but it seems to be "
+            " of type bytes (not a list)"
         )
 
         assert len(self.cells) == 128, (
-            f"You are supposed to pass a full cell list with 128 elements to this function, but got list of length {len(self.cells)}"  # noqa: E501
+            f"You are supposed to pass a full cell list with 128 elements to this function, "
+            f"but got list of length {len(self.cells)}"
         )
 
         assert len(deletion_indices) < 129, (
-            f"You can't delete more than every cell (max len of deletion indices list is 128), but you passed a deletion indices list of length {len(deletion_indices)}"  # noqa: E501
+            f"You can't delete more than every cell (max len of deletion indices list is 128), "
+            f"but you passed a deletion indices list of length {len(deletion_indices)}"
         )
         for i in deletion_indices:
             assert 0 <= i <= 127, f"Expected integers in range [0, 127], but got: {i}"
@@ -319,26 +324,36 @@ def delete_cells_then_recover_them(self, deletion_indices: list[int]):
 
         recovered_cells, recovered_proofs = ckzg.recover_cells_and_kzg_proofs(
             remaining_indices, remaining_cells, Blob._trusted_setup
-        )  # on success returns two lists of len 128  # noqa: E501
+        )  # on success returns two lists of len 128
 
         # determine success/failure
         assert len(recovered_cells) == len(self.cells), (
-            f"Failed to recover cell list. Original cell list had length {len(self.cells)} but recovered cell list has length {len(recovered_cells)}"  # noqa: E501
+            f"Failed to recover cell list. Original cell list had length {len(self.cells)} but "
+            f"recovered cell list has length {len(recovered_cells)}"
         )
         assert len(recovered_proofs) == len(self.proof), (
-            f"Failed to recover proofs list. Original proofs list had length {len(self.proof)} but recovered proofs list has length {len(recovered_proofs)}"  # noqa: E501
+            f"Failed to recover proofs list. Original proofs list had length {len(self.proof)} "
+            f"but recovered proofs list has length {len(recovered_proofs)}"
         )
 
         for i in range(len(recovered_cells)):
             assert self.cells[i] == recovered_cells[i], (
-                f"Failed to correctly restore missing cells. At index {i} original cell was 0x{self.cells[i].hex()} but reconstructed cell does not match: 0x{recovered_cells[i].hex()}"  # noqa: E501
+                f"Failed to correctly restore missing cells. At index {i} original cell was "
+                f"0x{self.cells[i].hex()} but reconstructed cell does not match: "
+                f"0x{recovered_cells[i].hex()}"
             )
             assert self.proof[i] == recovered_proofs[i], (
-                f"Failed to correctly restore missing proofs. At index {i} original proof was 0x{self.proof[i].hex()} but reconstructed proof does not match: 0x{recovered_proofs[i].hex()}"  # noqa: E501
+                f"Failed to correctly restore missing proofs. At index {i} original proof was "
+                f"0x{self.proof[i].hex()} but reconstructed proof does not match: "
+                f"0x{recovered_proofs[i].hex()}"
             )
 
     class ProofCorruptionMode(Enum):
-        """Define what the proof corruption modes do. For Osaka and later each Bytes object in the list is manipulated this way."""  # noqa: E501
+        """
+        Define what the proof corruption modes do.
+
+        For Osaka and later each Bytes object in the list is manipulated this way.
+        """
 
         CORRUPT_FIRST_BYTE = 1  # corrupts a single byte (index 0)
         CORRUPT_LAST_BYTE = 2  # corrupts a single byte (last valid index)
@@ -470,12 +485,14 @@ def corrupt_byte(b: bytes) -> Bytes:
 #  osaka
 testseed = 55
 ddd: Blob = Blob.from_fork(Osaka, testseed + 10)
+assert isinstance(ddd.proof[0], bytes)  # mypy fix
 oldValue = ddd.proof[0][5]
 for m in Blob.ProofCorruptionMode:
     ddd.corrupt_proof(m)
 print("proof corruption works (osaka):", oldValue != ddd.proof[0][5])
 #   prague
 eeeeeeeeee: Blob = Blob.from_fork(Prague, testseed + 11)
+assert isinstance(eeeeeeeeee.proof[5], int)  # mypy fix
 oldValue = eeeeeeeeee.proof[5]
 for m in Blob.ProofCorruptionMode:
     eeeeeeeeee.corrupt_proof(m)