内容社区系统开发文档（下）

目录

4 系统测试

4.1 数据库测试设计

测试流程图

测试策略矩阵

4.2 数据库测试内容 

测试点 1：数据项增删改查（CRUD）操作 和 数据一致性测试

一、基础数据初始化 

二、用户表测试（users）

1. 新增用户

2. 更新用户信息

3. 删除用户

三、内容表测试（contents）

1. 发布内容

2. 更新内容

3. 删除内容（级联验证）

四、评论系统测试（comments）

1. 发表评论

2. 逻辑删除评论

3. 物理删除评论

五、互动系统测试（interactions）

1. 点赞内容

2. 取消点赞（删除互动记录）

3. 收藏内容（唯一性约束测试）

六、关注系统测试（follows）

1. 关注用户

2. 取消关注

3. 统计粉丝数和关注数（通过存储过程）

七、标签系统测试（tags & content_tags）

1. 新增标签

2. 关联内容和标签

3. 删除标签（级联删除关联关系）

八、存储过程与函数测试

九、触发器完整性验证

1. 验证互动触发器

2. 验证评论逻辑删除触发器

十、异常场景测试

1. 外键约束失败

测试点 2：数据表的并发操作

测试点 3：边界条件测试（表满 / 空状态）

测试点4：针对存储过程的接口测试

测试点5：结合业务逻辑做关联表的接口测试

---

4 系统测试

4.1 数据库测试设计

测试流程图

测试策略矩阵

测试类型	测试目标	测试工具/方法
​​功能测试​​	CRUD操作有效性验证	SQL脚本
​​性能测试​​	高并发场景稳定性验证	ThreadPoolExecutor
​​数据完整性​​	级联删除/触发器验证	事务回滚测试
​​容错测试​​	无效数据插入/空表操作	边界值测试

4.2 数据库测试内容 

测试点 1：数据项增删改查（CRUD）操作 和 数据一致性测试

测试内容：验证用户表、内容表等表的数据增删改查功能。 验证外键约束、触发器逻辑是否保证数据一致性（如删除内容时级联删除关联评论）。

一、基础数据初始化 

-- ----------------------------
-- 一、基础数据初始化
-- ------------------------------ 插入测试用户
INSERT INTO users (username, email, password_hash)
VALUES
('test_user1', 'user1@example.com', SHA2('password1', 256)),
('test_user2', 'user2@example.com', SHA2('password2', 256));-- 插入测试标签
INSERT INTO tags (tag_name)
VALUES ('科技'), ('生活'), ('搞笑');-- 插入测试内容（用户1发布图文和视频）
INSERT INTO contents (user_id, content_type, title, text_content)
VALUES
(1, 'text', '测试图文内容', '这是一段测试文本'),
(1, 'video', '测试视频内容', NULL);-- 插入多媒体文件（关联内容ID=1和2）
INSERT INTO media_files (content_id, file_type, file_url)
VALUES
(1, 'image', 'http://example.com/image1.jpg'),
(2, 'video', 'http://example.com/video1.mp4');-- 插入内容-标签关联（内容1关联科技和生活标签）
INSERT INTO content_tags (content_id, tag_id)
VALUES
(1, 1), (1, 2);

二、用户表测试（users）

1. 新增用户

-- ----------------------------
-- 二、用户表测试（users）
-- ----------------------------# 1. 新增用户
-- 正常新增（唯一邮箱/合法格式）
INSERT INTO users (username, email, password_hash)
VALUES ('test_user3', 'user3@example.com', SHA2('password3', 256));-- 异常测试：重复用户名
INSERT INTO users (username, email, password_hash)
VALUES ('test_user1', 'user4@example.com', SHA2('password4', 256)); -- 应报错（UNIQUE约束）-- 异常测试：非法邮箱格式
INSERT INTO users (username, email, password_hash)
VALUES ('test_user4', 'invalid-email', SHA2('password4', 256)); -- 应报错（CHECK约束）

 

2. 更新用户信息

# 2. 更新用户信息
-- 修改头像和简介
UPDATE users
SET avatar_url='http://example.com/new_avatar.jpg', bio='新简介'
WHERE user_id=1;-- 验证更新结果
SELECT username, avatar_url, bio FROM users WHERE user_id=1;

3. 删除用户

# 3. 删除用户
-- 删除用户（级联影响：内容、评论、互动、关注关系等）
DELETE FROM users WHERE user_id=3;-- 验证关联数据是否被级联删除（如contents中user_id=3的记录应被删除）
SELECT * FROM contents WHERE user_id=3; -- 应无结果

三、内容表测试（contents）

1. 发布内容

-- ----------------------------
-- 三、内容表测试（contents）
-- ----------------------------
# 1. 发布内容
-- 发布短视频（无文本内容）
INSERT INTO contents (user_id, content_type, title)
VALUES (2, 'video', '测试短视频');-- 验证内容类型和字段完整性
SELECT content_type, text_content FROM contents WHERE content_id=3; -- 应显示video和NULL

2. 更新内容

# 2. 更新内容
-- 修改标题和浏览数
UPDATE contents 
SET title='更新后的标题', view_count=200
WHERE content_id=1;-- 验证更新时间是否自动更新（updated_at字段）
SELECT title, view_count, updated_at, now() FROM contents WHERE content_id=1;

3. 删除内容（级联验证）

# 3. 删除内容（级联验证）
SELECT * FROM media_files WHERE content_id=1;
-- ----------------------------
SELECT * FROM content_tags WHERE content_id=1;
-- 删除内容ID=1
DELETE FROM contents WHERE content_id=1;-- 验证关联数据是否被级联删除：
-- 1. media_files中content_id=1的记录应被删除
SELECT * FROM media_files WHERE content_id=1; -- 应无结果
-- 2. content_tags中content_id=1的记录应被删除
SELECT * FROM content_tags WHERE content_id=1; -- 应无结果

四、评论系统测试（comments）

1. 发表评论

-- ----------------------------
-- 四、评论系统测试（comments）
-- ----------------------------
# 1. 发表评论
-- 发表一级评论（父评论ID为空）
INSERT INTO comments (content_id, user_id, comment_text)
VALUES (2, 2, '这是一条评论');-- 发表回复评论（父评论ID=1）
INSERT INTO comments (content_id, user_id, parent_comment_id, comment_text)
VALUES (2, 1, 1, '这是一条回复');-- 验证评论数是否自动增加（触发器生效）
SELECT comment_count FROM contents WHERE content_id=2; -- 应显示2

2. 逻辑删除评论

# 2. 逻辑删除评论
-- 将评论ID=1标记为删除（is_deleted=true）
UPDATE comments
SET is_deleted=TRUE
WHERE comment_id=1;-- 验证评论数是否减少（触发器trg_contents_after_comment_logical_delete生效）
SELECT comment_count FROM contents WHERE content_id=2; -- 应显示1（原2条，逻辑删除1条）

3. 物理删除评论

# 3. 物理删除评论
-- 删除评论ID=2
DELETE FROM comments WHERE comment_id=2;-- 验证评论数是否进一步减少
SELECT comment_count FROM contents WHERE content_id=2; -- 应显示0

五、互动系统测试（interactions）

1. 点赞内容

-- ----------------------------
-- 五、互动系统测试（interactions）
-- ----------------------------
# 1. 点赞内容
-- 用户2点赞内容ID=2
INSERT INTO interactions (user_id, content_id, interaction_type)
VALUES (2, 2, 'like');-- 验证点赞数是否增加（触发器生效）
SELECT like_count FROM contents WHERE content_id=2; -- 应显示1

2. 取消点赞（删除互动记录）

# 2. 取消点赞（删除互动记录）
-- 删除用户2对内容ID=2的点赞记录
DELETE FROM interactions
WHERE user_id=2 AND content_id=2 AND interaction_type='like';-- 验证点赞数是否减少
SELECT like_count FROM contents WHERE content_id=2; -- 应显示0

