SQL测试用例¶
测试目标: 验证SQL语句的正确性和性能 测试类型: 语法测试、功能测试、性能测试 涉及组件: SELECT、INSERT、UPDATE、DELETE、JOIN、子查询
📋 测试概述¶
测试目标¶
- 语法测试: 验证SQL语法正确性
- 功能测试: 验证SQL功能正确性
- 性能测试: 评估SQL执行效率
- 边界测试: 测试边界条件
测试环境¶
- 数据库: MySQL 8.0+, PostgreSQL 13+
- 测试框架: pytest
- 数据生成: Faker
🧪 测试用例列表¶
1. SELECT语句测试¶
测试用例1.1: 基础SELECT¶
测试目标: 验证基础SELECT语句
测试代码:
Python
import pytest
import mysql.connector
from mysql.connector import Error
def test_basic_select():
"""测试基础SELECT"""
# 创建连接
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 执行SELECT
cursor.execute("SELECT * FROM users LIMIT 10")
results = cursor.fetchall()
# 验证结果
assert isinstance(results, list)
assert len(results) <= 10
if results:
assert 'id' in results[0]
assert 'name' in results[0]
print("✓ 基础SELECT测试通过")
finally:
cursor.close()
conn.close()
预期结果: 查询成功返回结果
测试用例1.2: WHERE子句¶
测试目标: 验证WHERE子句
测试代码:
Python
def test_where_clause():
"""测试WHERE子句"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 测试等于
cursor.execute("SELECT * FROM users WHERE id = 1")
results = cursor.fetchall()
assert len(results) == 1
# 测试大于
cursor.execute("SELECT * FROM users WHERE age > 18")
results = cursor.fetchall()
assert all(r['age'] > 18 for r in results)
# 测试IN
cursor.execute("SELECT * FROM users WHERE status IN ('active', 'pending')")
results = cursor.fetchall()
assert all(r['status'] in ['active', 'pending'] for r in results)
# 测试LIKE
cursor.execute("SELECT * FROM users WHERE name LIKE 'A%'")
results = cursor.fetchall()
assert all(r['name'].startswith('A') for r in results)
print("✓ WHERE子句测试通过")
finally:
cursor.close()
conn.close()
预期结果: 所有WHERE条件正确过滤数据
测试用例1.3: ORDER BY和LIMIT¶
测试目标: 验证ORDER BY和LIMIT
测试代码:
Python
def test_order_by_limit():
"""测试ORDER BY和LIMIT"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 测试ORDER BY ASC
cursor.execute("SELECT * FROM users ORDER BY age ASC LIMIT 5")
results = cursor.fetchall()
ages = [r['age'] for r in results]
assert ages == sorted(ages)
# 测试ORDER BY DESC
cursor.execute("SELECT * FROM users ORDER BY created_at DESC LIMIT 5")
results = cursor.fetchall()
assert len(results) == 5
# 测试多列排序
cursor.execute("SELECT * FROM users ORDER BY status, age DESC LIMIT 10")
results = cursor.fetchall()
assert len(results) == 10
print("✓ ORDER BY和LIMIT测试通过")
finally:
cursor.close()
conn.close()
预期结果: 结果按指定顺序排序
测试用例1.4: 聚合函数¶
测试目标: 验证聚合函数
测试代码:
Python
def test_aggregate_functions():
"""测试聚合函数"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 测试COUNT
cursor.execute("SELECT COUNT(*) as count FROM users")
result = cursor.fetchone()
assert 'count' in result
assert result['count'] >= 0
# 测试SUM
cursor.execute("SELECT SUM(age) as total_age FROM users")
result = cursor.fetchone()
assert 'total_age' in result
# 测试AVG
cursor.execute("SELECT AVG(age) as avg_age FROM users")
result = cursor.fetchone()
assert 'avg_age' in result
assert result['avg_age'] > 0
# 测试MAX和MIN
cursor.execute("SELECT MAX(age) as max_age, MIN(age) as min_age FROM users")
result = cursor.fetchone()
assert 'max_age' in result
assert 'min_age' in result
assert result['max_age'] >= result['min_age']
# 测试GROUP BY
cursor.execute("SELECT status, COUNT(*) as count FROM users GROUP BY status")
results = cursor.fetchall()
assert len(results) > 0
assert all('status' in r and 'count' in r for r in results)
print("✓ 聚合函数测试通过")
finally:
cursor.close()
conn.close()
预期结果: 聚合函数正确计算
2. JOIN操作测试¶
测试用例2.1: INNER JOIN¶
测试目标: 验证INNER JOIN
测试代码:
Python
def test_inner_join():
"""测试INNER JOIN"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 执行INNER JOIN
query = """
SELECT u.id, u.name, o.id as order_id, o.amount
FROM users u
INNER JOIN orders o ON u.id = o.user_id
LIMIT 10
"""
cursor.execute(query)
results = cursor.fetchall()
# 验证结果
assert isinstance(results, list) # isinstance检查对象类型
assert len(results) <= 10
for result in results:
assert 'id' in result
assert 'name' in result
assert 'order_id' in result
assert 'amount' in result
print("✓ INNER JOIN测试通过")
finally:
cursor.close()
conn.close()
预期结果: 只返回匹配的行
测试用例2.2: LEFT JOIN¶
测试目标: 验证LEFT JOIN
测试代码:
Python
def test_left_join():
"""测试LEFT JOIN"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 执行LEFT JOIN
query = """
SELECT u.id, u.name, o.id as order_id
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
LIMIT 10
"""
cursor.execute(query)
results = cursor.fetchall()
# 验证结果
assert len(results) > 0
# LEFT JOIN应该返回所有左表行
# 即使右表没有匹配
for result in results:
assert 'id' in result
assert 'name' in result
# order_id可能为NULL
assert 'order_id' in result
print("✓ LEFT JOIN测试通过")
finally:
cursor.close()
conn.close()
预期结果: 返回所有左表行
测试用例2.3: 多表JOIN¶
测试目标: 验证多表JOIN
测试代码:
Python
def test_multiple_join():
"""测试多表JOIN"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 执行多表JOIN
query = """
SELECT u.id, u.name, o.id as order_id, p.name as product_name
FROM users u
INNER JOIN orders o ON u.id = o.user_id
INNER JOIN order_items oi ON o.id = oi.order_id
INNER JOIN products p ON oi.product_id = p.id
LIMIT 10
"""
cursor.execute(query)
results = cursor.fetchall()
# 验证结果
assert len(results) <= 10
for result in results:
assert 'id' in result
assert 'name' in result
assert 'order_id' in result
assert 'product_name' in result
print("✓ 多表JOIN测试通过")
finally:
cursor.close()
conn.close()
预期结果: 正确连接多个表
3. INSERT语句测试¶
测试用例3.1: 基础INSERT¶
测试目标: 验证基础INSERT语句
测试代码:
Python
def test_basic_insert():
"""测试基础INSERT"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 插入数据
query = """
INSERT INTO users (name, email, age, status)
VALUES ('Test User', 'test@example.com', 25, 'active')
"""
cursor.execute(query)
conn.commit()
# 验证插入
assert cursor.rowcount == 1
# 查询验证
cursor.execute("SELECT * FROM users WHERE name = 'Test User'")
result = cursor.fetchone()
assert result is not None
assert result['name'] == 'Test User'
assert result['email'] == 'test@example.com'
# 清理测试数据
cursor.execute("DELETE FROM users WHERE name = 'Test User'")
conn.commit()
print("✓ 基础INSERT测试通过")
finally:
cursor.close()
conn.close()
预期结果: 数据成功插入
测试用例3.2: 批量INSERT¶
测试目标: 验证批量INSERT
测试代码:
Python
def test_batch_insert():
"""测试批量INSERT"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 批量插入数据
query = """
INSERT INTO users (name, email, age, status)
VALUES (%s, %s, %s, %s)
"""
data = [
('User 1', 'user1@example.com', 20, 'active'),
('User 2', 'user2@example.com', 25, 'active'),
('User 3', 'user3@example.com', 30, 'active'),
]
cursor.executemany(query, data)
conn.commit()
# 验证插入
assert cursor.rowcount == 3
# 查询验证
cursor.execute("SELECT COUNT(*) as count FROM users WHERE name LIKE 'User %'")
result = cursor.fetchone()
assert result['count'] >= 3
# 清理测试数据
cursor.execute("DELETE FROM users WHERE name LIKE 'User %'")
conn.commit()
print("✓ 批量INSERT测试通过")
finally:
cursor.close()
conn.close()
预期结果: 批量数据成功插入
4. UPDATE语句测试¶
测试用例4.1: 基础UPDATE¶
测试目标: 验证基础UPDATE语句
测试代码:
Python
def test_basic_update():
"""测试基础UPDATE"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 先插入测试数据
cursor.execute("""
INSERT INTO users (name, email, age, status)
VALUES ('Update Test', 'update@example.com', 25, 'active')
""")
conn.commit()
# 更新数据
query = "UPDATE users SET age = 26 WHERE name = 'Update Test'"
cursor.execute(query)
conn.commit()
# 验证更新
assert cursor.rowcount == 1
# 查询验证
cursor.execute("SELECT age FROM users WHERE name = 'Update Test'")
result = cursor.fetchone()
assert result['age'] == 26
# 清理测试数据
cursor.execute("DELETE FROM users WHERE name = 'Update Test'")
conn.commit()
print("✓ 基础UPDATE测试通过")
finally:
cursor.close()
conn.close()
预期结果: 数据成功更新
测试用例4.2: 条件UPDATE¶
测试目标: 验证条件UPDATE
测试代码:
Python
def test_conditional_update():
"""测试条件UPDATE"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 插入测试数据
cursor.executemany("""
INSERT INTO users (name, email, age, status)
VALUES (%s, %s, %s, %s)
""", [
('Test 1', 'test1@example.com', 20, 'active'),
('Test 2', 'test2@example.com', 25, 'active'),
('Test 3', 'test3@example.com', 30, 'active'),
])
conn.commit()
# 条件更新
query = "UPDATE users SET status = 'inactive' WHERE age > 25"
cursor.execute(query)
conn.commit()
# 验证更新
assert cursor.rowcount >= 1
# 查询验证
cursor.execute("SELECT COUNT(*) as count FROM users WHERE status = 'inactive' AND name LIKE 'Test %'")
result = cursor.fetchone()
assert result['count'] >= 1
# 清理测试数据
cursor.execute("DELETE FROM users WHERE name LIKE 'Test %'")
conn.commit()
print("✓ 条件UPDATE测试通过")
finally:
cursor.close()
conn.close()
预期结果: 符合条件的数据被更新
5. DELETE语句测试¶
测试用例5.1: 基础DELETE¶
测试目标: 验证基础DELETE语句
测试代码:
Python
def test_basic_delete():
"""测试基础DELETE"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 插入测试数据
cursor.execute("""
INSERT INTO users (name, email, age, status)
VALUES ('Delete Test', 'delete@example.com', 25, 'active')
""")
conn.commit()
# 删除数据
query = "DELETE FROM users WHERE name = 'Delete Test'"
cursor.execute(query)
conn.commit()
# 验证删除
assert cursor.rowcount == 1
# 查询验证
cursor.execute("SELECT COUNT(*) as count FROM users WHERE name = 'Delete Test'")
result = cursor.fetchone()
assert result['count'] == 0
print("✓ 基础DELETE测试通过")
finally:
cursor.close()
conn.close()
预期结果: 数据成功删除
测试用例5.2: 条件DELETE¶
测试目标: 验证条件DELETE
测试代码:
Python
def test_conditional_delete():
"""测试条件DELETE"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 插入测试数据
cursor.executemany("""
INSERT INTO users (name, email, age, status)
VALUES (%s, %s, %s, %s)
""", [
('Test 1', 'test1@example.com', 20, 'active'),
('Test 2', 'test2@example.com', 25, 'active'),
('Test 3', 'test3@example.com', 30, 'active'),
])
conn.commit()
# 条件删除
query = "DELETE FROM users WHERE age > 25 AND name LIKE 'Test %'"
cursor.execute(query)
conn.commit()
# 验证删除
assert cursor.rowcount >= 1
# 查询验证
cursor.execute("SELECT COUNT(*) as count FROM users WHERE age > 25 AND name LIKE 'Test %'")
result = cursor.fetchone()
assert result['count'] == 0
# 清理剩余测试数据
cursor.execute("DELETE FROM users WHERE name LIKE 'Test %'")
conn.commit()
print("✓ 条件DELETE测试通过")
finally:
cursor.close()
conn.close()
预期结果: 符合条件的数据被删除
6. 子查询测试¶
测试用例6.1: 标量子查询¶
测试目标: 验证标量子查询
测试代码:
Python
def test_scalar_subquery():
"""测试标量子查询"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 使用子查询
query = """
SELECT * FROM users
WHERE age > (SELECT AVG(age) FROM users)
LIMIT 10
"""
cursor.execute(query)
results = cursor.fetchall()
# 验证结果
assert len(results) <= 10
# 获取平均年龄
cursor.execute("SELECT AVG(age) as avg_age FROM users")
avg_age = cursor.fetchone()['avg_age']
# 验证所有结果年龄都大于平均值
for result in results:
assert result['age'] > avg_age
print("✓ 标量子查询测试通过")
finally:
cursor.close()
conn.close()
预期结果: 子查询正确执行
测试用例6.2: EXISTS子查询¶
测试目标: 验证EXISTS子查询
测试代码:
Python
def test_exists_subquery():
"""测试EXISTS子查询"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try:
# 使用EXISTS子查询
query = """
SELECT u.id, u.name
FROM users u
WHERE EXISTS (
SELECT 1 FROM orders o
WHERE o.user_id = u.id
)
LIMIT 10
"""
cursor.execute(query)
results = cursor.fetchall()
# 验证结果
assert len(results) <= 10 # assert断言：条件为False时抛出AssertionError
for result in results:
assert 'id' in result
assert 'name' in result
print("✓ EXISTS子查询测试通过")
finally:
cursor.close()
conn.close()
预期结果: EXISTS子查询正确执行
7. 性能测试¶
测试用例7.1: 查询性能测试¶
测试目标: 测试查询执行时间
测试代码:
Python
import time
def test_query_performance():
"""测试查询性能"""
conn = mysql.connector.connect(
host='localhost',
user='test_user',
password='test_password',
database='test_db'
)
cursor = conn.cursor(dictionary=True)
try: # try/except捕获异常
# 测试简单查询
query = "SELECT * FROM users LIMIT 1000"
start_time = time.time()
cursor.execute(query)
results = cursor.fetchall()
end_time = time.time()
execution_time = end_time - start_time
print(f"简单查询执行时间: {execution_time:.4f}s")
print(f"返回行数: {len(results)}")
# 验证性能
assert execution_time < 1.0, "查询执行时间过长"
# 测试复杂查询
complex_query = """
SELECT u.id, u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name
HAVING COUNT(o.id) > 5
LIMIT 100
"""
start_time = time.time()
cursor.execute(complex_query)
results = cursor.fetchall()
end_time = time.time()
execution_time = end_time - start_time
print(f"复杂查询执行时间: {execution_time:.4f}s")
print(f"返回行数: {len(results)}")
# 验证性能
assert execution_time < 5.0, "复杂查询执行时间过长"
print("✓ 查询性能测试通过")
finally:
cursor.close()
conn.close()
预期结果: 查询在合理时间内完成
📊 测试执行¶
运行所有测试¶
Bash
# 运行所有测试
pytest tests/test_sql.py -v
# 运行特定测试
pytest tests/test_sql.py::test_basic_select -v
# 生成覆盖率报告
pytest tests/test_sql.py --cov=sql --cov-report=html
✅ 验证方法¶
1. 自动化验证¶
- 运行所有测试用例
- 检查断言是否通过
- 记录测试结果
2. 性能基准¶
- 建立性能基准
- 监控查询性能变化
- 优化慢查询
3. 数据验证¶
- 验证查询结果正确性
- 检查数据完整性
- 验证约束条件
📝 测试报告¶
测试报告应包含：
- 测试概览
- 测试用例数量
- 通过/失败统计
-
执行时间
-
详细结果
- 每个测试用例的结果
- 查询执行时间
-
性能指标
-
问题分析
- 失败原因分析
- 性能瓶颈分析
- 优化建议
测试完成标准: 所有测试用例通过 推荐测试频率: 每次SQL更新 测试维护周期: 每周