数据库测试：数据验证与数据清理策略，让Bug无处遁形_千锋教育

-- 验证用户表插入
SELECT COUNT(*) FROM users WHERE email = 'test@example.com';
-- 预期返回1

-- 验证密码字段加密存储（非明文）
SELECT password_hash FROM users WHERE id = 1001;
-- 预期是哈希值，如 '5e884898da28047151d0e56f8dc62928'

-- 事务A：开启事务，更新数据但不提交
BEGIN TRANSACTION;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
-- 先不提交

-- 事务B：在另一个会话中读取
SELECT balance FROM accounts WHERE id = 1;
-- 如果读到更新后的值（脏读），说明隔离级别太低

   -- 先禁用外键检查（MySQL）
   SET FOREIGN_KEY_CHECKS = 0;
   TRUNCATE TABLE orders;
   TRUNCATE TABLE order_items;
   SET FOREIGN_KEY_CHECKS = 1;

   -- 测试开始
   START TRANSACTION;
   -- 执行测试操作（插入、更新等）
   INSERT INTO products (name, price) VALUES ('测试商品', 99.99);
   -- 测试结束，自动回滚
   ROLLBACK;

   -- 插入测试数据时标记
   INSERT INTO logs (message, test_run_id) VALUES ('测试日志', 'RUN_20231027_001');
   
   -- 清理时精准删除
   DELETE FROM logs WHERE test_run_id = 'RUN_20231027_001';

-- 安全检查：确认当前是测试库
SELECT DATABASE();  -- 预期返回 'test_db' 而非 'prod_db'
IF DATABASE() = 'prod_db' THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '禁止在生产库执行清理操作';
END IF;

import pymysql
import pytest

# 数据库连接配置
DB_CONFIG = {
    'host': 'localhost',
    'user': 'test_user',
    'password': 'test_pass',
    'database': 'test_db'
}

def test_user_registration():
    """验证用户注册后数据正确写入"""
    conn = pymysql.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cursor:
            # 模拟注册操作（通常由你的被测系统完成）
            # 这里只做验证
            cursor.execute("SELECT email, status FROM users WHERE id = 1001")
            result = cursor.fetchone()
            
            # 断言
            assert result is not None, "用户未插入"
            assert result[0] == 'newuser@test.com', f"邮箱错误: {result[0]}"
            assert result[1] == 'active', f"状态错误: {result[1]}"
    finally:
        conn.close()

class DataFactory:
    def __init__(self, db_conn):
        self.conn = db_conn
        self.created_ids = []
    
    def create_user(self, email='test@test.com', status='active'):
        """创建测试用户，并记录ID"""
        with self.conn.cursor() as cursor:
            cursor.execute(
                "INSERT INTO users (email, status) VALUES (%s, %s)",
                (email, status)
            )
            self.conn.commit()
            user_id = cursor.lastrowid
            self.created_ids.append(('users', user_id))
        return user_id
    
    def clean_up(self):
        """清理所有创建的数据（逆序删除）"""
        for table, record_id in reversed(self.created_ids):
            with self.conn.cursor() as cursor:
                cursor.execute(f"DELETE FROM {table} WHERE id = %s", (record_id,))
        self.conn.commit()
        self.created_ids.clear()

factory = DataFactory(db_conn)
user_id = factory.create_user()
# 执行测试...
factory.clean_up()  # 自动清理

-- 开启事务模拟
START TRANSACTION;

-- 插入订单
INSERT INTO orders (user_id, product_id, quantity) VALUES (1, 101, 2);

-- 扣减库存
UPDATE products SET stock = stock - 2 WHERE id = 101;

-- 验证库存不为负
SELECT stock FROM products WHERE id = 101;
-- 预期：stock >= 0（如果为负，说明并发控制失效）

COMMIT;

-- 删除用户前，记录关联数据数量
SELECT COUNT(*) FROM orders WHERE user_id = 100;
SELECT COUNT(*) FROM comments WHERE user_id = 100;

-- 执行删除（假设有级联约束）
DELETE FROM users WHERE id = 100;

-- 验证级联删除是否生效
SELECT COUNT(*) FROM orders WHERE user_id = 100;  -- 预期0
SELECT COUNT(*) FROM comments WHERE user_id = 100; -- 预期0

-- 模拟更新
UPDATE orders SET status = 'shipped' WHERE id = 500;

-- 验证updated_at是否大于created_at
SELECT id, created_at, updated_at 
FROM orders 
WHERE id = 500 
  AND updated_at <= created_at;
-- 预期返回0行（如果返回，说明时间戳逻辑有bug）

-- 约束：库存不能为负
ALTER TABLE products 
ADD CONSTRAINT chk_stock_non_negative CHECK (stock >= 0);

-- 触发器：自动更新updated_at
CREATE TRIGGER update_orders_timestamp 
BEFORE UPDATE ON orders
FOR EACH ROW
SET NEW.updated_at = NOW();

# 测试前，从基线重建数据库
mysql -u test -p test_db < baseline_v1.0.sql

# 执行测试
pytest test_suite.py

# 测试后，自动清理
mysql -u test -p test_db < truncate_all.sql