Testing

This reference covers testing strategies for Substreams applications, from unit testing individual handlers to integration testing complete modules with real blockchain data.

Tip: The substreams-testing agent skill gives your AI coding assistant expert knowledge on unit testing, integration testing, performance testing, and CI/CD patterns for Substreams.

Why Testing Matters

Blockchain data is immutable and mistakes are permanent. Testing is critical because:

• Small bugs amplify across millions of blocks

• Complex transformations introduce edge cases

• DeFi applications depend on data accuracy

• Parallel execution can create race conditions

Unit Testing

The substreams::testing module provides utilities for testing map handlers directly without WASM compilation overhead.

The map! Macro

The substreams::testing::map! macro allows you to invoke map handlers directly in unit tests:

use substreams::errors::Error;
use substreams_ethereum::pb::eth::v2::Block;

mod pb;
use pb::myproject::{Events, Event};

#[substreams::handlers::map]
pub fn map_events(block: Block) -> Result<Events, Error> {
    let events: Vec<Event> = block.logs()
        .filter(|log| !log.topics.is_empty())
        .filter_map(|log| parse_event(log).ok())
        .collect();

    Ok(Events { events })
}

#[substreams::handlers::map]
pub fn filter_events(
    event_type: String,
    events: Events,
) -> Result<Events, Error> {
    let filtered: Vec<Event> = events.events
        .into_iter()
        .filter(|e| e.event_type == event_type)
        .collect();

    Ok(Events { events: filtered })
}

#[cfg(test)]
mod tests {
    use super::*;
    use substreams::testing;

    #[test]
    fn test_map_events() {
        // Arrange
        let block = create_test_block();

        // Act
        let result = testing::map!(map_events(block)).unwrap();

        // Assert
        assert!(!result.events.is_empty());
    }

    #[test]
    fn test_map_events_empty_block() {
        let empty_block = Block::default();

        let result = testing::map!(map_events(empty_block)).unwrap();

        assert!(result.events.is_empty());
    }

    #[test]
    fn test_chained_handlers() {
        let block = create_test_block();

        // Chain multiple handlers together
        let all_events = testing::map!(map_events(block)).unwrap();
        let transfers = testing::map!(filter_events("transfer".to_string(), all_events)).unwrap();

        assert!(transfers.events.iter().all(|e| e.event_type == "transfer"));
    }

    fn create_test_block() -> Block {
        // Create a test block with sample data
        Block {
            number: 17000000,
            // ... populate with test data
            ..Default::default()
        }
    }
}

The clock() Function

For modules that depend on block time or need clock information:

use substreams::testing;

#[test]
fn test_time_dependent_logic() {
    let clock = testing::clock();

    // Use clock in time-dependent tests
    let result = process_with_time(data, clock);

    assert!(result.is_ok());
}

Opting Out of Testable Generation

Map handlers automatically generate testable functions that the map! macro uses. In rare cases where you don't want this behavior:

#[substreams::handlers::map(no_testable)]
pub fn my_handler(block: Block) -> Result<Output, Error> {
    // This handler won't generate a testable function
    // You'll need to extract logic to separate functions for testing
}

Test Data

Using Real Blockchain Data

Use the firecore tool to extract real blocks for testing:

firecore tools firehose-single-block-client mainnet.eth.streamingfast.io:443 17000000 \
    --output=bytes \
    --bytes-encoding=base64 \
    > ./src/testdata/ethereum_mainnet_17000000.binpb.base64

Load the block in your tests:

use buffa::Message;
use std::fs;
use substreams_ethereum::pb::eth::v2::Block;

fn load_test_block(path: &str) -> Block {
    let base64_data = fs::read_to_string(path).expect("Failed to read test file");
    let bytes = base64::decode(base64_data.trim()).expect("Failed to decode base64");
    Block::decode_from_slice(&bytes).expect("Failed to decode protobuf")
}

#[test]
fn test_with_real_block() {
    let block = load_test_block("./src/testdata/ethereum_mainnet_17000000.binpb.base64");

    let result = testing::map!(map_events(block)).unwrap();

    // Assertions based on known block contents
    assert!(result.events.len() > 0);
}

Constructing Test Blocks

For controlled testing scenarios, construct blocks programmatically:

use buffa::MessageField;
use substreams_ethereum::pb::eth::v2::{Block, TransactionTrace, TransactionReceipt, Log};

fn create_block_with_transfer() -> Block {
    Block {
        number: 17000000,
        hash: hex::decode("abc123...").unwrap(),
        transaction_traces: vec![
            TransactionTrace {
                hash: hex::decode("def456...").unwrap(),
                receipt: MessageField::some(TransactionReceipt {
                    logs: vec![
                        Log {
                            address: hex::decode("a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48").unwrap(),
                            topics: vec![
                                // Transfer(address,address,uint256)
                                hex::decode("ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef").unwrap(),
                            ],
                            data: vec![],
                            ..Default::default()
                        }
                    ],
                    ..Default::default()
                }),
                ..Default::default()
            }
        ],
        ..Default::default()
    }
}

Test Data Builder Pattern

For complex test scenarios, use a builder pattern:

struct TestBlockBuilder {
    block: Block,
}

impl TestBlockBuilder {
    fn new(number: u64) -> Self {
        Self {
            block: Block {
                number,
                ..Default::default()
            }
        }
    }

    fn with_transfer(mut self, from: &str, to: &str, amount: u64) -> Self {
        // Add transfer log to block
        self
    }

    fn build(self) -> Block {
        self.block
    }
}

