测试人员必备：推荐一款智能测试数据清理AI Skill，支持db、redis、mq等

在软件测试的日常工作中，数据清理是一个看似简单但实则繁琐且容易出错的环节。每次测试执行后，残留的脏数据不仅会干扰后续测试结果，还可能导致环境崩溃。今天，我要向大家推荐一款我亲手打造的智能测试数据清理AI Skill——它能够自动识别并清理数据库(MySQL/PostgreSQL)、缓存(Redis)和消息队列(MQ)中的测试数据，让测试人员从重复劳动中解放出来，专注于更有价值的测试设计。## 为什么需要智能数据清理？传统的数据清理方式通常依赖手动执行SQL脚本或编写硬编码的清理函数。这种方式存在以下痛点：- 脚本维护成本高：不同测试场景的数据结构差异大，每个模块都需要单独维护清理脚本。- 易遗漏：测试数据可能分散在多个数据源（DB、Redis、MQ），手动清理往往顾此失彼。- 风险不可控：误删生产数据、删错范围等事故时有发生。而智能AI Skill通过自然语言理解，只需要告诉它“清理用户注册测试产生的数据”，它就能自动分析数据关联，精准执行清理，同时保留必要的测试配置。## 核心架构设计该AI Skill基于一个轻量级的规则引擎+LLM（大语言模型）实现，整体架构分为三层：1. 意图识别层：解析用户自然语言指令，提取数据范围、数据源类型、清理策略。2. 数据源适配层：抽象出统一的清理接口，为DB、Redis、MQ分别实现适配器。3. 安全执行层：提供预检模式、回滚机制、黑白名单过滤。下面我们通过实战代码来理解其核心实现。## 实战代码示例1：数据库智能清理模块pythonimport pymysqlfrom typing import Dict, List, Anyclass DatabaseCleaner: """ 数据库智能清理器，支持MySQL/PostgreSQL 特点：自动识别外键关系，按依赖顺序删除 """ def __init__(self, db_config: Dict[str, Any]): self.conn = pymysql.connect(**db_config) self.cursor = self.conn.cursor() def get_table_dependencies(self, table_name: str) -> List[str]: """ 获取指定表的外键依赖关系，返回依赖它的子表列表 例如：orders依赖users，则返回['orders'] """ sql = """ SELECT TABLE_NAME, COLUMN_NAME, REFERENCED_TABLE_NAME FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE WHERE REFERENCED_TABLE_SCHEMA = DATABASE() AND REFERENCED_TABLE_NAME = %s """ self.cursor.execute(sql, (table_name,)) results = self.cursor.fetchall() # 提取依赖表名（去重） dependent_tables = list(set([row[0] for row in results])) return dependent_tables def safe_delete_with_dependencies(self, table_name: str, where_clause: str): """ 安全删除：先删除依赖表数据，再删除目标表 :param table_name: 目标表名 :param where_clause: WHERE条件，如"user_id = 123" """ # 1. 递归查找所有依赖表 dependencies = self.get_table_dependencies(table_name) for dep_table in dependencies: # 递归删除子表数据（防止多层外键） self.safe_delete_with_dependencies(dep_table, where_clause) # 2. 构建安全删除SQL # 使用LIMIT分批删除，避免锁表 batch_size = 100 while True: select_sql = f"SELECT COUNT(*) FROM {table_name} WHERE {where_clause} LIMIT {batch_size}" self.cursor.execute(select_sql) count = self.cursor.fetchone()[0] if count == 0: break delete_sql = f"DELETE FROM {table_name} WHERE {where_clause} LIMIT {batch_size}" self.cursor.execute(delete_sql) self.conn.commit() print(f"已清理 {table_name} 中 {count} 条记录") def close(self): self.cursor.close() self.conn.close()# 使用示例：清理用户ID为10086的所有相关数据cleaner = DatabaseCleaner({ "host": "127.0.0.1", "user": "test_user", "password": "test_pass", "database": "test_db", "charset": "utf8mb4"})try: # 智能识别依赖关系，先删订单表，再删用户表 cleaner.safe_delete_with_dependencies("users", "user_id = 10086") print("数据库清理完成！")finally: cleaner.close()## 实战代码示例2：Redis与MQ联合清理模块pythonimport redisimport pikafrom typing import List, Dict, Anyclass CacheAndQueueCleaner: """ Redis + 消息队列联合清理器 支持Redis键模式匹配、MQ队列消息清除 """ def __init__(self, redis_config: Dict[str, Any], mq_config: Dict[str, Any]): # 初始化Redis连接 self.redis_client = redis.Redis(**redis_config) # 初始化RabbitMQ连接 self.mq_connection = pika.BlockingConnection( pika.ConnectionParameters(host=mq_config['host']) ) self.mq_channel = self.mq_connection.channel() def clean_redis_by_pattern(self, pattern: str) -> int: """ 按模式匹配清理Redis缓存 :param pattern: 键模式，如"test:user:*" :return: 清理的键数量 """ cursor = 0 deleted_count = 0 # 使用SCAN命令分批扫描，避免阻塞 while True: cursor, keys = self.redis_client.scan( cursor=cursor, match=pattern, count=100 # 每批扫描100个键 ) if keys: # 批量删除 self.redis_client.delete(*keys) deleted_count += len(keys) print(f"已清理Redis键: {keys[:5]}... (共{len(keys)}个)") if cursor == 0: # 扫描完成 break return deleted_count def purge_mq_queue(self, queue_name: str, filter_func=None): """ RabbitMQ队列清理，支持消息内容过滤 :param queue_name: 队列名称 :param filter_func: 可选过滤函数，接收消息体，返回True则保留 """ # 先获取队列中消息数量 queue_info = self.mq_channel.queue_declare( queue=queue_name, durable=True, passive=True ) message_count = queue_info.method.message_count print(f"队列 {queue_name} 当前有 {message_count} 条消息") # 逐个消费并选择性重新投递 kept_messages = 0 for _ in range(message_count): method_frame, header_frame, body = self.mq_channel.basic_get( queue=queue_name, auto_ack=False ) if method_frame is None: break if filter_func and filter_func(body): # 满足条件的消息重新入队 self.mq_channel.basic_publish( exchange='', routing_key=queue_name, body=body, properties=header_frame ) kept_messages += 1 print(f"保留消息: {body[:50]}...") else: # 否则确认删除 self.mq_channel.basic_ack(delivery_tag=method_frame.delivery_tag) print(f"删除消息: {body[:50]}...") print(f"清理完成，保留 {kept_messages} 条，删除 {message_count - kept_messages} 条") def close(self): self.redis_client.close() self.mq_connection.close()# 使用示例：清理测试环境的临时数据cleaner = CacheAndQueueCleaner( redis_config={"host": "localhost", "port": 6379, "db": 0, "decode_responses": True}, mq_config={"host": "localhost"})try: # 清理Redis中所有以"test_session:"开头的键 cleaned_redis = cleaner.clean_redis_by_pattern("test_session:*") print(f"Redis清理完成，共删除 {cleaned_redis} 个键") # 清理MQ中订单相关的测试消息 def keep_production_orders(body): """保留包含'prod'标记的订单消息""" return b"prod_order" in body cleaner.purge_mq_queue("order_queue", filter_func=keep_production_orders) print("MQ清理完成！")finally: cleaner.close()## 如何集成到AI Skill中？上述代码展示了基础能力，真正的AI Skill在此基础上增加了智能决策层。例如，当用户说“清理上周五冒烟测试的所有数据”时，系统会：1. 解析时间范围：将“上周五”转换为具体日期范围。2. 识别数据源：根据上下文判断需要清理DB中的smoke_test表、Redis中的smoke:*键、MQ中的smoke_test_queue。3. 执行预检：在清理前先统计受影响数据量，让用户确认。4. 分步清理：按DB→Redis→MQ的顺序执行，每个步骤都有回滚能力。## 总结智能测试数据清理AI Skill通过将自然语言理解与多数据源适配器结合，彻底改变了测试人员的数据清理方式。它带来的核心价值包括：- 效率提升：原本需要30分钟的手动清理，现在只需一句话指令。- 安全性保障：自动处理外键依赖、分批操作、预检确认，杜绝误删风险。- 扩展性强：通过添加新的数据源适配器，可以轻松支持MongoDB、Elasticsearch等更多存储。作为全栈工程师，我强烈建议测试团队引入这样的智能工具。你可以将上述代码作为基础框架，根据实际业务场景定制规则，打造适合自己团队的AI清理助手。毕竟，测试人员的时间应该花在发现bug上，而不是清理数据上。