概述
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 - 1 加 1 等于 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..5ok 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.sqlBEGIN;-- ========== 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 textLANGUAGE plpgsql AS $$DECLAREv_email text;BEGINSELECT 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..12ok 1 - users 表应存在ok 2 - 应有 id 列ok 3 - 应有 email 列ok 4 - 应有 created_at 列ok 5 - id 应为 integerok 6 - email 非空ok 7 - 应有主键ok 8 - get_user_name(int) 存在ok 9 - 返回 textok 10 - users 表行数 >= 0ok 11 - 1+1 应等于 2ok 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 存在 |