Test Blobhash Opcode

Documentation for tests/cancun/eip4844_blobs/test_blobhash_opcode.py.

Generate fixtures for these test cases for Cancun with:

Cancun only:

fill -v tests/cancun/eip4844_blobs/test_blobhash_opcode.py --fork=Cancun --evm-bin=/path/to/evm-tool-dev-version

For all forks up to and including Cancun:

fill -v tests/cancun/eip4844_blobs/test_blobhash_opcode.py --until=Cancun --evm-bin=/path/to/evm-tool-dev-version

Tests BLOBHASH opcode in EIP-4844: Shard Blob Transactions

Test cases for the BLOBHASH opcode in EIP-4844: Shard Blob Transactions.

Adding a new test

Add a function that is named test_<test_name> and takes at least the following arguments:

• blockchain_test
• pre
• tx
• post

Additional custom pytest.fixture fixtures can be added and parametrized for new test cases.

There is no specific structure to follow within this test module.

Spec dataclass

Parameters from the EIP-4844 specifications as defined at https://eips.ethereum.org/EIPS/eip-4844#parameters

If the parameter is not currently used within the tests, it is commented out.

Source code in tests/cancun/eip4844_blobs/spec.py

@dataclass(frozen=True)
class Spec:
    """
    Parameters from the EIP-4844 specifications as defined at
    https://eips.ethereum.org/EIPS/eip-4844#parameters

    If the parameter is not currently used within the tests, it is commented
    out.
    """

    BLOB_TX_TYPE = 0x03
    FIELD_ELEMENTS_PER_BLOB = 4096
    BLS_MODULUS = 0x73EDA753299D7D483339D80809A1D80553BDA402FFFE5BFEFFFFFFFF00000001
    BLOB_COMMITMENT_VERSION_KZG = 1
    POINT_EVALUATION_PRECOMPILE_ADDRESS = 20
    POINT_EVALUATION_PRECOMPILE_GAS = 50_000
    MAX_DATA_GAS_PER_BLOCK = 786432
    TARGET_DATA_GAS_PER_BLOCK = 393216
    MIN_DATA_GASPRICE = 1
    DATA_GASPRICE_UPDATE_FRACTION = 3338477
    # MAX_VERSIONED_HASHES_LIST_SIZE = 2**24
    # MAX_CALLDATA_SIZE = 2**24
    # MAX_ACCESS_LIST_SIZE = 2**24
    # MAX_ACCESS_LIST_STORAGE_KEYS = 2**24
    # MAX_TX_WRAP_COMMITMENTS = 2**12
    # LIMIT_BLOBS_PER_TX = 2**12
    DATA_GAS_PER_BLOB = 2**17
    HASH_OPCODE_BYTE = 0x49
    HASH_GAS_COST = 3

    @classmethod
    def kzg_to_versioned_hash(
        cls,
        kzg_commitment: bytes | int,  # 48 bytes
        blob_commitment_version_kzg: Optional[bytes | int] = None,
    ) -> bytes:
        """
        Calculates the versioned hash for a given KZG commitment.
        """
        if blob_commitment_version_kzg is None:
            blob_commitment_version_kzg = cls.BLOB_COMMITMENT_VERSION_KZG
        if isinstance(kzg_commitment, int):
            kzg_commitment = kzg_commitment.to_bytes(48, "big")
        if isinstance(blob_commitment_version_kzg, int):
            blob_commitment_version_kzg = blob_commitment_version_kzg.to_bytes(1, "big")
        return blob_commitment_version_kzg + sha256(kzg_commitment).digest()[1:]

    @classmethod
    def fake_exponential(cls, factor: int, numerator: int, denominator: int) -> int:
        """
        Used to calculate the data gas cost.
        """
        i = 1
        output = 0
        numerator_accumulator = factor * denominator
        while numerator_accumulator > 0:
            output += numerator_accumulator
            numerator_accumulator = (numerator_accumulator * numerator) // (denominator * i)
            i += 1
        return output // denominator

    @classmethod
    def calc_excess_data_gas(cls, parent: BlockHeaderDataGasFields) -> int:
        """
        Calculate the excess data gas for a block given the excess data gas
        and data gas used from the parent block header.
        """
        if parent.excess_data_gas + parent.data_gas_used < cls.TARGET_DATA_GAS_PER_BLOCK:
            return 0
        else:
            return parent.excess_data_gas + parent.data_gas_used - cls.TARGET_DATA_GAS_PER_BLOCK

    # Note: Currently unused.
    # @classmethod
    # def get_total_data_gas(cls, tx: Transaction) -> int:
    #     """
    #     Calculate the total data gas for a transaction.
    #     """
    #     if tx.blob_versioned_hashes is None:
    #         return 0
    #     return cls.DATA_GAS_PER_BLOB * len(tx.blob_versioned_hashes)

    @classmethod
    def get_data_gasprice(cls, *, excess_data_gas: int) -> int:
        """
        Calculate the data gas price from the excess.
        """
        return cls.fake_exponential(
            cls.MIN_DATA_GASPRICE,
            excess_data_gas,
            cls.DATA_GASPRICE_UPDATE_FRACTION,
        )

