@kysera/testing

Testing utilities for Kysera - transaction isolation, factories, seeding, and test helpers.

Installation

Install as a development dependency:

npm install --save-dev @kysera/testing

Overview

Dependencies: @kysera/core Peer Dependencies: kysely >=0.29.0; optional: @kysera/executor, better-sqlite3

:::info Package Type This is a utility package for testing. It's not part of the Repository/DAL pattern - it provides testing helpers that work with Kysely instances directly. :::

Key Features

• Transaction Rollback Testing - Automatic rollback for isolated, fast tests
• Database Cleanup Strategies - Multiple strategies for cleaning test databases
• Test Data Factories - Generate test data with sensible defaults
• Database Seeding - Composable seeders for consistent test data
• Test Helpers - Utilities for assertions, waiting, and snapshots

Quick Start

import { testInTransaction, createFactory } from '@kysera/testing'

const createUser = createFactory({
  email: () => `user-${Date.now()}@example.com`,
  name: 'Test User',
  role: 'user'
})

it('creates user', async () => {
  await testInTransaction(db, async trx => {
    const userData = createUser({ name: 'Alice' })
    const user = await trx.insertInto('users').values(userData).returningAll().executeTakeFirst()
    expect(user?.name).toBe('Alice')
  })
  // Database automatically rolled back - no cleanup needed!
})

Transaction Testing

testInTransaction()

Test in a transaction that automatically rolls back. Fastest testing approach.

import { testInTransaction } from '@kysera/testing'

it('creates and queries user', async () => {
  await testInTransaction(db, async trx => {
    await trx.insertInto('users').values({ email: 'test@example.com', name: 'Test' }).execute()
    const user = await trx
      .selectFrom('users')
      .where('email', '=', 'test@example.com')
      .selectAll()
      .executeTakeFirst()
    expect(user?.name).toBe('Test')
  })
  // Automatically rolled back
})

testWithSavepoints()

Test with savepoints for nested transaction testing.

import { testWithSavepoints } from '@kysera/testing'

it('handles nested operations', async () => {
  await testWithSavepoints(db, async trx => {
    await createUserWithProfile(trx, userData)
    // Verify results...
  })
})

An optional third logger argument (defaults to silentLogger) receives warnings when an unexpected error occurs while rolling back to the savepoint.

testWithIsolation()

Test with specific transaction isolation level.

import { testWithIsolation } from '@kysera/testing'

it('handles serializable isolation', async () => {
  await testWithIsolation(db, 'serializable', async trx => {
    // Test behavior under serializable isolation
  })
})

Isolation Levels: 'read uncommitted', 'read committed', 'repeatable read', 'serializable'

Uses kysely's dialect-aware setIsolationLevel() under the hood — the correct statements are emitted per database (MySQL sets the level before starting the transaction; a raw SET TRANSACTION inside an active MySQL transaction would fail). SQLite's driver ignores isolation settings (single-writer).

Database Cleanup

cleanDatabase()

Clean database using specified strategy.

import { cleanDatabase } from '@kysera/testing'

// Truncate - fast bulk cleanup
afterEach(async () => {
  await cleanDatabase(db, 'truncate', ['users', 'orders', 'order_items'])
})

// Delete - requires FK-safe order (children first)
afterEach(async () => {
  await cleanDatabase(db, 'delete', ['order_items', 'orders', 'users'])
})

// Options form - pass the dialect explicitly (recommended)
afterEach(async () => {
  await cleanDatabase(db, 'truncate', {
    dialect: 'postgres',
    tables: ['users', 'orders', 'order_items']
  })
})

The third argument is either the table list or a CleanupOptions object ({ tables, dialect?, logger? }). When dialect is omitted, cleanDatabase attempts to detect it from the Kysely instance; passing it explicitly is recommended. The tables list is required for the 'delete' and 'truncate' strategies.

Strategies:

• 'transaction' - No cleanup (use with testInTransaction)
• 'delete' - DELETE FROM each table (medium speed, FK-safe order required)
• 'truncate' - TRUNCATE TABLE (fastest bulk clean, handles FKs automatically)

Security Features:

