diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index e917df5d..417302ad 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -76,8 +76,13 @@ CATALOG_DRAFT_GUIDANCE_OPTION = "catalog_draft_guidance" # mask_columns:证据面板遮罩的列,驱动不认识它,拼进连接串会被当成未知参数拒掉 # catalog_profile:数据剖析设置,同上;它还是个对象,拼进去连 URL 都不成形 # catalog_draft_guidance:数据目录的起草要求,一段中文,拼进去连接直接失败 +# introspect_scope:探查范围的规则(data/source_objects.SCOPE_OPTION),运行时按需把范围外的表加进来时也会自动写; +# catalog_maintenance:定时维护的步骤(data/source_maintenance.MAINTENANCE_OPTION)。这两项都是对象,以前漏在名单外, +# 设过之后驱动 connect() 收到不认识的关键字参数,这个数据源上每一条查询都失败。 +# 在别的模块声明的两项这里写字面量(那两个模块反过来要用数据层,导入会成环);tests/test_datasource_option_keys.py 把 +# app/data 下所有 *_OPTION 常量扫出来核对,新加的配置项忘了进这份名单,那条测试先报 _NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION, - CATALOG_DRAFT_GUIDANCE_OPTION}) + CATALOG_DRAFT_GUIDANCE_OPTION, "introspect_scope", "catalog_maintenance"}) _MASK_SPLIT = re.compile(r"[,,、;;\n]+") diff --git a/backend/tests/test_datasource_option_keys.py b/backend/tests/test_datasource_option_keys.py new file mode 100644 index 00000000..27f47d6a --- /dev/null +++ b/backend/tests/test_datasource_option_keys.py @@ -0,0 +1,47 @@ +"""数据源 options 里 AgentLab 自己的配置项一律不交给驱动(data/engine._NON_DRIVER_OPTIONS)。 + +2026-10 部署上撞过:探查范围(introspect_scope,大库那一波加的,运行时按需加入表时也会自动写)、定时维护的步骤 +(catalog_maintenance)没进摘除名单,被当成驱动参数拼进连接串,MySQL 驱动 connect() 收到不认识的关键字参数,这个数据源上 +每一条查询都失败(「connect() got an unexpected keyword argument 'introspect_scope'」)。名单是手写的,以后再加配置项照样会 +漏,所以这里把 app/data 下所有 *_OPTION 常量(数据源 options 的键都这样声明)扫出来逐个核对:新加一项忘了进名单,这条测试先报。 +""" +from __future__ import annotations + +import re +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from app.data.engine import _NON_DRIVER_OPTIONS, build_url + +DATA_DIR = Path(__file__).resolve().parents[1] / "app" / "data" +_DECL = re.compile(r'^([A-Z_]*_OPTION)\s*(?::\s*\w+\s*)?=\s*"([^"]+)"', re.M) + + +def _declared() -> dict[str, str]: + out: dict[str, str] = {} + for path in sorted(DATA_DIR.glob("*.py")): + for name, key in _DECL.findall(path.read_text(encoding="utf-8")): + out[f"{path.stem}.{name}"] = key + return out + + +def test_every_declared_option_key_is_kept_away_from_the_driver(): + declared = _declared() + assert {"source_objects.SCOPE_OPTION", "source_maintenance.MAINTENANCE_OPTION"} <= set(declared), declared + missing = {where: key for where, key in declared.items() if key not in _NON_DRIVER_OPTIONS} + assert not missing, f"这些数据源配置项会被拼进连接串交给驱动:{missing}" + + +@pytest.mark.parametrize("kind", ["mysql", "postgresql", "oracle"]) +def test_app_options_never_reach_the_connection_url(kind): + keys = set(_declared().values()) | set(_NON_DRIVER_OPTIONS) + app_options = {key: {"mode": "rules"} if key in ("introspect_scope", "catalog_maintenance", "catalog_profile") else "x" + for key in keys} + src = SimpleNamespace(kind=kind, host="db.example", port=None, database="shop", username="reader", password=None, + options={"charset": "utf8mb4", **app_options}) + url = build_url(src) + assert "charset=utf8mb4" in url, "驱动参数照旧拼进去" + for key in keys: + assert f"{key}=" not in url, (key, url)