refactor(specs/static_state): Fork string parsing (#1977)

* refactor(specs/static_state): Fork string parsing

* fix: tox

* docs: Changelog

* fix: review comments

* fix: spelling again

Author: marioevz

docs/CHANGELOG.md

@@ -78,6 +78,7 @@ Users can select any of the artifacts depending on their testing needs for their
 - 🔀 Disabled writing debugging information to the EVM "dump directory" to improve performance. To obtain debug output, the `--evm-dump-dir` flag must now be explicitly set. As a consequence, the now redundant `--skip-evm-dump` option was removed ([#1874](https://github.com/ethereum/execution-spec-tests/pull/1874)).
 - ✨ Generate unique addresses with Python for compatible static tests, instead of using hard-coded addresses from legacy static test fillers ([#1781](https://github.com/ethereum/execution-spec-tests/pull/1781)).
 - ✨ Added support for the `--benchmark-gas-values` flag in the `fill` command, allowing a single genesis file to be used across different gas limit settings when generating fixtures. ([#1895](https://github.com/ethereum/execution-spec-tests/pull/1895)).
+- ✨ Static tests can now specify a maximum fork where they should be filled for ([#1977](https://github.com/ethereum/execution-spec-tests/pull/1977)).
 
 #### `consume`
 

src/ethereum_test_specs/static_state/expect_section.py

@@ -1,13 +1,13 @@
 """Expect section structure of ethereum/tests fillers."""
 
-from enum import Enum
-from typing import Annotated, Any, Dict, List, Mapping, Union
+import re
+from enum import StrEnum
+from typing import Annotated, Any, Dict, Iterator, List, Mapping, Set, Union
 
 from pydantic import (
     BaseModel,
     BeforeValidator,
     Field,
-    ValidationInfo,
     ValidatorFunctionWrapHandler,
     field_validator,
     model_validator,
@@ -22,7 +22,7 @@
     Storage,
 )
 from ethereum_test_exceptions import TransactionExceptionInstanceOrList
-from ethereum_test_forks import get_forks
+from ethereum_test_forks import Fork, get_forks
 from ethereum_test_types import Alloc
 
 from .common import (
@@ -86,6 +86,14 @@ def resolve(self, tags: TagDict) -> Storage:
                 storage[resolved_key] = value
         return storage
 
+    def __contains__(self, key: Address) -> bool:
+        """Check if the storage contains a key."""
+        return key in self.root
+
+    def __iter__(self) -> Iterator[ValueOrCreateTagInFiller]:  # type: ignore[override]
+        """Iterate over the storage."""
+        return iter(self.root)
+
 
 class AccountInExpectSection(BaseModel, TagDependentData):
     """Class that represents an account in expect section filler."""
@@ -127,66 +135,116 @@ def resolve(self, tags: TagDict) -> Account:
         return Account(**account_kwargs)
 
 
-class CMP(Enum):
+class CMP(StrEnum):
     """Comparison action."""
 
-    GT = 1
-    LT = 2
-    LE = 3
-    GE = 4
-    EQ = 5
-
-
-def parse_networks(fork_with_operand: str) -> List[str]:
-    """Parse fork_with_operand `>=Cancun` into [Cancun, Prague, ...]."""
-    parsed_forks: List[str] = []
-    all_forks_by_name = [fork.name() for fork in get_forks()]
-
-    action: CMP = CMP.EQ
-    fork: str = fork_with_operand
-    if fork_with_operand[:1] == "<":
-        action = CMP.LT
-        fork = fork_with_operand[1:]
-    if fork_with_operand[:1] == ">":
-        action = CMP.GT
-        fork = fork_with_operand[1:]
-    if fork_with_operand[:2] == "<=":
-        action = CMP.LE
-        fork = fork_with_operand[2:]
-    if fork_with_operand[:2] == ">=":
-        action = CMP.GE
-        fork = fork_with_operand[2:]
-
-    if action == CMP.EQ:
-        fork = fork_with_operand
-
-    # translate unsupported fork names
-    if fork == "EIP158":
-        fork = "Byzantium"
-
-    if action == CMP.EQ:
-        parsed_forks.append(fork)
-        return parsed_forks
-
-    try:
-        # print(all_forks_by_name)
-        idx = all_forks_by_name.index(fork)
-        # ['Frontier', 'Homestead', 'Byzantium', 'Constantinople', 'ConstantinopleFix',
-        #  'Istanbul', 'MuirGlacier', 'Berlin', 'London', 'ArrowGlacier', 'GrayGlacier',
-        #  'Paris', 'Shanghai', 'Cancun', 'Prague', 'Osaka']
-    except ValueError:
-        raise ValueError(f"Unsupported fork: {fork}") from Exception
-
-    if action == CMP.GE:
-        parsed_forks = all_forks_by_name[idx:]
-    elif action == CMP.GT:
-        parsed_forks = all_forks_by_name[idx + 1 :]
-    elif action == CMP.LE:
-        parsed_forks = all_forks_by_name[: idx + 1]
-    elif action == CMP.LT:
-        parsed_forks = all_forks_by_name[:idx]
-
-    return parsed_forks
+    LE = "<="
+    GE = ">="
+    LT = "<"
+    GT = ">"
+    EQ = "="
+
+
+class ForkConstraint(BaseModel):
+    """Single fork with an operand."""
+
+    operand: CMP
+    fork: Fork
+
+    @field_validator("fork", mode="before")
+    @classmethod
+    def parse_fork_synonyms(cls, value: Any):
+        """Resolve fork synonyms."""
+        if value == "EIP158":
+            value = "Byzantium"
+        return value
+
+    @model_validator(mode="before")
+    @classmethod
+    def parse_from_string(cls, data: Any) -> Any:
+        """Parse a fork with operand from a string."""
+        if isinstance(data, str):
+            for cmp in CMP:
+                if data.startswith(cmp):
+                    fork = data.removeprefix(cmp)
+                    return {
+                        "operand": cmp,
+                        "fork": fork,
+                    }
+            return {
+                "operand": CMP.EQ,
+                "fork": data,
+            }
+        return data
+
+    def match(self, fork: Fork) -> bool:
+        """Return whether the fork satisfies the operand evaluation."""
+        match self.operand:
+            case CMP.LE:
+                return fork <= self.fork
+            case CMP.GE:
+                return fork >= self.fork
+            case CMP.LT:
+                return fork < self.fork
+            case CMP.GT:
+                return fork > self.fork
+            case CMP.EQ:
+                return fork == self.fork
+            case _:
+                raise ValueError(f"Invalid operand: {self.operand}")
+
+
+class ForkSet(EthereumTestRootModel):
+    """Set of forks."""
+
+    root: Set[Fork]
+
+    @model_validator(mode="before")
+    @classmethod
+    def parse_from_list_or_string(cls, value: Any) -> Set[Fork]:
+        """Parse fork_with_operand `>=Cancun` into {Cancun, Prague, ...}."""
+        fork_set: Set[Fork] = set()
+        if not isinstance(value, list):
+            value = [value]
+
+        for fork_with_operand in value:
+            matches = re.findall(r"(<=|<|>=|>|=)([^<>=]+)", fork_with_operand)
+            if matches:
+                all_fork_constraints = [
+                    ForkConstraint.model_validate(f"{op}{fork.strip()}") for op, fork in matches
+                ]
+            else:
+                all_fork_constraints = [ForkConstraint.model_validate(fork_with_operand.strip())]
+
+            for fork in get_forks():
+                for f in all_fork_constraints:
+                    if not f.match(fork):
+                        # If any constraint does not match, skip adding
+                        break
+                else:
+                    # All constraints match, add the fork to the set
+                    fork_set.add(fork)
+
+        return fork_set
+
+    def __hash__(self) -> int:
+        """Return the hash of the fork set."""
+        h = hash(None)
+        for fork in sorted([str(f) for f in self]):
+            h ^= hash(fork)
+        return h
+
+    def __contains__(self, fork: Fork) -> bool:
+        """Check if the fork set contains a fork."""
+        return fork in self.root
+
+    def __iter__(self) -> Iterator[Fork]:  # type: ignore[override]
+        """Iterate over the fork set."""
+        return iter(self.root)
+
+    def __len__(self) -> int:
+        """Return the length of the fork set."""
+        return len(self.root)
 
 
 class ResultInFiller(EthereumTestRootModel, TagDependentData):
@@ -228,44 +286,61 @@ def resolve(self, tags: TagDict) -> Alloc:
             post[resolved_address] = account.resolve(tags)
         return post
 
+    def __contains__(self, address: Address) -> bool:
+        """Check if the result contains an address."""
+        return address in self.root
+
+    def __iter__(self) -> Iterator[AddressOrCreateTagInFiller]:  # type: ignore[override]
+        """Iterate over the result."""
+        return iter(self.root)
+
+    def __len__(self) -> int:
+        """Return the length of the result."""
+        return len(self.root)
+
+
+class ExpectException(EthereumTestRootModel):
+    """Expect exception model."""
+
+    root: Dict[ForkSet, TransactionExceptionInstanceOrList]
+
+    def __getitem__(self, fork: Fork) -> TransactionExceptionInstanceOrList:
+        """Get an expectation for a given fork."""
+        for k in self.root:
+            if fork in k:
+                return self.root[k]
+        raise KeyError(f"Fork {fork} not found in expectations.")
+
+    def __contains__(self, fork: Fork) -> bool:
+        """Check if the expect exception contains a fork."""
+        return fork in self.root
+
+    def __iter__(self) -> Iterator[ForkSet]:  # type: ignore[override]
+        """Iterate over the expect exception."""
+        return iter(self.root)
+
+    def __len__(self) -> int:
+        """Return the length of the expect exception."""
+        return len(self.root)
+
 
 class ExpectSectionInStateTestFiller(CamelModel):
     """Expect section in state test filler."""
 
     indexes: Indexes = Field(default_factory=Indexes)
-    network: List[str]
+    network: ForkSet
     result: ResultInFiller
-    expect_exception: Dict[str, TransactionExceptionInstanceOrList] | None = None
-
-    @field_validator("network", mode="before")
-    @classmethod
-    def parse_networks(cls, network: List[str], info: ValidationInfo) -> List[str]:
-        """Parse networks into array of forks."""
-        forks: List[str] = []
-        for net in network:
-            forks.extend(parse_networks(net))
-        return forks
-
-    @field_validator("expect_exception", mode="before")
-    @classmethod
-    def parse_expect_exception(
-        cls, expect_exception: Dict[str, str] | None, info: ValidationInfo
-    ) -> Dict[str, str] | None:
-        """Parse operand networks in exceptions."""
-        if expect_exception is None:
-            return expect_exception
-
-        parsed_expect_exception: Dict[str, str] = {}
-        for fork_with_operand, exception in expect_exception.items():
-            forks: List[str] = parse_networks(fork_with_operand)
-            for fork in forks:
-                if fork in parsed_expect_exception:
-                    raise ValueError(
-                        "Expect exception has redundant fork with multiple exceptions!"
-                    )
-                parsed_expect_exception[fork] = exception
-
-        return parsed_expect_exception
+    expect_exception: ExpectException | None = None
+
+    def model_post_init(self, __context):
+        """Validate that the expectation is coherent."""
+        if self.expect_exception is None:
+            return
+        all_forks: Set[Fork] = set()
+        for current_fork_set in self.expect_exception:
+            for fork in current_fork_set:
+                assert fork not in all_forks
+                all_forks.add(fork)
 
     def has_index(self, d: int, g: int, v: int) -> bool:
         """Check if there is index set in indexes."""

src/ethereum_test_specs/static_state/state_static.py

@@ -1,6 +1,6 @@
 """Ethereum General State Test filler static test spec parser."""
 
-from typing import Callable, ClassVar, List, Self, Union
+from typing import Callable, ClassVar, List, Self, Set, Union
 
 import pytest
 from _pytest.mark.structures import ParameterSet
@@ -156,7 +156,7 @@ def test_state_vectors(
         ):
             for expect in self.expect:
                 if expect.has_index(d, g, v):
-                    if fork.name() in expect.network:
+                    if fork in expect.network:
                         tx_tag_dependencies = self.transaction.tag_dependencies()
                         result_tag_dependencies = expect.result.tag_dependencies()
                         all_dependencies = {**tx_tag_dependencies, **result_tag_dependencies}
@@ -165,7 +165,7 @@ def test_state_vectors(
                         exception = (
                             None
                             if expect.expect_exception is None
-                            else expect.expect_exception[fork.name()]
+                            else expect.expect_exception[fork]
                         )
                         tx = self.transaction.get_transaction(tags, d, g, v, exception)
                         post = expect.result.resolve(tags)
@@ -193,9 +193,7 @@ def test_state_vectors(
 
     def get_valid_at_forks(self) -> List[str]:
         """Return list of forks that are valid for this test."""
-        fork_list: List[str] = []
+        fork_set: Set[Fork] = set()
         for expect in self.expect:
-            for fork in expect.network:
-                if fork not in fork_list:
-                    fork_list.append(fork)
-        return fork_list
+            fork_set.update(expect.network)
+        return sorted([str(f) for f in fork_set])