• SQL injection prevention - Table names are validated with a strict identifier regex (must start with a letter or underscore; letters, digits, and underscores only) before being interpolated into the raw TRUNCATE statements
• Identifier quoting - The MySQL and MSSQL truncate paths additionally quote the validated name (backticks / square brackets); the PostgreSQL path interpolates it unquoted, so safety rests on the regex validation rather than on escaping

Test Data Factories

createFactory()

Create a generic test data factory.

import { createFactory } from '@kysera/testing'

const createUser = createFactory({
  email: () => `user-${Date.now()}@example.com`,
  name: 'Test User',
  role: 'user'
})

const user1 = createUser() // Use defaults
const admin = createUser({ role: 'admin' }) // Override

createMany()

Create multiple instances.

import { createMany } from '@kysera/testing'

const users = createMany(createUser, 5)
const admins = createMany(createUser, 3, i => ({
  name: `Admin ${i + 1}`,
  role: 'admin'
}))

createSequenceFactory()

Factory with built-in sequence counter.

import { createSequenceFactory } from '@kysera/testing'

const createUser = createSequenceFactory(seq => ({
  id: seq,
  email: `user-${seq}@example.com`,
  name: `User ${seq}`
}))

const user1 = createUser() // { id: 1, email: 'user-1@...' }
const user2 = createUser() // { id: 2, email: 'user-2@...' }

Database Seeding

seedDatabase()

Seed database with test data.

import { seedDatabase } from '@kysera/testing'

beforeAll(async () => {
  await seedDatabase(db, async trx => {
    await trx
      .insertInto('users')
      .values([
        { email: 'alice@example.com', name: 'Alice' },
        { email: 'bob@example.com', name: 'Bob' }
      ])
      .execute()
  })
})

composeSeeders()

Combine multiple seed functions.

import { composeSeeders, seedDatabase, type SeedFunction } from '@kysera/testing';

const seedUsers: SeedFunction<DB> = async (trx) => {
  await trx.insertInto('users').values([...]).execute();
};

const seedPosts: SeedFunction<DB> = async (trx) => {
  await trx.insertInto('posts').values([...]).execute();
};

const seedAll = composeSeeders([seedUsers, seedPosts]);

beforeAll(async () => {
  await seedDatabase(db, seedAll);
});

Test Helpers

waitFor()

Wait for a condition to be true.

import { waitFor } from '@kysera/testing'

await waitFor(async () => {
  const user = await db
    .selectFrom('users')
    .where('email', '=', 'test@example.com')
    .executeTakeFirst()
  return user !== undefined
})

// With options
await waitFor(async () => (await getProcessedCount()) >= 10, {
  timeout: 10000,
  interval: 200,
  timeoutMessage: 'Jobs did not complete'
})

snapshotTable()

Snapshot table state for comparison.

import { snapshotTable } from '@kysera/testing'

const before = await snapshotTable(db, 'users')
await createUser(db, userData)
const after = await snapshotTable(db, 'users')
expect(after.length).toBe(before.length + 1)

countRows()

Count rows in a table.

import { countRows } from '@kysera/testing'

const count = await countRows(db, 'users')
expect(count).toBe(5)

assertRowExists()

Assert that a row exists.

import { assertRowExists } from '@kysera/testing'

const user = await assertRowExists(db, 'users', { email: 'test@example.com' })
expect(user.name).toBe('Test User')

assertRowNotExists()

Assert that no row exists.

import { assertRowNotExists } from '@kysera/testing'

await deleteUser(db, userId)
await assertRowNotExists(db, 'users', { id: userId })

TypeScript Types

type IsolationLevel = 'read uncommitted' | 'read committed' | 'repeatable read' | 'serializable'
type CleanupStrategy = 'truncate' | 'transaction' | 'delete'
type FactoryFunction<T> = (overrides?: Partial<T>) => T
type SeedFunction<DB> = (trx: Transaction<DB>) => Promise<void>

// Per-field defaults for createFactory: plain values or zero-arg functions
type FactoryDefaults<T extends Record<string, unknown>> = {
  [K in keyof T]: T[K] | (() => T[K])
}

// Re-exported from @kysera/core for convenience
type Dialect = 'postgres' | 'mysql' | 'sqlite' | 'mssql'

interface WaitForOptions {
  timeout?: number // Default: 5000
  interval?: number // Default: 100
  timeoutMessage?: string
}

