Back to Details

pgsql-test-seeding

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Seeding Test Databases with pgsql-test

使用 pgsql-test 为测试数据库植入数据

Load test data efficiently using loadJson, loadSql, and loadCsv methods. Create maintainable, realistic test fixtures.
借助 loadJson、loadSql 和 loadCsv 方法高效加载测试数据，创建可维护、贴近真实场景的测试夹具。

When to Apply

适用场景

Use this skill when:
  • Setting up test data for database tests
  • Loading fixtures from JSON, SQL, or CSV files
  • Seeding data that respects or bypasses RLS
  • Creating per-test or shared test data
出现以下需求时可使用该技能：
  • 为数据库测试准备测试数据
  • 从 JSON、SQL 或 CSV 文件加载测试夹具
  • 植入需要遵循或绕过 RLS 的数据
  • 创建单测试专用或多测试共享的测试数据

Seeding Methods Overview

植入方法概览

MethodBest ForRLS Behavior
loadJson()
Inline data, small datasetsRespects RLS (use 
pg
to bypass)
loadSql()
Complex data, version-controlled fixturesRespects RLS (use 
pg
to bypass)
loadCsv()
Large datasets, spreadsheet exportsBypasses RLS (uses COPY)
方法适用场景RLS 行为
loadJson()
行内数据、小型数据集遵循 RLS（使用 
pg
可绕过）
loadSql()
复杂数据、版本管控的测试夹具遵循 RLS（使用 
pg
可绕过）
loadCsv()
大型数据集、电子表格导出数据绕过 RLS（使用 COPY 命令）

Seeding with loadJson()

使用 loadJson() 植入数据

Best for inline test data. Clean, readable, and type-safe.
typescript
import { getConnections, PgTestClient } from 'pgsql-test';

let pg: PgTestClient;
let db: PgTestClient;
let teardown: () => Promise<void>;

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  // Seed using superuser to bypass RLS
  await pg.loadJson({
    'app.users': [
      {
        id: '550e8400-e29b-41d4-a716-446655440001',
        email: 'alice@example.com',
        name: 'Alice'
      },
      {
        id: '550e8400-e29b-41d4-a716-446655440002',
        email: 'bob@example.com',
        name: 'Bob'
      }
    ],
    'app.posts': [
      {
        id: 'post-1',
        title: 'First Post',
        owner_id: '550e8400-e29b-41d4-a716-446655440001'
      },
      {
        id: 'post-2',
        title: 'Second Post',
        owner_id: '550e8400-e29b-41d4-a716-446655440002'
      }
    ]
  });
});

afterAll(async () => {
  await teardown();
});
Key features:
  • Schema-qualified table names: 
    'app.users'
  • Explicit UUIDs for referential integrity
  • Multiple tables in one call
  • Order matters for foreign keys
最适合行内测试数据，代码简洁、可读性高且类型安全。
typescript
import { getConnections, PgTestClient } from 'pgsql-test';

let pg: PgTestClient;
let db: PgTestClient;
let teardown: () => Promise<void>;

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  // 使用超级用户账号植入数据以绕过 RLS
  await pg.loadJson({
    'app.users': [
      {
        id: '550e8400-e29b-41d4-a716-446655440001',
        email: 'alice@example.com',
        name: 'Alice'
      },
      {
        id: '550e8400-e29b-41d4-a716-446655440002',
        email: 'bob@example.com',
        name: 'Bob'
      }
    ],
    'app.posts': [
      {
        id: 'post-1',
        title: 'First Post',
        owner_id: '550e8400-e29b-41d4-a716-446655440001'
      },
      {
        id: 'post-2',
        title: 'Second Post',
        owner_id: '550e8400-e29b-41d4-a716-446655440002'
      }
    ]
  });
});

afterAll(async () => {
  await teardown();
});
核心特性：
  • 带 schema 限定的表名：
    'app.users'
  • 显式 UUID 保证参照完整性
  • 单次调用可植入多张表数据
  • 数据顺序对有外键关联的表有影响

Seeding with loadSql()

使用 loadSql() 植入数据

Best for complex data or version-controlled fixtures.
Create 
__tests__/fixtures/seed.sql
:
sql
-- Insert users
INSERT INTO app.users (id, email, name) VALUES
  ('550e8400-e29b-41d4-a716-446655440001', 'alice@example.com', 'Alice'),
  ('550e8400-e29b-41d4-a716-446655440002', 'bob@example.com', 'Bob'),
  ('550e8400-e29b-41d4-a716-446655440003', 'charlie@example.com', 'Charlie');

-- Insert posts with foreign key references
INSERT INTO app.posts (id, title, owner_id) VALUES
  ('post-1', 'Alice Post 1', '550e8400-e29b-41d4-a716-446655440001'),
  ('post-2', 'Alice Post 2', '550e8400-e29b-41d4-a716-446655440001'),
  ('post-3', 'Bob Post', '550e8400-e29b-41d4-a716-446655440002');
