"""配置层回归测试（不依赖 pytest） python tests/test_config.py 覆盖： - config.yml 能被加载，关键配置项存在且类型正确； - 点号取值 / 布尔与整数转换； - 目录解析（相对路径基于项目根）； - 环境变量覆盖 config.yml（临时试验用）； - 敏感项不在 config.yml 里（只能在 .env）。 """ import os import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) import yaml # noqa: E402 import config # noqa: E402 # 测试不硬编码"当前配的是哪个模型/哪条路由"——那是用户随时会改的。 # 这里直接用 yaml 独立读一遍 config.yml，与 config.py 的解析结果对比： # 校验的是"解析机制正确"，而不是"值恰好等于某个名字"。 with open(ROOT / 'config.yml', encoding='utf-8') as _f: RAW = yaml.safe_load(_f) def raw(dotted, default=None): node = RAW for key in dotted.split('.'): if not isinstance(node, dict) or key not in node: return default node = node[key] return node # 按 config.yml 自己的 routes/endpoints 推导"应要求哪些密钥"（独立算法） _expected_secrets = {'MYSQL_PASSWORD'} for _role in ('asr', 'correct', 'split'): _ep = (RAW.get('endpoints') or {}).get((RAW.get('routes') or {}).get(_role), {}) if _ep.get('api_key_env'): _expected_secrets.add(_ep['api_key_env']) def main(): failed = 0 def check(desc, got, want): nonlocal failed if got == want: print(f" ✓ {desc}: {got!r}") else: failed += 1 print(f" ✗ {desc}: 期望 {want!r}，实际 {got!r}") print("config.yml 加载:") check('配置文件存在', config.LOADED_CONFIG_FILE is not None, True) check('数据库端口', config.get_int('mysql.port'), 13306) check('数据库库名', config.get('mysql.database'), 'myquant') check('ASR 模型', config.model('asr_model'), 'paraformer-realtime-v2') check('校对模型（与 config.yml 一致）', config.model('correct_model'), raw('models.correct_model')) check('切分模型（与 config.yml 一致）', config.model('split_model'), raw('models.split_model')) check('校对开关(默认1)', config.get_bool('llm_correct.enabled'), True) check('清理开关(默认1)', config.get_bool('cleanup.after_daily_run'), True) check('仅成功时清理', config.get_bool('cleanup.only_on_success'), True) check('语言提示', config.get_list('asr.language_hints'), ['zh', 'en']) print("\n目录解析:") check('video_dir 基于项目根', config.video_dir(), ROOT / 'xwlb_video') check('audio_dir 基于项目根', config.audio_dir(), ROOT / 'audio_processing') print("\n环境变量覆盖优先于 config.yml:") os.environ['DASHSCOPE_LLM_MODEL'] = 'qwen-max-test' os.environ['LLM_CORRECT_ENABLED'] = '0' cfg = config.reload_config() check('模型被环境变量覆盖', cfg['models']['correct_model'], 'qwen-max-test') check('开关被环境变量覆盖', config.get_bool('llm_correct.enabled'), False) del os.environ['DASHSCOPE_LLM_MODEL'], os.environ['LLM_CORRECT_ENABLED'] config.reload_config() check('恢复 config.yml 值', config.model('correct_model'), raw('models.correct_model')) print("\n接入点与路由（换供应商的核心）:") check('asr 路由（与 config.yml 一致）', config.route('asr'), raw('routes.asr')) check('correct 路由（与 config.yml 一致）', config.route('correct'), raw('routes.correct')) check('correct 接入点类型（与 config.yml 一致）', config.endpoint_for('correct')[1]['kind'], raw(f"endpoints.{raw('routes.correct')}.kind")) check('split 接入点类型', config.endpoint_for('split')[1]['kind'], 'openai') check('split 请求地址拼接', config.openai_url('split'), 'https://api.deepseek.com/v1/chat/completions') check('密钥变量名来自接入点', config.api_key_env('split'), 'DEEPSEEK_API_KEY') check('必需敏感项 = 路由用到的接入点密钥 + MySQL 口令', sorted(config.required_secrets()), sorted(_expected_secrets)) check('默认配置无问题', config.validate_endpoints(), []) import dashscope check('dashscope SDK 地址已按配置生效', dashscope.base_http_api_url, config.endpoint('dashscope')['http_base_url']) check('dashscope WS 地址已按配置生效', dashscope.base_websocket_api_url, config.endpoint('dashscope')['websocket_base_url']) print("\n切换供应商（把校对接管到 OpenAI 兼容接入点）:") os.environ['XWLB_ROUTE_CORRECT'] = 'deepseek' os.environ['DASHSCOPE_LLM_MODEL'] = 'deepseek-chat' config.reload_config() check('correct 已切到 openai 类型', config.endpoint_for('correct')[1]['kind'], 'openai') check('correct 请求地址随之改变', config.openai_url('correct'), 'https://api.deepseek.com/v1/chat/completions') check('correct 模型名可独立覆盖', config.model('correct_model'), 'deepseek-chat') check('密钥变量名随供应商变化', config.api_key_env('correct'), 'DEEPSEEK_API_KEY') # 把 ASR 也切到非 dashscope 接入点：应被校验拦下（实时识别协议不支持） os.environ['XWLB_ROUTE_ASR'] = 'deepseek' config.reload_config() check('非 dashscope 的 ASR 被校验提示', any('仅支持 kind=dashscope' in p for p in config.validate_endpoints()), True) check('不需要 dashscope 密钥了', config.required_secrets(), ['DEEPSEEK_API_KEY', 'MYSQL_PASSWORD']) # 指向不存在的接入点：必须给出可读错误 os.environ['XWLB_ROUTE_SPLIT'] = 'not-exist' config.reload_config() try: config.endpoint_for('split') check('未知接入点应报错', '未报错', 'RuntimeError') except RuntimeError as e: check('未知接入点报错可读', '没有该定义' in str(e), True) # 地址也可用环境变量覆盖 os.environ['XWLB_ROUTE_CORRECT'] = 'deepseek' os.environ['DEEPSEEK_BASE_URL'] = 'https://my-gateway.example.com/llm' config.reload_config() check('base_url 可被环境变量覆盖', config.openai_url('correct'), 'https://my-gateway.example.com/llm/chat/completions') for key in ('XWLB_ROUTE_CORRECT', 'XWLB_ROUTE_ASR', 'XWLB_ROUTE_SPLIT', 'DASHSCOPE_LLM_MODEL', 'DEEPSEEK_BASE_URL'): os.environ.pop(key, None) config.reload_config() check('恢复 config.yml 路由', [config.route(r) for r in ('asr', 'correct', 'split')], [raw('routes.asr'), raw('routes.correct'), raw('routes.split')]) print("\n敏感项与 config.yml 隔离:") import yaml yml = yaml.safe_load((ROOT / 'config.yml').read_text(encoding='utf-8')) def walk(node, path=''): for key, value in (node or {}).items(): here = f'{path}.{key}' if path else str(key) if isinstance(value, dict): yield from walk(value, here) else: yield here, value leaves = list(walk(yml)) # 1) 配置项名不得是口令/密钥类（注意排除 max_tokens 这类合法项） risky_names = {'password', 'passwd', 'secret', 'secret_key', 'api_key', 'apikey', 'token', 'access_token', 'auth_token'} risky = [p for p, _ in leaves if (lambda last: last in risky_names or last.endswith(('_password', '_secret', '_api_key')))( p.split('.')[-1].lower())] check('config.yml 无口令/密钥类配置项', risky, []) # 2) 配置值不得等于 .env 中的真实密钥（防误粘贴） real_secrets = {v for v in (os.getenv('MYSQL_PASSWORD'), os.getenv('DASHSCOPE_API_KEY'), os.getenv('DEEPSEEK_API_KEY')) if v} leaked = [p for p, v in leaves if isinstance(v, str) and v in real_secrets] check('config.yml 未泄漏 .env 中的密钥值', leaked, []) check('db_config 口令来自环境变量', config.db_config()['password'] == os.getenv('MYSQL_PASSWORD'), True) print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}") return 1 if failed else 0 if __name__ == '__main__': sys.exit(main())