当前位置：首页 > 技术干货 > 软件测试技术干货 > 数据库测试：如何验证数据一致性和完整性_千锋教育

数据库测试：如何验证数据一致性和完整性_千锋教育

来源：千锋教育

发布时间：2026-07-03 18:31:20

01 数据一致性：为什么99%的测试都漏掉了这个环节

数据一致性，不是“数据看起来对不对”。它指的是：同一份数据，在系统的任何角落、任何时刻、任何操作下，都保持逻辑上的统一。

举个血淋淋的例子：电商下单。用户点击“下单”，订单表写入一条记录，库存表扣除一个数量。如果订单写入成功，库存扣减失败——用户付了钱，商家没货发。如果订单写入失败，库存扣减成功——用户没买成，库存却少了。这就是数据不一致。

为什么99%的测试人员会漏掉这个环节？三个原因：

1. 只测功能，不测数据流。测试人员盯着界面“下单成功”的弹窗，却从没检查过库存表是否真的减了1。

2. 只测单表，不测关联。订单表、支付表、物流表，每个表单独测都没问题，组合起来就崩。

3. 只测正常，不测异常。网络断了、事务回滚了、并发来了——这些场景下的数据状态，没人验证。

数据一致性的本质，是事务的ACID属性。A（原子性）：操作要么全做，要么全不做。C（一致性）：操作前后，数据约束不被破坏。I（隔离性）：并发操作互不干扰。D（持久性）：操作一旦完成，结果永久保存。

测试数据一致性，就是验证这四个属性在真实场景下是否成立。

ACID不验证，数据就是薛定谔的猫

02 实战验证：用SQL和代码锁定数据一致性

理论说完了，上真家伙。我们用一个经典的“银行转账”场景来演示。

场景：用户A向用户B转账100元。A账户余额500，B账户余额200。转账后，A余额应为400，B余额应为300。

表结构：

sql

CREATE TABLE accounts (
    user_id INT PRIMARY KEY,
    balance DECIMAL(10,2) NOT NULL,
    version INT DEFAULT 0
);

INSERT INTO accounts VALUES (1, 500.00, 0), (2, 200.00, 0);

2.1 原子性验证

原子性要求：A扣钱和B加钱，要么都成功，要么都失败。不能出现A扣了钱，B没收到。

测试方法：在事务中故意制造失败，然后检查数据状态。

sql

-- 开启事务
BEGIN;
-- 从A扣钱
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
-- 模拟失败：插入一个违反约束的数据
INSERT INTO accounts VALUES (3, -100.00, 0); -- 假设balance不能为负，但此处强制插入
-- 回滚事务
ROLLBACK;
-- 验证：A的余额是否还是500？
SELECT balance FROM accounts WHERE user_id = 1;
-- 预期结果：500.00

测试要点：

►在事务中插入一个必然失败的操作，验证回滚后所有数据恢复原状。

►检查日志，确认没有部分提交的记录。

2.2 一致性验证

一致性要求：转账前后，总余额不变。500+200=700，转账后400+300=700。

测试方法：计算转账前后的总余额，对比是否相等。

sql

-- 转账前总余额
SELECT SUM(balance) AS total_before FROM accounts;
-- 执行转账（此处用伪代码，实际在应用层执行）
-- UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
-- UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
-- 转账后总余额
SELECT SUM(balance) AS total_after FROM accounts;
-- 断言：total_before == total_after

测试要点：

►不只看单个账户，要看全局约束。

►检查是否有账户余额变成负数（如果业务规则不允许）。

►验证账户数量是否变化（没有凭空多出或消失账户）。

2.3 隔离性验证

隔离性要求：并发转账时，数据不会被破坏。

测试方法：用两个线程同时向同一账户转账。

模拟两个线程同时从A向B转账100元

t1 = threading.Thread(target=transfer, args=(1, 2, 100))

t2 = threading.Thread(target=transfer, args=(1, 2, 100))

t1.start()

