Testing pipes

@subsquid/pipes/testing provides utilities for testing pipes end-to-end without hitting a real portal:

mockPortal(responses) / finalizedMockPortal(responses): spin up a local HTTP server that plays back scripted portal responses. Both resolve to { url, close() }; pass url as the stream’s portal.
readAll(stream): drain a stream into a flat array of data items.

@subsquid/pipes/testing/evm adds typed EVM helpers on top:

encodeEvent({ abi, eventName, address, args }): encode an event log from a viem-style ABI. args are fully typed from the ABI.
mockBlock({ transactions }): build a portal block; metadata (number, hash, timestamp) is auto-generated with an incrementing block counter.
resetMockBlockCounter(): reset the auto-incrementing block numbers between tests.
mockEvmPortalStream({ blocks, finalized? }): shorthand that wraps blocks in a mock portal serving one 200 response.

Example (Vitest)

import { commonAbis, evmEventDecoder, evmPortalStream } from '@subsquid/pipes/evm'
import { type MockPortal, readAll } from '@subsquid/pipes/testing'
import {
  encodeEvent,
  mockEvmPortalStream,
  mockBlock,
  resetMockBlockCounter,
} from '@subsquid/pipes/testing/evm'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'

const ERC20_ABI = [
  {
    type: 'event' as const,
    name: 'Transfer',
    inputs: [
      { name: 'from', type: 'address', indexed: true },
      { name: 'to', type: 'address', indexed: true },
      { name: 'value', type: 'uint256', indexed: false },
    ],
  },
] as const

const USDC = '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48' as const
const ALICE = '0x7a250d5630b4cf539739df2c5dacb4c659f2488d' as const
const BOB = '0xc82e11e709deb68f3631fc165ebd8b4e3fc3d18f' as const

describe('EVM pipe', () => {
  let portal: MockPortal

  beforeEach(() => {
    resetMockBlockCounter()
  })

  afterEach(async () => {
    await portal?.close()
  })

  it('decodes ERC20 transfers from mock blocks', async () => {
    // 1. Encode events; args are fully typed from the ABI
    const transfer = encodeEvent({
      abi: ERC20_ABI,
      eventName: 'Transfer',
      address: USDC,
      args: { from: ALICE, to: BOB, value: 1_000_000n },
    })

    // 2. Build mock blocks and serve them from a mock portal
    portal = await mockEvmPortalStream({
      blocks: [mockBlock({ transactions: [{ logs: [transfer] }] })],
    })

    // 3. Create the pipe exactly as in production, with the mock portal URL
    const stream = evmPortalStream({
      id: 'test',
      portal: portal.url,
      outputs: evmEventDecoder({
        range: { from: 0, to: 1 },
        events: { transfers: commonAbis.erc20.events.Transfer },
      }),
    }).pipe((batch) => batch.transfers)

    // 4. Collect all output and assert
    const transfers = await readAll(stream)

    expect(transfers).toHaveLength(1)
    expect(transfers[0].event.from).toBe(ALICE)
    expect(transfers[0].event.value).toBe(1_000_000n)
    expect(transfers[0].contract).toBe(USDC)
  })
})

Custom .pipe() transforms are tested the same way: chain them onto the stream and assert on the collected output. Full runnable example: 14.writing-testing.example.ts.

Mock response types

mockPortal accepts an array of scripted responses, played back in order:

type MockResponse =
  | { statusCode: 204 }                                                  // empty response
  | { statusCode: 200, data: [...], head?: { finalized?, latest? } }     // data blocks
  | { statusCode: 409, data: { previousBlocks: [...] } }                 // fork signal
  | { statusCode: 500 | 503 }                                            // server error

The 409 variant lets you test fork handling deterministically, and 500/503 let you exercise retry behavior.

Overview

Pipes SDK

Squid SDK

Alternative clients

Example (Vitest)

Mock response types

​Example (Vitest)

​Mock response types

Example (Vitest)

Mock response types