Best Practices

1. Use Transaction Rollback for Speed

// Fast - automatic rollback
await testInTransaction(db, async trx => {
  /* test */
})

// Slower - manual cleanup
await createUser(db, userData)
await cleanDatabase(db, 'truncate', ['users'])

2. Define Factories Once

// factories.ts
export const createUser = createFactory({
  email: () => `user-${Date.now()}@example.com`,
  name: 'Test User'
})

// test file
import { createUser } from './factories'

3. Compose Seeders

// seeders.ts
export const seedUsers: SeedFunction<DB> = async (trx) => { ... };
export const seedPosts: SeedFunction<DB> = async (trx) => { ... };
export const seedAll = composeSeeders([seedUsers, seedPosts]);

4. Choose Right Cleanup Strategy

• Transaction: Fastest (use with testInTransaction)
• Truncate: Fast bulk cleanup, handles FKs automatically
• Delete: Medium speed, requires FK-safe order

Plugin Testing

Utilities for testing Kysera plugins in isolation and integration scenarios.

createMockPlugin()

Creates a mock plugin for testing plugin interactions and execution order.

import { createMockPlugin } from '@kysera/testing'

const mockPlugin = createMockPlugin('test-plugin', {
  onIntercept: (qb, ctx) => {
    console.log(`Intercepted ${ctx.operation} on ${ctx.table}`)
    return qb // Return unmodified
  },
  priority: 100
})

const executor = await createExecutor(db, [mockPlugin, softDeletePlugin()])

// Run some queries
await executor.selectFrom('users').selectAll().execute()

// Check recorded operations
expect(mockPlugin.operations).toHaveLength(1)
expect(mockPlugin.operations[0].operation).toBe('select')
expect(mockPlugin.operations[0].table).toBe('users')

// Reset tracking
mockPlugin.reset()

Returns:

// Anonymous intersection type — there is no named MockPlugin interface to import
Plugin & {
  operations: RecordedOperation[]
  reset: () => void
}

interface RecordedOperation {
  operation: 'select' | 'insert' | 'update' | 'delete' | 'replace' | 'merge'
  table: string
  timestamp: Date
  metadata: Record<string, unknown>
}

spyOnPlugin()

Wraps an existing plugin to record all operations while preserving original behavior.

import { spyOnPlugin } from '@kysera/testing'
import { softDeletePlugin } from '@kysera/soft-delete'

const spiedPlugin = spyOnPlugin(softDeletePlugin())

const executor = await createExecutor(db, [spiedPlugin])

await executor.deleteFrom('users').where('id', '=', 1).execute()

// Verify the plugin was called
expect(spiedPlugin.calls).toHaveLength(1)
expect(spiedPlugin.calls[0].operation).toBe('delete')

// Reset call tracking
spiedPlugin.reset()

Returns:

// Anonymous intersection type — there is no named SpiedPlugin interface to import
Plugin & {
  calls: RecordedOperation[]
  reset: () => void
}

assertPluginBehavior()

Asserts that a plugin behaves as expected for a given operation.

import { assertPluginBehavior } from '@kysera/testing'
import type { QueryBuilderContext } from '@kysera/executor'

const plugin = softDeletePlugin({ deletedAtColumn: 'deleted_at' })

const result = assertPluginBehavior(
  plugin,
  { where: () => mockQb }, // Mock query builder
  { operation: 'select', table: 'users', metadata: {} } as QueryBuilderContext,
  { shouldModifyQuery: true }
)

expect(result.intercepted).toBe(true)
expect(result.modified).toBe(true)
expect(result.error).toBeUndefined()

Returns:

interface PluginTestResult {
  intercepted: boolean
  modified: boolean
  error?: Error
}

createInMemoryDatabase()

Creates an in-memory SQLite database for fast, isolated plugin tests.

import { createInMemoryDatabase } from '@kysera/testing'

const db = await createInMemoryDatabase<MyDB>(`
  CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email TEXT NOT NULL,
    name TEXT,
    deleted_at TEXT
  );
  CREATE TABLE posts (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    user_id INTEGER
  )
`)

const executor = await createExecutor(db, [softDeletePlugin()])

// Run tests against in-memory database
await executor.insertInto('users').values({ email: 'test@example.com', name: 'Test' }).execute()