t2.start()

t1.join()

t2.join()

验证最终余额

conn = pymysql.connect(host='localhost', user='test', password='test', db='test')

cursor = conn.cursor()

cursor.execute("SELECT balance FROM accounts WHERE user_id = 1")

print("A余额:", cursor.fetchone()[0]) # 预期：300（500-100-100）

cursor.execute("SELECT balance FROM accounts WHERE user_id = 2")

print("B余额:", cursor.fetchone()[0]) # 预期：400（200+100+100）

text


**测试要点**：
- 重复执行100次，看是否出现数据不一致。
- 检查是否有死锁发生，死锁后数据是否回滚到一致状态。
- 使用不同隔离级别（READ COMMITTED、REPEATABLE READ、SERIALIZABLE）对比结果。

<blockquote style="margin:16px 0;padding:12px 20px;background:#F5F5F5;border-left:4px solid #9E9E9E;color:#616161;font-style:italic;border-radius:0 6px 6px 0;">**并发不测，数据就是纸糊的墙**</blockquote>

## 03 数据完整性：从约束到触发器，一个都不能少

数据完整性，是数据库的“法律”。它规定了数据长什么样、能有什么值、表与表之间怎么关联。

<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">3.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">1 实体完整性验证</span></div><div style="padding:16px 20px;"><p>实体完整性要求：每行数据都是唯一的、可识别的。核心就是<strong>主键</strong>和<strong>唯一约束</strong>。</p>
<p><strong>测试方法</strong>：尝试插入重复数据，验证约束生效。</p>
<div style="margin:16px 0;border-radius:8px;overflow:hidden;border:1px solid #E0E0E0;"><div style="background:#263238;color:#B0BEC5;padding:6px 16px;font-size:12px;font-family:Consolas,monospace;"><span style="text-transform:uppercase;letter-spacing:1px;">sql</span></div><pre style="background:#1E272E;color:#E0E0E0;padding:16px 20px;margin:0;overflow-x:auto;font-size:14px;line-height:1.7;font-family:Consolas,"Courier New",monospace;"><code>-- 创建带主键的表
CREATE TABLE users (
    id INT PRIMARY KEY,
    email VARCHAR(100) UNIQUE
);

-- 测试主键
INSERT INTO users VALUES (1, 'a@test.com');
INSERT INTO users VALUES (1, 'b@test.com');  -- 预期：失败，主键重复

-- 测试唯一约束
INSERT INTO users VALUES (2, 'a@test.com');  -- 预期：失败，email重复</code></pre></div>
<p><strong>测试要点</strong>：</p>
<ul style="list-style:none;padding-left:0;margin:12px 0;">
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>尝试插入NULL值到主键列（应失败）。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>尝试UPDATE主键为已存在的值（应失败）。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>批量插入时，验证约束是否逐行检查。</li>
</ul></div></div>
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">3.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">2 参照完整性验证</span></div><div style="padding:16px 20px;"><p>参照完整性要求：外键引用的数据必须存在。不能出现订单的客户ID指向一个不存在的客户。</p>
<p><strong>测试方法</strong>：操作外键约束，验证级联行为。</p>
<div style="margin:16px 0;border-radius:8px;overflow:hidden;border:1px solid #E0E0E0;"><div style="background:#263238;color:#B0BEC5;padding:6px 16px;font-size:12px;font-family:Consolas,monospace;"><span style="text-transform:uppercase;letter-spacing:1px;">sql</span></div><pre style="background:#1E272E;color:#E0E0E0;padding:16px 20px;margin:0;overflow-x:auto;font-size:14px;line-height:1.7;font-family:Consolas,"Courier New",monospace;"><code>-- 创建表
CREATE TABLE customers (
    id INT PRIMARY KEY,
    name VARCHAR(50)
);

CREATE TABLE orders (
    id INT PRIMARY KEY,
    customer_id INT,
    FOREIGN KEY (customer_id) REFERENCES customers(id)
);