3. 收藏内容（唯一性约束测试）

# 3. 收藏内容（唯一性约束测试）
-- 用户1收藏内容ID=2（正常插入）
INSERT INTO interactions (user_id, content_id, interaction_type)
VALUES (1, 2, 'collect');-- 重复收藏（应报错，唯一性约束）
INSERT INTO interactions (user_id, content_id, interaction_type)
VALUES (1, 2, 'collect'); -- 应失败

六、关注系统测试（follows）

1. 关注用户

-- ----------------------------
-- 六、关注系统测试（follows）
-- ----------------------------
# 1. 关注用户
-- 用户1关注用户2
INSERT INTO follows (follower_id, followed_id)
VALUES (1, 2);-- 验证关注关系是否存在
SELECT * FROM follows WHERE follower_id=1 AND followed_id=2; -- 应返回记录

2. 取消关注

# 2. 取消关注
-- 用户1取消关注用户2
DELETE FROM follows WHERE follower_id=1 AND followed_id=2;-- 验证关注关系是否删除
SELECT * FROM follows WHERE follower_id=1 AND followed_id=2; -- 应无结果

3. 统计粉丝数和关注数（通过存储过程）

# 3. 统计粉丝数和关注数（通过存储过程）
-- 调用get_user_info存储过程
CALL get_user_info('test_user1'); -- 结果应包含follower_count（0）和following_count（0）

七、标签系统测试（tags & content_tags）

1. 新增标签

-- ----------------------------
-- 七、标签系统测试（tags & content_tags）
-- ----------------------------
# 1. 新增标签
-- 新增标签“旅行”
INSERT INTO tags (tag_name)
VALUES ('旅行');-- 验证唯一性（重复插入应报错）
INSERT INTO tags (tag_name) VALUES ('旅行'); -- 应失败

2. 关联内容和标签

# 2. 关联内容和标签
-- 为内容ID=2添加“旅行”标签（tag_id=4）
INSERT INTO content_tags (content_id, tag_id)
VALUES (2, 4);-- 查询内容ID=2的标签
SELECT t.tag_name FROM tags t
JOIN content_tags ct ON t.tag_id=ct.tag_id
WHERE ct.content_id=2; -- 应返回“旅行”

3. 删除标签（级联删除关联关系）

# 3. 删除标签（级联删除关联关系）
-- 删除标签“旅行”（tag_id=4）
DELETE FROM tags WHERE tag_id=4;-- 验证content_tags中关联记录是否删除
SELECT * FROM content_tags WHERE tag_id=4; -- 应无结果

八、存储过程与函数测试

-- ----------------------------
-- 八、存储过程与函数测试
-- ----------------------------
# 1. 函数测试
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章', '这是一篇测试文章。');
select * from users where username='test_user1';
select * from contents where user_id=(select user_id from users where username='test_user1');
-- 测试user_content_count函数（用户1应有2条内容）
SELECT user_content_count('test_user1'); -- 应返回2-- 测试content_heat函数
SELECT content_heat(2); -- 应返回1

# 2. 存储过程测试（分页查询带标签的内容）
-- 插入测试数据（内容ID=2关联标签“科技”）
INSERT INTO content_tags (content_id, tag_id) VALUES (2, 1);-- 调用get_paginated_contents存储过程，查询标签为“科技”的内容
CALL get_paginated_contents('科技', NULL, NULL,1, 10, NULL, @total_count);

insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章2生活', '这是一篇测试文章2。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章3生活', '这是一篇测试文章3。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章4生活', '这是一篇测试文章4。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章5生活', '这是一篇测试文章5。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章6生活', '这是一篇测试文章6。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章7生活', '这是一篇测试文章7。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章8生活', '这是一篇测试文章8。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章9生活', '这是一篇测试文章9。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章10生活', '这是一篇测试文章10。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章11', '这是一篇生活测试文章11。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章12', '这是一篇生活测试文章12。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章13', '这是一篇生活测试文章13。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '测试文章14', '这是一篇生活测试文章14。');-- 调用get_paginated_contents存储过程，查询带关键字的内容
CALL get_paginated_contents(NULL, NULL, '生活',1, 10, NULL, @total_count);
-- ----------------------------
SELECT @total_count; -- 应返回13

-- 调用get_paginated_contents存储过程，其他测试
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '快乐', '这是快乐一篇生活测试文章15。');
insert into contents (user_id, content_type, title, text_content) values (1, 'text', '快乐', '豆包，混元，………………。');CALL get_paginated_contents(NULL, NULL, '快乐',1, 10, NULL, @total_count);
-- ----------------------------
CALL get_paginated_contents(NULL, NULL, '豆包',1, 10, NULL, @total_count);
-- ----------------------------
CALL get_paginated_contents(NULL, 'video', NULL,1, 10, NULL, @total_count);

# 调用get_user_info
CALL get_user_info('test_user1');CALL get_user_info('test_user2');

九、触发器完整性验证

1. 验证互动触发器

-- ----------------------------
-- 九、触发器完整性验证
-- ----------------------------
# 1. 验证互动触发器
-- 插入点赞记录，触发like_count+1
INSERT INTO interactions (user_id, content_id, interaction_type) VALUES (1, 2, 'like');
SELECT like_count FROM contents WHERE content_id=2; -- 应显示1-- 删除点赞记录，触发like_count-1
DELETE FROM interactions WHERE content_id=2 AND interaction_type='like';
SELECT like_count FROM contents WHERE content_id=2; -- 应显示0

2. 验证评论逻辑删除触发器

# 2. 验证评论逻辑删除触发器
-- 重新插入评论并逻辑删除
INSERT INTO comments (content_id, user_id, comment_text) VALUES (2, 1, '测试评论');
UPDATE comments SET is_deleted=TRUE WHERE comment_id=3;
SELECT comment_count FROM contents WHERE content_id=2; -- 应显示0（逻辑删除后计数减少）

十、异常场景测试

1. 外键约束失败

-- ------------------
-- 十、异常场景测试
-- ------------------
# 1. 外键约束失败
-- 插入内容时指定不存在的user_id=99
INSERT INTO contents (user_id, content_type, title) VALUES (99, 'text', '非法用户'); -- 应报错（外键约束）# 2. 唯一性约束冲突
-- 插入重复的邮箱
INSERT INTO users (username, email, password_hash) VALUES ('test_user4', 'user1@example.com', SHA2('pwd', 256)); -- 应报错（email唯一）

测试点 2：数据表的并发操作