#[test]
fn test_with_builder() {
    let block = TestBlockBuilder::new(17000000)
        .with_transfer("0xabc...", "0xdef...", 1000)
        .build();

    let result = testing::map!(map_events(block)).unwrap();
    assert_eq!(result.events.len(), 1);
}

Integration Testing

Test complete modules with real blockchain data to verify behavior across multiple blocks.

#[test]
fn test_balance_consistency() {
    let blocks = vec![
        load_test_block("./testdata/block_17000000.binpb.base64"),
        load_test_block("./testdata/block_17000001.binpb.base64"),
        load_test_block("./testdata/block_17000002.binpb.base64"),
    ];

    let mut total_transfers = 0u64;

    for block in blocks {
        let result = testing::map!(map_transfers(block)).unwrap();
        total_transfers += result.transfers.len() as u64;
    }

    assert!(total_transfers > 0, "Expected transfers across block range");
}

End-to-End Testing

Execute complete Substreams in the real execution environment using the CLI:

# Run module on a block range
substreams run -s 17000000 -t +100 map_events --network mainnet

# Validate with production mode caching
substreams run -s 17000000 -t +1000 map_events --production-mode

Wrap CLI commands in Rust tests for automated validation:

use std::process::Command;

#[test]
#[ignore] // Run with: cargo test -- --ignored
fn test_e2e_execution() {
    let output = Command::new("substreams")
        .args([
            "run",
            "-s", "17000000",
            "-t", "+10",
            "map_events",
            "--network", "mainnet",
            "-o", "json"
        ])
        .output()
        .expect("Failed to execute substreams");

    assert!(output.status.success());

    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("events"));
}

Performance Testing

Benchmarking with Criterion

Use the Criterion crate for reliable benchmarks:

// benches/module_bench.rs
use criterion::{black_box, criterion_group, criterion_main, Criterion};

fn bench_map_events(c: &mut Criterion) {
    let block = load_heavy_block(); // Block with many transactions

    c.bench_function("map_events", |b| {
        b.iter(|| {
            testing::map!(map_events(black_box(block.clone())))
        })
    });
}

criterion_group!(benches, bench_map_events);
criterion_main!(benches);

Production Mode Validation

Test with production mode to validate caching behavior and performance:

time substreams run -s 17000000 -t +1000 map_events --production-mode

Testing Best Practices

Test Structure

Follow the Given-When-Expect pattern for clarity:

#[test]
fn test_filter_events_by_address() {
    // Given: A block with multiple events
    let block = create_block_with_mixed_events();
    let target_address = "0x5acc84a3e955bdd76467d3348077d003f00ffb97";

    // When: Filtering by address
    let all_events = testing::map!(map_events(block)).unwrap();
    let filtered = testing::map!(filter_by_address(target_address.to_string(), all_events)).unwrap();

    // Expect: Only matching events returned
    assert!(filtered.events.len() > 0);
    for event in &filtered.events {
        assert_eq!(hex::encode(&event.address), target_address.trim_start_matches("0x"));
    }
}

Error Scenario Testing

Test malformed data and edge cases:

#[test]
fn test_handles_empty_topics() {
    let block = create_block_with_log(Log {
        topics: vec![], // No topics
        ..Default::default()
    });

    let result = testing::map!(map_events(block)).unwrap();

    // Should handle gracefully, not panic
    assert!(result.events.is_empty());
}

#[test]
fn test_handles_invalid_address() {
    let block = create_block_with_log(Log {
        address: vec![0x00], // Invalid: too short
        ..Default::default()
    });

    let result = testing::map!(map_events(block));

    // Verify error handling
    assert!(result.is_err() || result.unwrap().events.is_empty());
}

Reorganization Testing

For stores that maintain state, test chain reorganization scenarios:

#[test]
fn test_store_handles_reorg() {
    // Process blocks on main chain
    let block_a = create_block(100, "hash_a");
    let block_b = create_block(101, "hash_b");

    // Simulate reorg: different block at same height
    let block_b_prime = create_block(101, "hash_b_prime");

    // Verify store correctly handles the reorganization
    // ...
}

CI/CD Integration

GitHub Actions Example

name: Test Substreams

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: dtolnay/rust-action@stable
        with:
          targets: wasm32-unknown-unknown

      - name: Run unit tests
        run: cargo test

      - name: Check WASM compilation
        run: cargo build --target wasm32-unknown-unknown --release

      - name: Run clippy
        run: cargo clippy -- -D warnings

      - name: Check formatting
        run: cargo fmt -- --check

Pre-commit Hooks

Automate quality checks before commits:

#!/bin/bash
# .git/hooks/pre-commit

cargo fmt -- --check || exit 1
cargo clippy -- -D warnings || exit 1
cargo test || exit 1
cargo build --target wasm32-unknown-unknown --release || exit 1

Summary

Test Level	Purpose	Tools
Unit	Test individual handlers	substreams::testing::map!
Integration	Test with real blockchain data	firecore, test fixtures
End-to-End	Validate complete pipeline	substreams CLI
Performance	Benchmark and optimize	Criterion, production mode

Start with comprehensive unit tests, add integration tests with real data, and validate with end-to-end tests. Automate everything through CI/CD.

Cargo.toml Requirements

[dependencies]
buffa = "0.9.2"
substreams = "0.8.0"
substreams-ethereum = "0.12.0"  # Or appropriate chain package

[dev-dependencies]
hex = "0.4"
base64 = "0.21"

# For benchmarks
[dev-dependencies.criterion]
version = "0.5"
features = ["html_reports"]

[[bench]]
name = "module_bench"
harness = false