// Clean up
await db.destroy()

:::note Dependency Requires better-sqlite3 as a dev dependency:

pnpm add -D better-sqlite3

:::

createPluginTestHarness()

Creates a structured test harness for plugin integration testing with setup, execute, verify, and teardown phases.

import { createPluginTestHarness, createMockPlugin } from '@kysera/testing'
import { softDeletePlugin } from '@kysera/soft-delete'
import { timestampsPlugin } from '@kysera/timestamps'

const harness = createPluginTestHarness({
  plugins: [softDeletePlugin(), timestampsPlugin()],
  schema: `
    CREATE TABLE posts (
      id INTEGER PRIMARY KEY,
      title TEXT,
      deleted_at TEXT,
      created_at TEXT,
      updated_at TEXT
    )
  `,
  seedData: async (executor) => {
    await executor.insertInto('posts')
      .values({ title: 'Seed Post' })
      .execute()
  }
})

// Setup: creates in-memory DB, applies schema, runs seeds
await harness.setup()

// Execute: run test operations
const result = await harness.execute(async (executor) => {
  return executor.insertInto('posts')
    .values({ title: 'Test Post' })
    .returningAll()
    .executeTakeFirst()
})

// Verify: run assertions
harness.verify(result, (r) => {
  expect(r.title).toBe('Test Post')
  expect(r.created_at).toBeDefined()
  expect(r.updated_at).toBeDefined()
})

// Access raw database if needed
const db = harness.getDb()

// Teardown: clean up resources
await harness.teardown()

Harness Methods:

Method	Description
setup()	Creates database, applies schema, runs seedData
execute(fn)	Executes test function with executor
verify(result, assertions)	Runs assertions on result
getDb()	Returns raw Kysely instance
teardown()	Destroys database, cleans up resources

createTestExecutor()

One-shot alternative to the harness: wraps an existing test database (e.g. from createInMemoryDatabase) in a plugin-aware executor. With debug: true, a low-priority recorder plugin is appended that captures every intercepted operation on the returned operations array.

import { createInMemoryDatabase, createTestExecutor } from '@kysera/testing'
import { softDeletePlugin } from '@kysera/soft-delete'

const db = await createInMemoryDatabase<MyDB>(`
  CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT, deleted_at TEXT)
`)

const { executor, operations, cleanup } = await createTestExecutor({
  db,
  plugins: [softDeletePlugin()],
  debug: true
})

// Queries through `executor` are intercepted by the plugins;
// queries through `db` bypass them
const users = await executor.selectFrom('users').selectAll().execute()
expect(operations[0]?.operation).toBe('select')

await cleanup() // destroys the underlying database

function createTestExecutor<DB>(
  options: CreateTestExecutorOptions<DB>
): Promise<TestExecutorResult<DB>>

Requires the optional peer @kysera/executor (the executor is created via createExecutor under the hood).

Plugin Testing Types

interface RecordedOperation {
  operation: 'select' | 'insert' | 'update' | 'delete' | 'replace' | 'merge'
  table: string
  timestamp: Date
  metadata: Record<string, unknown>
}

interface PluginTestResult {
  intercepted: boolean
  modified: boolean
  error?: Error
}

interface PluginAssertionOptions {
  expectedOperation?: QueryBuilderContext['operation']
  expectedTable?: string
  shouldModifyQuery?: boolean
}

// Mock executor context for unit-testing plugin hooks without a database
interface MockOperationContext {
  operation: QueryBuilderContext['operation']
  table: string
  executor: Kysely<unknown> | Transaction<unknown>
}

// Options for createTestExecutor()
interface CreateTestExecutorOptions<DB> {
  db: Kysely<DB> // database to wrap (e.g. from createInMemoryDatabase)
  plugins: Plugin[] // plugins to apply to the executor
  debug?: boolean // record intercepted operations on the result's `operations`
}

// Result of createTestExecutor()
interface TestExecutorResult<DB> {
  executor: KyseraExecutor<DB> // plugin-aware executor
  db: Kysely<DB> // the underlying database (bypasses plugins)
  operations: RecordedOperation[] // empty unless debug: true
  cleanup: () => Promise<void> // destroys the underlying database
}