-- 测试插入
INSERT INTO orders VALUES (1, 999);  -- 预期：失败，客户ID不存在

-- 测试删除
INSERT INTO customers VALUES (1, 'Alice');
INSERT INTO orders VALUES (1, 1);
DELETE FROM customers WHERE id = 1;  -- 预期：失败，存在关联订单（如果未设置级联删除）

-- 测试级联行为（如果设置了ON DELETE CASCADE）
-- DELETE FROM customers WHERE id = 1;  -- 预期：成功，同时删除关联订单</code></pre></div>
<p><strong>测试要点</strong>：</p>
<ul style="list-style:none;padding-left:0;margin:12px 0;">
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试所有级联选项：CASCADE、SET NULL、RESTRICT、NO ACTION。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试多级外键关联（A→B→C），验证级联删除是否完整。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试循环引用（A→B→A），验证数据库是否能处理。</li>
</ul></div></div>
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">3.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">3 域完整性验证</span></div><div style="padding:16px 20px;"><p>域完整性要求：列的值必须在合法范围内。比如年龄不能为负，邮箱必须有@符号。</p>
<p><strong>测试方法</strong>：使用CHECK约束或触发器，验证非法数据被拒绝。</p>
<div style="margin:16px 0;border-radius:8px;overflow:hidden;border:1px solid #E0E0E0;"><div style="background:#263238;color:#B0BEC5;padding:6px 16px;font-size:12px;font-family:Consolas,monospace;"><span style="text-transform:uppercase;letter-spacing:1px;">sql</span></div><pre style="background:#1E272E;color:#E0E0E0;padding:16px 20px;margin:0;overflow-x:auto;font-size:14px;line-height:1.7;font-family:Consolas,"Courier New",monospace;"><code>-- 使用CHECK约束
CREATE TABLE employees (
    id INT PRIMARY KEY,
    age INT CHECK (age &gt;= 18 AND age &lt;= 65),
    salary DECIMAL(10,2) CHECK (salary &gt; 0)
);

-- 测试
INSERT INTO employees VALUES (1, 16, 5000);  -- 预期：失败，年龄小于18
INSERT INTO employees VALUES (2, 30, -100);  -- 预期：失败，薪资为负

-- 使用触发器验证更复杂的规则
CREATE TRIGGER check_salary_before_insert
BEFORE INSERT ON employees
FOR EACH ROW
BEGIN
    IF NEW.salary &lt; 0 THEN
        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '薪资不能为负';
    END IF;
END;</code></pre></div>
<p><strong>测试要点</strong>：</p>
<ul style="list-style:none;padding-left:0;margin:12px 0;">
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试边界值：最小值、最大值、刚好在边界上的值。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试NULL值：允许NULL的列，NULL是否被约束检查通过。</li>
<li style="position:relative;padding:4px 0 4px 24px;line-height:1.8;"><span style="color:#2196F3;font-weight:bold;position:absolute;left:4px;">&#9658;</span>测试数据类型：插入字符串到数字列、插入超长字符串。</li>
</ul>
<p>> <strong>约束不验证，数据就是无底洞</strong></p></div></div>
## 04 自动化测试：用脚本把数据验证变成日常

手动验证数据一致性，一次两次可以，天天做就是自杀。必须自动化。

<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">4.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">1 数据库测试框架：pytest + SQL</span></div><div style="padding:16px 20px;"><p>用Python的pytest框架，结合SQL查询，实现自动化数据验证。</p></div></div>
    # 记录转账前总余额
    cursor.execute("SELECT SUM(balance) FROM accounts")
    total_before = cursor.fetchone()[0]
    
    # 执行转账（实际项目中，这里调用业务接口）
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
    db_conn.commit()
    
    # 记录转账后总余额
    cursor.execute("SELECT SUM(balance) FROM accounts")
    total_after = cursor.fetchone()[0]
    
    # 验证
    assert total_before == total_after, f"总余额不一致：{total_before} vs {total_after}"

