From 7febf2bcdb8e08c88e27fc98b34f5ee82c70b9c3 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 12:06:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=A4=E7=A7=8D=E5=BA=93=E8=A1=8C=E4=B8=BA?= =?UTF-8?q?=E5=86=8D=E5=AF=B9=E9=BD=90=E4=B8=80=E8=BD=AE=EF=BC=88=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=20NUL=E3=80=81R-a4=E3=80=81R-a6=E3=80=81R-a7=E3=80=81?= =?UTF-8?q?R-a10=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 编排者的决定：PostgreSQL 下文本列写入时去掉 NUL 字符（before_cursor_execute 钩子，只装在应用库引擎上，看所有字符串参数，没有 NUL 的原样交回），SQLite 不变；同 JSON 列的处理 - R-a4 应用库地址里的 sslmode= 换成 asyncpg 的 ssl=（取值同一套），sslrootcert 等证书参数、application_name 这类 asyncpg 不收的参数启动时就报错并说怎么写（证书用 PGSSLROOTCERT 等环境变量）；断点库连接串换回 sslmode= - R-a6 不分大小写的搜索换成 db/portable.contains_ci：SQLite 上同 ilike（lower LIKE lower），PostgreSQL 上模式带 COLLATE "default"（列是 COLLATE "C"，列上的 ILIKE 只认 ASCII，É/é、Ж/ж、全角Ａ/ａ 匹配不上）；运行列表按名字、对象索引按注释两处都改；改正 db/postgres 里那句说明 - R-a7 JSON 守卫改成编译层：tests/jsonguard 挂在应用库编译器上（conftest，两种库、跑全量时一直生效），整个 JSON 值出现在 = / != / IN / IS DISTINCT FROM / 大小比较、SELECT DISTINCT、GROUP BY、ORDER BY、UNION（去重的几种）里就报错；按类型认，变量别名、跨行 order_by、select(Run.origin).distinct()、JSON 下标的比较都抓得到，测试拿这些反例证明，json_eq、IS NULL、UNION ALL 不受影响。原来的逐行正则测试删掉 - R-a10 json_eq 两种库同一语义、按 JSON 类型比：字符串只等于字符串，数只等于数（1 等于 1.0），True / False 只等于 JSON 的 true / false（SQLite 上用 json_type 分开，以前 1 能匹配 true）；非标量 TypeError。json_text 两种库都是「字符串取本身、别的取 JSON 写法」（SQLite 上同 PostgreSQL 的 ->>），数 1 的 json_text 两边都是 '1' - db/postgres 说明里写清 NULL 排序补写的范围（只认直接按列排的） --- backend/app/api/runs.py | 16 ++-- backend/app/data/source_objects.py | 6 +- backend/app/db/portable.py | 91 +++++++++++++++--- backend/app/db/postgres.py | 37 +++++++- backend/app/db/urls.py | 28 +++++- backend/tests/conftest.py | 6 ++ backend/tests/jsonguard.py | 63 +++++++++++++ backend/tests/test_db_portable.py | 147 +++++++++++++++++++++++++---- 8 files changed, 342 insertions(+), 52 deletions(-) create mode 100644 backend/tests/jsonguard.py diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index bd87fc14..b2b96eb8 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -213,7 +213,7 @@ async def list_runs( """ stmt = _filtered(select(Run), workflow_id=workflow_id, status=status, run_class=run_class, q=q, schedule_id=schedule_id, origin_kind=origin_kind) - stmt = stmt.order_by(Run.created_at.desc()).limit(limit) + stmt = stmt.order_by(Run.created_at.desc(), Run.id.desc()).limit(limit) if before is not None: stmt = stmt.where(Run.created_at < before) return list((await session.execute(stmt)).scalars()) @@ -265,7 +265,10 @@ def _filtered(stmt: Any, *, workflow_id: str | None, status: str | None, run_cla if classes := _csv(run_class): stmt = stmt.where(Run.run_class.in_(classes)) if q and q.strip(): - stmt = stmt.where(Run.workflow_name.ilike(f"%{_like(q.strip())}%", escape="\\")) + # 不分大小写（db/portable.contains_ci）：PostgreSQL 的文本列按字节排序（COLLATE "C"），列上的 ILIKE 只认 ASCII 的大小写 + from app.db.portable import contains_ci + + stmt = stmt.where(contains_ci(Run.workflow_name, q.strip())) if schedule_id: stmt = stmt.where(Run.schedule_id == schedule_id) if origins := _csv(origin_kind): @@ -281,11 +284,6 @@ def _csv(value: str | None) -> list[str]: return [v.strip() for v in (value or "").split(",") if v.strip()] -def _like(text: str) -> str: - """用户输入里的 % 和 _ 按字面匹配，不当通配符。""" - return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") - - @router.get("/{run_id}", response_model=RunOut) async def get_run(run_id: str, session: AsyncSession = Depends(get_session)) -> RunOut: run = await session.get(Run, run_id) @@ -343,7 +341,7 @@ async def list_artifacts( ) -> list[dict[str, Any]]: """这次运行产出的全部工件引用。审阅出具物时从这里下钻到完整证据。""" rows = await session.execute( - select(Artifact).where(Artifact.run_id == run_id).order_by(Artifact.created_at) + select(Artifact).where(Artifact.run_id == run_id).order_by(Artifact.created_at, Artifact.id) ) return [ { @@ -753,7 +751,7 @@ async def list_approvals( .join(Task, Task.approval_id == Approval.id, isouter=True) # 别人的个人应用刷新、个人蓝图试运行上的审批不列（只给拥有者，同运行列表） .where(or_(Run.id.is_(None), not_(hidden_runs(current_principal())))) - .order_by(Approval.created_at.desc()) + .order_by(Approval.created_at.desc(), Approval.id.desc()) .limit(limit) ) if status != "all" and (statuses := _csv(status)): diff --git a/backend/app/data/source_objects.py b/backend/app/data/source_objects.py index 65d78ee5..9d58d1b8 100644 --- a/backend/app/data/source_objects.py +++ b/backend/app/data/source_objects.py @@ -47,6 +47,7 @@ from app.core.errors import describe_exception, first_line from app.data.names import name_key from app.db.base import SessionLocal, utcnow from app.db.models_sources import SourceObject +from app.db.portable import contains_ci logger = logging.getLogger(__name__) @@ -886,9 +887,8 @@ async def search_index(session: AsyncSession, row: Any, query: str, *, limit: in return [], 0 key = name_key(text.rsplit(".", 1)[-1]) filt = (*_scope_filter(row.id, index_schema(row)), SourceObject.removed_at.is_(None), - # 名字比的是 name_key（ASCII 已转小写）；注释要不分大小写：SQLite 的 LIKE 本来就不分（ASCII），PostgreSQL 分 - or_(SourceObject.name_key.contains(key, autoescape=True), - SourceObject.comment.icontains(text, autoescape=True))) + # 名字比的是 name_key（ASCII 已转小写）；注释要不分大小写（db/portable.contains_ci：两种库都折 Unicode 的大小写） + or_(SourceObject.name_key.contains(key, autoescape=True), contains_ci(SourceObject.comment, text))) total = int((await session.execute(select(func.count()).select_from(SourceObject).where(*filt))).scalar() or 0) rows = (await session.execute( select(SourceObject.name, SourceObject.name_key, SourceObject.kind, SourceObject.comment, diff --git a/backend/app/db/portable.py b/backend/app/db/portable.py index 4d643d1d..054df487 100644 --- a/backend/app/db/portable.py +++ b/backend/app/db/portable.py @@ -2,12 +2,14 @@ 只给应用库（agentlab.db / AGENTLAB_DATABASE_URL）用；发给业务数据源的 SQL 不走这里。 -**按 JSON 列里的键筛选。** SQLite 用 json_extract：取出来是 SQL 标量（字符串是文本、数字是数字、true 是 1、 -没有这个键或值为 null 是 NULL），和 Python 值比是带类型的比较（数字 3 不等于字符串 "3"）。PostgreSQL 的 JSON 列是 -存原文的 json（db/base.PortableJSON），用 -> / ->> 取，要比较的只把取出来的那一小段转成 JSONB： +**按 JSON 列里的键筛选。** 两种库同一套语义，按 JSON 的类型比（SQLite 的 json_extract 单独用时不分：true 取出来 +是 1，和 1 比相等；PostgreSQL 的 JSON 列是存原文的 json，db/base.PortableJSON，比较时只把取出来的那一小段转成 JSONB）： -- json_eq：PostgreSQL 上按 JSONB 相等比（`(col -> 'k')::jsonb = '3'::jsonb`），同样带类型，和 SQLite 一致； -- json_text：只要文本（字符串原样、数字是它的写法），PostgreSQL 上是 ->>； +- json_eq(列, 键, 值)：字符串只等于同样的字符串；数只等于数（1 和 1.0 相等）；True / False 只等于 JSON 的 true / false + （不等于 1 / 0）；None 是「没有这个键或值为 null」。只收这几种标量； +- json_text：这个键的值当文本——字符串是它本身，数、true / false、数组、对象是它的 JSON 写法（同 PostgreSQL 的 ->>， + SQLite 上用 json_type 分开、非字符串取 -> 的写法），没有这个键或值为 null 是 NULL。所以数 1 的 json_text 是 '1'， + 两种库都是； - json_flag：标记位「真」（true、非零的数）：没有这个键、null、false、0 都算假； - json_is_string：这个键的值是不是字符串（SQLite 的 json_type 叫 text，PostgreSQL 的 jsonb_typeof 叫 string）。 @@ -16,6 +18,11 @@ **插入，撞了唯一键就什么都不做。** insert_ignore 按会话连着的库挑 sqlite / postgresql 的 insert，两边的 on_conflict_do_nothing 是同一个接口。 +**不分大小写地找子串。** contains_ci：SQLite 上是 lower(列) LIKE lower(模式)（同 SQLAlchemy 的 ilike；大小写怎么折 +看 SQLite 的构建，本机 conda 的 3.53 连 É/é、全角Ａ/ａ 都认）。PostgreSQL 上应用库的文本列是 COLLATE "C"（db/postgres.py）， +列上的 ILIKE、lower() 只认 ASCII 的大小写；所以模式带 COLLATE "default"，按建库时的区域设置折大小写（en_US.UTF-8、 +C.UTF-8 都认 Unicode；建库用的是纯 C 区域的话只认 ASCII）。用户输入里的 %、_、反斜杠按字面匹配。 + **撞的是哪个唯一约束。** unique_violation 认 IntegrityError：SQLite 的报错是「UNIQUE constraint failed: 表.列」， PostgreSQL（asyncpg）的是 23505，表名、DETAIL 里的「Key (列)=」分别带着。不靠约束名（命名规则改了也认得出）。 """ @@ -64,12 +71,15 @@ class json_text(FunctionElement[str]): class _JsonEq(FunctionElement[bool]): + """kind 是 text / number / true / false，写进语句（语句缓存按它区分）；true / false 不用绑定参数。""" + type = Boolean() inherit_cache = True name = "json_eq" - def __init__(self, column: Any, key: str, value: Any) -> None: - super().__init__(column, _key(key), bindparam(None, value, type_=_JsonScalar())) + def __init__(self, column: Any, key: str, kind: str, value: Any = None) -> None: + extra = [] if kind in ("true", "false") else [bindparam(None, value, type_=_JsonScalar())] + super().__init__(column, _key(key), literal_column(kind), *extra) class _JsonNull(FunctionElement[bool]): @@ -82,10 +92,18 @@ class _JsonNull(FunctionElement[bool]): def json_eq(column: Any, key: str, value: Any) -> Any: - """JSON 列顶层键的值等于 value（带类型的比较，见模块说明）。value 为 None 时是「没有这个键或值为 null」。 + """JSON 列顶层键的值等于 value（按 JSON 的类型比，见模块说明）。value 为 None 时是「没有这个键或值为 null」。 - None 用另一个表达式：语句缓存不按绑定参数的值区分，同一个表达式不能因为值是 None 编译成两样。""" - return _JsonNull(column, key) if value is None else _JsonEq(column, key, value) + 值的种类写进语句：语句缓存不按绑定参数的值区分，同一个表达式不能因为值的种类不同编译成两样。""" + if value is None: + return _JsonNull(column, key) + if isinstance(value, bool): + return _JsonEq(column, key, "true" if value else "false") + if isinstance(value, (int, float)): + return _JsonEq(column, key, "number", value) + if isinstance(value, str): + return _JsonEq(column, key, "text", value) + raise TypeError(f"json_eq 只比标量（字符串、数、布尔、None），不收 {type(value).__name__}") class json_flag(FunctionElement[bool]): @@ -120,14 +138,23 @@ def _parts(element: Any, compiler: Any, **kw: Any) -> tuple[str, str, list[Any]] @compiles(json_text) def _json_text_sqlite(element: Any, compiler: Any, **kw: Any) -> str: + # 同 PostgreSQL 的 ->>：字符串取它本身，别的取 JSON 写法（-> 要 SQLite 3.38 以上） col, key, _ = _parts(element, compiler, **kw) - return f"json_extract({col}, '$.{key}')" + path = f"'$.{key}'" + return (f"(CASE json_type({col}, {path}) WHEN 'text' THEN json_extract({col}, {path}) WHEN 'null' THEN NULL " + f"ELSE ({col} -> {path}) END)") + + +_SQLITE_KIND = {"text": "= 'text'", "number": "IN ('integer', 'real')", "true": "= 'true'", "false": "= 'false'"} @compiles(_JsonEq) def _json_eq_sqlite(element: Any, compiler: Any, **kw: Any) -> str: - col, key, (value,) = _parts(element, compiler, **kw) - return f"(json_extract({col}, '$.{key}') = {compiler.process(value, **kw)})" + col, key, (kind, *value) = _parts(element, compiler, **kw) + typed = f"json_type({col}, '$.{key}') {_SQLITE_KIND[kind.name]}" + if not value: + return f"({typed})" + return f"({typed} AND json_extract({col}, '$.{key}') = {compiler.process(value[0], **kw)})" @compiles(_JsonNull) @@ -164,8 +191,10 @@ def _json_text_pg(element: Any, compiler: Any, **kw: Any) -> str: @compiles(_JsonEq, "postgresql") def _json_eq_pg(element: Any, compiler: Any, **kw: Any) -> str: - col, key, (value,) = _parts(element, compiler, **kw) - return f"({_got(col, key)} = CAST({compiler.process(value, **kw)} AS JSONB))" + # JSONB 的相等本来就按类型：'1' = '1.0'，'1' ≠ 'true'，'"1"' ≠ '1' + col, key, (kind, *value) = _parts(element, compiler, **kw) + right = f"'{kind.name}'::jsonb" if not value else f"CAST({compiler.process(value[0], **kw)} AS JSONB)" + return f"({_got(col, key)} = {right})" @compiles(_JsonNull, "postgresql") @@ -187,6 +216,38 @@ def _json_is_string_pg(element: Any, compiler: Any, **kw: Any) -> str: return f"(jsonb_typeof({_got(col, key)}) = 'string')" +# ---- 不分大小写的子串 ---- + + +def _escape_like(text: str) -> str: + return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +class _ContainsCI(FunctionElement[bool]): + type = Boolean() + inherit_cache = True + name = "contains_ci" + + +def contains_ci(column: Any, text: str) -> Any: + """列里含有 text，不分大小写（见模块说明）。""" + return _ContainsCI(column, bindparam(None, f"%{_escape_like(text)}%", type_=String())) + + +@compiles(_ContainsCI) +def _contains_ci_sqlite(element: Any, compiler: Any, **kw: Any) -> str: + column, pattern = list(element.clauses) + return (f"(lower({compiler.process(column, **kw)}) LIKE lower({compiler.process(pattern, **kw)}) " + "ESCAPE '\\')") + + +@compiles(_ContainsCI, "postgresql") +def _contains_ci_pg(element: Any, compiler: Any, **kw: Any) -> str: + column, pattern = list(element.clauses) + return (f"({compiler.process(column, **kw)} ILIKE ({compiler.process(pattern, **kw)} COLLATE \"default\") " + "ESCAPE '\\')") + + # ---- 插入，撞了唯一键就跳过 ---- diff --git a/backend/app/db/postgres.py b/backend/app/db/postgres.py index 6a15ab24..3921c60b 100644 --- a/backend/app/db/postgres.py +++ b/backend/app/db/postgres.py @@ -12,13 +12,17 @@ VARCHAR(n) 超长直接报错、INTEGER 只有 32 位。模型里的 String(n) **文本怎么比大小、怎么排序。** SQLite 缺省按字节比（BINARY）；PostgreSQL 按建库时的区域设置（常见的 en_US.UTF-8 不分 大小写地先比字母、忽略标点），同一列排出来的先后、`key < '…'` 这种比较的结果都会变。应用库上的文本列建表时一律带 COLLATE "C"（按字节比），和 SQLite 一致，不看建库时选了什么区域；LIKE 'xx%' 这种前缀匹配也因此用得上普通索引。 -ILIKE、lower() 照旧按库的字符集分大小写。 +代价是列上的 ILIKE、lower() 也按 C 的规则，只认 ASCII 的大小写（É/é、全角Ａ/ａ 不认）：要不分大小写地找，用 +db/portable.contains_ci，模式带 COLLATE "default"，按建库时的区域设置折大小写。 **ORDER BY 里 NULL 的位置。** SQLite 升序 NULL 在前、降序 NULL 在后，PostgreSQL 正好相反。应用里两百多处排序， 可空的列（开始时间、到期时间、发布时间……）排出来的先后要和 SQLite 一致：可空列上没写 nulls_first / nulls_last 的 ASC 补 NULLS FIRST、DESC 补 NULLS LAST，ORDER BY 里直接写列名（隐含 ASC）的同样补上。不可空的列不动—— -补了会让索引用不上，又没有 NULL 可排。窗口函数 OVER (ORDER BY …) 里同样补。原生 text() 里写的 ORDER BY 不经过 -这里，要自己写明（db/base 的去重语句里就写了）。 +补了会让索引用不上，又没有 NULL 可排。窗口函数 OVER (ORDER BY …) 里同样补。 + +只认直接写的列（Column，含它的 asc() / desc()）。这几种不补，两种库排出来的先后可能不同，要紧的地方自己写明 +nulls_first / nulls_last：label 出来的聚合（max(...) 之类）、coalesce / case 这类表达式、外连接进来的表上的列（模型里 +不可空，外连接没连上时却是 NULL）、原生 text() 里的 ORDER BY（db/base 的去重语句里写明了）。 """ from __future__ import annotations @@ -83,6 +87,33 @@ class AppCompiler(PGCompiler_asyncpg): return text +def _has_nul(value: Any) -> bool: + return isinstance(value, str) and "\x00" in value + + +def _clean_row(params: Any) -> Any: + if isinstance(params, dict): + return {k: v.replace("\x00", "") if _has_nul(v) else v for k, v in params.items()} \ + if any(_has_nul(v) for v in params.values()) else params + if isinstance(params, (list, tuple)) and any(_has_nul(v) for v in params): + return type(params)(v.replace("\x00", "") if _has_nul(v) else v for v in params) + return params + + +def strip_nul(conn: Any, cursor: Any, statement: str, parameters: Any, context: Any, + executemany: bool) -> tuple[str, Any]: + """写进 PostgreSQL 之前去掉文本参数里的 NUL 字符（before_cursor_execute，retval=True）。 + + PostgreSQL 的 text / varchar 不收 \\x00（报 invalid byte sequence），SQLite 照存：一段带 NUL 的模型输出、一个从二进制 + 文件里粘出来的问题，在 SQLite 上写得进去，在 PostgreSQL 上整条写入失败。所以和 JSON 列（db/base.pg_json_dumps）一样 + 去掉。只看参数里的字符串，bytes（LargeBinary）不动；没有 NUL 的参数原样交回，不多一次拷贝。""" + if executemany: + if any(_clean_row(row) is not row for row in parameters): + return statement, [_clean_row(row) for row in parameters] + return statement, parameters + return statement, _clean_row(parameters) + + def install(dialect: Any) -> None: """装到应用库引擎的方言实例上。""" dialect.statement_compiler = AppCompiler diff --git a/backend/app/db/urls.py b/backend/app/db/urls.py index 62064ac9..cf7f16e1 100644 --- a/backend/app/db/urls.py +++ b/backend/app/db/urls.py @@ -15,6 +15,13 @@ _CHECKPOINT_SCHEMES = {"postgresql", "postgres", "postgresql+psycopg", "postgres #: asyncpg 和 libpq 对 SSL 的参数名不一样：asyncpg 是 ssl=require，libpq 是 sslmode=require（取值同一套） _ASYNCPG_TO_LIBPQ = {"ssl": "sslmode"} +#: libpq 写法的 SSL 证书参数：asyncpg 不收这几个连接参数（SQLAlchemy 原样交过去就是 TypeError），改用同名的环境变量 +_LIBPQ_SSL_FILES = {"sslrootcert": "PGSSLROOTCERT", "sslcert": "PGSSLCERT", "sslkey": "PGSSLKEY", + "sslcrl": "PGSSLCRL", "sslpassword": "PGSSLPASSWORD"} +_SSL_MODES = {"disable", "allow", "prefer", "require", "verify-ca", "verify-full"} +#: 应用库地址里认的参数。SQLAlchemy 把地址里的参数原样交给 asyncpg.connect，它不认的（application_name、connect_timeout +#: 这类 libpq 写法）连的时候才报 TypeError；这里在启动时就说清楚 +_APP_PARAMS = {"ssl", "prepared_statement_cache_size", "target_session_attrs", "async_fallback"} #: 只给 SQLAlchemy / asyncpg 看的参数，libpq 不认（psycopg 连的时候会报「invalid connection option」） _ASYNCPG_ONLY = {"prepared_statement_cache_size", "prepared_statement_name_func", "command_timeout", "async_fallback"} @@ -32,9 +39,26 @@ def _parse(raw: str, schemes: set[str], example: str) -> URL: def normalize_database_url(raw: str) -> str: - """AGENTLAB_DATABASE_URL → postgresql+asyncpg://…（密码原样保留）。认不出的写法抛 ValueError。""" + """AGENTLAB_DATABASE_URL → postgresql+asyncpg://…（密码原样保留）。认不出的写法抛 ValueError。 + + SSL：libpq 的 sslmode= 换成 asyncpg 的 ssl=（取值同一套：disable / allow / prefer / require / verify-ca / verify-full）； + 证书文件（sslrootcert 之类）asyncpg 不收连接参数，要用 PGSSLROOTCERT 等环境变量（asyncpg 和断点库的 libpq 都认）。""" url = _parse(raw, _APP_SCHEMES, "postgresql+asyncpg://用户:密码@主机:5432/库名") - return url.set(drivername="postgresql+asyncpg").render_as_string(hide_password=False) + query = dict(url.query) + if files := [k for k in query if k in _LIBPQ_SSL_FILES]: + raise ValueError("地址里不能写 " + "、".join(files) + "：asyncpg 不收这几个参数，改设环境变量 " + + "、".join(_LIBPQ_SSL_FILES[k] for k in files)) + if "sslmode" in query: + if "ssl" in query: + raise ValueError("ssl= 和 sslmode= 只写一个") + mode = str(query.pop("sslmode")) + if mode not in _SSL_MODES: + raise ValueError(f"认不出 sslmode={mode}（可选 {' / '.join(sorted(_SSL_MODES))}）") + query["ssl"] = mode + if other := sorted(k for k in query if k not in _APP_PARAMS): + raise ValueError("地址里的参数 " + "、".join(other) + " 不认（asyncpg 不收）；能写的只有 " + + "、".join(sorted(_APP_PARAMS | {"sslmode"}))) + return url.set(drivername="postgresql+asyncpg", query=query).render_as_string(hide_password=False) def normalize_checkpoint_url(raw: str) -> str: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index cd074b0a..0a51b2a2 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -71,6 +71,12 @@ os.environ.setdefault("AGENTLAB_SCHEDULER", "0") import pytest # noqa: E402 +# 整个 JSON 值做比较、去重、分组、排序时编译就报错（tests/jsonguard.py）：PostgreSQL 的 json 不支持，SQLite 的测试也拦下 +from app.db.base import engine as _guarded_engine # noqa: E402 +from tests.jsonguard import install as _json_guard # noqa: E402 + +_json_guard(_guarded_engine.sync_engine.dialect) + @pytest.fixture(scope="session", autouse=True) def _prepare_schema(): diff --git a/backend/tests/jsonguard.py b/backend/tests/jsonguard.py new file mode 100644 index 00000000..6dbb11e3 --- /dev/null +++ b/backend/tests/jsonguard.py @@ -0,0 +1,63 @@ +"""测试时挂在应用库编译器上的守卫：整个 JSON 值做比较、去重、分组、排序就报错（两种库都挂，跑全量时一直生效）。 + +PostgreSQL 的 json 没有相等和大小的运算符：`=`、`!=`、`IN`、`IS DISTINCT FROM`、`<` 之类，SELECT DISTINCT、GROUP BY、 +ORDER BY、UNION（去重的那几种）碰到 json 类型的值都报错；SQLite 照常执行。只在 PostgreSQL 上跑到才发现太晚：这里在编译 +那一刻就拦下，SQLite 的测试也拦得到。按表达式的类型认，不按写法认：跨行的 order_by、变量别名、 +`select(Run.origin).distinct()`、JSON 的下标（`Run.origin["k"] == 1`，取出来的还是 json）都算。按键筛选用 db/portable 的 +json_eq 等（类型是布尔、文本），不受影响。 +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import JSON +from sqlalchemy.sql import operators +from sqlalchemy.sql.elements import UnaryExpression + + +class JsonMisuse(AssertionError): + """整个 JSON 值被拿去比较、去重、分组或排序。""" + + +_COMPARE = {operators.eq, operators.ne, operators.in_op, operators.not_in_op, operators.is_distinct_from, + operators.is_not_distinct_from, operators.lt, operators.le, operators.gt, operators.ge} + + +def _json(element: Any) -> bool: + while isinstance(element, UnaryExpression) and element.modifier is not None: + element = element.element + return isinstance(getattr(element, "type", None), JSON) + + +class JsonGuard: + """混进编译器类里用：class Guarded(JsonGuard, 原来的编译器)。""" + + def visit_binary(self, binary: Any, **kw: Any) -> str: + if binary.operator in _COMPARE and (_json(binary.left) or _json(binary.right)): + raise JsonMisuse(f"整个 JSON 值做了比较（{binary.operator.__name__}）：PostgreSQL 的 json 没有这个运算符，" + "按键筛选用 db/portable.json_eq 等") + return super().visit_binary(binary, **kw) # type: ignore[misc] + + def visit_select(self, select: Any, **kw: Any) -> str: + if select._distinct and any(_json(c) for c in select.selected_columns): + raise JsonMisuse("SELECT DISTINCT 里有 JSON 列：PostgreSQL 的 json 没有相等运算符，去不了重") + if any(_json(c) for c in select._group_by_clauses): + raise JsonMisuse("GROUP BY 了 JSON 值：PostgreSQL 的 json 没有相等运算符") + if any(_json(c) for c in select._order_by_clauses): + raise JsonMisuse("ORDER BY 了 JSON 值：PostgreSQL 的 json 没有大小") + return super().visit_select(select, **kw) # type: ignore[misc] + + def visit_compound_select(self, cs: Any, **kw: Any) -> str: + keyword = str(getattr(cs.keyword, "value", cs.keyword)).upper() + if "ALL" not in keyword and any(_json(c) for c in cs.selected_columns): + raise JsonMisuse(f"{keyword or 'UNION'} 去重时碰到 JSON 列：PostgreSQL 的 json 没有相等运算符") + return super().visit_compound_select(cs, **kw) # type: ignore[misc] + + +def install(dialect: Any) -> None: + """挂到应用库引擎的方言实例上（PostgreSQL 时是 db/postgres 换过的编译器，SQLite 时是原来的）。""" + base = dialect.statement_compiler + if issubclass(base, JsonGuard): + return + dialect.statement_compiler = type(f"JsonGuarded{base.__name__}", (JsonGuard, base), {}) diff --git a/backend/tests/test_db_portable.py b/backend/tests/test_db_portable.py index 6374077f..1d0e134e 100644 --- a/backend/tests/test_db_portable.py +++ b/backend/tests/test_db_portable.py @@ -45,6 +45,21 @@ def test_every_migrated_column_is_in_the_models(): assert missing == [] +def test_every_not_null_migrated_column_gets_a_value_for_old_rows(): + """迁移补的列模型里是 NOT NULL 的，存量行不能留 NULL（PostgreSQL 的表照模型带 NOT NULL，导入时整行插不进）：要么补列时 + 带 DEFAULT 回填，要么登记在 _NOT_NULL_BACKFILLS 里每次启动按模型的缺省值填。""" + bad = [] + for table, column, ddl in db_base._COLUMN_MIGRATIONS: + col = Base.metadata.tables[table].c[column] + if col.nullable: + continue + if not re.search(r"\bDEFAULT\b", ddl, re.I) and (table, column) not in db_base._NOT_NULL_BACKFILLS: + bad.append((table, column)) + assert bad == [] + for table, column in db_base._NOT_NULL_BACKFILLS: + assert db_base.model_default(Base.metadata.tables[table].c[column])[0], (table, column) + + def test_checkpoint_tables_do_not_clash_with_app_tables(): """断点和应用表放在同一个库、同一个 schema：两边的表名不能撞。名字从两个 saver 自己的建表语句里取。""" import inspect as pyinspect @@ -58,22 +73,40 @@ def test_checkpoint_tables_do_not_clash_with_app_tables(): assert created & set(Base.metadata.tables) == set() -def test_no_query_compares_sorts_or_groups_a_whole_json_column(): - """PostgreSQL 的 json 没有相等运算符：整列做 ==、!=、in_、DISTINCT、ORDER BY、GROUP BY 都会报错（SQLite 不报）。 - 按里面某个键筛选走 db/portable 的 json_eq 等。扫一遍应用代码里「模型.JSON列」后面跟着这几样的写法。""" - from pathlib import Path +def test_the_compile_guard_catches_whole_json_values_however_they_are_written(): + """整个 JSON 值做比较、去重、分组、排序：编译时就报错（tests/jsonguard.py，conftest 挂在应用库引擎上，跑全量时一直生效）。 + 按表达式的类型认，逐行正则抓不到的写法也抓得到；按键筛选的 json_eq 等、IS NULL、UNION ALL 不受影响。""" + from sqlalchemy import union, union_all - from sqlalchemy import JSON + from tests.jsonguard import JsonMisuse - columns = sorted({f"{m.class_.__name__}.{a.key}" for m in Base.registry.mappers for a in m.column_attrs - if isinstance(a.columns[0].type, JSON)}) - names = "|".join(re.escape(c) for c in columns) - bad = re.compile(rf"(? bool: + return bool((await session.execute(select(func.count()).select_from(Run).where( + Run.workflow_name.like(f"大小写-{tag}-%"), contains_ci(Run.workflow_name, needle)))).scalar()) + + assert await hit("report") and await hit("REPORT_100%") + assert not await hit("report-100") and not await hit("100%%") + assert await hit("éclair") == bool(folds) and await hit("жук") == bool(folds) and await hit("ａｂ") == bool(folds) + assert await hit("Éclair") and await hit("Жук") + + async def test_timestamps_read_back_as_aware_utc(): from datetime import datetime, timedelta, timezone