132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试数据库迁移功能
|
||
"""
|
||
import sqlite3
|
||
import shutil
|
||
from pathlib import Path
|
||
from db_migration import DatabaseMigration, initialize_database, get_database_info
|
||
|
||
DB_PATH = Path(__file__).parent / "experiments.db"
|
||
BACKUP_PATH = Path(__file__).parent / "experiments.db.test_backup"
|
||
|
||
print("=" * 80)
|
||
print("数据库迁移功能测试")
|
||
print("=" * 80)
|
||
|
||
# 1. 备份当前数据库
|
||
print("\n步骤 1: 备份当前数据库")
|
||
if DB_PATH.exists():
|
||
shutil.copy(DB_PATH, BACKUP_PATH)
|
||
print(f"✅ 已备份到: {BACKUP_PATH}")
|
||
else:
|
||
print("⚠️ 数据库文件不存在，将创建新数据库")
|
||
|
||
# 2. 获取当前数据库信息
|
||
print("\n步骤 2: 当前数据库信息")
|
||
if DB_PATH.exists():
|
||
info = get_database_info()
|
||
if info:
|
||
print(f" 版本: {info['version']}")
|
||
print(f" 表数量: {len(info['tables'])}")
|
||
|
||
for table in info['tables']:
|
||
if table['name'] == 'experiments':
|
||
print(f"\n experiments 表:")
|
||
print(f" 列数: {table['column_count']}")
|
||
print(f" 列名: {', '.join(table['columns'])}")
|
||
else:
|
||
print(" 数据库不存在")
|
||
|
||
# 3. 测试迁移功能
|
||
print("\n步骤 3: 运行数据库初始化")
|
||
print("-" * 80)
|
||
success = initialize_database()
|
||
print("-" * 80)
|
||
|
||
if success:
|
||
print("✅ 数据库初始化成功")
|
||
else:
|
||
print("❌ 数据库初始化失败")
|
||
|
||
# 4. 验证迁移结果
|
||
print("\n步骤 4: 验证迁移结果")
|
||
info = get_database_info()
|
||
if info:
|
||
print(f" 当前版本: {info['version']}")
|
||
|
||
for table in info['tables']:
|
||
if table['name'] == 'experiments':
|
||
print(f"\n experiments 表:")
|
||
print(f" 列数: {table['column_count']}")
|
||
print(f" 列名:")
|
||
for col in table['columns']:
|
||
print(f" - {col}")
|
||
|
||
# 检查必要字段
|
||
required_fields = [
|
||
'id', 'work_order_no', 'config_json',
|
||
'start_ts', 'end_ts', 'is_paused',
|
||
'created_at', 'script_data'
|
||
]
|
||
|
||
missing = [f for f in required_fields if f not in table['columns']]
|
||
|
||
if missing:
|
||
print(f"\n ⚠️ 缺失字段: {', '.join(missing)}")
|
||
else:
|
||
print(f"\n ✅ 所有必要字段都存在")
|
||
|
||
# 5. 测试添加列功能
|
||
print("\n步骤 5: 测试添加列功能")
|
||
migration = DatabaseMigration()
|
||
migration.connect()
|
||
|
||
# 测试添加一个测试列
|
||
test_column = "test_migration_column"
|
||
print(f" 尝试添加测试列: {test_column}")
|
||
|
||
if migration.column_exists("experiments", test_column):
|
||
print(f" 列 {test_column} 已存在")
|
||
else:
|
||
success = migration.add_column("experiments", test_column, "TEXT", "test_value")
|
||
if success:
|
||
print(f" ✅ 成功添加列: {test_column}")
|
||
|
||
# 验证
|
||
if migration.column_exists("experiments", test_column):
|
||
print(f" ✅ 验证成功: 列已存在")
|
||
else:
|
||
print(f" ❌ 验证失败: 列不存在")
|
||
|
||
# 删除测试列（通过重建表）
|
||
print(f" 清理测试列...")
|
||
# 注意：SQLite 不支持 DROP COLUMN，这里只是演示
|
||
# 实际使用中，测试列会保留在数据库中
|
||
else:
|
||
print(f" ❌ 添加列失败")
|
||
|
||
migration.close()
|
||
|
||
# 6. 总结
|
||
print("\n" + "=" * 80)
|
||
print("测试总结")
|
||
print("=" * 80)
|
||
|
||
final_info = get_database_info()
|
||
if final_info:
|
||
print(f"✅ 数据库版本: {final_info['version']}")
|
||
print(f"✅ 表数量: {len(final_info['tables'])}")
|
||
|
||
for table in final_info['tables']:
|
||
if table['name'] == 'experiments':
|
||
print(f"✅ experiments 表列数: {table['column_count']}")
|
||
|
||
print("\n提示:")
|
||
print(" - 原始数据库已备份到: experiments.db.test_backup")
|
||
print(" - 如需恢复，请手动复制备份文件")
|
||
print(" - 测试列 'test_migration_column' 已添加（如果之前不存在）")
|
||
|
||
print("\n" + "=" * 80)
|