Testing

plasma’s test suite runs entirely on Node — no browser, no Cloudflare edge. Your app’s tests can do the same. This guide walks through the two harnesses.

Client-side — `fake-indexeddb`

pnpm add -D fake-indexeddb vitest

Import once at the top of your test file:

import "fake-indexeddb/auto"

That’s it — plasma’s client engine transparently uses the polyfilled indexedDB, IDBKeyRange, IDBOpenDBRequest, etc.

A minimal mutator test

import "fake-indexeddb/auto"
import { createDb, defineSchema, eq, id, int, table, text } from "@sh1n4ps/plasma-core"
import { createIdbEngine } from "@sh1n4ps/plasma-client"
import { describe, expect, it } from "vitest"

const todos = table("todos", { id: id(), title: text(), done: int().default(0) })
const schema = defineSchema({ todos })

async function makeDb() {
  const engine = await createIdbEngine({
    schema,
    dbName: `test-${Math.random().toString(36).slice(2, 8)}`,
  })
  return createDb({ schema, engine })
}

describe("markDone mutator", () => {
  it("flips the done flag", async () => {
    const db = await makeDb()
    await db.insert(todos).values({ id: "t1", title: "hi" })
    await db.update(todos).set({ done: 1 }).where(eq(todos.id, "t1"))

    const rows = await db.select().from(todos)
    expect(rows[0]?.done).toBe(1)
  })
})

createIdbEngine is the standalone engine constructor — no PlasmaClient scaffolding, no sync loop. Perfect for unit-testing mutator logic against the same query builder your app uses.

Testing live queries deterministically

The IVM setup is async — subscribe(cb) returns before the initial callback has fired. Use whenReady():

it("emits the initial snapshot", async () => {
  const db = await makeDb()
  await db.insert(todos).values({ id: "t1", title: "hi" })

  const live = db.select().from(todos).live()
  const seen: number[] = []
  live.subscribe((rows) => seen.push(rows.length))

  await live.whenReady?.()
  expect(seen).toEqual([1])
})

Without whenReady(), a setTimeout(0) wait races the IDB seed read on a busy CI machine.

Server-side — Miniflare + `@cloudflare/vitest-pool-workers`

The Worker-side tests need a D1 emulator and a workerd runtime. Cloudflare ships @cloudflare/vitest-pool-workers which gives you both.

pnpm add -D @cloudflare/vitest-pool-workers wrangler

vitest.config.ts:

import { defineConfig } from "vitest/config"

export default defineConfig({
  test: {
    poolOptions: {
      workers: {
        wrangler: { configPath: "./wrangler.jsonc" },
      },
    },
    projects: [
      {
        test: {
          name: "server",
          include: ["packages/server/test-workerd/**/*.test.ts"],
          poolOptions: {
            workers: {
              main: "./packages/server/test-workerd/entry.ts",
              miniflare: {
                d1Databases: { DB: "test-db" },
                r2Buckets: { BUCKET: "test-bucket" },
              },
            },
          },
        },
      },
    ],
  },
})

Your wrangler.jsonc declares the D1 and R2 bindings the test workers use — Miniflare emulates them in-process.

Seeding data

Use the same ensureSchema you’d use in production:

import { env } from "cloudflare:test"
import { ensureSchema, fromD1 } from "@sh1n4ps/plasma-server"

beforeEach(async () => {
  const executor = fromD1(env.DB)
  await ensureSchema({ schema, executor })
})

env.DB is Miniflare’s D1 emulator. env.BUCKET is Miniflare’s R2 emulator. They start empty on every test worker boot; you seed the tables you need.

End-to-end push/pull

You can drive the sync handler with actual Request objects:

import { createSyncHandler } from "@sh1n4ps/plasma-server"

it("push then pull round-trips a mutation", async () => {
  const executor = fromD1(env.DB)
  await ensureSchema({ schema, executor })

  const handler = createSyncHandler({
    schema, mutators, executor,
    schemaVersion: "v1",
    auth: async () => ({ ok: true, clientGroupID: "g", clientID: "c", ctx: {} }),
  })

  const pushRes = await handler(new Request("https://plasma.test/sync/push", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      protocolVersion: 1,
      schemaVersion: "v1",
      clientGroupID: "g",
      clientID: "c",
      mutations: [{ id: 1, name: "createTodo", args: { id: "t1", title: "hi", updatedAt: 0 } }],
    }),
  }))
  expect(pushRes.status).toBe(200)

  const pullRes = await handler(new Request("https://plasma.test/sync/pull?protocolVersion=1&schemaVersion=v1&clientGroupID=g&clientID=c"))
  const body = await pullRes.json()
  expect(body.patch).toHaveLength(1)
})

This runs against the same sync-handler.ts code path production uses, including auth, schema mismatch, envelope validation, and CRDT merge.

Testing React components with `PlasmaProvider`

For component tests that use useLiveQuery / useMutation, wrap with PlasmaProvider and construct the client with the fake IDB:

import { PlasmaProvider } from "@sh1n4ps/plasma-react"
import { createPlasmaClient } from "@sh1n4ps/plasma-client"
import { render } from "@testing-library/react"

const client = createPlasmaClient({
  schema, mutators,
  dbName: `test-${Math.random()}`,
  endpoint: "https://plasma.test/sync",
  clientGroupID: "test",
  schemaVersion: "v1",
  getContext: async () => ({ userId: "test" }),
  fetcher: async () => new Response(JSON.stringify({
    cookie: null,
    patch: [],
    lastMutationIDs: {},
    hasMore: false,
  })),   // stub every push/pull
  pollIntervalMs: 60_000,
  retry: { maxAttempts: 1 },
})

render(
  <PlasmaProvider client={client}>
    <TodoList />
  </PlasmaProvider>,
)

The fetcher stub lets you test the client without a running Worker. If you need to test push/pull behaviour, hand-craft the responses.

Deterministic timing patterns

Two patterns show up frequently:

await live.whenReady?.() — before asserting on the first snapshot delivery.
await new Promise(r => setTimeout(r, 0)) — after triggering a mutation, to let the reactive hub’s synchronous emit finish propagating.

Neither pattern needs a magic number. whenReady awaits a real Promise; a zero-tick microtask is enough for the emit.