170 lines
5.0 KiB
Python
170 lines
5.0 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
SQL Server状态功能测试脚本
|
|
"""
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
def test_database_schema():
|
|
"""测试数据库schema是否包含sqlserver_status字段"""
|
|
print("=" * 60)
|
|
print("测试: 数据库Schema")
|
|
print("=" * 60)
|
|
|
|
db_path = Path(__file__).parent / 'experiments.db'
|
|
|
|
if not db_path.exists():
|
|
print("⚠ 数据库文件不存在，将在程序首次运行时创建")
|
|
return False
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
cursor = conn.cursor()
|
|
|
|
# 获取表结构
|
|
cursor.execute("PRAGMA table_info(experiments)")
|
|
columns = cursor.fetchall()
|
|
|
|
column_names = [col[1] for col in columns]
|
|
|
|
# 验证新字段
|
|
if 'sqlserver_status' in column_names:
|
|
print("✓ sqlserver_status 字段存在")
|
|
|
|
# 查询是否有数据
|
|
cursor.execute("SELECT COUNT(*) FROM experiments WHERE sqlserver_status IS NOT NULL")
|
|
count = cursor.fetchone()[0]
|
|
print(f"✓ 有 {count} 条记录包含SQL Server状态")
|
|
|
|
# 显示状态统计
|
|
cursor.execute("""
|
|
SELECT sqlserver_status, COUNT(*)
|
|
FROM experiments
|
|
WHERE sqlserver_status IS NOT NULL
|
|
GROUP BY sqlserver_status
|
|
""")
|
|
stats = cursor.fetchall()
|
|
|
|
if stats:
|
|
print("\n状态统计:")
|
|
for status, count in stats:
|
|
print(f" {status}: {count} 条")
|
|
|
|
conn.close()
|
|
return True
|
|
else:
|
|
print("⚠ sqlserver_status 字段不存在（将在程序启动时自动添加）")
|
|
conn.close()
|
|
return False
|
|
|
|
def test_verify_function():
|
|
"""测试验证函数"""
|
|
print("\n" + "=" * 60)
|
|
print("测试: 验证函数")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from sqlserver_writer import verify_data_in_sqlserver
|
|
print("✓ verify_data_in_sqlserver 函数导入成功")
|
|
|
|
# 测试函数签名
|
|
import inspect
|
|
sig = inspect.signature(verify_data_in_sqlserver)
|
|
params = list(sig.parameters.keys())
|
|
expected_params = ['order_no', 'start_time', 'end_time', 'sqlserver_config']
|
|
|
|
if params == expected_params:
|
|
print(f"✓ 函数参数正确: {params}")
|
|
else:
|
|
print(f"⚠ 函数参数不匹配")
|
|
print(f" 期望: {expected_params}")
|
|
print(f" 实际: {params}")
|
|
|
|
return True
|
|
except ImportError as e:
|
|
print(f"✗ 导入失败: {e}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
return False
|
|
|
|
def test_ui_columns():
|
|
"""测试UI列数"""
|
|
print("\n" + "=" * 60)
|
|
print("测试: UI列配置")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
# 读取ui_main.py文件
|
|
ui_file = Path(__file__).parent / 'ui_main.py'
|
|
|
|
if not ui_file.exists():
|
|
print("✗ ui_main.py 文件不存在")
|
|
return False
|
|
|
|
with open(ui_file, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# 检查列数
|
|
if 'setColumnCount(17)' in content:
|
|
print("✓ 表格列数已更新为17列")
|
|
else:
|
|
print("⚠ 表格列数可能未正确更新")
|
|
|
|
# 检查列标题
|
|
if '"数据库状态"' in content:
|
|
print("✓ 列标题包含「数据库状态」")
|
|
else:
|
|
print("⚠ 列标题可能缺少「数据库状态」")
|
|
|
|
# 检查状态更新函数
|
|
if '_update_sqlserver_status' in content:
|
|
print("✓ _update_sqlserver_status 函数存在")
|
|
else:
|
|
print("⚠ _update_sqlserver_status 函数可能缺失")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""运行所有测试"""
|
|
print("\n" + "=" * 60)
|
|
print("SQL Server状态功能测试")
|
|
print("=" * 60 + "\n")
|
|
|
|
results = []
|
|
|
|
results.append(("数据库Schema", test_database_schema()))
|
|
results.append(("验证函数", test_verify_function()))
|
|
results.append(("UI列配置", test_ui_columns()))
|
|
|
|
print("\n" + "=" * 60)
|
|
print("测试结果汇总")
|
|
print("=" * 60)
|
|
|
|
for name, passed in results:
|
|
status = "✓ 通过" if passed else "⚠ 需要检查"
|
|
print(f"{name}: {status}")
|
|
|
|
all_passed = all(result[1] for result in results)
|
|
|
|
if all_passed:
|
|
print("\n✓ 所有测试通过！")
|
|
else:
|
|
print("\n⚠ 部分测试需要检查")
|
|
|
|
print("\n下一步:")
|
|
print("1. 启动程序: python main.py")
|
|
print("2. 点击「开始工单」并完成一个实验")
|
|
print("3. 点击「保存数据」按钮")
|
|
print("4. 查看实验历史列表中的「数据库状态」列")
|
|
print("5. 验证状态显示为「已入库」（绿色）或其他状态")
|
|
print()
|
|
|
|
return 0 if all_passed else 1
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|