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

Author: felix314159

src/ethereum_test_base_types/composite_types.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar, Dict, List, SupportsBytes, Type, TypeAlias
 
-from pydantic import Field, PrivateAttr, TypeAdapter
+from pydantic import BaseModel, Field, PrivateAttr, TypeAdapter
 
 from .base_types import Address, Bytes, Hash, HashInt, HexNumber, ZeroPaddedHexNumber
 from .conversions import BytesConvertible, NumberConvertible
@@ -487,7 +487,7 @@ class ForkBlobSchedule(CamelModel):
 
 
 class BlobSchedule(EthereumTestRootModel[Dict[str, ForkBlobSchedule]]):
-    """Blob schedule configuration dictionary."""
+    """Blob schedule configuration dictionary. Key is fork name."""
 
     root: Dict[str, ForkBlobSchedule] = Field(default_factory=dict, validate_default=True)
 
@@ -503,6 +503,36 @@ def last(self) -> ForkBlobSchedule | None:
             return None
         return list(self.root.values())[-1]
 
-    def __getitem__(self, key: str) -> ForkBlobSchedule:
+    def __getitem__(self, key: str) -> ForkBlobSchedule | None:
         """Return the schedule for a given fork."""
-        return self.root[key]
+        return self.root.get(key)
+
+
+class TimestampBlobSchedule(BaseModel):
+    """
+    Contains a list of dictionaries. Each dictionary is a scheduled BPO fork.
+    Each dictionary's key is the activation timestamp and the values are a ForkBlobSchedule
+    object with the fields max, target and base_fee_update_fraction.
+    """
+
+    root: List[Dict[int, ForkBlobSchedule]] = Field(default_factory=list, validate_default=True)
+
+    @classmethod
+    def add_schedule(cls, activation_timestamp: int, schedule: ForkBlobSchedule):
+        """Add a schedule to the schedule list."""
+        assert activation_timestamp > -1
+        assert schedule.max_blobs_per_block > 0
+        assert schedule.base_fee_update_fraction > 0
+        assert schedule.target_blobs_per_block > 0
+
+        # ensure that the timestamp of each scheduled bpo fork is unique
+        existing_keys: set = set()
+        for d in cls.root:
+            existing_keys.update(d.keys())
+        assert activation_timestamp not in existing_keys, (
+            f"No duplicate activation forks allowed: Timestamp {activation_timestamp} already "
+            "exists in the current schedule."
+        )
+
+        # add a scheduled bpo fork
+        cls.root.append({activation_timestamp: schedule})

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,51 @@
 logger = get_logger(__name__)
 
 
+class BPO_Parameters(Enum):  # noqa: N801
+    """Define BPO keys for IDE autocomplete."""
+
+    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:  # noqa: D417
+    """
+    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

tests/osaka/eip7892_bpo/__init__.py

@@ -0,0 +1 @@
+"""EIP-7892 Tests."""

tests/osaka/eip7892_bpo/conftest.py

@@ -0,0 +1 @@
+"""Pytest (plugin) definitions local to EIP-7892 tests."""

tests/osaka/eip7892_bpo/spec.py

@@ -0,0 +1,24 @@
+"""Defines EIP-7892 specification constants and functions."""
+
+from dataclasses import dataclass
+
+# Base the spec on EIP-4844 which EIP-7892 extends
+from ...cancun.eip4844_blobs.spec import Spec as EIP4844Spec
+
+
+@dataclass(frozen=True)
+class ReferenceSpec:
+    """Defines the reference spec version and git path."""
+
+    git_path: str
+    version: str
+
+
+ref_spec_7892 = ReferenceSpec("EIPS/eip-7892.md", "e42c14f83052bfaa8c38832dcbc46e357dd1a1d9")
+
+
+@dataclass(frozen=True)
+class Spec(EIP4844Spec):
+    """Parameters from the EIP-7892 specifications."""
+
+    pass

tests/osaka/eip7892_bpo/test_bpo.py

@@ -0,0 +1,61 @@
+"""abstract: Test [EIP-7892: Blob Parameter Only Hardforks](https://eips.ethereum.org/EIPS/eip-7892)."""
+
+import pytest
+
+from ethereum_test_base_types.composite_types import ForkBlobSchedule, TimestampBlobSchedule
+from ethereum_test_forks import Fork
+from ethereum_test_tools import (
+    Alloc,
+    Block,
+    BlockchainTestFiller,
+)
+from ethereum_test_types import Environment
+
+from .spec import ref_spec_7892  # type: ignore
+
+REFERENCE_SPEC_GIT_PATH = ref_spec_7892.git_path
+REFERENCE_SPEC_VERSION = ref_spec_7892.version
+
+
+@pytest.mark.valid_from("Osaka")
+def test_bpo_schedule(
+    blockchain_test: BlockchainTestFiller,
+    pre: Alloc,
+    post: Alloc,
+    env: Environment,
+    fork: Fork,
+):
+    """Test whether clients correctly set provided BPO schedules."""
+    bpo_schedule = TimestampBlobSchedule()
+    # below ensure that there is a timestamp difference of at least 3 between each scheduled fork
+    bpo_schedule.add_schedule(
+        1234, ForkBlobSchedule(max=6, target=5, base_fee_update_fraction=5007716)
+    )
+    bpo_schedule.add_schedule(
+        2345, ForkBlobSchedule(max=4, target=3, base_fee_update_fraction=5007716)
+    )
+
+    blocks = []
+    for schedule_dict in bpo_schedule.root:
+        for t in schedule_dict:
+            # add block before bpo
+            blocks.append(Block(timestamp=t - 1))
+            # add block at bpo
+            blocks.append(Block(timestamp=t))
+            # add block after bpo
+            blocks.append(Block(timestamp=t + 1))
+
+    # amount of created blocks = 3 * len(bpo_schedule.root)
+    assert len(blocks) == 3 * len(bpo_schedule.root)
+
+    # TODO:
+    # for each block the client should report the current values of: max, target and base_fee_update_fraction  # noqa: E501
+    # we need to signal to the client that the expected response is according to the bpo_schedule defined above  # noqa: E501
+
+    blockchain_test(
+        genesis_environment=env,
+        pre=pre,
+        post=post,
+        blocks=blocks,
+        bpo_schedule=bpo_schedule,
+    )