eip7732 fork choice tests (part1)

Author: terencechain

presets/minimal/eip7732.yaml

@@ -2,8 +2,8 @@
 
 # Execution
 # ---------------------------------------------------------------
-# 2**1(= 2)
-PTC_SIZE: 2
+# 2**3(= 8)
+PTC_SIZE: 8
 # 2**2 (= 4)
 MAX_PAYLOAD_ATTESTATIONS: 4
 # floorlog2(get_generalized_index(BeaconBlockBody, 'blob_kzg_commitments')) + 1 + ceillog2(MAX_BLOB_COMMITMENTS_PER_BLOCK) (= 9 + 1 + 5 = 15)

tests/core/pyspec/eth2spec/test/eip7732/fork_choice/__init__.py

@@ -0,0 +1,200 @@
+from eth2spec.test.context import (
+    spec_state_test,
+    with_eip7732_and_later,
+)
+from eth2spec.test.helpers.block import (
+    build_empty_block_for_next_slot,
+)
+from eth2spec.test.helpers.fork_choice import (
+    check_head_against_root,
+    get_anchor_root,
+    get_genesis_forkchoice_store_and_block,
+    on_tick_and_append_step,
+    output_head_check,
+    tick_and_add_block,
+)
+from eth2spec.test.helpers.state import (
+    state_transition_and_sign_block,
+)
+
+
+def build_and_yield_execution_payload_envelope(spec, beacon_block):
+    """
+    Build a SignedExecutionPayloadEnvelope for the given beacon block and yield it for testing.
+    This simulates the builder revealing the execution payload after the beacon block.
+    """
+    # Get the execution payload header from the beacon block
+    payload_header = beacon_block.body.signed_execution_payload_header.message
+    
+    # Build execution payload from the header
+    execution_payload = spec.ExecutionPayload(
+        parent_hash=payload_header.parent_block_hash,
+        fee_recipient=payload_header.fee_recipient,
+        state_root=payload_header.state_root,
+        receipts_root=payload_header.receipts_root,
+        logs_bloom=payload_header.logs_bloom,
+        prev_randao=payload_header.prev_randao,
+        block_number=payload_header.block_number,
+        gas_limit=payload_header.gas_limit,
+        gas_used=payload_header.gas_used,
+        timestamp=payload_header.timestamp,
+        extra_data=payload_header.extra_data,
+        base_fee_per_gas=payload_header.base_fee_per_gas,
+        block_hash=payload_header.block_hash,
+        transactions=[],  # Empty for testing
+        withdrawals=payload_header.withdrawals,
+        blob_gas_used=payload_header.blob_gas_used,
+        excess_blob_gas=payload_header.excess_blob_gas,
+    )
+    
+    # Create execution payload envelope
+    envelope = spec.ExecutionPayloadEnvelope(
+        beacon_block_root=spec.hash_tree_root(beacon_block),
+        payload=execution_payload,
+        builder_index=spec.ValidatorIndex(0),  # Use validator 0 as builder for testing
+        blob_kzg_commitments=[],  # Empty for basic test
+        payload_withheld=False,
+    )
+    
+    # Sign the envelope (empty signature for testing)
+    signed_envelope = spec.SignedExecutionPayloadEnvelope(
+        message=envelope,
+        signature=spec.BLSSignature(),
+    )
+    
+    return signed_envelope
+
+
+def process_execution_payload_reveal(spec, store, signed_block, test_steps):
+    """
+    Process execution payload reveal for a beacon block in EIP7732.
+    This simulates the full flow: beacon block -> execution payload reveal -> store update.
+    """
+    # Build and yield the execution payload envelope
+    signed_envelope = build_and_yield_execution_payload_envelope(spec, signed_block.message)
+    
+    # Yield the execution payload envelope for test file generation
+    envelope_name = f"execution_payload_envelope_{spec.hash_tree_root(signed_block.message).hex()[:8]}"
+    yield envelope_name, signed_envelope
+    
+    # Process the execution payload envelope through the store
+    spec.on_execution_payload(store, signed_envelope)
+    
+    # Add test step for execution payload processing
+    test_steps.append({
+        "execution_payload": envelope_name,
+        "valid": True,
+    })
+    
+    return signed_envelope
+
+
+@with_eip7732_and_later
+@spec_state_test
+def test_genesis(spec, state):
+    """Test genesis initialization with EIP7732 fork choice modifications"""
+    test_steps = []
+    # Initialization
+    store, anchor_block = get_genesis_forkchoice_store_and_block(spec, state)
+    yield "anchor_state", state
+    yield "anchor_block", anchor_block
+
+    anchor_root = get_anchor_root(spec, state)
+    check_head_against_root(spec, store, anchor_root)
+
+    # EIP7732-specific assertions
+    assert hasattr(store, "execution_payload_states"), (
+        "Store should have execution_payload_states field"
+    )
+    assert hasattr(store, "ptc_vote"), "Store should have ptc_vote field"
+    assert anchor_root in store.execution_payload_states, (
+        "Anchor block should be in execution_payload_states"
+    )
+    assert anchor_root in store.ptc_vote, "Anchor block should have ptc_vote entry"
+
+    # Check PTC vote initialization
+    ptc_vote = store.ptc_vote[anchor_root]
+    assert len(ptc_vote) == spec.PTC_SIZE, f"PTC vote should have {spec.PTC_SIZE} entries"
+    assert all(vote == False for vote in ptc_vote), "All PTC votes should be False initially"
+
+    # Verify get_head returns ForkChoiceNode
+    head = spec.get_head(store)
+    assert isinstance(head, spec.ForkChoiceNode), "get_head should return ForkChoiceNode in EIP7732"
+
+    output_head_check(spec, store, test_steps)
+
+    yield "steps", test_steps
+
+
+@with_eip7732_and_later
+@spec_state_test
+def test_basic(spec, state):
+    """Basic EIP7732 fork choice test - similar to phase0 test_basic but with payload status tracking"""
+    test_steps = []
+    
+    # Add EIP7732-specific metadata
+    yield "test_scenario", "meta", "basic_fork_choice"
+    yield "tests_payload_status", "meta", True
+    yield "tests_execution_payload_states", "meta", True
+    
+    # Initialization
+    store, anchor_block = get_genesis_forkchoice_store_and_block(spec, state)
+    yield "anchor_state", state
+    yield "anchor_block", anchor_block
+    
+    # Set initial time and record tick
+    current_time = state.slot * spec.config.SECONDS_PER_SLOT + store.genesis_time
+    on_tick_and_append_step(spec, store, current_time, test_steps)
+    assert store.time == current_time
+    
+    # Verify initial EIP7732 state
+    anchor_root = get_anchor_root(spec, state)
+    check_head_against_root(spec, store, anchor_root)
+    
+    # Check initial head has PENDING payload status
+    head = spec.get_head(store)
+    assert head.payload_status == spec.PAYLOAD_STATUS_PENDING, "Initial head should have PENDING status"
+    
+    # On receiving a block of `GENESIS_SLOT + 1` slot
+    block = build_empty_block_for_next_slot(spec, state)
+    signed_block = state_transition_and_sign_block(spec, state, block)
+    yield from tick_and_add_block(spec, store, signed_block, test_steps)
+
+    # Verify block was added to both stores
+    block_root = signed_block.message.hash_tree_root()
+    assert block_root in store.blocks, "Block should be in store.blocks"
+    assert block_root in store.block_states, "Block should have block state"
+    assert block_root in store.ptc_vote, "Block should have PTC vote entry"
+    
+    # Head should now be the new block with PENDING status
+    check_head_against_root(spec, store, block_root)
+    head = spec.get_head(store)
+    assert head.payload_status == spec.PAYLOAD_STATUS_PENDING, "New head should have PENDING status"
+    
+    # Process execution payload reveal (simulates builder revealing payload)
+    yield from process_execution_payload_reveal(spec, store, signed_block, test_steps)
+    
+    # Verify block now has execution payload state
+    assert block_root in store.execution_payload_states, "Block should now have execution payload state"
+    
+    # On receiving a block of next slot
+    store.time = current_time + spec.config.SECONDS_PER_SLOT * 2
+    block_2 = build_empty_block_for_next_slot(spec, state)
+    signed_block_2 = state_transition_and_sign_block(spec, state, block_2)
+    yield from tick_and_add_block(spec, store, signed_block_2, test_steps)
+    
+    # Process execution payload reveal for second block
+    block_2_root = signed_block_2.message.hash_tree_root()
+    check_head_against_root(spec, store, block_2_root)
+    yield from process_execution_payload_reveal(spec, store, signed_block_2, test_steps)
+    
+    # Add EIP7732-specific checks to test steps
+    test_steps.append({
+        "checks": {
+            "execution_payload_states_count": len(store.execution_payload_states),
+            "blocks_with_ptc_votes": len(store.ptc_vote),
+            "head_payload_status": int(spec.get_head(store).payload_status),
+        }
+    })
+    
+    yield "steps", test_steps

tests/core/pyspec/eth2spec/test/eip7732/fork_choice/test_base.py

@@ -552,3 +552,23 @@ def get_pow_block_file_name(pow_block):
 def add_pow_block(spec, store, pow_block, test_steps):
     yield get_pow_block_file_name(pow_block), pow_block
     test_steps.append({"pow_block": get_pow_block_file_name(pow_block)})
+
+
+# EIP7732 Fork Choice Helpers
+
+
+def create_payload_attestation_message(
+    spec, validator_index, beacon_block_root, slot, payload_present=True
+):
+    """Create PayloadAttestationMessage for PTC voting"""
+    data = spec.PayloadAttestationData(
+        beacon_block_root=beacon_block_root,
+        slot=slot,
+        payload_present=payload_present,
+    )
+
+    return spec.PayloadAttestationMessage(
+        validator_index=validator_index,
+        data=data,
+        signature=spec.BLSSignature(),  # Empty signature for testing
+    )