mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
数据源的探查范围、定时维护不再被拼进连接串:data/engine._NON_DRIVER_OPTIONS 补上 introspect_scope、catalog_maintenance。以前漏在名单外,设过探查范围(运行时按需把范围外的表加进来时也会自动写)或定时维护的 MySQL / PostgreSQL / Oracle 数据源,驱动 connect() 收到不认识的关键字参数,每一条查询都失败;加守卫测试,把 app/data 下所有 *_OPTION 常量扫出来逐个核对都不进连接串
This commit is contained in:
@@ -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]+")
|
||||
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user