test_storage_access_cold()¶

@pytest.mark.parametrize(
    "storage_action,tx_result",
    [
        pytest.param(
            StorageAction.READ, TransactionResult.SUCCESS, id="SSLOAD"
        ),
        pytest.param(
            StorageAction.WRITE_SAME_VALUE,
            TransactionResult.SUCCESS,
            id="SSTORE same value",
        ),
        pytest.param(
            StorageAction.WRITE_SAME_VALUE,
            TransactionResult.REVERT,
            id="SSTORE same value, revert",
        ),
        pytest.param(
            StorageAction.WRITE_SAME_VALUE,
            TransactionResult.OUT_OF_GAS,
            id="SSTORE same value, out of gas",
        ),
        pytest.param(
            StorageAction.WRITE_NEW_VALUE,
            TransactionResult.SUCCESS,
            id="SSTORE new value",
        ),
        pytest.param(
            StorageAction.WRITE_NEW_VALUE,
            TransactionResult.REVERT,
            id="SSTORE new value, revert",
        ),
        pytest.param(
            StorageAction.WRITE_NEW_VALUE,
            TransactionResult.OUT_OF_GAS,
            id="SSTORE new value, out of gas",
        ),
    ],
)
@pytest.mark.parametrize("absent_slots", [True, False])
def test_storage_access_cold(
    benchmark_test: BenchmarkTestFiller,
    pre: Alloc,
    fork: Fork,
    storage_action: StorageAction,
    absent_slots: bool,
    gas_benchmark_value: int,
    tx_gas_limit: int,
    tx_result: TransactionResult,
) -> None:
    """
    Benchmark cold storage slot accesses using EIP-7702 delegation.

    The authority EOA delegates to:
    - StorageInitializer: storage[i] = i for each slot (absent_slots=False)
    - BenchmarkExecutor: performs the benchmark operation (SLOAD/SSTORE)
    """
    executor_code = create_benchmark_executor(
        storage_action, absent_slots, tx_result
    )
    initializer_code = create_storage_initializer()

    authority = pre.fund_eoa(amount=0)
    initializer_addr = pre.deploy_contract(code=initializer_code)
    executor_addr = pre.deploy_contract(code=executor_code)

    # Calldata generator for both the executor and initializer.
    def calldata_generator(
        iteration_count: int, start_iteration: int
    ) -> bytes:
        return Hash(start_iteration) + Hash(iteration_count)

    # Number of slots that can be processed in the execution phase
    num_target_slots = sum(
        executor_code.tx_iterations_by_gas_limit(
            fork=fork,
            gas_limit=gas_benchmark_value,
            calldata=calldata_generator,
            recipient_type=RecipientType.DELEGATION_7702,
            outcome=tx_result,
        )
    )

    blocks = []
    delegation_sender = pre.fund_eoa()

    # Setup phase: initialize storage slots (only if absent_slots=False)
    with TestPhaseManager.setup():
        setup_txs = []
        authority_nonce = 0
        if not absent_slots:
            setup_txs.append(
                Transaction(
                    to=delegation_sender,
                    gas_limit=tx_gas_limit,
                    sender=delegation_sender,
                    authorization_list=[
                        AuthorizationTuple(
                            address=initializer_addr,
                            nonce=authority_nonce,
                            signer=authority,
                        ),
                    ],
                )
            )
            authority_nonce += 1

            setup_txs += list(
                initializer_code.transactions_by_total_iteration_count(
                    fork=fork,
                    total_iterations=num_target_slots,
                    sender=pre.fund_eoa(),
                    to=authority,
                    start_iteration=1,
                    calldata=calldata_generator,
                    recipient_type=RecipientType.DELEGATION_7702,
                )
            )

        setup_txs.append(
            Transaction(
                to=delegation_sender,
                gas_limit=tx_gas_limit,
                sender=delegation_sender,
                authorization_list=[
                    AuthorizationTuple(
                        address=executor_addr,
                        nonce=authority_nonce,
                        signer=authority,
                    ),
                ],
            )
        )
        blocks.append(Block(txs=setup_txs))

    # Execution phase: run benchmark
    # For absent_slots=False, authority has storage, triggering refund
    expected_gas_used = 0

    with TestPhaseManager.execution():
        # One gas short so the out-of-gas variants cannot terminate cleanly.
        tx_gas_limit_delta = (
            -1 if tx_result == TransactionResult.OUT_OF_GAS else 0
        )
        exec_txs = list(
            executor_code.transactions_by_gas_limit(
                fork=fork,
                gas_limit=gas_benchmark_value,
                sender=pre.fund_eoa(),
                to=authority,
                calldata=calldata_generator,
                start_iteration=1,
                recipient_type=RecipientType.DELEGATION_7702,
                tx_gas_limit_delta=tx_gas_limit_delta,
                outcome=tx_result,
            )
        )
        for exec_tx in exec_txs:
            expected_gas_used += exec_tx.gas_cost

    blocks.append(Block(txs=exec_txs))

    benchmark_test(
        blocks=blocks,
        expected_benchmark_gas_used=expected_gas_used,
    )