'''
# 数据库并发压力测试系统## 项目概述
本系统是针对内容社区型数据库设计的并发压力测试工具，可模拟高并发场景下的用户行为，验证数据库的事务处理能力、数据一致性和性能表现。系统基于Python开发，支持12种典型社交场景的并发测试。## 功能特性
- 支持8种核心社交操作测试
- 模拟20-50并发线程压力
- 提供自动化数据一致性验证
- 支持MySQL事务隔离级别测试
- 包含异常场景测试（级联删除、外键约束等）## 技术栈
- 语言: Python 3.12.7
- 数据库: MySQL 8.0.31
- 并发库: `concurrent.futures`
- ORM: `PyMySQL`
- 日志: `logging`模块## 测试场景
| 测试用例名称                  | 并发操作数 | 验证要点                     |
|----------------------------|----------|----------------------------|
| `t_concurrent_nested_comments` | 50+20    | 嵌套评论创建/删除的父子关系一致性    |
| `t_concurrent_hot_content`    | 500      | 热点内容的高强度混合操作处理能力     |
| `t_concurrent_user_deletion` | 10+20    | 用户级联删除与关联数据原子性       |
| `t_concurrent_tag_creation`  | 20       | 唯一约束下的标签创建冲突处理       |## 使用说明
1. 数据库配置
```python
# 修改DB_CONFIG配置
DB_CONFIG = {'host': 'localhost',  # 数据库地址'user': 'root',       # 数据库账号'password': '1234',   # 数据库密码'database': 'contentcommunity'  # 测试数据库名称
}
'''import sys
import threading
import time
import random
import pymysql
from pymysql import cursors
import logging
from concurrent.futures import ThreadPoolExecutor
import hashlib# 在文件开头修改日志配置
logging.basicConfig(level=logging.DEBUG,  # 改为DEBUG级别format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)# 数据库配置
DB_CONFIG = {'host': 'localhost','user': 'root','password': '1234','database': 'contentcommunity','charset': 'utf8mb4','cursorclass': cursors.DictCursor,'autocommit': False
}# 测试配置
t_CONFIG = {'num_threads': 20,'t_duration': 300,'user_prefix': 'testuser_'
}class Colors:GREEN = '\033[92m'RED = '\033[91m'YELLOW = '\033[93m'BLUE = '\033[94m'END = '\033[0m'BOLD = '\033[1m'def get_connection():"""获取数据库连接"""return pymysql.connect(**DB_CONFIG)def setup_t_data():"""初始化测试数据"""conn = get_connection()try:with conn.cursor() as cursor:# 创建测试用户for i in range(10):username = f"{t_CONFIG['user_prefix']}{i}"email = f"{username}@test.com"password_hash = hashlib.sha256(f"password{i}".encode()).hexdigest()cursor.execute("""INSERT IGNORE INTO users (username, email, password_hash, avatar_url)VALUES (%s, %s, %s, 'default_avatar.jpg')""", (username, email, password_hash))# 创建测试标签tags = ['科技', '美食', '旅行', '编程', '摄影']for tag in tags:cursor.execute("INSERT IGNORE INTO tags (tag_name) VALUES (%s)", (tag,))conn.commit()finally:conn.close()def get_random_user():"""获取随机用户"""conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT user_id FROM users ORDER BY RAND() LIMIT 1")return cursor.fetchone()['user_id']finally:conn.close()def create_t_content(user_id):"""创建测试内容"""conn = get_connection()try:with conn.cursor() as cursor:content_type = random.choice(['text', 'image', 'video'])cursor.execute("""INSERT INTO contents (user_id, content_type, title, text_content)VALUES (%s, %s, %s, %s)""", (user_id, content_type, f"测试标题_{time.time()}", "测试内容"))content_id = cursor.lastrowidif content_type != 'text':cursor.execute("""INSERT INTO media_files (content_id, file_type, file_url)VALUES (%s, %s, %s)""", (content_id, content_type, f"https://example.com/{content_id}"))conn.commit()return content_idexcept Exception as e:conn.rollback()raise efinally:conn.close()def create_comment(content_id, user_id):"""创建评论并返回评论ID"""conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("""INSERT INTO comments (content_id, user_id, comment_text)VALUES (%s, %s, '父评论')""", (content_id, user_id))conn.commit()return cursor.lastrowidexcept pymysql.Error as e:conn.rollback()raise efinally:conn.close()def t_concurrent_nested_comments():"""测试嵌套评论的并发创建"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始嵌套评论测试，内容ID：{content_id}{Colors.END}")# 先创建父评论parent_comment_id = create_comment(content_id, get_random_user())def nested_comment_task():conn = get_connection()try:user_id = get_random_user()with conn.cursor() as cursor:cursor.execute("""INSERT INTO comments (content_id, user_id, parent_comment_id, comment_text)VALUES (%s, %s, %s, '子评论')""", (content_id, user_id, parent_comment_id))conn.commit()except pymysql.IntegrityError as e:# 处理父评论可能被删除的情况if e.args[0] == 1452:  # 外键约束错误conn.rollback()logger.debug(f"父评论已被删除，跳过子评论创建")else:raisefinally:conn.close()# 执行并发测试（包含可能的父评论删除操作）with ThreadPoolExecutor(max_workers=20) as executor:# 创建子评论任务comment_futures = [executor.submit(nested_comment_task) for _ in range(50)]# 添加随机删除父评论的任务def delete_parent_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("""DELETE FROM comments WHERE comment_id = %s""", (parent_comment_id,))conn.commit()finally:conn.close()# 随机提交删除任务if random.random() < 0.3:  # 30%概率触发删除executor.submit(delete_parent_task)# 验证结果conn = get_connection()try:with conn.cursor() as cursor:# 检查父评论是否存在cursor.execute("SELECT 1 FROM comments WHERE comment_id = %s", (parent_comment_id,))parent_exists = cursor.fetchone() is not None# 获取实际评论数cursor.execute("""SELECT COUNT(*) as total,COUNT(parent_comment_id) as child_countFROM comments WHERE content_id = %s""", (content_id,))counts = cursor.fetchone()# 验证计数器一致性cursor.execute("SELECT comment_count FROM contents WHERE content_id = %s", (content_id,))db_count = cursor.fetchone()['comment_count']# 计算预期值expected_total = counts['child_count'] + (1 if parent_exists else 0)valid = (db_count == expected_total andcounts['total'] == expected_total)return validfinally:conn.close()def t_concurrent_likes():"""测试并发点赞操作"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始点赞测试，内容ID：{content_id}{Colors.END}")def like_task():conn = get_connection()try:user_id = get_random_user()with conn.cursor() as cursor:# 检查是否已点赞cursor.execute("""SELECT 1 FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'like'""", (user_id, content_id))if cursor.fetchone():cursor.execute("""DELETE FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'like'""", (user_id, content_id))else:cursor.execute("""INSERT INTO interactions (user_id, content_id, interaction_type)VALUES (%s, %s, 'like')""", (user_id, content_id))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(like_task) for _ in range(100)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT like_count FROM contents WHERE content_id = %s", (content_id,))actual_likes = cursor.fetchone()['like_count']cursor.execute("SELECT COUNT(*) FROM interactions WHERE content_id = %s", (content_id,))interaction_count = cursor.fetchone()['COUNT(*)']return actual_likes == interaction_countfinally:conn.close()def t_concurrent_comments():"""测试并发评论操作"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始评论测试，内容ID：{content_id}{Colors.END}")def comment_task():conn = get_connection()try:user_id = get_random_user()with conn.cursor() as cursor:cursor.execute("""INSERT INTO comments (content_id, user_id, comment_text)VALUES (%s, %s, %s)""", (content_id, user_id, "测试评论"))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(comment_task) for _ in range(50)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT comment_count FROM contents WHERE content_id = %s", (content_id,))actual_comments = cursor.fetchone()['comment_count']cursor.execute("SELECT COUNT(*) FROM comments WHERE content_id = %s", (content_id,))comment_count = cursor.fetchone()['COUNT(*)']return actual_comments == comment_countfinally:conn.close()def t_concurrent_follows():"""测试并发关注操作"""user1 = get_random_user()user2 = get_random_user()logger.info(f"{Colors.BLUE}▶ 开始关注测试，用户：{user1} -> {user2}{Colors.END}")def follow_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("""SELECT 1 FROM follows WHERE follower_id = %s AND followed_id = %s""", (user1, user2))if cursor.fetchone():cursor.execute("""DELETE FROM follows WHERE follower_id = %s AND followed_id = %s""", (user1, user2))else:cursor.execute("""INSERT INTO follows (follower_id, followed_id)VALUES (%s, %s)""", (user1, user2))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(follow_task) for _ in range(50)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT COUNT(*) FROM follows WHERE follower_id = %s AND followed_id = %s", (user1, user2))actual_follows = cursor.fetchone()['COUNT(*)']return actual_follows in (0, 1)finally:conn.close()def t_concurrent_content_creation():"""测试并发内容创建"""logger.info(f"{Colors.BLUE}▶ 开始内容创建测试{Colors.END}")results = []def create_task():try:user_id = get_random_user()content_id = create_t_content(user_id)return content_id is not Noneexcept Exception as e:return False# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(create_task) for _ in range(50)]results = [f.result() for f in futures]return all(results)def t_concurrent_tagging():"""测试并发标签关联"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始标签测试，内容ID：{content_id}{Colors.END}")def tag_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT tag_id FROM tags ORDER BY RAND() LIMIT 1")tag_id = cursor.fetchone()['tag_id']cursor.execute("""INSERT IGNORE INTO content_tags (content_id, tag_id)VALUES (%s, %s)""", (content_id, tag_id))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(tag_task) for _ in range(20)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT COUNT(DISTINCT tag_id) FROM content_tags WHERE content_id = %s", (content_id,))unique_tags = cursor.fetchone()['COUNT(DISTINCT tag_id)']return unique_tags > 0finally:conn.close()def t_concurrent_updates():"""测试并发更新操作"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始更新测试，内容ID：{content_id}{Colors.END}")def update_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("""UPDATE contents SET view_count = view_count + 1 WHERE content_id = %s""", (content_id,))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(update_task) for _ in range(100)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT view_count FROM contents WHERE content_id = %s", (content_id,))view_count = cursor.fetchone()['view_count']return view_count == 100finally:conn.close()def t_concurrent_collects():"""测试并发收藏操作"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始收藏测试，内容ID：{content_id}{Colors.END}")def collect_task():conn = get_connection()try:user_id = get_random_user()with conn.cursor() as cursor:cursor.execute("""SELECT 1 FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'collect'""", (user_id, content_id))if cursor.fetchone():cursor.execute("""DELETE FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'collect'""", (user_id, content_id))else:cursor.execute("""INSERT INTO interactions (user_id, content_id, interaction_type)VALUES (%s, %s, 'collect')""", (user_id, content_id))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(collect_task) for _ in range(100)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT collect_count FROM contents WHERE content_id = %s", (content_id,))actual_collects = cursor.fetchone()['collect_count']cursor.execute("SELECT COUNT(*) FROM interactions WHERE content_id = %s", (content_id,))interaction_count = cursor.fetchone()['COUNT(*)']return actual_collects == interaction_countfinally:conn.close()# 在原有代码基础上新增以下测试函数
def t_concurrent_user_updates():"""测试并发用户资料更新"""user_id = get_random_user()logger.info(f"{Colors.BLUE}▶ 开始用户资料更新测试，用户ID：{user_id}{Colors.END}")def update_task():conn = get_connection()try:with conn.cursor() as cursor:# 模拟更新头像和简介new_avatar = f"avatar_{random.randint(1,100)}.jpg"new_bio = f"Bio updated at {time.time()}"cursor.execute("""UPDATE users SET avatar_url = %s, bio = %s WHERE user_id = %s""", (new_avatar, new_bio, user_id))conn.commit()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(update_task) for _ in range(50)]# 验证最终一致性conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT avatar_url, bio FROM users WHERE user_id = %s", (user_id,))result = cursor.fetchone()return result['avatar_url'].startswith('avatar_') and len(result['bio']) > 0finally:conn.close()def t_concurrent_content_deletion():"""测试并发内容删除（测试级联删除触发器）"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始内容删除测试，内容ID：{content_id}{Colors.END}")def delete_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("DELETE FROM contents WHERE content_id = %s", (content_id,))conn.commit()except pymysql.IntegrityError as e:conn.rollback()finally:conn.close()# 并发删除和创建相关数据with ThreadPoolExecutor(max_workers=20) as executor:# 启动删除线程delete_futures = [executor.submit(delete_task) for _ in range(10)]# 同时创建相关数据测试并发冲突def create_related_data():conn = get_connection()try:user_id = get_random_user()# 测试创建评论conn.cursor().execute("""INSERT INTO comments (content_id, user_id, comment_text)VALUES (%s, %s, '测试评论')""", (content_id, user_id))# 测试添加标签conn.cursor().execute("""INSERT INTO content_tags (content_id, tag_id)SELECT %s, tag_id FROM tags ORDER BY RAND() LIMIT 1""", (content_id,))conn.commit()except pymysql.Error:conn.rollback()finally:conn.close()data_futures = [executor.submit(create_related_data) for _ in range(20)]# 验证最终删除结果conn = get_connection()try:with conn.cursor() as cursor:# 检查内容是否存在cursor.execute("SELECT 1 FROM contents WHERE content_id = %s", (content_id,))content_exists = cursor.fetchone() is not None# 检查关联数据cursor.execute("SELECT 1 FROM comments WHERE content_id = %s", (content_id,))comments_exist = cursor.fetchone() is not Nonecursor.execute("SELECT 1 FROM content_tags WHERE content_id = %s", (content_id,))tags_exist = cursor.fetchone() is not Nonereturn not (content_exists or comments_exist or tags_exist)finally:conn.close()def t_concurrent_tag_creation():"""测试并发标签创建（测试唯一约束）"""tag_name = f"t_tag_{int(time.time())}"logger.info(f"{Colors.BLUE}▶ 开始标签创建测试，标签名：{tag_name}{Colors.END}")def create_tag_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("""INSERT INTO tags (tag_name) VALUES (%s)""", (tag_name,))conn.commit()except pymysql.IntegrityError:conn.rollback()finally:conn.close()# 执行并发测试with ThreadPoolExecutor(max_workers=20) as executor:futures = [executor.submit(create_tag_task) for _ in range(20)]# 验证结果conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("SELECT COUNT(*) FROM tags WHERE tag_name = %s", (tag_name,))count = cursor.fetchone()['COUNT(*)']return count == 1finally:conn.close()def t_concurrent_user_deletion():"""测试并发用户删除（测试外键约束和级联删除）"""user_id = get_random_user()content_id = create_t_content(user_id)logger.info(f"{Colors.BLUE}▶ 开始用户删除测试，用户ID：{user_id}{Colors.END}")def delete_task():conn = get_connection()try:with conn.cursor() as cursor:cursor.execute("DELETE FROM users WHERE user_id = %s", (user_id,))conn.commit()except pymysql.Error:conn.rollback()finally:conn.close()def create_data_task():conn = get_connection()try:# 尝试创建关联数据with conn.cursor() as cursor:# 创建关注关系cursor.execute("""INSERT INTO follows (follower_id, followed_id)VALUES (%s, %s)""", (user_id, get_random_user()))# 创建互动cursor.execute("""INSERT INTO interactions (user_id, content_id, interaction_type)VALUES (%s, %s, 'like')""", (user_id, content_id))conn.commit()except pymysql.Error:conn.rollback()finally:conn.close()# 并发删除和创建关联数据with ThreadPoolExecutor(max_workers=20) as executor:del_futures = [executor.submit(delete_task) for _ in range(10)]data_futures = [executor.submit(create_data_task) for _ in range(20)]# 验证删除结果conn = get_connection()try:with conn.cursor() as cursor:# 检查用户是否存在cursor.execute("SELECT 1 FROM users WHERE user_id = %s", (user_id,))user_exists = cursor.fetchone() is not None# 检查关联数据cursor.execute("SELECT 1 FROM contents WHERE user_id = %s", (user_id,))content_exists = cursor.fetchone() is not Nonecursor.execute("SELECT 1 FROM follows WHERE follower_id = %s", (user_id,))follows_exist = cursor.fetchone() is not Nonereturn not (user_exists or content_exists or follows_exist)finally:conn.close()def t_concurrent_hot_content():"""测试热门内容的极端并发"""content_id = create_t_content(get_random_user())logger.info(f"{Colors.BLUE}▶ 开始热门内容测试，内容ID：{content_id}{Colors.END}")def hot_content_task():conn = get_connection()try:with conn.cursor() as cursor:# 随机选择操作类型operation = random.choice(['like', 'unlike','comment','collect', 'uncollect'])user_id = get_random_user()if operation in ['like', 'collect']:# 插入互动cursor.execute("""INSERT IGNORE INTO interactions (user_id, content_id, interaction_type)VALUES (%s, %s, %s)""", (user_id, content_id, operation))elif operation in ['unlike', 'uncollect']:# 删除互动cursor.execute("""DELETE FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = %s""", (user_id, content_id, operation.replace('un','')))elif operation == 'comment':# 创建评论cursor.execute("""INSERT INTO comments (content_id, user_id, comment_text)VALUES (%s, %s, '热门内容评论')""", (content_id, user_id))conn.commit()finally:conn.close()# 执行高强度并发with ThreadPoolExecutor(max_workers=50) as executor:futures = [executor.submit(hot_content_task) for _ in range(500)]# 验证最终一致性conn = get_connection()try:with conn.cursor() as cursor:# 获取内容统计cursor.execute("""SELECT like_count, collect_count,comment_count FROM contents WHERE content_id = %s""", (content_id,))content_stats = cursor.fetchone()# 获取实际互动数据cursor.execute("""SELECT SUM(IF(interaction_type='like',1,0)) as actual_likes,SUM(IF(interaction_type='collect',1,0)) as actual_collects,(SELECT COUNT(*) FROM comments WHERE content_id=%s) as actual_commentsFROM interactions WHERE content_id=%s""", (content_id, content_id))actual_stats = cursor.fetchone()# 验证计数器一致性return (content_stats['like_count'] == actual_stats['actual_likes'] andcontent_stats['collect_count'] == actual_stats['actual_collects'] andcontent_stats['comment_count'] == actual_stats['actual_comments'])finally:conn.close()def run_tests():"""运行所有测试用例"""tests = [t_concurrent_likes,t_concurrent_collects,t_concurrent_comments,t_concurrent_follows,t_concurrent_content_creation,t_concurrent_tagging,t_concurrent_updates,t_concurrent_user_updates,     # 新增t_concurrent_content_deletion,  # 新增t_concurrent_nested_comments,   # 新增 t_concurrent_tag_creation,      # 新增t_concurrent_user_deletion,     # 新增t_concurrent_hot_content        # 新增]passed = 0failed = 0results = []logger.info(f"\n{Colors.BOLD}{Colors.BLUE}=== 开始并发测试 ({len(tests)} 个测试用例) ==={Colors.END}\n")for test in tests:try:start = time.time()result = test()duration = time.time() - startif result:status = f"{Colors.GREEN}✓ 通过{Colors.END}"passed += 1else:status = f"{Colors.RED}✗ 失败{Colors.END}"failed += 1logger.info(f"{Colors.BOLD}{test.__name__}: {status}")logger.info(f"耗时: {duration:.2f}s{Colors.END}")results.append((test.__name__, result, duration))except Exception as e:logger.error(f"{Colors.RED}测试异常: {str(e)}{Colors.END}")failed += 1results.append((test.__name__, False, 0))# 打印总结报告logger.info(f"\n{Colors.BOLD}{Colors.BLUE}=== 测试总结 ==={Colors.END}")logger.info(f"总用例数: {len(tests)}")logger.info(f"{Colors.GREEN}通过用例: {passed}{Colors.END}")logger.info(f"{Colors.RED}失败用例: {failed}{Colors.END}")logger.info(f"通过率: {(passed/len(tests))*100:.1f}%")return passed == len(tests)if __name__ == "__main__":try:# 初始化测试数据setup_t_data()# 运行测试套件success = run_tests()# 返回适当的退出码sys.exit(0 if success else 1)except Exception as e:logger.error(f"{Colors.RED}测试执行失败: {str(e)}{Colors.END}")sys.exit(1)

测试点 3：边界条件测试（表满 / 空状态）

测试执行说明：
1. 完整性和隔离性：每个测试用例使用独立事务，确保测试环境隔离
2. 异常处理：通过IF EXISTS子句处理可能不存在的对象
3. 结果验证：
   - 使用ROW_COUNT()验证影响行数
   - 检查外键约束是否阻止非法操作
   - 验证触发器在空表场景下的稳定性
4. 覆盖范围：
   - 所有用户提供的表结构
   - 级联删除约束
   - 触发器空触发场景
   - 基础DDL/DML操作边界情况

-- 数据库删除与清空测试套件
-- 测试用例设计说明：
-- 1. 测试所有表的空表删除操作
-- 2. 测试空表记录删除操作
-- 3. 验证外键约束与触发器在空表场景下的表现-- -----------------------------------
-- 测试用例1：数据库级删除测试
-- -----------------------------------
-- 场景：空数据库删除验证
DROP DATABASE IF EXISTS contentcommunity;
SELECT '空数据库删除测试完成' AS test_result;

-- -----------------------------------
-- 测试用例2：表级删除测试
-- -----------------------------------
-- 初始化环境
CREATE DATABASE IF NOT EXISTS contentcommunity;
-- -----------------------
-- 重建表结构（保持空表状态）
-- 这里需要执行完整建表SQL
# 运行MySQL脚本【此处省略…………】
-- -----------------------
USE contentcommunity;-- 场景：空表删除操作验证
-- 注意按照依赖关系逆序删除
DROP TABLE IF EXISTS follows;
DROP TABLE IF EXISTS interactions;
DROP TABLE IF EXISTS comments;
DROP TABLE IF EXISTS content_tags;
DROP TABLE IF EXISTS tags;
DROP TABLE IF EXISTS media_files;
DROP TABLE IF EXISTS contents;
DROP TABLE IF EXISTS users;
SELECT '空表删除操作验证完成' AS test_result;

-- -----------------------------------
-- 测试用例3：空表记录删除测试
-- -----------------------------------
-- 重建表结构（保持空表状态）
-- 这里需要执行完整建表SQL
# 运行MySQL脚本【此处省略…………】-- 场景：验证空表数据删除操作
START TRANSACTION;
-- 测试数据删除语句
DELETE FROM follows;
DELETE FROM interactions;
DELETE FROM comments;
DELETE FROM content_tags;
DELETE FROM tags;
DELETE FROM media_files;
DELETE FROM contents;
DELETE FROM users;
-- 验证影响行数
SELECT ROW_COUNT() AS affected_rows;
COMMIT;
-- -------------------------------------
SELECT '空表记录删除操作验证完成' AS test_result;

-- -----------------------------------
-- 测试用例4：空表约束验证
-- -----------------------------------
-- 重建表结构（保持空表状态）
-- 这里需要执行完整建表SQL
# 运行MySQL脚本【此处省略…………】-- 验证外键约束在空表场景下的表现
-- 必须按照实际表结构提供完整字段值
INSERT INTO users (username,email,password_hash
) VALUES ('test_user','test@example.com','a665a45920422f9d417e4867efdc4fb8a04a1f3fff1fa07e998e86f7f7a27ae3' -- 123的SHA256哈希
);-- 验证级联删除（需补充关联表结构）
-- 创建关联内容测试数据
INSERT INTO contents (user_id,content_type,title
) VALUES ((SELECT user_id FROM users WHERE username = 'test_user'),'text','测试内容'
);-- 执行级联删除验证
DELETE FROM users WHERE username = 'test_user';-- 验证关联内容是否被级联删除
SELECT COUNT(*) AS remaining_contents
FROM contents
WHERE user_id = (SELECT user_id FROM users WHERE username = 'test_user');-- -----------------------------------------
SELECT '空表约束验证完成' AS test_result;

-- -----------------------------------------
-- 测试用例5：触发器空表测试
-- -----------------------------------------
-- 验证空表触发器的稳定性
-- 创建测试触发器（示例）
DELIMITER $$
CREATE TRIGGER trg_test
BEFORE DELETE ON users
FOR EACH ROW
BEGINDELETE FROM contents WHERE user_id = OLD.user_id;
END$$
DELIMITER ;-- 执行空表删除操作
DELETE FROM users; -- 验证触发器不会报错
DROP TRIGGER trg_test;
SELECT '空表触发器验证完成' AS test_result;

测试点4：针对存储过程的接口测试

1. 初始化设置：禁用外键检查，清空测试数据并重置自增 ID，然后启用外键检查。

   -- ----------------------------
   -- 内容社区系统存储过程测试脚本
   -- ------------------------------ 使用内容社区数据库
   USE contentcommunity;-- 禁用外键检查，以便于测试数据的插入和删除
   SET FOREIGN_KEY_CHECKS = 0;-- 清空测试数据（如果存在）
   DELETE FROM interactions;
   DELETE FROM comments;
   DELETE FROM content_tags;
   DELETE FROM media_files;
   DELETE FROM contents;
   DELETE FROM tags;
   DELETE FROM follows;
   DELETE FROM users;-- 重置自增ID
   ALTER TABLE users AUTO_INCREMENT = 1;
   ALTER TABLE contents AUTO_INCREMENT = 1;
   ALTER TABLE media_files AUTO_INCREMENT = 1;
   ALTER TABLE tags AUTO_INCREMENT = 1;
   ALTER TABLE comments AUTO_INCREMENT = 1;
   ALTER TABLE interactions AUTO_INCREMENT = 1;
   ALTER TABLE follows AUTO_INCREMENT = 1;-- 启用外键检查
   SET FOREIGN_KEY_CHECKS = 1;-- 开始测试
   SELECT '开始测试存储过程' AS '测试状态';

2. 测试数据准备：创建 3 个测试用户、7 个标签、5 个内容（包括文本、图片和视频）、6 条评论、13 条互动记录和 5 条关注关系。

   -- ----------------------------
   -- 测试数据准备
   -- ----------------------------
   INSERT INTO users (username, email, password_hash, avatar_url, bio, registration_date, last_login) VALUES
   ('test_user1', 'test1@example.com', SHA2('password1', 256), 'https://example.com/avatar1.jpg', '测试用户1', NOW() - INTERVAL 30 DAY, NOW() - INTERVAL 1 DAY),
   ('test_user2', 'test2@example.com', SHA2('password2', 256), 'https://example.com/avatar2.jpg', '测试用户2', NOW() - INTERVAL 20 DAY, NOW() - INTERVAL 2 DAY),
   ('test_user3', 'test3@example.com', SHA2('password3', 256), 'https://example.com/avatar3.jpg', '测试用户3', NOW() - INTERVAL 10 DAY, NOW() - INTERVAL 3 DAY);INSERT INTO tags (tag_name) VALUES
   ('技术'), ('生活'), ('美食'), ('旅行'), ('摄影'), ('编程'), ('AI');INSERT INTO contents (user_id, content_type, title, text_content, created_at, view_count, like_count, comment_count, collect_count) VALUES
   (1, 'text', 'Python编程技巧分享', '这是一篇关于Python编程的技巧分享文章...', NOW() - INTERVAL 5 DAY, 100, 20, 5, 10),
   (1, 'image', '美丽的自然风光', '', NOW() - INTERVAL 4 DAY, 150, 30, 7, 15),
   (2, 'video', '美食制作教程', '', NOW() - INTERVAL 3 DAY, 200, 40, 10, 20),
   (2, 'text', '旅行见闻', '分享我的旅行经历和见闻...', NOW() - INTERVAL 2 DAY, 80, 15, 3, 8),
   (3, 'image', '城市建筑摄影', '', NOW() - INTERVAL 1 DAY, 120, 25, 6, 12);INSERT INTO media_files (content_id, file_type, file_url) VALUES
   (2, 'image', 'https://example.com/image1.jpg'),
   (3, 'video', 'https://example.com/video1.mp4'),
   (5, 'image', 'https://example.com/image2.jpg');INSERT INTO content_tags (content_id, tag_id) VALUES
   (1, 1), (1, 6),
   (2, 4), (2, 5),
   (3, 3),
   (4, 4),
   (5, 5);INSERT INTO comments (content_id, user_id, parent_comment_id, comment_text, created_at, like_count) VALUES
   (1, 2, NULL, '非常实用的技巧！', NOW() - INTERVAL 4 DAY, 5),
   (1, 3, NULL, '感谢分享！', NOW() - INTERVAL 3 DAY, 3),
   (2, 1, NULL, '风景真美！', NOW() - INTERVAL 3 DAY, 7),
   (2, 3, 3, '请问这是在哪里拍摄的？', NOW() - INTERVAL 2 DAY, 2),
   (3, 1, NULL, '看起来很好吃！', NOW() - INTERVAL 2 DAY, 4),
   (3, 2, NULL, '请问用的什么食材？', NOW() - INTERVAL 1 DAY, 3);INSERT INTO interactions (user_id, content_id, interaction_type, interacted_at) VALUES
   (2, 1, 'like', NOW() - INTERVAL 4 DAY),
   (3, 1, 'like', NOW() - INTERVAL 3 DAY),
   (1, 2, 'like', NOW() - INTERVAL 3 DAY),
   (3, 2, 'like', NOW() - INTERVAL 2 DAY),
   (1, 3, 'like', NOW() - INTERVAL 2 DAY),
   (2, 3, 'like', NOW() - INTERVAL 1 DAY),
   (3, 4, 'like', NOW() - INTERVAL 1 DAY),
   (1, 5, 'like', NOW() - INTERVAL 1 DAY),
   (2, 5, 'like', NOW() - INTERVAL 1 DAY),
   (2, 1, 'collect', NOW() - INTERVAL 3 DAY),
   (3, 2, 'collect', NOW() - INTERVAL 2 DAY),
   (1, 3, 'collect', NOW() - INTERVAL 2 DAY);INSERT INTO follows (follower_id, followed_id, created_at) VALUES
   (2, 1, NOW() - INTERVAL 5 DAY),
   (3, 1, NOW() - INTERVAL 4 DAY),
   (1, 2, NOW() - INTERVAL 3 DAY),
   (3, 2, NOW() - INTERVAL 2 DAY),
   (1, 3, NOW() - INTERVAL 1 DAY);

3. 存储过程测试：

   • get_paginated_contents 存储过程测试：
     • 无筛选条件的分页查询

       -- ----------------------------
       -- 测试 get_paginated_contents 存储过程
       -- ----------------------------
       SELECT '测试 get_paginated_contents 存储过程' AS '测试状态';-- 测试1：无筛选条件，第一页，每页2条记录
       SELECT '测试1：无筛选条件，第一页，每页2条记录' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents(NULL, NULL, NULL, 1, 2, NULL, @total_count);
       SELECT @total_count AS '总记录数';

       

     • 按标签、内容类型、关键词和用户名的单独筛选

       -- 测试2：按标签筛选（技术）
       SELECT '测试2：按标签筛选（技术）' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents('技术', NULL, NULL,1, 10, NULL, @total_count);
       SELECT @total_count AS '总记录数';-- 测试3：按内容类型筛选（text）
       SELECT '测试3：按内容类型筛选（text）' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents(NULL, 'text', NULL, 1, 10, NULL, @total_count);
       SELECT @total_count AS '总记录数';-- 测试4：按关键词筛选（Python）
       SELECT '测试4：按关键词筛选（Python）' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents(NULL, NULL, 'Python', 1, 10, NULL, @total_count);
       SELECT @total_count AS '总记录数';-- 测试5：按用户名筛选（test_user2）
       SELECT '测试5：按用户名筛选（test_user2）' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents(NULL, NULL, NULL, 1, 10, 'test_user2', @total_count);
       SELECT @total_count AS '总记录数';

       

     • 组合筛选

       -- 测试6：组合筛选（标签+内容类型+关键词）
       SELECT '测试6：组合筛选（标签+内容类型+关键词）' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents('技术,编程', 'text', '技巧',1, 10, NULL, @total_count);
       -- ---------------------------------------------------------------------------------------------
       SELECT @total_count AS '总记录数';

       

     • 不存在的筛选条件测试

       -- 测试7：不存在的筛选条件
       SELECT '测试7：不存在的筛选条件' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents('不存在的标签', NULL, NULL,1, 10, NULL, @total_count);
       -- ---------------------------------------------------------------------------------------------
       SELECT @total_count AS '总记录数';

       

     • 分页功能测试

       -- 测试8：分页测试，第二页，每页2条记录
       SELECT '测试8：分页测试，第二页，每页2条记录' AS '测试用例';
       SET @total_count = 0;
       CALL get_paginated_contents(NULL, NULL, NULL,2, 2, NULL, @total_count);
       -- ---------------------------------------------------------------------------------------------
       SELECT @total_count AS '总记录数';

       

   • get_user_info 存储过程测试：
     • 查询存在的用户

       -- ----------------------------
       -- 测试 get_user_info 存储过程
       -- ----------------------------
       SELECT '测试 get_user_info 存储过程' AS '测试状态';-- 测试1：查询存在的用户（test_user1）
       SELECT '测试1：查询存在的用户（test_user1）' AS '测试用例';
       -- ---------------------------------------------------------------------------------------------
       CALL get_user_info('test_user1');-- 测试2：查询存在的用户（test_user2）
       SELECT '测试2：查询存在的用户（test_user2）' AS '测试用例';
       -- ---------------------------------------------------------------------------------------------
       CALL get_user_info('test_user2');-- 测试3：查询存在的用户（test_user3）
       SELECT '测试3：查询存在的用户（test_user3）' AS '测试用例';
       -- ---------------------------------------------------------------------------------------------
       CALL get_user_info('test_user3');
       

       

     • 查询不存在的用户

       -- 测试4：查询不存在的用户
       SELECT '测试4：查询不存在的用户' AS '测试用例';
       -- ---------------------------------------------------------------------------------------------
       CALL get_user_info('不存在的用户');

       

4. 测试结果验证：显示测试完成信息。

-- ----------------------------
-- 测试结果验证
-- ----------------------------
SELECT '测试完成' AS '测试状态';

测试点5：结合业务逻辑做关联表的接口测试

import hashlibimport pymysql
import unittest
import random
import string# 数据库配置
DB_CONFIG = {'host': 'localhost','user': 'root','password': '1234','database': 'contentcommunity','charset': 'utf8mb4','autocommit': True
}class DatabaseTester(unittest.TestCase):@classmethoddef setUpClass(cls):cls.conn = pymysql.connect(**DB_CONFIG)cls.cursor = cls.conn.cursor(pymysql.cursors.DictCursor)print("数据库连接成功")@classmethoddef tearDownClass(cls):cls.cursor.close()cls.conn.close()print("数据库连接关闭")def setUp(self):# 测试前清理相关数据self.clean_up()print("测试环境准备完成")def tearDown(self):# 测试后清理相关数据self.clean_up()print("测试环境清理完成")def clean_up(self):# 清理顺序需考虑外键约束，从子表到父表self.cursor.execute("DELETE FROM follows")self.cursor.execute("DELETE FROM interactions")self.cursor.execute("DELETE FROM comments")self.cursor.execute("DELETE FROM content_tags")self.cursor.execute("DELETE FROM media_files")self.cursor.execute("DELETE FROM contents")self.cursor.execute("DELETE FROM tags")self.cursor.execute("DELETE FROM users WHERE username != 'admin'")def generate_random_string(self, length=10):letters = string.ascii_letters + string.digitsreturn ''.join(random.choice(letters) for _ in range(length))def test_user_registration_and_login(self):"""测试用户注册和登录功能"""# 生成随机测试用户username = "test_user_" + self.generate_random_string()email = username + "@example.com"password = "TestPassword123!"# 注册用户self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(username, email, hashlib.sha256(password.encode()).hexdigest()))# 验证注册成功self.cursor.execute("SELECT * FROM users WHERE username = %s", (username,))user = self.cursor.fetchone()self.assertIsNotNone(user)self.assertEqual(user['username'], username)self.assertEqual(user['email'], email)# 模拟登录self.cursor.execute("SELECT password_hash FROM users WHERE username = %s", (username,))result = self.cursor.fetchone()self.assertIsNotNone(result)self.assertEqual(result['password_hash'], hashlib.sha256(password.encode()).hexdigest())def test_content_creation_and_management(self):"""测试内容创建和管理功能"""# 创建测试用户test_username = "content_user_" + self.generate_random_string()test_email = test_username + "@example.com"test_password = "ContentPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(test_username, test_email, hashlib.sha256(test_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (test_username,))user_id = self.cursor.fetchone()['user_id']# 创建内容title = "测试文章标题"content = "这是测试文章内容"self.cursor.execute("INSERT INTO contents (user_id, content_type, title, text_content) VALUES (%s, %s, %s, %s)",(user_id, 'text', title, content))self.cursor.execute("SELECT LAST_INSERT_ID()")content_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 验证内容创建self.cursor.execute("SELECT * FROM contents WHERE content_id = %s", (content_id,))created_content = self.cursor.fetchone()self.assertIsNotNone(created_content)self.assertEqual(created_content['title'], title)self.assertEqual(created_content['text_content'], content)# 编辑内容new_title = "更新后的测试标题"new_content = "更新后的测试内容"self.cursor.execute("UPDATE contents SET title = %s, text_content = %s WHERE content_id = %s",(new_title, new_content, content_id))# 验证内容更新self.cursor.execute("SELECT * FROM contents WHERE content_id = %s", (content_id,))updated_content = self.cursor.fetchone()self.assertEqual(updated_content['title'], new_title)self.assertEqual(updated_content['text_content'], new_content)# 删除内容self.cursor.execute("DELETE FROM contents WHERE content_id = %s", (content_id,))# 验证内容删除self.cursor.execute("SELECT * FROM contents WHERE content_id = %s", (content_id,))deleted_content = self.cursor.fetchone()self.assertIsNone(deleted_content)def test_media_file_management(self):"""测试多媒体文件管理功能"""# 创建测试用户和内容test_username = "media_user_" + self.generate_random_string()test_email = test_username + "@example.com"test_password = "MediaPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(test_username, test_email, hashlib.sha256(test_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (test_username,))user_id = self.cursor.fetchone()['user_id']self.cursor.execute("INSERT INTO contents (user_id, content_type, title) VALUES (%s, %s, %s)",(user_id, 'image', "测试图片内容"))self.cursor.execute("SELECT LAST_INSERT_ID()")content_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 添加多媒体文件file_url = "https://example.com/test-image.jpg"self.cursor.execute("INSERT INTO media_files (content_id, file_type, file_url) VALUES (%s, %s, %s)",(content_id, 'image', file_url))# 验证文件添加self.cursor.execute("SELECT * FROM media_files WHERE content_id = %s", (content_id,))media_file = self.cursor.fetchone()self.assertIsNotNone(media_file)self.assertEqual(media_file['file_url'], file_url)# 删除多媒体文件（通过删除内容级联删除）self.cursor.execute("DELETE FROM contents WHERE content_id = %s", (content_id,))# 验证文件删除self.cursor.execute("SELECT * FROM media_files WHERE content_id = %s", (content_id,))deleted_media = self.cursor.fetchone()self.assertIsNone(deleted_media)def test_tag_management(self):"""测试标签管理功能"""# 创建测试标签tag_name1 = "测试标签1_" + self.generate_random_string()tag_name2 = "测试标签2_" + self.generate_random_string()self.cursor.execute("INSERT INTO tags (tag_name) VALUES (%s)", (tag_name1,))self.cursor.execute("INSERT INTO tags (tag_name) VALUES (%s)", (tag_name2,))# 验证标签创建self.cursor.execute("SELECT * FROM tags WHERE tag_name = %s", (tag_name1,))tag1 = self.cursor.fetchone()self.assertIsNotNone(tag1)self.cursor.execute("SELECT * FROM tags WHERE tag_name = %s", (tag_name2,))tag2 = self.cursor.fetchone()self.assertIsNotNone(tag2)# 创建测试内容并关联标签test_username = "tag_user_" + self.generate_random_string()test_email = test_username + "@example.com"test_password = "TagPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(test_username, test_email, hashlib.sha256(test_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (test_username,))user_id = self.cursor.fetchone()['user_id']self.cursor.execute("INSERT INTO contents (user_id, content_type, title) VALUES (%s, %s, %s)",(user_id, 'text', "测试标签内容"))self.cursor.execute("SELECT LAST_INSERT_ID()")content_id = self.cursor.fetchone()['LAST_INSERT_ID()']self.cursor.execute("INSERT INTO content_tags (content_id, tag_id) VALUES (%s, %s)",(content_id, tag1['tag_id']))self.cursor.execute("INSERT INTO content_tags (content_id, tag_id) VALUES (%s, %s)",(content_id, tag2['tag_id']))# 验证标签关联self.cursor.execute("""SELECT t.tag_name FROM content_tags ctJOIN tags t ON ct.tag_id = t.tag_idWHERE ct.content_id = %s""", (content_id,))associated_tags = self.cursor.fetchall()self.assertEqual(len(associated_tags), 2)self.assertIn(tag_name1, [tag['tag_name'] for tag in associated_tags])self.assertIn(tag_name2, [tag['tag_name'] for tag in associated_tags])# 删除内容（级联删除标签关联）self.cursor.execute("DELETE FROM contents WHERE content_id = %s", (content_id,))# 验证标签关联删除self.cursor.execute("""SELECT * FROM content_tags WHERE content_id = %s""", (content_id,))remaining_tags = self.cursor.fetchall()self.assertEqual(len(remaining_tags), 0)def test_comment_system(self):"""测试评论系统功能"""# 创建测试用户和内容test_username = "comment_user_" + self.generate_random_string()test_email = test_username + "@example.com"test_password = "CommentPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(test_username, test_email, hashlib.sha256(test_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (test_username,))user_id = self.cursor.fetchone()['user_id']self.cursor.execute("INSERT INTO contents (user_id, content_type, title) VALUES (%s, %s, %s)",(user_id, 'text', "测试评论内容"))self.cursor.execute("SELECT LAST_INSERT_ID()")content_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 发布评论comment_text = "这是测试评论"self.cursor.execute("INSERT INTO comments (content_id, user_id, comment_text) VALUES (%s, %s, %s)",(content_id, user_id, comment_text))self.cursor.execute("SELECT LAST_INSERT_ID()")comment_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 验证评论发布self.cursor.execute("SELECT * FROM comments WHERE comment_id = %s", (comment_id,))posted_comment = self.cursor.fetchone()self.assertIsNotNone(posted_comment)self.assertEqual(posted_comment['comment_text'], comment_text)# 发布子评论reply_text = "这是子评论"self.cursor.execute("INSERT INTO comments (content_id, user_id, comment_text, parent_comment_id) VALUES (%s, %s, %s, %s)",(content_id, user_id, reply_text, comment_id))self.cursor.execute("SELECT LAST_INSERT_ID()")reply_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 验证子评论发布self.cursor.execute("SELECT * FROM comments WHERE comment_id = %s", (reply_id,))posted_reply = self.cursor.fetchone()self.assertIsNotNone(posted_reply)self.assertEqual(posted_reply['comment_text'], reply_text)self.assertEqual(posted_reply['parent_comment_id'], comment_id)# 删除主评论（级联删除子评论）self.cursor.execute("DELETE FROM comments WHERE comment_id = %s", (comment_id,))# 验证评论删除self.cursor.execute("SELECT * FROM comments WHERE comment_id = %s", (comment_id,))deleted_comment = self.cursor.fetchone()self.assertIsNone(deleted_comment)self.cursor.execute("SELECT * FROM comments WHERE comment_id = %s", (reply_id,))deleted_reply = self.cursor.fetchone()self.assertIsNone(deleted_reply)def test_interaction_system(self):"""测试用户互动系统功能"""# 创建测试用户和内容test_username = "interaction_user_" + self.generate_random_string()test_email = test_username + "@example.com"test_password = "InteractionPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(test_username, test_email, hashlib.sha256(test_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (test_username,))user_id = self.cursor.fetchone()['user_id']self.cursor.execute("INSERT INTO contents (user_id, content_type, title) VALUES (%s, %s, %s)",(user_id, 'text', "测试互动内容"))self.cursor.execute("SELECT LAST_INSERT_ID()")content_id = self.cursor.fetchone()['LAST_INSERT_ID()']# 点赞内容self.cursor.execute("INSERT INTO interactions (user_id, content_id, interaction_type) VALUES (%s, %s, %s)",(user_id, content_id, 'like'))# 验证点赞self.cursor.execute("""SELECT * FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'like'""", (user_id, content_id))like_interaction = self.cursor.fetchone()self.assertIsNotNone(like_interaction)# 收藏内容self.cursor.execute("INSERT INTO interactions (user_id, content_id, interaction_type) VALUES (%s, %s, %s)",(user_id, content_id, 'collect'))# 验证收藏self.cursor.execute("""SELECT * FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'collect'""", (user_id, content_id))collect_interaction = self.cursor.fetchone()self.assertIsNotNone(collect_interaction)# 取消点赞self.cursor.execute("""DELETE FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'like'""", (user_id, content_id))# 验证点赞取消self.cursor.execute("""SELECT * FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'like'""", (user_id, content_id))canceled_like = self.cursor.fetchone()self.assertIsNone(canceled_like)# 取消收藏self.cursor.execute("""DELETE FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'collect'""", (user_id, content_id))# 验证收藏取消self.cursor.execute("""SELECT * FROM interactions WHERE user_id = %s AND content_id = %s AND interaction_type = 'collect'""", (user_id, content_id))canceled_collect = self.cursor.fetchone()self.assertIsNone(canceled_collect)def test_following_system(self):"""测试关注系统功能"""# 创建两个测试用户follower_username = "follower_user_" + self.generate_random_string()follower_email = follower_username + "@example.com"follower_password = "FollowerPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(follower_username, follower_email, hashlib.sha256(follower_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (follower_username,))follower_id = self.cursor.fetchone()['user_id']followed_username = "followed_user_" + self.generate_random_string()followed_email = followed_username + "@example.com"followed_password = "FollowedPassword123!"self.cursor.execute("INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s)",(followed_username, followed_email, hashlib.sha256(followed_password.encode()).hexdigest()))self.cursor.execute("SELECT user_id FROM users WHERE username = %s", (followed_username,))followed_id = self.cursor.fetchone()['user_id']# 关注操作self.cursor.execute("INSERT INTO follows (follower_id, followed_id) VALUES (%s, %s)",(follower_id, followed_id))# 验证关注self.cursor.execute("""SELECT * FROM follows WHERE follower_id = %s AND followed_id = %s""", (follower_id, followed_id))follow_record = self.cursor.fetchone()self.assertIsNotNone(follow_record)# 取消关注self.cursor.execute("""DELETE FROM follows WHERE follower_id = %s AND followed_id = %s""", (follower_id, followed_id))# 验证取消关注self.cursor.execute("""SELECT * FROM follows WHERE follower_id = %s AND followed_id = %s""", (follower_id, followed_id))unfollow_record = self.cursor.fetchone()self.assertIsNone(unfollow_record)if __name__ == "__main__":unittest.main()

测试代码：

1. 完整的测试环境准备和清理机制，确保每次测试都在干净的环境中运行

2. 覆盖了所有主要的业务逻辑，包括用户管理、内容管理、多媒体文件管理、标签管理、评论系统、互动系统和关注系统

3. 严格遵循数据库约束规则，处理好外键约束和唯一键冲突问题

4. 测试用例设计全面，包含创建、读取、更新和删除操作

5. 使用随机数据生成测试用户和内容，避免测试数据冲突

6. 每个测试用例都包含数据验证步骤，确保数据库操作符合预期