Test a source

Use fixture responses and the in-memory adapters to test source behavior before connecting the reader to a live destination.

Choose the test boundary

Boundary	Use for	Does not establish
Reader plus fixture client	Mapping, pagination termination, time bounds, and provider errors	Durable checkpoint ordering
runIngestion plus memory adapters	Commit ordering, failures, resumption, and selected stream behavior	Actual ClickHouse encoding, engine merges, or deduplication-window settings
An isolated development database	Schema compatibility, JSON encoding, and query results after replay	Every future provider response or production load profile

Import the testing adapters from @chkit/plugin-ingest/testing. runIngestion, selectStreams, and the ClickHouse runtime adapters are public APIs; embedding them means the host supplies the journal, destination, cancellation, and other runtime settings normally provided by the CLI.

Example: a lost acknowledgement

Save as src/ingestion.test.ts and run bun test src/ingestion.test.ts. The memory destination accepts the second row, then the simulated transport reports failure. A new run must resume from the previous durable checkpoint and replay the ambiguous write without adding a duplicate:

import { expect, test } from 'bun:test'
import { cursorState, definePipeline, defineStream, rawRows, rawTable, runIngestion, selectStreams } from '@chkit/plugin-ingest'
import { createMemoryDestination, createMemoryJournal } from '@chkit/plugin-ingest/testing'

test('replays an acknowledged write whose response was lost', async () => {
  const source = [{ id: '1' }, { id: '2' }, { id: '3' }]
  const stream = defineStream({
    id: 'fixture.items',
    destination: rawTable({ database: 'default', name: 'items_raw' }),
    batchSize: 1,
    incremental: cursorState({
      id: 'fixture.offset', version: 1,
      parse(raw) {
        if (typeof raw !== 'number' || !Number.isSafeInteger(raw) || raw < 0) {
          throw new Error('Invalid offset')
        }
        return raw
      },
    }),
    async *read({ state }) {
      // A fixed fixture has stable offsets; a mutable API may not.
      for (let offset = state ?? 0; offset < source.length; offset += 1) {
        yield { rows: rawRows(source.slice(offset, offset + 1), (item) => item.id), state: offset + 1 }
      }
    },
  })
  // Disable reader restarts so the first execution exposes the write failure.
  const pipeline = definePipeline({ id: 'fixture', streams: [stream], retry: { retries: 0 } })
  const selected = selectStreams([pipeline], [])
  const journal = createMemoryJournal()
  const destination = createMemoryDestination()
  const failed = await runIngestion({ selected, backfill: undefined }, {
    journal,
    destination: {
      async insert(input) {
        await destination.insert(input)
        if (input.rows[0]?.id === '2') throw new Error('acknowledgement lost')
      },
    },
  })
  expect(failed.ok).toBe(false)
  expect((await journal.readCheckpoint(stream.id)).envelope?.state).toBe(1)

  const resumed = await runIngestion({ selected, backfill: undefined }, { journal, destination })
  expect(resumed.ok).toBe(true)
  expect((await journal.readCheckpoint(stream.id)).envelope?.state).toBe(3)
  expect(destination.tables.get('default.items_raw')?.map((row) => row.id)).toEqual(['1', '2', '3'])
}, 10_000)

The memory destination deduplicates tokens without a time limit. It does not model ReplacingMergeTree merges or the destination’s real deduplication window; verify those against the intended ClickHouse target.

Cases to cover for a real source

• Empty input, several pages, an empty page with a continuation, and a repeated continuation.
• Inclusive/exclusive timestamp boundaries, late updates, overlap, and explicit backfill bounds.
• Provider cursor expiration, malformed saved state, and terminal cursor behavior.
• Transient failures, rate limiting, permanent auth failures, and cancellation.
• Failure after rows are saved but before progress is journaled; reruns and later successful syncs.
• Missing fields, large records, current state versus history, and replay after a tombstone.
• Child edits/deletes that do not change the parent timestamp; a failed child fetch must not publish an incomplete root as complete.
• Deletion discovery and checkpointing; a partial reconciliation scan must not mark unseen records as deleted.

Start with chkit ingest list and chkit check --offline to verify exports and required metadata columns. These checks do not prove that a provider reader or live table behaves correctly.

Related pages

• Readers and pagination: request and source contracts.
• Incremental syncs: durable state boundaries.
• Scheduling and recovery: interpret production outcomes.