印刻数据库

外观

PostgreSQL 测试场景设计 ​

核心概念 ​

测试场景设计是指根据业务需求和系统特性，设计合理的测试用例和测试流程，以验证数据库系统的功能、性能、安全性和可靠性。主要涉及以下核心概念：

  • 测试场景：模拟真实业务流程的测试用例集合
  • 测试用例：验证单个功能点或性能指标的具体测试步骤
  • 测试数据：用于执行测试的数据集
  • 测试环境：执行测试的硬件和软件环境
  • 测试指标：用于评估测试结果的量化标准

测试场景设计原则 ​

1. 全面性原则 ​

  • 覆盖所有核心功能和业务流程
  • 考虑正常、异常和边界情况
  • 覆盖不同的使用场景和用户角色

2. 真实性原则 ​

  • 模拟真实的业务负载和访问模式
  • 使用接近生产环境的测试数据
  • 测试环境配置与生产环境保持一致

3. 可重复性原则 ​

  • 测试用例可重复执行，结果可验证
  • 测试数据可重现
  • 测试环境可恢复

4. 优先级原则 ​

  • 按照功能重要性和使用频率确定测试优先级
  • 优先测试核心业务流程和高风险功能

5. 可扩展性原则 ​

  • 测试用例设计应易于扩展和维护
  • 支持新增功能和业务流程的测试

功能测试场景设计 ​

1. 数据定义语言（DDL）测试 ​

场景1：表创建测试 ​

测试目标：验证表创建语法和约束是否正确

测试用例：

sql
-- 1. 创建基本表
CREATE TABLE test_users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    email VARCHAR(100) NOT NULL UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 2. 创建带外键约束的表
CREATE TABLE test_orders (
    id SERIAL PRIMARY KEY,
    user_id INT REFERENCES test_users(id) ON DELETE CASCADE,
    amount DECIMAL(10,2) NOT NULL,
    order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 3. 创建带检查约束的表
CREATE TABLE test_products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    price DECIMAL(10,2) CHECK (price > 0),
    stock INT CHECK (stock >= 0)
);

