1  from copy import deepcopy
 2  
 3  from eth2spec.phase0.spec import (
 4      get_beacon_proposer_index,
 5      cache_state,
 6      advance_slot,
 7      process_block_header,
 8  )
 9  from eth2spec.test.context import spec_state_test, expect_assertion_error, always_bls
10  from eth2spec.test.helpers.block import (
11      build_empty_block_for_next_slot,
12      sign_block
13  )
14  from eth2spec.test.helpers.state import next_slot
15  
16  
17  def prepare_state_for_header_processing(state):
18      cache_state(state)
19      advance_slot(state)
20  
21  
22  def run_block_header_processing(state, block, valid=True):
23      """
24      Run ``process_block_header``, yielding:
25        - pre-state ('pre')
26        - block ('block')
27        - post-state ('post').
28      If ``valid == False``, run expecting ``AssertionError``
29      """
30      prepare_state_for_header_processing(state)
31  
32      yield 'pre', state
33      yield 'block', block
34  
35      if not valid:
36          expect_assertion_error(lambda: process_block_header(state, block))
37          yield 'post', None
38          return
39  
40      process_block_header(state, block)
41      yield 'post', state
42  
43  
44  @spec_state_test
45  def test_success_block_header(state):
46      block = build_empty_block_for_next_slot(state, signed=True)
47      yield from run_block_header_processing(state, block)
48  
49  
50  @always_bls
51  @spec_state_test
52  def test_invalid_sig_block_header(state):
53      block = build_empty_block_for_next_slot(state)
54      yield from run_block_header_processing(state, block, valid=False)
55  
56  
57  @spec_state_test
58  def test_invalid_slot_block_header(state):
59      block = build_empty_block_for_next_slot(state)
60      block.slot = state.slot + 2  # invalid slot
61      sign_block(state, block)
62  
63      yield from run_block_header_processing(state, block, valid=False)
64  
65  
66  @spec_state_test
67  def test_invalid_previous_block_root(state):
68      block = build_empty_block_for_next_slot(state)
69      block.previous_block_root = b'\12' * 32  # invalid prev root
70      sign_block(state, block)
71  
72      yield from run_block_header_processing(state, block, valid=False)
73  
74  
75  @spec_state_test
76  def test_proposer_slashed(state):
77      # use stub state to get proposer index of next slot
78      stub_state = deepcopy(state)
79      next_slot(stub_state)
80      proposer_index = get_beacon_proposer_index(stub_state)
81  
82      # set proposer to slashed
83      state.validator_registry[proposer_index].slashed = True
84  
85      block = build_empty_block_for_next_slot(state, signed=True)
86  
87      yield from run_block_header_processing(state, block, valid=False)