eth2.0-specs/tests

Getting Started with Consensus Spec Tests

Getting Started

Creating the environment

Use an OS that has Python 3.8 or above. For example, Debian 11 (bullseye)

1. Install the packages you need:

   sudo apt install -y make git wget python3-venv gcc python3-dev
   

2. Download the latest consensus specs

   git clone https://github.com/ethereum/consensus-specs.git
   cd consensus-specs
   

3. Create the specifications and tests:

   make install_test
   make pyspec
   

To read more about creating the environment, see here.

Running your first test

1. Enter the virtual Python environment:

   cd ~/consensus-specs
   . venv/bin/activate
   

2. Run a sanity check test:

   cd tests/core/pyspec
   python -m pytest -k test_empty_block_transition --fork Merge eth2spec
   

3. The output should be similar to:

   ============================= test session starts ==============================
   platform linux -- Python 3.9.2, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
   rootdir: /home/qbzzt1/consensus-specs
   plugins: cov-2.12.1, forked-1.3.0, xdist-2.3.0
   collected 629 items / 626 deselected / 3 selected
   
   eth2spec/test/merge/sanity/test_blocks.py .                              [ 33%]
   eth2spec/test/phase0/sanity/test_blocks.py ..                            [100%]
   
   =============================== warnings summary ===============================
   ../../../venv/lib/python3.9/site-packages/cytoolz/compatibility.py:2
     /home/qbzzt1/consensus-specs/venv/lib/python3.9/site-packages/cytoolz/compatibility.py:2: 
   DeprecationWarning: The toolz.compatibility module is no longer needed in Python 3 and has 
   been deprecated. Please import these utilities directly from the standard library. This 
   module will be removed in a future release.
       warnings.warn("The toolz.compatibility module is no longer "
   
   -- Docs: https://docs.pytest.org/en/stable/warnings.html
   ================ 3 passed, 626 deselected, 1 warning in 16.81s =================   
   

The "Hello, World" of Consensus Spec Tests

One of the test_empty_block_transition tests is implemented by a function with the same name located in ~/consensus-specs/tests/core/pyspec/eth2spec/test/phase0/sanity/test_blocks.py. To learn how consensus spec tests are written, let's go over the code:

@with_all_phases

This decorator specifies that this test is applicable to all the phases of consensus layer development. These phases are similar to forks (Istanbul, Berlin, London, etc.) in the execution blockchain. If you are interested, you can see the definition of this decorator here.

@spec_state_test

This decorator specifies that this test is a state transition test, and that it does not include a transition between different forks.

def test_empty_block_transition(spec, state):

This type of test receives two parameters:

• specs: The protocol specifications
• state: The genesis state before the test

    pre_slot = state.slot

A slot is a unit of time (every 12 seconds in mainnet), for which a specific validator (selected randomly but in a deterministic manner) is a proposer. The proposer can propose a block during that slot.

    pre_eth1_votes = len(state.eth1_data_votes)
    pre_mix = spec.get_randao_mix(state, spec.get_current_epoch(state))

Store some values to check later that certain updates happened.

    yield 'pre', state

In Python yield is used by generators. However, for our purposes we can treat it as a partial return statement that doesn't stop the function's processing, only adds to a list of return values. Here we add two values, the string 'pre' and the initial state, to the list of return values.

You can read more about test generators and how the are used here.

    block = build_empty_block_for_next_slot(spec, state)

The state contains the last block, which is necessary for building up the next block (every block needs to have the hash of the previous one in a blockchain).

    signed_block = state_transition_and_sign_block(spec, state, block)

Create a block signed by the appropriate proposer and advance the state.

    yield 'blocks', [signed_block]
    yield 'post', state

More yield statements. The output of a consensus test is:

1. 'pre'
2. The state before the test was run
3. 'blocks'
4. A list of signed blocks
5. 'post'
6. The state after the test

    # One vote for the eth1
    assert len(state.eth1_data_votes) == pre_eth1_votes + 1

    # Check that the new parent root is correct
    assert spec.get_block_root_at_slot(state, pre_slot) == signed_block.message.parent_root
    
    # Random data changed
    assert spec.get_randao_mix(state, spec.get_current_epoch(state)) != pre_mix

Finally we assertions that test the transition was legitimate. In this case we have three assertions:

1. One item was added to eth1_data_votes
2. The new block's parent_root is the same as the block in the previous location
3. The random data that every block includes was changed.

New Tests

The easiest way to write a new test is to copy and modify an existing one. For example, lets write a test where the first slot of the beacon chain is empty (because the assigned proposer is offline, for example), and then there's an empty block in the second slot.

We already know how to accomplish most of what we need for this test, but the only way we know to advance the state is state_transition_and_sign_block, a function that also puts a block into the slot. So let's see if the function's definition tells us how to advance the state without a block.

First, we need to find out where the function is located. Run:

find . -name '*.py' -exec grep 'def state_transition_and_sign_block' {} \; -print

And you'll find that the function is defined in eth2spec/test/helpers/state.py. Looking in that file, we see that the second function is:

def next_slot(spec, state):
    """
    Transition to the next slot.
    """
    spec.process_slots(state, state.slot + 1)

This looks like exactly what we need. So we add this call before we create the empty block:

.
.
.
    yield 'pre', state

    next_slot(spec, state)

    block = build_empty_block_for_next_slot(spec, state)
.
.
.

That's it. Our new test works (copy test_empty_block_transition, rename it, add the next_slot call, and then run it to verify this).

Tests Designed to Fail

It is important to make sure that the system rejects invalid input, so our next step is to deal with cases where the protocol is supposed to reject something. To see such a test, look at test_prev_slot_block_transition (in the same file we used previously, ~/consensus-specs/tests/core/pyspec/eth2spec/test/phase0/sanity/test_blocks.py).

@with_all_phases
@spec_state_test
def test_prev_slot_block_transition(spec, state):
    spec.process_slots(state, state.slot + 1)
    block = build_empty_block(spec, state, slot=state.slot)

Build an empty block for the current slot.

    proposer_index = spec.get_beacon_proposer_index(state)

Get the identity of the current proposer, the one for this slot.

    spec.process_slots(state, state.slot + 1)

Transition to the new slot, which naturally has a different proposer.

    yield 'pre', state
    expect_assertion_error(lambda: transition_unsigned_block(spec, state, block))

Specify that the function transition_unsigned_block will cause an assertion error. You can see this function in ~/consensus-specs/tests/core/pyspec/eth2spec/test/helpers/block.py, and one of the tests is that the block must be for this slot:

assert state.slot == block.slot

Because we use lambda notation, the test does not call transition_unsigned_block here. Instead, this is a function parameter that can be called later.

    block.state_root = state.hash_tree_root()

Set the block's state root to the current state hash tree root, which identifies this block as belonging to this slot (even though it was created for the previous slot).

    signed_block = sign_block(spec, state, block, proposer_index=proposer_index)

Notice that proposer_index is the variable we set earlier, before we advanced the slot with spec.process_slots(state, state.slot + 1). It is not the proposer for the current state.

    yield 'blocks', [signed_block]
    yield 'post', None   # No post state, signifying it errors out

This is the way we specify that a test is designed to fail - failed tests have no post state, because the processing mechanism errors out before creating it.

Attestation Tests

The consensus layer doesn't provide any direct functionality to end users. It does not execute EVM programs or store user data. It exists to provide a secure source of information about the latest verified block hash of the execution layer.

For every slot a validator is randomly selected as the proposer. The proposer proposes a block for the current head of the consensus layer chain (built on the previous block). That block includes the hash of the proposed new head of the execution layer.

For every slot there is also a randomly selected committee of validators that needs to vote whether the new consensus layer block is valid, which requires the proposed head of the execution chain to also be a valid block. These votes are called attestations, and they are sent as independent messages. The proposer for a block is able to include attestations from previous slots, which is how they get on chain to form consensus, reward honest validators, etc.

You can see a simple successful attestation test here: Lets go over it line by line.

@with_all_phases
@spec_state_test
def test_success(spec, state):
    attestation = get_valid_attestation(spec, state, signed=True)

This function creates a valid attestation (which can then be modified to make it invalid if needed). To see an attestion "from the inside" we need to follow it.

 def get_valid_attestation(spec,
                          state,
                          slot=None,
                          index=None,
                          filter_participant_set=None,
                          signed=False):

Only two parameters, spec and state are required. However, there are four other parameters that can affect the attestation created by this function.

    # If filter_participant_set filters everything, the attestation has 0 participants, and cannot be signed.
    # Thus strictly speaking invalid when no participant is added later.
    if slot is None:
        slot = state.slot
    if index is None:
        index = 0

Default values. Normally we want to choose the current slot, and out of the proposers and committees that it can have, we want the first one.

    attestation_data = build_attestation_data(
        spec, state, slot=slot, index=index
    )

Build the actual attestation. You can see this function here to see the exact data in an attestation.

   beacon_committee = spec.get_beacon_committee(
       state,
       attestation_data.slot,
       attestation_data.index,
   )

This is the committee that is supposed to approve or reject the proposed block.

    committee_size = len(beacon_committee)
    aggregation_bits = Bitlist[spec.MAX_VALIDATORS_PER_COMMITTEE](*([0] * committee_size))

There's a bit for every committee member to see if it approves or not.

    attestation = spec.Attestation(
        aggregation_bits=aggregation_bits,
        data=attestation_data,
    )
    # fill the attestation with (optionally filtered) participants, and optionally sign it
    fill_aggregate_attestation(spec, state, attestation, signed=signed, filter_participant_set=filter_participant_set)

   return attestation  

    next_slots(spec, state, spec.MIN_ATTESTATION_INCLUSION_DELAY)

Attestations have to appear after the block they attest for, so we advance spec.MIN_ATTESTATION_INCLUSION_DELAY slots before creating the block that includes the attestation. Currently a single block is sufficient, but that may change in the future.

    yield from run_attestation_processing(spec, state, attestation)

This function processes the attestation and returns the result.

Adding an Attestation Test

Attestations can't happen in the same block as the one about which they are attesting, or in a block that is after the block is finalized. This is specified as part of the specs, in the process_attestation function (which is created from the spec by the make pyspec command you ran earlier). Here is the relevant code fragment:

def process_attestation(state: BeaconState, attestation: Attestation) -> None:
    data = attestation.data
    assert data.target.epoch in (get_previous_epoch(state), get_current_epoch(state))
    assert data.target.epoch == compute_epoch_at_slot(data.slot)
    assert data.slot + MIN_ATTESTATION_INCLUSION_DELAY <= state.slot <= data.slot + SLOTS_PER_EPOCH
    ...

In the last line you can see two conditions being asserted:

1. data.slot + MIN_ATTESTATION_INCLUSION_DELAY <= state.slot which verifies that the attestation doesn't arrive too early.
2. state.slot <= data.slot + SLOTS_PER_EPOCH which verifies that the attestation doesn't arrive too late.

This is how the consensus layer tests deal with edge cases, by asserting the conditions required for the values to be legitimate. In the case of these particular conditions, they are tested here. One test checks what happens if the attestation is too early, and another if it is too late.

However, it is not enough to ensure we reject invalid blocks. It is also necessary to ensure we accept all valid blocks. You saw earlier a test (test_success) that tested that being MIN_ATTESTATION_INCLUSION_DELAY after the data for which we attest is enough. Now we'll write a similar test that verifies that being SLOTS_PER_EPOCH away is still valid. To do this, we modify the test_after_epoch_slots function. We need two changes:

1. Call transition_to_slot_via_block with one less slot to advance
2. Don't tell run_attestation_processing to return an empty post state.

The modified function is:

@with_all_phases
@spec_state_test
def test_almost_after_epoch_slots(spec, state):
    attestation = get_valid_attestation(spec, state, signed=True)

    # increment to latest inclusion slot (not beyond it)
    transition_to_slot_via_block(spec, state, state.slot + spec.SLOTS_PER_EPOCH)

    yield from run_attestation_processing(spec, state, attestation)

Add this function to the file consensus-specs/tests/core/pyspec/eth2spec/test/phase0/block_processing/test_process_attestation.py, and run the test:

cd ~/consensus-specs
. venv/bin/activate
cd tests/core/pyspec
python -m pytest -k almost_after --fork Merge eth2spec

You should see it ran successfully (although you might get a warning, you can ignore it)

How are These Tests Used?

So far we've ran tests against the formal specifications. This is a way to check the specifications are what we expect, but it doesn't actually check the beacon chain clients. The way these tests get applied by clients is that every few weeks new test specifications are released, in a format documented here. All the consensus layer clients implement test-runners that consume the test vectors in this standard format.

---

Original version by Ori Pomerantz