test_blobhash_gas_cost(pre, template_tx, blocks, post, tx_type, blockchain_test)

Tests BLOBHASH opcode gas cost using a variety of indexes.

Asserts that the gas consumption of the BLOBHASH opcode is correct by ensuring it matches HASH_OPCODE_GAS = 3. Includes both valid and invalid random index sizes from the range [0, 2**256-1], for tx types 2 and 3.

Source code in tests/cancun/eip4844_blobs/test_blobhash_opcode.py

@pytest.mark.parametrize("tx_type", [0, 1, 2, 3])
def test_blobhash_gas_cost(
    pre,
    template_tx,
    blocks,
    post,
    tx_type,
    blockchain_test: BlockchainTestFiller,
):
    """
    Tests `BLOBHASH` opcode gas cost using a variety of indexes.

    Asserts that the gas consumption of the `BLOBHASH` opcode is correct by ensuring
    it matches `HASH_OPCODE_GAS = 3`. Includes both valid and invalid random
    index sizes from the range `[0, 2**256-1]`, for tx types 2 and 3.
    """
    assert (
        Op.BLOBHASH.int() == Spec.HASH_OPCODE_BYTE
    ), "Opcodes blobhash byte doesn't match that defined in the spec"
    gas_measures_code = [
        CodeGasMeasure(
            code=Op.BLOBHASH(i),
            overhead_cost=3,
            extra_stack_items=1,
        )
        for i in blobhash_index_values
    ]
    for i, gas_code in enumerate(gas_measures_code):
        address = to_address(0x100 + i * 0x100)
        pre[address] = Account(code=gas_code)
        blocks.append(
            Block(
                txs=[
                    template_tx.with_fields(
                        ty=tx_type,
                        nonce=i,
                        to=address,
                        gas_price=10 if tx_type < 2 else None,
                        access_list=[] if tx_type >= 2 else None,
                        max_priority_fee_per_gas=10 if tx_type >= 2 else None,
                        blob_versioned_hashes=random_blob_hashes[
                            0 : SpecHelpers.target_blobs_per_block()
                        ]
                        if tx_type >= 3
                        else None,
                    )
                ]
            )
        )
        post[address] = Account(storage={0: Spec.HASH_GAS_COST})
    blockchain_test(
        pre=pre,
        blocks=blocks,
        post=post,
    )

SpecHelpers dataclass

Define parameters and helper functions that are tightly coupled to the 4844 spec but not strictly part of it.

Source code in tests/cancun/eip4844_blobs/spec.py