-- 4. 创建带索引的表
CREATE TABLE test_posts (
    id SERIAL PRIMARY KEY,
    title VARCHAR(200) NOT NULL,
    content TEXT,
    author_id INT REFERENCES test_users(id),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_posts_author ON test_posts(author_id);
CREATE INDEX idx_posts_created ON test_posts(created_at DESC);

场景2：表修改测试 ​

测试目标：验证表结构修改功能

测试用例：

sql
-- 1. 添加列
ALTER TABLE test_users ADD COLUMN phone VARCHAR(20);

-- 2. 修改列类型
ALTER TABLE test_users ALTER COLUMN phone TYPE VARCHAR(30);

-- 3. 修改列默认值
ALTER TABLE test_users ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP;

-- 4. 添加约束
ALTER TABLE test_users ADD CONSTRAINT chk_email CHECK (email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}$');

-- 5. 删除列
ALTER TABLE test_users DROP COLUMN phone;

-- 6. 删除表
DROP TABLE IF EXISTS test_products;
DROP TABLE IF EXISTS test_orders;
DROP TABLE IF EXISTS test_posts;
DROP TABLE IF EXISTS test_users;

2. 数据操作语言（DML）测试 ​

场景1：数据插入测试 ​

测试目标：验证数据插入功能

测试用例：

sql
-- 准备测试表
CREATE TABLE test_insert (
    id SERIAL PRIMARY KEY,
    col1 INT NOT NULL,
    col2 VARCHAR(50),
    col3 TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 1. 单条数据插入
INSERT INTO test_insert (col1, col2) VALUES (1, 'test1');

-- 2. 多条数据插入
INSERT INTO test_insert (col1, col2) VALUES 
    (2, 'test2'),
    (3, 'test3'),
    (4, 'test4');

-- 3. 复制插入
INSERT INTO test_insert (col1, col2) 
SELECT col1 * 10, col2 || '_copy' FROM test_insert;

-- 4. 使用DEFAULT值插入
INSERT INTO test_insert (col1) VALUES (5);

-- 清理测试数据
DROP TABLE test_insert;

场景2：数据查询测试 ​

测试目标：验证数据查询功能

测试用例：

sql
-- 准备测试数据
CREATE TABLE test_query (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age INT,
    department VARCHAR(50),
    salary DECIMAL(10,2),
    hired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO test_query (name, age, department, salary) VALUES
    ('张三', 25, '技术部', 8000),
    ('李四', 30, '技术部', 10000),
    ('王五', 28, '市场部', 7000),
    ('赵六', 35, '财务部', 9000),
    ('孙七', 26, '技术部', 8500);

-- 1. 基本查询
SELECT * FROM test_query;
SELECT name, salary FROM test_query;

-- 2. 条件查询
SELECT * FROM test_query WHERE department = '技术部';
SELECT * FROM test_query WHERE age > 28;
SELECT * FROM test_query WHERE salary BETWEEN 7000 AND 9000;

-- 3. 排序查询
SELECT * FROM test_query ORDER BY salary DESC;
SELECT * FROM test_query ORDER BY department, age ASC;

-- 4. 分组查询
SELECT department, COUNT(*) as count, AVG(salary) as avg_salary 
FROM test_query 
GROUP BY department;

-- 5. 连接查询（自连接示例）
SELECT a.name as employee, b.name as manager 
FROM test_query a 
LEFT JOIN test_query b ON a.manager_id = b.id;

-- 清理测试数据
DROP TABLE test_query;

3. 事务测试 ​

场景：事务完整性测试 ​

测试目标：验证事务的ACID特性

测试用例：

sql
-- 准备测试表
CREATE TABLE test_transaction (
    id SERIAL PRIMARY KEY,
    balance INT DEFAULT 0
);

INSERT INTO test_transaction (balance) VALUES (1000);

-- 1. 测试事务提交
BEGIN;
UPDATE test_transaction SET balance = balance - 100 WHERE id = 1;
UPDATE test_transaction SET balance = balance + 100 WHERE id = 1;
COMMIT;

-- 验证结果
SELECT * FROM test_transaction;  -- 应返回 1000

-- 2. 测试事务回滚
BEGIN;
UPDATE test_transaction SET balance = balance - 200 WHERE id = 1;
ROLLBACK;

-- 验证结果
SELECT * FROM test_transaction;  -- 应返回 1000

-- 3. 测试事务隔离级别
-- 设置隔离级别
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;

BEGIN;
UPDATE test_transaction SET balance = balance + 300 WHERE id = 1;
-- 在另一个会话中查询，应看不到未提交的修改
-- COMMIT 或 ROLLBACK

-- 清理测试数据
DROP TABLE test_transaction;

性能测试场景设计 ​

1. 基准测试场景 ​

场景1：pgbench 基准测试 ​

测试目标：建立数据库性能基准

测试用例：

bash
# 1. 初始化测试数据
pgbench -i -s 10 postgres

# 2. 执行默认基准测试
pgbench -c 10 -j 2 -T 60 postgres

# 3. 执行只读测试
pgbench -c 20 -j 4 -T 120 -S postgres

# 4. 执行自定义脚本测试
cat > custom_benchmark.sql << EOF
BEGIN;
SELECT * FROM pgbench_accounts WHERE aid = 1;
UPDATE pgbench_accounts SET abalance = abalance + 1 WHERE aid = 1;
SELECT * FROM pgbench_tellers WHERE tid = 2;
UPDATE pgbench_tellers SET tbalance = tbalance + 1 WHERE tid = 2;
SELECT * FROM pgbench_branches WHERE bid = 3;
UPDATE pgbench_branches SET bbalance = bbalance + 1 WHERE bid = 3;
INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (2, 3, 1, 1, CURRENT_TIMESTAMP);
END;
EOF

pgbench -c 15 -j 3 -T 90 -f custom_benchmark.sql postgres

场景2：OLTP 负载测试 ​

测试目标：测试数据库在高并发OLTP场景下的性能

测试用例：

bash
# 使用 sysbench 进行 OLTP 读写测试
sysbench --db-driver=pgsql --pgsql-host=localhost --pgsql-port=5432 --pgsql-user=postgres --pgsql-password=postgres --pgsql-db=postgres --table-size=1000000 --tables=10 oltp_read_write prepare

sysbench --db-driver=pgsql --pgsql-host=localhost --pgsql-port=5432 --pgsql-user=postgres --pgsql-password=postgres --pgsql-db=postgres --table-size=1000000 --tables=10 --threads=20 --time=300 --report-interval=10 oltp_read_write run

sysbench --db-driver=pgsql --pgsql-host=localhost --pgsql-port=5432 --pgsql-user=postgres --pgsql-password=postgres --pgsql-db=postgres --table-size=1000000 --tables=10 oltp_read_write cleanup

2. 特定场景性能测试 ​

场景1：大数据量查询测试 ​

测试目标：测试数据库在大数据量下的查询性能

测试用例：

sql
-- 准备测试数据
CREATE TABLE test_big_data (
    id SERIAL PRIMARY KEY,
    data_column TEXT,
    category VARCHAR(50),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 插入100万条测试数据
INSERT INTO test_big_data (data_column, category) 
SELECT 
    md5(random()::text || clock_timestamp()::text) AS data_column,
    'category_' || (random() * 9)::int AS category
FROM generate_series(1, 1000000);

-- 创建索引
CREATE INDEX idx_test_big_data_category ON test_big_data(category);
CREATE INDEX idx_test_big_data_created ON test_big_data(created_at);

-- 测试查询性能
EXPLAIN ANALYZE SELECT * FROM test_big_data WHERE category = 'category_5';
EXPLAIN ANALYZE SELECT category, COUNT(*) FROM test_big_data GROUP BY category;
EXPLAIN ANALYZE SELECT * FROM test_big_data WHERE created_at > NOW() - INTERVAL '1 hour';

-- 清理测试数据
DROP TABLE test_big_data;

场景2：并发写入测试 ​

测试目标：测试数据库在高并发写入场景下的性能

测试用例：

python
import psycopg2
import threading
import time

# 数据库连接参数
DB_PARAMS = {
    'host': 'localhost',
    'port': 5432,
    'dbname': 'postgres',
    'user': 'postgres',
    'password': 'postgres'
}

# 测试表创建
def create_test_table():
    conn = psycopg2.connect(**DB_PARAMS)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS test_concurrent_write (
            id SERIAL PRIMARY KEY,
            data TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    cursor.close()
    conn.close()

# 并发写入函数
def concurrent_write(thread_id, num_records):
    conn = psycopg2.connect(**DB_PARAMS)
    cursor = conn.cursor()
    for i in range(num_records):
        data = f'thread_{thread_id}_record_{i}'
        cursor.execute('INSERT INTO test_concurrent_write (data) VALUES (%s)', (data,))
    conn.commit()
    cursor.close()
    conn.close()

# 执行并发写入测试
def run_concurrent_write_test():
    create_test_table()
    
    # 测试参数
    num_threads = 10
    num_records_per_thread = 1000
    
    # 开始时间
    start_time = time.time()
    
    # 创建并启动线程
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(target=concurrent_write, args=(i, num_records_per_thread))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()
    
    # 结束时间
    end_time = time.time()
    
    # 计算结果
    total_records = num_threads * num_records_per_thread
    elapsed_time = end_time - start_time
    tps = total_records / elapsed_time
    
    print(f"测试完成：")
    print(f"  线程数：{num_threads}")
    print(f"  每条线程记录数：{num_records_per_thread}")
    print(f"  总记录数：{total_records}")
    print(f"  耗时：{elapsed_time:.2f}秒")
    print(f"  TPS：{tps:.2f}")

if __name__ == "__main__":
    run_concurrent_write_test()

安全测试场景设计 ​

1. 权限测试 ​

场景：用户权限管理测试 ​

测试目标：验证用户权限管理功能

测试用例：

sql
-- 1. 创建测试用户
CREATE USER test_user WITH PASSWORD 'test_password';
CREATE USER test_readonly WITH PASSWORD 'readonly_password';

-- 2. 创建测试表
CREATE TABLE test_secure (
    id SERIAL PRIMARY KEY,
    sensitive_data TEXT,
    public_data TEXT
);

-- 3. 插入测试数据
INSERT INTO test_secure (sensitive_data, public_data) VALUES 
    ('机密数据1', '公开数据1'),
    ('机密数据2', '公开数据2');

-- 4. 分配权限
GRANT SELECT ON test_secure TO test_readonly;
GRANT ALL ON test_secure TO test_user;

-- 5. 测试权限
-- 以 test_readonly 用户登录
-- SELECT * FROM test_secure;  -- 应成功
-- INSERT INTO test_secure (sensitive_data, public_data) VALUES ('测试', '测试');  -- 应失败

-- 以 test_user 用户登录
-- SELECT * FROM test_secure;  -- 应成功
-- INSERT INTO test_secure (sensitive_data, public_data) VALUES ('测试', '测试');  -- 应成功
-- DELETE FROM test_secure WHERE id = 1;  -- 应成功

-- 6. 测试行级安全性
CREATE POLICY test_policy ON test_secure FOR SELECT 
    USING (public_data IS NOT NULL);

ALTER TABLE test_secure ENABLE ROW LEVEL SECURITY;

-- 7. 清理测试数据
DROP TABLE test_secure;
DROP USER test_user;
DROP USER test_readonly;

2. 注入攻击测试 ​

场景：SQL注入防护测试 ​

测试目标：验证应用程序对SQL注入攻击的防护能力

测试用例：

python
import psycopg2

# 安全的参数化查询
def safe_query(user_input):
    conn = psycopg2.connect(
        host='localhost',
        port=5432,
        dbname='postgres',
        user='postgres',
        password='postgres'
    )
    cursor = conn.cursor()
    
    # 使用参数化查询，防止SQL注入
    query = "SELECT * FROM users WHERE username = %s"
    cursor.execute(query, (user_input,))
    
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    
    return result

# 不安全的字符串拼接
def unsafe_query(user_input):
    conn = psycopg2.connect(
        host='localhost',
        port=5432,
        dbname='postgres',
        user='postgres',
        password='postgres'
    )
    cursor = conn.cursor()
    
    # 不安全的字符串拼接，容易受到SQL注入攻击
    query = f"SELECT * FROM users WHERE username = '{user_input}'"
    cursor.execute(query)
    
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    
    return result

# 测试SQL注入
if __name__ == "__main__":
    # 恶意输入
    malicious_input = "' OR 1=1 --"
    
    print("测试安全查询：")
    try:
        result = safe_query(malicious_input)
        print(f"  结果数量：{len(result)}")
    except Exception as e:
        print(f"  错误：{e}")
    
    print("\n测试不安全查询：")
    try:
        result = unsafe_query(malicious_input)
        print(f"  结果数量：{len(result)}")
        print("  警告：不安全查询容易受到SQL注入攻击！")
    except Exception as e:
        print(f"  错误：{e}")

测试场景设计最佳实践 ​

1. 测试数据设计 ​

  • 真实数据模拟：使用接近生产环境的数据分布和特征
  • 数据量规划：根据测试目标确定合适的数据量
  • 数据多样性：包含不同类型和长度的数据
  • 数据可维护性：测试数据易于生成、清理和恢复

2. 测试环境设计 ​

  • 环境隔离：测试环境与开发、生产环境隔离
  • 配置一致性：测试环境配置与生产环境保持一致
  • 环境监控：监控测试环境的资源使用情况
  • 环境恢复：建立测试环境的快速恢复机制

3. 测试执行设计 ​

  • 测试顺序：按照功能依赖关系确定测试执行顺序
  • 测试并行：合理利用并行测试提高测试效率
  • 测试监控：实时监控测试执行过程和资源使用情况
  • 测试日志：详细记录测试执行过程和结果

4. 测试结果分析 ​

  • 结果验证：与预期结果进行对比验证
  • 性能分析：分析性能瓶颈和优化空间
  • 问题定位：准确定位测试中发现的问题
  • 报告生成：生成详细的测试报告

常见问题处理 ​

  • 问题1：测试数据生成困难 解决方法：

    • 使用工具生成测试数据（如 generate_series、pgbench 等）
    • 从生产环境导出脱敏数据
    • 使用测试数据生成工具（如 Mockaroo、DataFaker 等）

  • 问题2：测试环境资源不足 解决方法：

    • 优化测试用例，减少资源消耗
    • 增加测试环境资源
    • 使用云环境进行测试
    • 采用分布式测试策略

  • 问题3：测试结果不稳定 解决方法：

    • 确保测试环境稳定，无其他负载
    • 增加测试执行时间和次数
    • 优化测试用例，减少外部依赖
    • 采用统计方法分析测试结果

  • 问题4：测试用例维护困难 解决方法：

    • 采用模块化设计，提高测试用例的复用性
    • 使用测试框架管理测试用例
    • 建立测试用例的版本控制
    • 定期 review 和更新测试用例

常见问题（FAQ） ​

Q1：如何设计有效的性能测试场景？ ​

A1：设计有效性能测试场景的方法：

  • 分析真实业务负载和访问模式
  • 确定关键性能指标（TPS、响应时间、资源使用率等）
  • 设计逐步递增的负载测试
  • 模拟真实的并发用户数
  • 考虑不同的业务场景组合

Q2：如何确定测试数据量？ ​

A2：确定测试数据量的方法：

  • 根据生产环境数据量确定
  • 考虑数据增长趋势
  • 结合测试目标和环境资源
  • 对于性能测试，建议使用接近生产环境的数据量

Q3：如何设计安全测试场景？ ​

A3：设计安全测试场景的方法：

  • 覆盖常见的安全漏洞类型（SQL注入、权限绕过、数据泄露等）
  • 模拟真实的攻击场景
  • 考虑不同的攻击向量
  • 结合业务场景设计安全测试

Q4：如何提高测试效率？ ​

A4：提高测试效率的方法：

  • 自动化测试执行
  • 并行测试
  • 优化测试用例设计
  • 合理利用测试环境资源
  • 建立测试用例的复用机制

Q5：如何确保测试覆盖的全面性？ ​

A5：确保测试覆盖全面性的方法：

  • 基于需求和业务流程设计测试场景
  • 使用测试覆盖工具分析测试覆盖情况
  • 考虑正常、异常和边界情况
  • 结合静态分析和动态测试
  • 定期进行测试用例 review