feat: add SSZ static vector generator

Author: RazorClient

packages/testing/src/execution_testing/ssz/__init__.py

@@ -2,10 +2,17 @@
 SSZ container types and helpers for the REST+SSZ Engine API.
 """
 
-from typing import Any, Mapping, Sequence, Type, TypeVar
+from typing import Any, Mapping, Sequence
 
 from remerkleable.core import View
 
+from .codec import (
+    decode_bytes,
+    decode_value,
+    encode_bytes,
+    encode_value,
+    hash_tree_root,
+)
 from .constants import (
     MAX_BYTES_PER_EXECUTION_REQUEST,
     MAX_BYTES_PER_TX,
@@ -96,23 +103,6 @@
     VersionedHash,
 )
 
-ViewT = TypeVar("ViewT", bound=View)
-
-
-def encode_bytes(value: View) -> bytes:
-    """Serialize an SSZ value to its canonical byte encoding."""
-    return value.encode_bytes()
-
-
-def decode_bytes(ssz_type: Type[ViewT], data: bytes) -> ViewT:
-    """Deserialize ``data`` into an SSZ value of ``ssz_type``."""
-    return ssz_type.decode_bytes(data)
-
-
-def hash_tree_root(value: View) -> bytes:
-    """Return the 32-byte SSZ `hash_tree_root` of an SSZ value."""
-    return bytes(value.hash_tree_root())
-
 
 def _build(cls: Any, fork: str, candidates: Mapping[str, Any]) -> View:
     kwargs = {}
@@ -284,8 +274,10 @@ def _opt_bytes(value: bytes | None) -> bytes | None:
     "VersionedHash",
     "Withdrawal",
     "decode_bytes",
+    "decode_value",
     "deterministic_seed",
     "encode_bytes",
+    "encode_value",
     "envelope_bytes",
     "get_random_ssz_object",
     "hash_tree_root",

packages/testing/src/execution_testing/ssz/codec.py

@@ -0,0 +1,81 @@
+"""
+SSZ (de)serialization for this package's containers.
+"""
+
+from typing import Any, Type, TypeVar
+
+from remerkleable.basic import boolean, uint
+from remerkleable.bitfields import Bitlist, Bitvector
+from remerkleable.byte_arrays import ByteList, ByteVector
+from remerkleable.complex import Container, List
+from remerkleable.core import View
+
+ViewT = TypeVar("ViewT", bound=View)
+
+
+def encode_bytes(value: View) -> bytes:
+    """Serialize an SSZ value to its canonical byte encoding."""
+    return value.encode_bytes()
+
+
+def decode_bytes(ssz_type: Type[ViewT], data: bytes) -> ViewT:
+    """Deserialize ``data`` into an SSZ value of ``ssz_type``."""
+    return ssz_type.decode_bytes(data)
+
+
+def hash_tree_root(value: View) -> bytes:
+    """Return the 32-byte SSZ `hash_tree_root` of an SSZ value."""
+    return bytes(value.hash_tree_root())
+
+
+def encode_value(value: Any) -> Any:
+    """
+    Encode an SSZ value to its canonical value tree.
+    """
+    if isinstance(value, Container):
+        return {
+            field: encode_value(getattr(value, field))
+            for field in type(value).fields()
+        }
+    if isinstance(value, (Bitvector, Bitlist)):
+        return "0x" + value.encode_bytes().hex()
+    if isinstance(value, List):
+        return [encode_value(element) for element in value]
+    if isinstance(value, (ByteVector, ByteList)):
+        return "0x" + bytes(value).hex()
+    if isinstance(value, boolean):
+        return bool(value)
+    if isinstance(value, uint):
+        return str(int(value))
+    raise TypeError(f"unsupported SSZ value type: {type(value)!r}")
+
+
+def decode_value(ssz_type: Any, obj: Any) -> Any:
+    """
+    Decode a canonical value tree into an SSZ value of ``ssz_type``.
+    """
+    if issubclass(ssz_type, Container):
+        fields = ssz_type.fields()
+        return ssz_type(
+            **{
+                field: decode_value(field_type, obj[field])
+                for field, field_type in fields.items()
+            }
+        )
+    if issubclass(ssz_type, (Bitvector, Bitlist)):
+        return ssz_type.decode_bytes(bytes.fromhex(_strip_0x(obj)))
+    if issubclass(ssz_type, List):
+        element_type = ssz_type.element_cls()
+        return ssz_type(*(decode_value(element_type, item) for item in obj))
+    if issubclass(ssz_type, (ByteVector, ByteList)):
+        return ssz_type(bytes.fromhex(_strip_0x(obj)))
+    if issubclass(ssz_type, boolean):
+        return ssz_type(bool(obj))
+    if issubclass(ssz_type, uint):
+        return ssz_type(int(obj))
+    raise TypeError(f"unsupported SSZ type: {ssz_type!r}")
+
+
+def _strip_0x(value: str) -> str:
+    """Return ``value`` without a leading ``0x`` prefix."""
+    return value[2:] if value.startswith("0x") else value

packages/testing/src/execution_testing/ssz/random_value.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 from remerkleable.basic import boolean, uint
+from remerkleable.bitfields import Bitlist, Bitvector
 from remerkleable.byte_arrays import ByteList, ByteVector
 from remerkleable.complex import Container, List
 from remerkleable.core import View
@@ -109,6 +110,36 @@ def get_random_ssz_object(
             return _max_basic_value(typ)
         else:
             return _random_basic_value(rng, typ)
+    elif issubclass(typ, Bitvector):
+        # Bit vectors are fixed length; no cap applies.
+        length = typ.vector_length()
+        if mode == RandomizationMode.mode_zero:
+            return typ(*([False] * length))
+        elif mode == RandomizationMode.mode_max:
+            return typ(*([True] * length))
+        else:
+            return typ(*(bool(rng.getrandbits(1)) for _ in range(length)))
+    elif issubclass(typ, Bitlist):
+        cap = min(max_bytes_length, typ.limit())
+        if mode == RandomizationMode.mode_nil_count:
+            length = 0
+        elif mode == RandomizationMode.mode_max_count:
+            length = cap
+        elif mode in (
+            RandomizationMode.mode_one_count,
+            RandomizationMode.mode_zero,
+            RandomizationMode.mode_max,
+        ):
+            length = min(1, typ.limit())
+        else:
+            length = rng.randint(0, cap)
+        if mode == RandomizationMode.mode_zero:
+            bits = [False] * length
+        elif mode == RandomizationMode.mode_max:
+            bits = [True] * length
+        else:
+            bits = [bool(rng.getrandbits(1)) for _ in range(length)]
+        return typ(*bits)
     elif issubclass(typ, List):
         limit = max_list_length
         if typ.limit() < limit:

packages/testing/src/execution_testing/ssz/tests/test_vectors.py

@@ -0,0 +1,66 @@
+"""
+Self-tests for the SSZ vector generator.
+"""
+
+from pathlib import Path
+
+import yaml
+
+from .. import decode_bytes, decode_value, encode_bytes, hash_tree_root
+from ..vectors.generate import (
+    CONTAINERS,
+    case_dir,
+    iter_cases,
+    write_case,
+)
+
+CASES = list(iter_cases())
+
+
+def test_generator_produces_cases() -> None:
+    """Sanity: the generator actually produces cases."""
+    assert CASES
+
+
+def test_generation_is_deterministic() -> None:
+    """Two independent generations are identical (seeded determinism)."""
+    again = [case for _, _, _, case in iter_cases()]
+    assert again == [case for _, _, _, case in CASES]
+
+
+def test_each_case_is_internally_consistent() -> None:
+    """
+    ``serialized`` decodes and re-encodes to itself, its ``hash_tree_root``
+    equals ``root``, and the canonical ``value`` parses back to the same
+    bytes. This is a self-test of the generator, not a check against an
+    independent oracle.
+    """
+    for container_name, _, _, case in CASES:
+        ssz_type = CONTAINERS[container_name]
+        serialized = case["serialized"]
+        decoded = decode_bytes(ssz_type, serialized)
+        assert encode_bytes(decoded) == serialized, container_name
+        assert "0x" + hash_tree_root(decoded).hex() == case["root"]
+        from_value = decode_value(ssz_type, case["value"])
+        assert encode_bytes(from_value) == serialized
+
+
+def test_case_has_expected_keys() -> None:
+    """Each generated case carries exactly the documented keys."""
+    for _, _, _, case in CASES:
+        assert set(case) == {"value", "serialized", "root"}
+
+
+def test_write_case_emits_three_files(tmp_path: Path) -> None:
+    """``write_case`` writes the ssz_static three-file layout, uncompressed."""
+    container_name, suite, case_index, case = CASES[0]
+    directory = case_dir(tmp_path, container_name, suite, case_index)
+    write_case(directory, case)
+
+    assert (directory / "serialized.ssz").read_bytes() == case["serialized"]
+    assert yaml.safe_load((directory / "value.yaml").read_text()) == (
+        case["value"]
+    )
+    assert yaml.safe_load((directory / "roots.yaml").read_text()) == {
+        "root": case["root"]
+    }

packages/testing/src/execution_testing/ssz/vectors/.gitignore

@@ -0,0 +1,2 @@
+# Generated SSZ vectors are a release artifact, never committed (see README.md).
+**/case_*/

packages/testing/src/execution_testing/ssz/vectors/__init__.py

@@ -0,0 +1 @@
+"""Cross-client SSZ static test vectors and their generator."""