Load in tests:
typescript
import path from 'path';

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  await pg.loadSql([
    path.join(__dirname, 'fixtures/seed.sql')
  ]);
});
Multiple SQL files:
typescript
await pg.loadSql([
  path.join(__dirname, 'fixtures/users.sql'),
  path.join(__dirname, 'fixtures/posts.sql'),
  path.join(__dirname, 'fixtures/comments.sql')
]);
Files execute in order, so put parent tables first.
最适合复杂数据或版本管控的测试夹具。
创建 
__tests__/fixtures/seed.sql
：
sql
-- 插入用户数据
INSERT INTO app.users (id, email, name) VALUES
  ('550e8400-e29b-41d4-a716-446655440001', 'alice@example.com', 'Alice'),
  ('550e8400-e29b-41d4-a716-446655440002', 'bob@example.com', 'Bob'),
  ('550e8400-e29b-41d4-a716-446655440003', 'charlie@example.com', 'Charlie');

-- 插入带外键关联的帖子数据
INSERT INTO app.posts (id, title, owner_id) VALUES
  ('post-1', 'Alice Post 1', '550e8400-e29b-41d4-a716-446655440001'),
  ('post-2', 'Alice Post 2', '550e8400-e29b-41d4-a716-446655440001'),
  ('post-3', 'Bob Post', '550e8400-e29b-41d4-a716-446655440002');
在测试中加载：
typescript
import path from 'path';

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  await pg.loadSql([
    path.join(__dirname, 'fixtures/seed.sql')
  ]);
});
加载多个 SQL 文件：
typescript
await pg.loadSql([
  path.join(__dirname, 'fixtures/users.sql'),
  path.join(__dirname, 'fixtures/posts.sql'),
  path.join(__dirname, 'fixtures/comments.sql')
]);
文件按顺序执行，因此需要将父表的加载逻辑放在子表前面。

Seeding with loadCsv()

使用 loadCsv() 植入数据

Best for large datasets or spreadsheet exports.
Create 
__tests__/fixtures/users.csv
:
csv
id,email,name
550e8400-e29b-41d4-a716-446655440001,alice@example.com,Alice
550e8400-e29b-41d4-a716-446655440002,bob@example.com,Bob
550e8400-e29b-41d4-a716-446655440003,charlie@example.com,Charlie
Load in tests:
typescript
import path from 'path';

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  await pg.loadCsv({
    'app.users': path.join(__dirname, 'fixtures/users.csv'),
    'app.posts': path.join(__dirname, 'fixtures/posts.csv')
  });
});
Important: 
loadCsv()
uses PostgreSQL's COPY command, which bypasses RLS. Always use 
pg
(superuser) client for CSV loading.
最适合大型数据集或电子表格导出数据。
创建 
__tests__/fixtures/users.csv
：
csv
id,email,name
550e8400-e29b-41d4-a716-446655440001,alice@example.com,Alice
550e8400-e29b-41d4-a716-446655440002,bob@example.com,Bob
550e8400-e29b-41d4-a716-446655440003,charlie@example.com,Charlie
在测试中加载：
typescript
import path from 'path';

beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  await pg.loadCsv({
    'app.users': path.join(__dirname, 'fixtures/users.csv'),
    'app.posts': path.join(__dirname, 'fixtures/posts.csv')
  });
});
注意： 
loadCsv()
使用 PostgreSQL 的 COPY 命令，会绕过 RLS。加载 CSV 时请始终使用 
pg
（超级用户）客户端。

Combining Seeding Strategies

组合多种植入策略

Mix methods based on data characteristics:
typescript
beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  // 1. Load large reference data from CSV
  await pg.loadCsv({
    'app.categories': path.join(__dirname, 'fixtures/categories.csv')
  });

  // 2. Load complex relationships from SQL
  await pg.loadSql([
    path.join(__dirname, 'fixtures/users-with-roles.sql')
  ]);

  // 3. Add test-specific data inline
  await pg.loadJson({
    'app.posts': [
      { title: 'Test Post', owner_id: testUserId, category_id: 1 }
    ]
  });
});
根据数据特性混合使用不同方法：
typescript
beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());

  // 1. 从 CSV 加载大型参考数据
  await pg.loadCsv({
    'app.categories': path.join(__dirname, 'fixtures/categories.csv')
  });

  // 2. 从 SQL 加载复杂关联数据
  await pg.loadSql([
    path.join(__dirname, 'fixtures/users-with-roles.sql')
  ]);

  // 3. 行内添加测试专用数据
  await pg.loadJson({
    'app.posts': [
      { title: 'Test Post', owner_id: testUserId, category_id: 1 }
    ]
  });
});

Per-Test Seeding

单测试专属数据植入

When different tests need different data, seed in 
beforeEach()
:
typescript
beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());
});

afterAll(async () => {
  await teardown();
});

beforeEach(async () => {
  await pg.beforeEach();
  await db.beforeEach();
});

afterEach(async () => {
  await db.afterEach();
  await pg.afterEach();
});

