Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/scripts/gen_test_case_reference.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
f"--until={GENERATE_UNTIL_FORK}",
"--checklist-doc-gen",
"--skip-index",
"--ignore=tests/ported_static",
"-m",
"not blockchain_test_engine and not benchmark",
"-s",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
)
from execution_testing.test_types import (
Alloc,
BlockAccessList,
Environment,
Transaction,
TransactionReceipt,
Expand Down Expand Up @@ -287,7 +286,7 @@ class Result(CamelModel):
blob_gas_used: HexNumber | None = None
requests_hash: Hash | None = None
requests: List[Bytes] | None = None
block_access_list: BlockAccessList | None = None
block_access_list: Bytes | None = None
block_access_list_hash: Hash | None = None
block_exception: Annotated[
BlockExceptionWithMessage | UndefinedException | None,
Expand Down
14 changes: 8 additions & 6 deletions packages/testing/src/execution_testing/specs/blockchain.py
Original file line number Diff line number Diff line change
Expand Up @@ -713,18 +713,21 @@ def generate_block_data(
)
requests_list = block.requests

# Decode BAL from RLP bytes provided by the transition tool.
t8n_bal_rlp = transition_tool_output.result.block_access_list
t8n_bal: BlockAccessList | None = None
if t8n_bal_rlp is not None:
t8n_bal = BlockAccessList.from_rlp(t8n_bal_rlp)

if self.fork.header_bal_hash_required(
block_number=header.number, timestamp=header.timestamp
):
assert (
transition_tool_output.result.block_access_list is not None
), (
assert t8n_bal is not None, (
"Block access list is required for this block but was not "
"provided by the transition tool"
)

rlp = transition_tool_output.result.block_access_list.rlp
computed_bal_hash = Hash(rlp.keccak256())
computed_bal_hash = Hash(t8n_bal.rlp.keccak256())
assert computed_bal_hash == header.block_access_list_hash, (
"Block access list hash in header does not match the "
f"computed hash from BAL: {header.block_access_list_hash} "
Expand All @@ -741,7 +744,6 @@ def generate_block_data(

# Process block access list - apply transformer if present for invalid
# tests
t8n_bal = transition_tool_output.result.block_access_list
bal = t8n_bal

# Always validate BAL structural integrity (ordering, duplicates)
Expand Down
Original file line number Diff line number Diff line change
@@ -1,20 +1,82 @@
"""Block Access List (BAL) for t8n tool communication and fixtures."""

from functools import cached_property
from typing import Any, List
from typing import Any, Callable, List, Sequence, Union

import ethereum_rlp as eth_rlp
from ethereum_rlp import Simple
from pydantic import Field

from execution_testing.base_types import Bytes, EthereumTestRootModel
from execution_testing.base_types import (
Address,
Bytes,
EthereumTestRootModel,
ZeroPaddedHexNumber,
)
from execution_testing.base_types.serialization import (
to_serializable_element,
)

from .account_changes import BalAccountChange
from .account_changes import (
BalAccountChange,
BalBalanceChange,
BalCodeChange,
BalNonceChange,
BalStorageChange,
BalStorageSlot,
)
from .exceptions import BlockAccessListValidationError


def _bytes_from_rlp(data: Simple) -> bytes:
"""Extract bytes from an RLP-decoded Simple value."""
assert isinstance(data, bytes), f"expected bytes, got {type(data)}"
return data


def _int_from_rlp(data: Simple) -> int:
"""Decode an RLP Simple value to int."""
raw = _bytes_from_rlp(data)
if len(raw) == 0:
return 0
return int.from_bytes(raw, "big")


def _seq_from_rlp(data: Simple) -> Sequence[Simple]:
"""Extract a sequence from an RLP-decoded Simple value."""
assert not isinstance(data, bytes), "expected sequence, got bytes"
return data


IndexedChange = Union[
BalBalanceChange, BalCodeChange, BalNonceChange, BalStorageChange
]


def _hex_from_rlp(data: Simple) -> ZeroPaddedHexNumber:
"""Decode an RLP Simple value to ZeroPaddedHexNumber."""
return ZeroPaddedHexNumber(_int_from_rlp(data))


def _decode_indexed_changes(
rlp_list: Simple,
cls: type[IndexedChange],
value_field: str,
value_fn: Callable[[Simple], Any] = _hex_from_rlp,
) -> list[IndexedChange]:
"""Decode a list of [block_access_index, value] RLP pairs."""
result: list[IndexedChange] = []
for item in _seq_from_rlp(rlp_list):
idx, val = _seq_from_rlp(item)
result.append(
cls(
block_access_index=_hex_from_rlp(idx),
**{value_field: value_fn(val)},
)
)
return result


class BlockAccessList(EthereumTestRootModel[List[BalAccountChange]]):
"""
Block Access List for t8n tool communication and fixtures.
Expand All @@ -37,6 +99,61 @@ class BlockAccessList(EthereumTestRootModel[List[BalAccountChange]]):

root: List[BalAccountChange] = Field(default_factory=list)

@classmethod
def from_rlp(cls, data: Bytes) -> "BlockAccessList":
"""
Decode an RLP-encoded block access list into a BlockAccessList.

The RLP structure per EIP-7928 is:
[
[address, storage_changes, storage_reads,
balance_changes, nonce_changes, code_changes],
...
]
"""
decoded = _seq_from_rlp(eth_rlp.decode(data))
accounts = []
for account_rlp in decoded:
fields = _seq_from_rlp(account_rlp)

storage_changes = []
for slot_entry in _seq_from_rlp(fields[1]):
slot_fields = _seq_from_rlp(slot_entry)
storage_changes.append(
BalStorageSlot(
slot=_hex_from_rlp(slot_fields[0]),
slot_changes=_decode_indexed_changes(
slot_fields[1],
BalStorageChange,
"post_value",
),
)
)

accounts.append(
BalAccountChange(
address=Address(_bytes_from_rlp(fields[0])),
storage_changes=storage_changes,
storage_reads=[
_hex_from_rlp(sr) for sr in _seq_from_rlp(fields[2])
],
balance_changes=_decode_indexed_changes(
fields[3], BalBalanceChange, "post_balance"
),
nonce_changes=_decode_indexed_changes(
fields[4], BalNonceChange, "post_nonce"
),
code_changes=_decode_indexed_changes(
fields[5],
BalCodeChange,
"new_code",
value_fn=lambda v: Bytes(_bytes_from_rlp(v)),
),
)
)

return cls(root=accounts)

def to_list(self) -> List[Any]:
"""Return the list for RLP encoding per EIP-7928."""
return to_serializable_element(self.root)
Expand Down
73 changes: 4 additions & 69 deletions src/ethereum_spec_tools/evm_tools/t8n/t8n_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -354,72 +354,6 @@ def update(self, t8n: "T8N", block_env: Any, block_output: Any) -> None:
block_output.block_access_list
)

@staticmethod
def _block_access_list_to_json(account_changes: Any) -> Any:
"""
Convert BlockAccessList to JSON format matching the Pydantic models.
"""
json_account_changes = []
for account in account_changes:
account_data: Dict[str, Any] = {
"address": "0x" + account.address.hex()
}

if account.storage_changes:
storage_changes = []
for slot_change in account.storage_changes:
slot_data: Dict[str, Any] = {
"slot": int(slot_change.slot),
"slotChanges": [],
}
for change in slot_change.changes:
slot_data["slotChanges"].append(
{
"blockAccessIndex": int(
change.block_access_index
),
"postValue": int(change.new_value),
}
)
storage_changes.append(slot_data)
account_data["storageChanges"] = storage_changes

if account.storage_reads:
account_data["storageReads"] = [
int(slot) for slot in account.storage_reads
]

if account.balance_changes:
account_data["balanceChanges"] = [
{
"blockAccessIndex": int(change.block_access_index),
"postBalance": int(change.post_balance),
}
for change in account.balance_changes
]

if account.nonce_changes:
account_data["nonceChanges"] = [
{
"blockAccessIndex": int(change.block_access_index),
"postNonce": int(change.new_nonce),
}
for change in account.nonce_changes
]

if account.code_changes:
account_data["codeChanges"] = [
{
"blockAccessIndex": int(change.block_access_index),
"newCode": "0x" + change.new_code.hex(),
}
for change in account.code_changes
]

json_account_changes.append(account_data)

return json_account_changes

def json_encode_receipts(self) -> Any:
"""
Encode receipts to JSON.
Expand Down Expand Up @@ -501,9 +435,10 @@ def to_json(self) -> Any:
data["blockException"] = self.block_exception

if self.block_access_list is not None:
# Convert BAL to JSON format
data["blockAccessList"] = self._block_access_list_to_json(
self.block_access_list
# Output BAL as RLP-encoded hex bytes; the testing framework
# handles JSON serialization.
data["blockAccessList"] = encode_to_hex(
rlp.encode(self.block_access_list)
)

if self.block_access_list_hash is not None:
Expand Down
3 changes: 2 additions & 1 deletion tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,8 @@ commands =
fill \
--skip-index \
--no-html \
--tb=no \
--tb=long \
-ra \
--show-capture=no \
--disable-warnings \
-m "json_loader and not derived_test" \
Expand Down
Loading