帮你快速理解、总结文档立即下载

pgTAP

最近更新时间:2026-07-21 09:53:00

我的收藏

概述

pgTAP 是 PostgreSQL 的单元测试框架,使用 PL/pgSQL 编写,提供了丰富的 TAP(Test Anything Protocol)断言函数。它使得在 SQL 脚本中编写数据库单元测试变得简单直观,支持表结构验证、数据断言、函数测试等多种场景。

前提条件

v17.10_r1.19、v18.4_r1.10及以上版本。
需要 plpgsql 扩展(默认已安装)。

创建插件

步骤1:安装扩展

CREATE EXTENSION IF NOT EXISTS pgtap;

步骤2:验证安装

postgres=# SELECT extname, extversion FROM pg_extension WHERE extname = 'pgtap';
extname | extversion
---------+------------
pgtap | 1.3.4
(1 row)
如需安装到指定 Schema:
CREATE EXTENSION pgtap SCHEMA tap;

核心概念

测试计划

每个测试套件以 plan() 开始,指定预期的测试数量,以 finish() 结束:
SELECT plan(3); -- 声明将运行 3 个测试

-- ... 测试用例 ...

SELECT * FROM finish();
如果不确定测试数量,可以使用 no_plan():
SELECT no_plan();

-- ... 任意数量的测试用例 ...

SELECT * FROM finish();

测试结果格式

每个断言输出一行 TAP 格式的结果:
ok 1 - 测试描述
not ok 2 - 失败的测试描述

功能函数详解

一、基本断言函数

ok( boolean, description )

最基础的断言函数,当表达式为 true 时测试通过。
SELECT ok(1 + 1 = 2, '1 加 1 等于 2');
输出:
ok 1 - 11 等于 2

is( got, expected, description )

判断两个值是否相等。
SELECT is(
(SELECT count(*)::int FROM pg_tables WHERE schemaname = 'public'),
5,
'public schema 应有 5 张表'
);

cmp_ok( got, operator, expected, description )

使用指定的操作符比较两个值。
SELECT cmp_ok(
(SELECT count(*)::int FROM orders),
'>=',
100,
'订单数应不少于 100'
);

matches( got, pattern, description )

正则匹配断言。
SELECT matches(
(SELECT version()),
'PostgreSQL 1[4-9]',
'版本号应为 PostgreSQL 14-19'
);

二、Schema 对象断言

表相关

函数
用途
has_table(schema, table, desc)
断言表存在
hasnt_table(schema, table, desc)
断言表不存在
has_column(table, column, desc)
断言列存在
col_type_is(table, column, type, desc)
断言列类型
col_not_null(table, column, desc)
断言列非空约束
col_has_default(table, column, desc)
断言列有默认值
col_default_is(table, column, default, desc)
断言列默认值
示例:
BEGIN;
SELECT plan(5);

-- 创建测试表
CREATE TABLE users (
id serial PRIMARY KEY,
email text NOT NULL,
created_at timestamptz DEFAULT now()
);

-- 验证 users 表结构
SELECT has_table('public', 'users', 'users 表应存在');
SELECT has_column('users', 'id', 'users 表应有 id 列');
SELECT col_type_is('users', 'id', 'integer', 'id 列应为 integer 类型');
SELECT col_not_null('users', 'email', 'email 列应有非空约束');
SELECT col_has_default('users', 'created_at', 'created_at 应有默认值');

SELECT * FROM finish();
ROLLBACK;
输出:
1..5
ok 1 - users 表应存在
ok 2 - users 表应有 id 列
ok 3 - id 列应为 integer 类型
ok 4 - email 列应有非空约束
ok 5 - created_at 应有默认值

索引相关

函数
用途
has_index(table, index, desc)
断言索引存在
hasnt_index(table, index, desc)
断言索引不存在
has_pk(table, desc)
断言有主键
has_fk(table, desc)
断言有外键
has_unique(table, desc)
断言有唯一约束
SELECT has_pk('users', 'users 表应有主键');
SELECT has_index('users', 'idx_users_email', 'users 表应有 email 索引');

函数/存储过程相关

函数
用途
has_function(schema, func, args, desc)
断言函数存在
function_returns(schema, func, args, type, desc)
断言函数返回类型
function_lang_is(schema, func, args, lang, desc)
断言函数语言
is_definer(schema, func, args, desc)
断言为 SECURITY DEFINER
SELECT has_function(
'public', 'calculate_total', ARRAY['integer'],
'calculate_total(int) 函数应存在'
);
SELECT function_returns(
'public', 'calculate_total', ARRAY['integer'], 'numeric',
'函数应返回 numeric 类型'
);

触发器相关

函数
用途
has_trigger(table, trigger, desc)
断言触发器存在
trigger_is(table, trigger, func, desc)
断言触发器关联的函数
SELECT has_trigger('orders', 'trg_orders_updated', '应有更新触发器');

扩展相关