describe('empty state tests', () => {
  it('handles no data gracefully', async () => {
    const result = await db.query('SELECT COUNT(*) FROM app.posts');
    expect(result.rows[0].count).toBe('0');
  });
});

describe('populated state tests', () => {
  beforeEach(async () => {
    await pg.loadJson({
      'app.posts': [
        { title: 'Test Post', owner_id: userId }
      ]
    });
  });

  it('finds existing posts', async () => {
    const result = await db.query('SELECT COUNT(*) FROM app.posts');
    expect(result.rows[0].count).toBe('1');
  });
});
当不同测试需要不同数据时，可在 
beforeEach()
中植入数据：
typescript
beforeAll(async () => {
  ({ pg, db, teardown } = await getConnections());
});

afterAll(async () => {
  await teardown();
});

beforeEach(async () => {
  await pg.beforeEach();
  await db.beforeEach();
});

afterEach(async () => {
  await db.afterEach();
  await pg.afterEach();
});

describe('空状态测试', () => {
  it('可以优雅处理无数据场景', async () => {
    const result = await db.query('SELECT COUNT(*) FROM app.posts');
    expect(result.rows[0].count).toBe('0');
  });
});

describe('有数据状态测试', () => {
  beforeEach(async () => {
    await pg.loadJson({
      'app.posts': [
        { title: 'Test Post', owner_id: userId }
      ]
    });
  });

  it('可以查询到已存在的帖子', async () => {
    const result = await db.query('SELECT COUNT(*) FROM app.posts');
    expect(result.rows[0].count).toBe('1');
  });
});

RLS-Aware Seeding

适配 RLS 的数据植入

When testing RLS, seed with the appropriate client:
typescript
// Bypass RLS for setup (use pg)
await pg.loadJson({
  'app.posts': [{ title: 'Admin Post', owner_id: adminId }]
});

// Respect RLS for user operations (use db with context)
db.setContext({
  role: 'authenticated',
  'request.jwt.claim.sub': userId
});

await db.loadJson({
  'app.posts': [{ title: 'User Post', owner_id: userId }]
});
测试 RLS 时，使用对应权限的客户端植入数据：
typescript
// 使用 pg 客户端绕过 RLS 完成初始化数据植入
await pg.loadJson({
  'app.posts': [{ title: 'Admin Post', owner_id: adminId }]
});

// 使用带上下文的 db 客户端遵循 RLS 执行用户操作
db.setContext({
  role: 'authenticated',
  'request.jwt.claim.sub': userId
});

await db.loadJson({
  'app.posts': [{ title: 'User Post', owner_id: userId }]
});

Fixture Organization

测试夹具组织

Recommended structure:
text
__tests__/
├── fixtures/
│   ├── users.csv
│   ├── posts.csv
│   ├── seed.sql
│   └── complex-scenario.sql
├── users.test.ts
├── posts.test.ts
└── rls.test.ts
推荐目录结构：
text
__tests__/
├── fixtures/
│   ├── users.csv
│   ├── posts.csv
│   ├── seed.sql
│   └── complex-scenario.sql
├── users.test.ts
├── posts.test.ts
└── rls.test.ts

Best Practices

最佳实践

  1. Use explicit IDs: Makes referential integrity predictable
  2. Order by dependencies: Parent tables before child tables
  3. Keep fixtures minimal: Only seed what tests need
  4. Use 
    pg
    for setup    : Bypass RLS during seeding
  5. Use 
    db
    for testing    : Enforce RLS during assertions
  6. Version control fixtures: SQL/CSV files in repo
  1. 使用显式 ID：保证参照完整性可预期
  2. 按依赖顺序加载：父表数据优先于子表加载
  3. 保持夹具轻量化：仅植入测试所需的最小数据量
  4. 初始化阶段使用 
    pg
    ：植入数据时绕过 RLS
  5. 测试阶段使用 
    db
    ：断言阶段强制执行 RLS 校验
  6. 测试夹具纳入版本管控：SQL/CSV 文件提交到代码仓库

Troubleshooting

故障排查

IssueSolution
Foreign key violationLoad parent tables first
RLS blocking insertsUse 
pg
client instead of 
db
CSV format errorsEnsure headers match column names
Data persists between testsCheck beforeEach/afterEach hooks
问题解决方案
外键约束报错优先加载父表数据
RLS 阻止插入操作使用 
pg
客户端替代 
db
客户端
CSV 格式错误确认表头和表列名完全匹配
测试间数据残留检查 beforeEach/afterEach 钩子逻辑

References

参考资料

  • Related skill: 
    pgsql-test-rls
    for RLS testing patterns
  • Related skill: 
    pgsql-test-exceptions
    for handling errors
  • Related skill: 
    pgpm-testing
    for general test setup
  • 相关技能：
    pgsql-test-rls
    RLS 测试模式
  • 相关技能：
    pgsql-test-exceptions
    错误处理
  • 相关技能：
    pgpm-testing
    通用测试配置