added bpo config + function to retrieve from it + unit test for that function

Author: felix314159

src/ethereum_test_types/blob_bpo_config.json

@@ -0,0 +1,37 @@
+{
+    "blobSchedule": {
+        "cancun": {
+            "target": 3,
+            "max": 6,
+            "baseFeeUpdateFraction": 3338477
+        },
+        "prague": {
+            "target": 6,
+            "max": 9,
+            "baseFeeUpdateFraction": 5007716
+        },
+        "osaka": {
+            "target": 6,
+            "max": 9,
+            "maxBlobsPerTx": 6,
+            "baseFeeUpdateFraction": 5007716
+        },
+        "bpo1": {
+            "target": 12,
+            "max": 16,
+            "maxBlobsPerTx": 12,
+            "baseFeeUpdateFraction": 5007716
+        },
+        "bpo2": {
+            "target": 16,
+            "max": 24,
+            "maxBlobsPerTx": 12,
+            "baseFeeUpdateFraction": 5007716
+        }
+    },
+"cancunTime": 0,
+"pragueTime": 0,
+"osakaTime": 1747387400,
+"bpo1Time": 1757387400,
+"bpo2Time": 1767387784
+}

src/ethereum_test_types/blob_types.py

@@ -1,5 +1,6 @@
 """Blob-related types for Ethereum tests."""
 
+import json
 import random
 from enum import Enum
 from hashlib import sha256
@@ -22,6 +23,49 @@
 logger = get_logger(__name__)
 
 
+class BPO_Parameters(Enum):
+    TARGET = "target"
+    MAX = "max"
+    BASE_FEE_UPDATE_FRACTION = "baseFeeUpdateFraction"
+    TIME = "Time"  # actually it is: <fork>Time
+
+
+def bpo_get_value(bpo_fork: str, bpo_parameter: BPO_Parameters) -> int:
+    """
+    Retrieve BPO values from the JSON config.
+
+    Arguments:
+    - bpo_fork: Any fork (e.g. cancun) or bpo forks (e.g. bpo1 or bpo2)
+    - bpo_parameter: Enum value that specifies what you want to access in the bpo config
+
+    Returns the retrieved int.
+
+    """
+    # ensure the bpo config exists and can be read
+    bpo_config_path = Path("src") / "ethereum_test_types" / "blob_bpo_config.json"
+    if not bpo_config_path.exists():
+        raise FileNotFoundError(f"Failed to find BPO config json: {bpo_config_path}")
+    with open(bpo_config_path, "r") as file:
+        bpo_config = json.load(file)
+
+    # force-lowercase the provided fork
+    bpo_fork = bpo_fork.lower()
+
+    # retrieve requested value
+    if bpo_parameter == BPO_Parameters.TARGET:
+        return bpo_config["blobSchedule"][bpo_fork][BPO_Parameters.TARGET.value]
+    elif bpo_parameter == BPO_Parameters.MAX:
+        return bpo_config["blobSchedule"][bpo_fork][BPO_Parameters.MAX.value]
+    elif bpo_parameter == BPO_Parameters.BASE_FEE_UPDATE_FRACTION:
+        return bpo_config["blobSchedule"][bpo_fork][BPO_Parameters.BASE_FEE_UPDATE_FRACTION.value]
+    elif bpo_parameter == BPO_Parameters.TIME:
+        return bpo_config[bpo_fork + BPO_Parameters.TIME.value]
+
+    raise NotImplementedError(
+        f"This function has not yet been updated to handle BPO Parameter: {bpo_parameter}"
+    )
+
+
 def clear_blob_cache(cached_blobs_folder_path: Path):
     """Delete all cached blobs."""
     if not cached_blobs_folder_path.is_dir():

src/ethereum_test_types/tests/test_blob_types.py

@@ -15,7 +15,13 @@
     ShanghaiToCancunAtTime15k,
 )
 
-from ..blob_types import CACHED_BLOBS_DIRECTORY, Blob, clear_blob_cache
+from ..blob_types import (
+    CACHED_BLOBS_DIRECTORY,
+    Blob,
+    BPO_Parameters,
+    bpo_get_value,
+    clear_blob_cache,
+)
 
 
 @pytest.mark.parametrize("seed", [0, 10, 100])
@@ -109,3 +115,24 @@ def test_transition_fork_blobs(
             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()}"
         )
+
+
+@pytest.mark.parametrize("bpo_fork", ["cancun", "prague", "osaka", "bpo1", "bpo2"])
+@pytest.mark.parametrize(
+    "bpo_parameter",
+    [
+        BPO_Parameters.TARGET,
+        BPO_Parameters.MAX,
+        BPO_Parameters.BASE_FEE_UPDATE_FRACTION,
+        BPO_Parameters.TIME,
+    ],
+)
+def test_bpo_parameter_lookup(bpo_fork, bpo_parameter):
+    """Tries looking up different values from the BPO configuration json."""
+    result = bpo_get_value(bpo_fork=bpo_fork, bpo_parameter=bpo_parameter)
+    print(
+        f"\nbpo_fork: {bpo_fork}\n"
+        f"bpo_parameter: {bpo_parameter}\n"
+        f"retrieved value from config: {result}\n"
+    )
+    # TODO: when one day actual bpo_config is known assert correction of retrieved values