def test_no_negative_balance(db_conn):
    """验证没有账户余额为负"""
    cursor = db_conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM accounts WHERE balance < 0")
    negative_count = cursor.fetchone()[0]
    assert negative_count == 0, f"存在{negative_count}个负余额账户"

4.2 数据完整性检查脚本

对于生产环境，可以写一个定时任务，检查数据完整性。

检查外键完整性：订单的客户必须存在

cursor.execute("""

SELECT o.id, o.customer_id

FROM orders o

LEFT JOIN customers c ON o.customer_id = c.id

WHERE c.id IS NULL

""")

orphan_orders = cursor.fetchall()

if orphan_orders:

errors.append(f"存在{len(orphan_orders)}个孤立订单")

检查数据一致性：订单总金额 = 订单明细金额之和

cursor.execute("""

SELECT o.id, o.total_amount, SUM(od.amount) as detail_sum

FROM orders o

JOIN order_details od ON o.id = od.order_id

GROUP BY o.id, o.total_amount

HAVING o.total_amount != SUM(od.amount)

""")

inconsistent_orders = cursor.fetchall()

if inconsistent_orders:

errors.append(f"存在{len(inconsistent_orders)}个金额不一致的订单")

检查唯一约束：邮箱不能重复

cursor.execute("""

SELECT email, COUNT(*) as cnt

FROM users

GROUP BY email

HAVING COUNT(*) > 1

""")

duplicate_emails = cursor.fetchall()

if duplicate_emails:

errors.append(f"存在{len(duplicate_emails)}个重复邮箱")

如果有错误，发送告警

if errors:

send_alert("\n".join(errors))

cursor.close()

conn.close()

def send_alert(message):

发送邮件或钉钉告警

print(f"[{datetime.now()}] 数据完整性检查失败：\n{message}")

定时任务（使用cron或apscheduler）

每天凌晨2点执行一次

check_data_integrity()

text


**自动化测试要点**：
- 每个测试函数只验证一个规则，方便定位问题。
- 使用fixture管理数据库连接，测试结束后自动关闭。
- 测试数据独立：每个测试用例创建自己的测试数据，互不干扰。
- 测试失败时，给出明确的错误信息，包括具体的数据ID。

<blockquote style="margin:16px 0;padding:12px 20px;background:#F5F5F5;border-left:4px solid #9E9E9E;color:#616161;font-style:italic;border-radius:0 6px 6px 0;">**自动化不跑，数据验证就是空话**</blockquote>

## 05 高级技巧：从快照到混沌工程，把数据测试做到极致

基础方法掌握了，但真实场景远比这复杂。以下三个高级技巧，帮你应对99%的棘手情况。

<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">5.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">1 数据快照对比</span></div><div style="padding:16px 20px;"><p>在复杂的分布式系统中，数据可能跨多个数据库、多个表。如何验证整个系统的数据一致性？<strong>数据快照</strong>。</p>
<p><strong>方法</strong>：在操作前后，分别对相关数据做快照，然后对比差异。</p>
<div style="margin:16px 0;border-radius:8px;overflow:hidden;border:1px solid #E0E0E0;"><div style="background:#263238;color:#B0BEC5;padding:6px 16px;font-size:12px;font-family:Consolas,monospace;"><span style="text-transform:uppercase;letter-spacing:1px;">sql</span></div><pre style="background:#1E272E;color:#E0E0E0;padding:16px 20px;margin:0;overflow-x:auto;font-size:14px;line-height:1.7;font-family:Consolas,"Courier New",monospace;"><code>-- 操作前快照
CREATE TABLE snapshot_before AS
SELECT 'orders' as table_name, COUNT(*) as row_count, SUM(total_amount) as total_amount FROM orders
UNION ALL
SELECT 'payments', COUNT(*), SUM(amount) FROM payments
UNION ALL
SELECT 'inventory', COUNT(*), SUM(quantity) FROM inventory;