@dataclass(frozen=True)
class SpecHelpers:
    """
    Define parameters and helper functions that are tightly coupled to the 4844
    spec but not strictly part of it.
    """

    BYTES_PER_FIELD_ELEMENT = 32

    @classmethod
    def max_blobs_per_block(cls) -> int:  # MAX_BLOBS_PER_BLOCK =
        """
        Returns the maximum number of blobs per block.
        """
        return Spec.MAX_DATA_GAS_PER_BLOCK // Spec.DATA_GAS_PER_BLOB

    @classmethod
    def target_blobs_per_block(cls) -> int:
        """
        Returns the target number of blobs per block.
        """
        return Spec.TARGET_DATA_GAS_PER_BLOCK // Spec.DATA_GAS_PER_BLOB

    @classmethod
    def calc_excess_data_gas_from_blob_count(
        cls, parent_excess_data_gas: int, parent_blob_count: int
    ) -> int:
        """
        Calculate the excess data gas for a block given the parent excess data gas
        and the number of blobs in the block.
        """
        parent_consumed_data_gas = parent_blob_count * Spec.DATA_GAS_PER_BLOB
        return Spec.calc_excess_data_gas(
            BlockHeaderDataGasFields(parent_excess_data_gas, parent_consumed_data_gas)
        )

    @classmethod
    def get_min_excess_data_gas_for_data_gas_price(cls, data_gas_price: int) -> int:
        """
        Gets the minimum required excess data gas value to get a given data gas cost in a block
        """
        current_excess_data_gas = 0
        current_data_gas_price = 1
        while current_data_gas_price < data_gas_price:
            current_excess_data_gas += Spec.DATA_GAS_PER_BLOB
            current_data_gas_price = Spec.get_data_gasprice(
                excess_data_gas=current_excess_data_gas
            )
        return current_excess_data_gas

    @classmethod
    def get_min_excess_data_blobs_for_data_gas_price(cls, data_gas_price: int) -> int:
        """
        Gets the minimum required excess data blobs to get a given data gas cost in a block
        """
        return (
            cls.get_min_excess_data_gas_for_data_gas_price(data_gas_price)
            // Spec.DATA_GAS_PER_BLOB
        )

test_blobhash_scenarios(pre, template_tx, blocks, post, scenario, blockchain_test)

Tests that the BLOBHASH opcode returns the correct versioned hash for various valid indexes.

Covers various scenarios with random blob_versioned_hash values within the valid range [0, 2**256-1].

Source code in tests/cancun/eip4844_blobs/test_blobhash_opcode.py

@pytest.mark.parametrize(
    "scenario",
    [
        "single_valid",
        "repeated_valid",
        "valid_invalid",
        "varied_valid",
    ],
)
def test_blobhash_scenarios(
    pre,
    template_tx,
    blocks,
    post,
    scenario: str,
    blockchain_test: BlockchainTestFiller,
):
    """
    Tests that the `BLOBHASH` opcode returns the correct versioned hash for
    various valid indexes.

    Covers various scenarios with random `blob_versioned_hash` values within
    the valid range `[0, 2**256-1]`.
    """
    TOTAL_BLOCKS = 5
    b_hashes_list = BlobhashScenario.create_blob_hashes_list(length=TOTAL_BLOCKS)
    blobhash_calls = BlobhashScenario.generate_blobhash_bytecode(scenario)
    for i in range(TOTAL_BLOCKS):
        address = to_address(0x100 + i * 0x100)
        pre[address] = Account(code=blobhash_calls)
        blocks.append(
            Block(
                txs=[
                    template_tx.with_fields(
                        ty=Spec.BLOB_TX_TYPE,
                        nonce=i,
                        to=address,
                        access_list=[],
                        max_priority_fee_per_gas=10,
                        blob_versioned_hashes=b_hashes_list[i],
                    )
                ]
            )
        )
        post[address] = Account(
            storage={
                index: b_hashes_list[i][index]
                for index in range(SpecHelpers.max_blobs_per_block())
            }
        )
    blockchain_test(
        pre=pre,
        blocks=blocks,
        post=post,
    )

test_blobhash_invalid_blob_index(pre, template_tx, blocks, post, blockchain_test, scenario)

Tests that the BLOBHASH opcode returns a zeroed bytes32 value for invalid indexes.

Includes cases where the index is negative (index < 0) or exceeds the maximum number of blob_versioned_hash values stored: (index >= len(tx.message.blob_versioned_hashes)).

It confirms that the returned value is a zeroed bytes32 for each case.

