logan
2984f11181
- 删除16个旧Python测试文件(970行→400行) - 重写conftest.py适配真实种子用户(admin/test123/test234/testtest/logan) - 5个精简测试文件覆盖核心API: - test_auth.py: 登录/鉴权/种子用户验证 (6 tests) - test_instances.py: 实例列表/详情/连接测试 (6 tests) - test_tickets.py: 工单CRUD/只读SQL拦截/auditor权限 (7 tests) - test_workflow.py: E2E完整审批流/拒绝/撤销 (3 tests) - test_permissions.py: 角色权限/执行权限/workflow权限 (6 tests) - 新增 cleanup_test_data.py 清理测试脏数据(支持--dry-run) - 全量29测试3.27s通过
88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cleanup test data from SQLite database.
|
|
Removes tickets, approvals, and test-created tables, preserving seed data.
|
|
|
|
Usage:
|
|
python tests/cleanup_test_data.py # full cleanup
|
|
python tests/cleanup_test_data.py --dry-run # preview only
|
|
"""
|
|
import sqlite3
|
|
import sys
|
|
import os
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(__file__), '..', 'backend', 'sqlite3.db')
|
|
|
|
# Tables in remote MySQL that were created by tests (DROP IF EXISTS)
|
|
TEST_MYSQL_TABLES = [
|
|
"_test_tbl", "_e2e_test", "_reject_test", "_cancel_test", "_perm_test",
|
|
"orders", "_ghost_test_del", "flow_test", "ghost_test",
|
|
]
|
|
|
|
|
|
def cleanup(dry_run=False):
|
|
db = sqlite3.connect(DB_PATH)
|
|
cur = db.cursor()
|
|
|
|
actions = []
|
|
|
|
# 1. Count and delete ticket-related data
|
|
for table in ["ticket_sql_items", "ticket_approvals", "tickets"]:
|
|
cur.execute(f"SELECT COUNT(*) FROM {table}")
|
|
count = cur.fetchone()[0]
|
|
if count > 0:
|
|
actions.append(f"DELETE {count} rows FROM {table}")
|
|
if not dry_run:
|
|
cur.execute(f"DELETE FROM {table}")
|
|
|
|
# 2. Count query history
|
|
cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='query_history'")
|
|
if cur.fetchone():
|
|
cur.execute("SELECT COUNT(*) FROM query_history")
|
|
count = cur.fetchone()[0]
|
|
if count > 0:
|
|
actions.append(f"DELETE {count} rows FROM query_history")
|
|
if not dry_run:
|
|
cur.execute("DELETE FROM query_history")
|
|
|
|
# 3. Reset auto-increment counters
|
|
if not dry_run:
|
|
for table in ["ticket_sql_items", "ticket_approvals", "tickets"]:
|
|
cur.execute(f"DELETE FROM sqlite_sequence WHERE name='{table}'")
|
|
|
|
if not dry_run:
|
|
db.commit()
|
|
db.close()
|
|
|
|
return actions
|
|
|
|
|
|
def main():
|
|
dry_run = "--dry-run" in sys.argv
|
|
if dry_run:
|
|
print("🔍 DRY RUN — preview only\n")
|
|
|
|
actions = cleanup(dry_run)
|
|
|
|
if not actions:
|
|
print("✅ 数据库干净，无需清理")
|
|
return
|
|
|
|
for a in actions:
|
|
prefix = " WOULD " if dry_run else " ✅ "
|
|
print(f"{prefix}{a}")
|
|
|
|
if dry_run:
|
|
print(f"\n共 {len(actions)} 项待清理 (使用不带 --dry-run 执行)")
|
|
else:
|
|
print(f"\n✅ 已清理 {len(actions)} 项")
|
|
|
|
# Print note about MySQL remote tables
|
|
print(f"\n⚠️ 注意: 远程 MySQL 测试表需手动清理:")
|
|
for t in TEST_MYSQL_TABLES:
|
|
print(f" DROP TABLE IF EXISTS opsxsql_test.{t};")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|