-- 执行操作（下单、支付、扣库存）

-- 操作后快照
CREATE TABLE snapshot_after AS
SELECT 'orders' as table_name, COUNT(*) as row_count, SUM(total_amount) as total_amount FROM orders
UNION ALL
SELECT 'payments', COUNT(*), SUM(amount) FROM payments
UNION ALL
SELECT 'inventory', COUNT(*), SUM(quantity) FROM inventory;

-- 对比差异
SELECT a.table_name, 
       a.row_count - b.row_count as row_diff,
       a.total_amount - b.total_amount as amount_diff
FROM snapshot_after a
JOIN snapshot_before b ON a.table_name = b.table_name;</code></pre></div></div></div>
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">5.</span><span style="font-size:17px;font-weight:700;color:#1565C0;">2 混沌工程式数据测试</span></div><div style="padding:16px 20px;"><p>主动注入故障，看数据库能否保持一致性。这才是真正的压力测试。</p></div></div>
        # 模拟50%概率的网络中断
        if random.random() < 0.5:
            raise Exception("模拟网络中断")
        
        cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
        cursor.execute("COMMIT")
    except:
        cursor.execute("ROLLBACK")
        # 验证数据是否回滚
        cursor.execute("SELECT balance FROM accounts WHERE user_id = 1")
        assert cursor.fetchone()[0] == 500, "数据未回滚"
        print("混沌测试通过：网络故障时数据回滚成功")
    finally:
        cursor.close()
        conn.close()

# 运行100次，看稳定性
for i in range(100):
    chaos_data_test()
    time.sleep(0.1)

5.3 性能与一致性的平衡

强一致性保证数据完整，但降低性能。最终一致性提升性能，但可能短暂不一致。如何测试？

方法：在并发场景下，测试不同一致性级别下的数据正确性。

测试最终一致性

def test_eventual_consistency():

"""验证在最终一致性模式下，数据最终会一致"""

模拟异步同步

conn1 = pymysql.connect(host='localhost', user='test', password='test', db='test')

conn2 = pymysql.connect(host='localhost', user='test', password='test', db='test')

cursor1 = conn1.cursor()

cursor2 = conn2.cursor()

在conn1中更新数据

cursor1.execute("UPDATE accounts SET balance = 600 WHERE user_id = 1")

conn1.commit()

立即在conn2中读取，可能读到旧值

cursor2.execute("SELECT balance FROM accounts WHERE user_id = 1")

immediate_balance = cursor2.fetchone()[0]

等待同步

time.sleep(1)

再次读取，应该读到新值

cursor2.execute("SELECT balance FROM accounts WHERE user_id = 1")

eventual_balance = cursor2.fetchone()[0]

最终一致性测试：最终应该一致

assert eventual_balance == 600, f"最终一致性失败：{eventual_balance} != 600"

print(f"最终一致性测试通过：即时读取={immediate_balance

声明：本站部分稿件版权来源于网络，如有侵犯版权，请及时联系我们。

JMeter vs LoadRunner：性能测试工具选型，90%的人选错了_千锋教育

移动测试框架大逃杀：Calabash、Detox、Maestro，谁是你的菜？_千锋教育

数据库测试：如何验证数据一致性和完整性_千锋教育

01 数据一致性：为什么99%的测试都漏掉了这个环节

02 实战验证：用SQL和代码锁定数据一致性

模拟两个线程同时从A向B转账100元

验证最终余额

检查外键完整性：订单的客户必须存在

检查数据一致性：订单总金额 = 订单明细金额之和

检查唯一约束：邮箱不能重复

如果有错误，发送告警

发送邮件或钉钉告警

定时任务（使用cron或apscheduler）

每天凌晨2点执行一次

测试最终一致性

模拟异步同步

在conn1中更新数据

立即在conn2中读取，可能读到旧值

等待同步

再次读取，应该读到新值

最终一致性测试：最终应该一致

最新问答资讯