聊一聊接口自动化测试如何验证数据库和数据表
接口自动化测试主要是验证API的请求和响应是否正确，但有时候，仅仅检查接口返回的数据可能不够，因为数据可能没有正确写入数据库，或者数据库中的状态没有更新。这时候就需要验证数据库了。那具体该怎么做呢？可能需要连接数据库，执行查询，然后检查结果是否符合预期。
当调用一个创建资源的接口，比如POST /users，这时候接口返回201 Created，但还要确认数据库里真的插入了一条新用户记录。这时候，可能需要连接到用户表，查询最新的记录，或者根据返回的ID去查。例如，用户注册可能涉及用户表，但如果有其他关联表，比如用户权限表、日志表等，可能也需要检查。所以，涉及的表格通常是根据接口的功能来决定的。比如，订单接口可能涉及订单表、订单项表、库存表等。
那么在自动化测试中，是如何组织这些验证呢？可能需要写一些数据库查询的代码，比如用Python的pytest框架，结合SQLAlchemy或者直接使用数据库驱动（如psycopg2、pymysql）来执行查询。然后，断言查询结果是否符合预期，比如记录的数量、某个字段的值是否正确。
️一、在验证数据库中的表需要考虑的问题如下几个️测试前后的数据库状态管理
比如，在测试前清理数据库，或者在测试后回滚事务，避免测试数据污染其他测试用例。或者使用测试专用的数据库，每次测试后都清理数据。
aspcms.cn️如何确定需要检查哪些表
可能需要根据业务逻辑和接口的文档来确定。例如，创建订单的接口，可能需要在订单表中新增一条记录，同时减少库存表中的库存数量，这时候需要检查这两个表。
️数据验证的精确性
比如，不仅要检查是否存在记录，还要检查各个字段的值是否正确，时间戳是否合理，关联的外键是否正确等。
️如何处理数据库的延迟
有时候接口返回成功，但数据库的写入可能因为异步操作而稍后完成，这时候可能需要重试机制，或者确保测试框架能够处理这种情况。
️不同环境的数据库配置
测试环境、预发布环境的数据库可能不同，自动化测试需要能够适配不同的环境配置。
️二、验证数据库的步骤️执行接口请求
发送API请求（如POST、PUT、DELETE）并获取响应数据，例如新创建资源的ID。
️确定涉及的数据表
根据接口功能分析影响的表。
下单流程可能涉及的表如下图。
️编写数据库查询
使用SQL或ORM工具查询相关表，如通过接口返回的ID检索记录：
sql
SELECT * FROM users WHERE id = {user_id};
执行查询并获取数据
使用数据库驱动（如Python的pymysql）或ORM（如SQLAlchemy）执行查询。
️断言验证
对比数据库结果与预期值，例如：
记录是否存在。
字段值是否正确（如用户名、邮箱）。
关联表数据是否更新（如库存减少）。
️清理测试数据
删除或回滚测试数据，避免影响后续测试。
️涉及的表类型
主表
直接由接口操作的核心表，如users、orders。
关联表
与主表有外键关系的表，如order_items关联orders。
日志/审计表
记录操作日志的表，如audit_logs。
状态表
存储业务状态的表，如inventory（库存状态）。
工具与代码示例（Python）
#python
import pytest
import requests
import pymysql
def test_create_user():
# 1. 调用接口
url = "http://api.example.com/users"
data = {"name": "John", "email": "john@example.com"}
response = requests.post(url, json=data)
assert response.status_code == 201
user_id = response.json()["id"]
# 2. 连接数据库验证
connection = pymysql.connect(host="test-db", user="root", password="", db="test")
try:
with connection.cursor() as cursor:
cursor.execute("SELECT name, email FROM users WHERE id = %s", (user_id,))
result = cursor.fetchone()
assert result == ("John", "john@example.com")
finally:
connection.close()
# 3. 清理数据（可选）
with connection.cursor() as cursor:
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
connection.commit()
️三、关键注意事项️环境隔离
使用独立的测试数据库，避免污染生产数据。
️事务管理
通过事务回滚（如pytest-django的transactional_db）自动清理数据。
️异步操作处理
若接口异步更新数据库，需添加重试机制：
#python
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def check_database():
# 查询数据库
️敏感数据处理
验证加密字段时，使用相同的哈希算法比对：
#python
import hashlib
hashed_password = hashlib.sha256("password123".encode()).hexdigest()
assert db_result["password"] == hashed_password
典型场景示例
场景1：创建资源
接口：POST /api/users
验证表：users
检查字段：name、email、created_at是否合理。
场景2：删除订单
接口：DELETE /api/orders/{id}
验证表：orders（检查is_deleted=1或记录不存在）。
场景3：更新库存
接口：POST /api/orders
验证表：inventory（确认库存数量减少）。
️四、验证的过程中需要注意的点测试数据库应该与接口服务使用的数据库一致，但最好是独立的测试实例，避免影响生产数据。
测试用例之间要隔离，避免因为数据残留导致测试失败。
数据库操作可能需要事务支持，在测试完成后回滚，避免留下测试数据。
对于复杂的业务逻辑，可能需要检查多个表的数据变化，确保数据一致性和完整性。
️五、验证过程中可能遇到的挑战测试数据准备：在测试前需要预置一些数据，例如测试删除订单前，需要先创建订单。这时候可能需要通过接口或者直接插入数据库来准备数据。
时间相关的字段，比如创建时间、更新时间，可能需要检查是否在合理的范围内，但不能依赖具体的时间值，因为每次测试运行的时间不同。这时候可以用范围检查，比如是否在调用接口之后的时间。
处理数据库的敏感信息，比如密码加密存储，测试时需要确保加密正确，比如不能明文存储密码，而是存储哈希值，这时候可能需要用相同的方法生成哈希，再进行比较。