Source code in tests/cancun/eip4844_blobs/test_blobhash_opcode.py

@pytest.mark.parametrize(
    "scenario",
    [
        "invalid_calls",
    ],
)
def test_blobhash_invalid_blob_index(
    pre,
    template_tx,
    blocks,
    post,
    blockchain_test: BlockchainTestFiller,
    scenario,
):
    """
    Tests that the `BLOBHASH` opcode returns a zeroed `bytes32` value for invalid
    indexes.

    Includes cases where the index is negative (`index < 0`) or
    exceeds the maximum number of `blob_versioned_hash` values stored:
    (`index >= len(tx.message.blob_versioned_hashes)`).

    It confirms that the returned value is a zeroed `bytes32` for each case.
    """
    TOTAL_BLOCKS = 5
    blobhash_calls = BlobhashScenario.generate_blobhash_bytecode(scenario)
    for i in range(TOTAL_BLOCKS):
        address = to_address(0x100 + i * 0x100)
        pre[address] = Account(code=blobhash_calls)
        blob_per_block = (i % SpecHelpers.max_blobs_per_block()) + 1
        blobs = [random_blob_hashes[blob] for blob in range(blob_per_block)]
        blocks.append(
            Block(
                txs=[
                    template_tx.with_fields(
                        ty=Spec.BLOB_TX_TYPE,
                        nonce=i,
                        to=address,
                        access_list=[],
                        max_priority_fee_per_gas=10,
                        blob_versioned_hashes=blobs,
                    )
                ]
            )
        )
        post[address] = Account(
            storage={
                index: (0 if index < 0 or index >= blob_per_block else blobs[index])
                for index in range(
                    -TOTAL_BLOCKS,
                    blob_per_block + (TOTAL_BLOCKS - (i % SpecHelpers.max_blobs_per_block())),
                )
            }
        )
    blockchain_test(
        pre=pre,
        blocks=blocks,
        post=post,
    )

test_blobhash_multiple_txs_in_block(pre, blob_tx, post, blockchain_test)

Tests that the BLOBHASH opcode returns the appropriate values when there is more than 1 blob tx type within a block (for tx types 2 and 3).

Scenarios involve tx type 3 followed by tx type 2 running the same code within a block, including the opposite.

Source code in tests/cancun/eip4844_blobs/test_blobhash_opcode.py

def test_blobhash_multiple_txs_in_block(
    pre,
    blob_tx,
    post,
    blockchain_test: BlockchainTestFiller,
):
    """
    Tests that the `BLOBHASH` opcode returns the appropriate values when there
    is more than 1 blob tx type within a block (for tx types 2 and 3).

    Scenarios involve tx type 3 followed by tx type 2 running the same code
    within a block, including the opposite.
    """
    blobhash_bytecode = BlobhashScenario.generate_blobhash_bytecode("single_valid")
    pre = {
        **pre,
        **{
            to_address(address): Account(code=blobhash_bytecode)
            for address in range(0x100, 0x500, 0x100)
        },
    }
    blocks = [
        Block(
            txs=[
                blob_tx(address=to_address(0x100), type=3, nonce=0),
                blob_tx(address=to_address(0x100), type=2, nonce=1),
            ]
        ),
        Block(
            txs=[
                blob_tx(address=to_address(0x200), type=2, nonce=2),
                blob_tx(address=to_address(0x200), type=3, nonce=3),
            ]
        ),
        Block(
            txs=[
                blob_tx(address=to_address(0x300), type=2, nonce=4),
                blob_tx(address=to_address(0x400), type=3, nonce=5),
            ],
        ),
    ]
    post = {
        to_address(address): Account(
            storage={i: random_blob_hashes[i] for i in range(SpecHelpers.max_blobs_per_block())}
        )
        if address in (0x200, 0x400)
        else Account(storage={i: 0 for i in range(SpecHelpers.max_blobs_per_block())})
        for address in range(0x100, 0x500, 0x100)
    }
    blockchain_test(
        pre=pre,
        blocks=blocks,
        post=post,
    )