# -*- 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()