91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试 SQLite 数据库迁移和功能
|
||
"""
|
||
import os
|
||
import sys
|
||
from database import get_database, migrate_from_json
|
||
from task_queue import get_task_queue, TaskStatus
|
||
|
||
def test_database():
|
||
"""测试数据库功能"""
|
||
print("=" * 60)
|
||
print("开始测试 SQLite 数据库功能")
|
||
print("=" * 60)
|
||
|
||
# 1. 测试数据库初始化
|
||
print("\n1. 测试数据库初始化...")
|
||
db = get_database()
|
||
print(f"✓ 数据库初始化成功: {db.db_path}")
|
||
|
||
# 2. 测试从 JSON 迁移数据
|
||
print("\n2. 测试数据迁移...")
|
||
json_file = "data/task_queue.json"
|
||
if os.path.exists(json_file):
|
||
count = migrate_from_json(json_file)
|
||
print(f"✓ 迁移了 {count} 个任务")
|
||
else:
|
||
print("! 未找到旧 JSON 文件，跳过迁移")
|
||
|
||
# 3. 测试任务队列功能
|
||
print("\n3. 测试任务队列功能...")
|
||
queue = get_task_queue()
|
||
|
||
# 3.1 添加测试任务
|
||
print("\n3.1 添加测试任务...")
|
||
task_id = queue.add_task(
|
||
url="https://baijiahao.baidu.com/u?app_id=test123",
|
||
months=3,
|
||
use_proxy=True,
|
||
username="test_user"
|
||
)
|
||
print(f"✓ 添加任务成功: {task_id}")
|
||
|
||
# 3.2 获取任务
|
||
print("\n3.2 获取任务...")
|
||
task = queue.get_task(task_id)
|
||
if task:
|
||
print(f"✓ 获取任务成功:")
|
||
print(f" - URL: {task['url']}")
|
||
print(f" - 状态: {task['status']}")
|
||
print(f" - 创建时间: {task['created_at']}")
|
||
|
||
# 3.3 更新任务状态
|
||
print("\n3.3 更新任务状态...")
|
||
queue.update_task_status(task_id, TaskStatus.PROCESSING)
|
||
task = queue.get_task(task_id)
|
||
print(f"✓ 更新状态成功: {task['status']}")
|
||
|
||
# 3.4 更新任务进度
|
||
print("\n3.4 更新任务进度...")
|
||
queue.update_task_progress(task_id, 50, "正在处理中...", 25)
|
||
task = queue.get_task(task_id)
|
||
print(f"✓ 更新进度成功: {task['progress']}%")
|
||
|
||
# 3.5 获取队列统计
|
||
print("\n3.5 获取队列统计...")
|
||
stats = queue.get_queue_stats()
|
||
print(f"✓ 队列统计:")
|
||
print(f" - 总任务数: {stats['total']}")
|
||
print(f" - 等待中: {stats['pending']}")
|
||
print(f" - 处理中: {stats['processing']}")
|
||
print(f" - 已完成: {stats['completed']}")
|
||
print(f" - 失败: {stats['failed']}")
|
||
|
||
# 3.6 获取所有任务
|
||
print("\n3.6 获取所有任务...")
|
||
all_tasks = queue.get_all_tasks()
|
||
print(f"✓ 获取所有任务成功，共 {len(all_tasks)} 个任务")
|
||
|
||
# 3.7 删除测试任务
|
||
print("\n3.7 删除测试任务...")
|
||
queue.delete_task(task_id)
|
||
print(f"✓ 删除任务成功: {task_id}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("所有测试通过！SQLite 数据库运行正常")
|
||
print("=" * 60)
|
||
|
||
if __name__ == "__main__":
|
||
test_database()
|