159 lines
7.5 KiB
Python
159 lines
7.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""元景 scense 测试用例域库表幂等初始化脚本。
|
||
|
||
修复 Bug [e2Lck6zIRCXB5g7_HPZn0]（环境根因）：
|
||
sd_test_cases 用例库不可达——list_cases 返回字段全空，scense 库无该表。
|
||
本脚本在应用 init() 阶段自动执行（也可独立运行），
|
||
用 CREATE TABLE IF NOT EXISTS 幂等创建测试用例域 3 张表：
|
||
sd_test_cases / sd_test_plans / sd_test_results
|
||
确保环境（重建/换库）后用例库自动可达，list_cases 不再返回全空。
|
||
|
||
用法：
|
||
1) 应用启动：app/scense.py init() 中调用 ensure_test_db()
|
||
2) 独立执行：python scripts/init_test_db.py（读取 .env 连接）
|
||
"""
|
||
import os
|
||
|
||
# 建表 DDL 与 scripts/../sql/sd_test_cases.sql 保持一致（幂等 IF NOT EXISTS）
|
||
_DDL_STATEMENTS = [
|
||
# ---------- sd_test_cases 测试用例表 ----------
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS `sd_test_cases` (
|
||
`id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||
`case_id` VARCHAR(64) NOT NULL COMMENT '用例编码（业务唯一，如 TC-0001）',
|
||
`case_name` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '用例名称（list_cases 读取字段，兼容 case_name 命名）',
|
||
`title` VARCHAR(255) NOT NULL COMMENT '用例标题',
|
||
`description` TEXT NULL COMMENT '用例描述',
|
||
`module` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '所属模块',
|
||
`plan_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '所属测试计划ID',
|
||
`priority` VARCHAR(16) NOT NULL DEFAULT 'P2' COMMENT '优先级：P0/P1/P2/P3',
|
||
`preconditions` TEXT NULL COMMENT '前置条件',
|
||
`steps` TEXT NULL COMMENT '测试步骤',
|
||
`expected_result` TEXT NULL COMMENT '预期结果',
|
||
`actual_result` TEXT NULL COMMENT '实际结果',
|
||
`status` VARCHAR(16) NOT NULL DEFAULT 'draft' COMMENT '用例状态：draft/active/closed',
|
||
`execute_status` VARCHAR(16) NOT NULL DEFAULT 'not_run' COMMENT '执行状态：not_run/pass/fail/blocked',
|
||
`created_by` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建人',
|
||
`updated_by` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '更新人',
|
||
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
|
||
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
|
||
PRIMARY KEY (`id`),
|
||
UNIQUE KEY `uk_case_id` (`case_id`),
|
||
KEY `idx_plan_id` (`plan_id`),
|
||
KEY `idx_module` (`module`),
|
||
KEY `idx_status` (`status`),
|
||
KEY `idx_execute_status` (`execute_status`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='测试用例表（list_cases 数据来源）'
|
||
""",
|
||
# ---------- sd_test_plans 测试计划表 ----------
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS `sd_test_plans` (
|
||
`id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||
`plan_id` VARCHAR(64) NOT NULL COMMENT '测试计划ID',
|
||
`plan_name` VARCHAR(255) NOT NULL COMMENT '计划名称',
|
||
`iteration_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '所属迭代ID',
|
||
`description` TEXT NULL COMMENT '计划描述',
|
||
`status` VARCHAR(16) NOT NULL DEFAULT 'draft' COMMENT '计划状态：draft/approved/in_progress/done',
|
||
`created_by` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建人',
|
||
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
|
||
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
|
||
PRIMARY KEY (`id`),
|
||
UNIQUE KEY `uk_plan_id` (`plan_id`),
|
||
KEY `idx_iteration_id` (`iteration_id`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='测试计划表'
|
||
""",
|
||
# ---------- sd_test_results 测试执行结果表 ----------
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS `sd_test_results` (
|
||
`id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||
`result_id` VARCHAR(64) NOT NULL COMMENT '执行结果ID',
|
||
`case_id` VARCHAR(64) NOT NULL COMMENT '用例编码',
|
||
`plan_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '所属测试计划ID',
|
||
`run_status` VARCHAR(16) NOT NULL DEFAULT 'not_run' COMMENT '执行结果：pass/fail/blocked/skipped',
|
||
`actual_result` TEXT NULL COMMENT '实际结果描述',
|
||
`bug_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '失败联动Bug ID',
|
||
`executed_by` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '执行人',
|
||
`executed_at` DATETIME NULL COMMENT '执行时间',
|
||
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
|
||
PRIMARY KEY (`id`),
|
||
UNIQUE KEY `uk_result_id` (`result_id`),
|
||
KEY `idx_case_id` (`case_id`),
|
||
KEY `idx_plan_id` (`plan_id`),
|
||
KEY `idx_run_status` (`run_status`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='测试执行结果表'
|
||
""",
|
||
]
|
||
|
||
|
||
def _dbname():
|
||
"""取主库名（多级兜底，不硬编码）。
|
||
|
||
优先级：
|
||
1) ServerEnv params['dbname']（应用运行期，appbase 写入）
|
||
2) conf/config.json 的 module_dbname 映射（独立部署脚本场景）
|
||
3) 环境变量 MYSQL_DATABASE
|
||
4) 默认 'scense'（单库应用约定）
|
||
"""
|
||
# 1) ServerEnv（应用内）
|
||
try:
|
||
from ahserver.serverenv import ServerEnv
|
||
env = ServerEnv()
|
||
if getattr(env, 'params', None) and env.params.get('dbname'):
|
||
return env.params.get('dbname')
|
||
except Exception:
|
||
pass
|
||
# 2) conf/config.json module_dbname 映射
|
||
try:
|
||
import json
|
||
from pathlib import Path
|
||
cfg_path = Path(__file__).resolve().parent.parent / 'conf' / 'config.json'
|
||
if cfg_path.exists():
|
||
cfg = json.loads(cfg_path.read_text(encoding='utf-8'))
|
||
mapping = cfg.get('module_dbname') or {}
|
||
if isinstance(mapping, dict) and mapping.get('scense'):
|
||
return mapping['scense']
|
||
if cfg.get('dbname'):
|
||
return cfg['dbname']
|
||
except Exception:
|
||
pass
|
||
# 3) 环境变量
|
||
if os.environ.get('MYSQL_DATABASE'):
|
||
return os.environ['MYSQL_DATABASE']
|
||
# 4) 默认
|
||
return 'scense'
|
||
|
||
|
||
def ensure_test_db():
|
||
"""幂等创建测试用例域表。重复执行安全；某条失败记录日志不中断后续。"""
|
||
from sqlor import sor
|
||
|
||
dbname = _dbname()
|
||
created = []
|
||
for ddl in _DDL_STATEMENTS:
|
||
try:
|
||
sor.sqlExe(dbname, ddl)
|
||
created.append(_table_name_of(ddl))
|
||
except Exception as e: # noqa: BLE001 - 初始化容错，不因单表失败阻塞应用启动
|
||
print('[init_test_db] create table failed: %s (%s)' % (_table_name_of(ddl), e))
|
||
print('[init_test_db] ensured tables in db[%s]: %s' % (dbname, created))
|
||
return created
|
||
|
||
|
||
def _table_name_of(ddl):
|
||
import re
|
||
m = re.search(r'CREATE TABLE IF NOT EXISTS `([^`]+)`', ddl)
|
||
return m.group(1) if m else '?'
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 独立执行：加载 .env 环境变量后调用
|
||
from pathlib import Path
|
||
env_path = Path(__file__).resolve().parent.parent / '.env'
|
||
if env_path.exists():
|
||
for line in env_path.read_text(encoding='utf-8').splitlines():
|
||
line = line.strip()
|
||
if line and not line.startswith('#') and '=' in line:
|
||
k, _, v = line.partition('=')
|
||
os.environ.setdefault(k.strip(), v.strip())
|
||
ensure_test_db()
|