@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: None (peer: kysely >=0.28.8)

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...
  })
})

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'

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'])
})

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 against database schema
• Safe identifier escaping - Uses dialect-specific escaping for table names
• Only whitelisted tables from the schema can be truncated/deleted

Deprecated: Dialect Detection

Dialect detection via Kysely internals is deprecated and will be removed in a future version. Always pass the dialect parameter explicitly:

// ❌ Deprecated - relies on internal Kysely APIs
await cleanDatabase(db, 'truncate', ['users'])

// ✅ Recommended - explicit dialect
await cleanDatabase(db, 'truncate', ['users'], { dialect: 'postgres' })

The automatic dialect detection may fail in future Kysely versions as it relies on internal APIs that are not part of Kysely's public contract.

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>

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:

interface MockPlugin extends 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:

interface SpiedPlugin extends 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()

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

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
}