函数
用途
has_extension(name, desc)
断言扩展已安装
hasnt_extension(name, desc)
断言扩展未安装
has_schema(name, desc)
断言 Schema 存在
has_role(name, desc)
断言角色存在
SELECT has_extension('pgtap', 'pgtap 扩展应已安装');
SELECT has_schema('public', 'public schema 应存在');

三、结果集断言

results_eq( sql1, sql2, description )

断言两个查询的结果集完全相同(含顺序)。
SELECT results_eq(
'SELECT id, name FROM users ORDER BY id LIMIT 2',
VALUES(1,′alice′),(2,′bob′),
'前两个用户应为 alice 和 bob'
);

set_eq( sql1, sql2, description )

断言两个查询的结果集相同(不考虑顺序)。
SELECT set_eq(
'SELECT name FROM departments',
VALUES(′Engineering′),(′Marketing′),(′Sales′),
'应有三个部门'
);

bag_eq( sql1, sql2, description )

断言两个结果集包含相同的行(允许重复,不考虑顺序)。

set_has( sql1, sql2, description )

断言第一个结果集包含第二个结果集的所有行。
SELECT set_has(
'SELECT name FROM users',
VALUES(′admin′),
'用户列表应包含 admin'
);

四、异常断言

throws_ok( sql, errcode, errmsg, description )

断言 SQL 执行会抛出指定的异常。
SELECT throws_ok(
'INSERT INTO users (id, name) VALUES (1, NULL)',
'23502',
NULL,
'插入空 name 应触发非空约束违反'
);

lives_ok( sql, description )

断言 SQL 可以正常执行不抛异常。
SELECT lives_ok(
'SELECT 1',
'SELECT 1 应正常执行'
);

完整测试示例

以下是一个完整的、自包含的测试脚本示例,展示了 pgTAP 的典型用法:
-- test_user_module.sql
BEGIN;

-- ========== 0. 创建测试对象 ==========
CREATE TABLE users (
id serial PRIMARY KEY,
email text NOT NULL,
created_at timestamptz DEFAULT now()
);

CREATE OR REPLACE FUNCTION get_user_name(p_id integer)
RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
v_email text;
BEGIN
SELECT email INTO v_email FROM users WHERE id = p_id;
RETURN coalesce(v_email, '<unknown>');
END;
$$;

SELECT plan(12);

-- ========== 1. Schema 验证 ==========
SELECT has_table('public', 'users', 'users 表应存在');
SELECT has_column('users', 'id', '应有 id 列');
SELECT has_column('users', 'email', '应有 email 列');
SELECT has_column('users', 'created_at', '应有 created_at 列');
SELECT col_type_is('users', 'id', 'integer', 'id 应为 integer');
SELECT col_not_null('users', 'email', 'email 非空');
SELECT has_pk('users', '应有主键');

-- ========== 2. 函数验证 ==========
SELECT has_function('public', 'get_user_name', ARRAY['integer'], 'get_user_name(int) 存在');
SELECT function_returns('public', 'get_user_name', ARRAY['integer'], 'text', '返回 text');

-- ========== 3. 数据验证 ==========
SELECT cmp_ok(
(SELECT count(*)::int FROM users),
'>=', 0,
'users 表行数 >= 0'
);

-- ========== 4. 结果集验证 ==========
SELECT results_eq(
'SELECT 1 + 1',
'SELECT 2',
'1+1 应等于 2'
);

-- ========== 5. 异常验证 ==========
SELECT throws_ok(
SELECT1/0,
'22012',
NULL,
'除零应抛异常'
);

SELECT * FROM finish();
ROLLBACK;

运行测试

psql -d mydb -f test_user_module.sql
预期输出:
1..12
ok 1 - users 表应存在
ok 2 - 应有 id 列
ok 3 - 应有 email 列
ok 4 - 应有 created_at 列
ok 5 - id 应为 integer
ok 6 - email 非空
ok 7 - 应有主键
ok 8 - get_user_name(int) 存在
ok 9 - 返回 text
ok 10 - users 表行数 >= 0
ok 11 - 1+1 应等于 2
ok 12 - 除零应抛异常

常用断言函数速查表

分类
函数
说明
基本
ok(bool, desc)
布尔断言
基本
is(got, expected, desc)
相等断言
基本
isnt(got, unexpected, desc)
不等断言
基本
cmp_ok(got, op, expected, desc)
操作符比较
基本
matches(got, regex, desc)
正则匹配
has_table(schema, table, desc)
表存在
has_column(table, col, desc)
列存在
col_type_is(table, col, type, desc)
列类型
col_not_null(table, col, desc)
非空约束
索引
has_index(table, idx, desc)
索引存在
索引
has_pk(table, desc)
主键存在
函数
has_function(schema, func, args, desc)
函数存在
结果集
results_eq(sql1, sql2, desc)
结果集相等(有序)
结果集
set_eq(sql1, sql2, desc)
集合相等(无序)
结果集
set_has(sql1, sql2, desc)
集合包含
异常
throws_ok(sql, errcode, msg, desc)
异常断言
异常
lives_ok(sql, desc)
正常执行断言
扩展
has_extension(name, desc)
扩展存在
Schema
has_schema(name, desc)
Schema 存在