d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
148 lines
6.1 KiB
Python
148 lines
6.1 KiB
Python
"""S04 只能持久化自己的结构定义，实例值由具名业务 owner 保存。"""
|
||
|
||
import ast
|
||
import re
|
||
from importlib.util import resolve_name
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-meta-owner-scope",
|
||
environment="离线架构检查",
|
||
given="当前元数据模块的导入与 SQL",
|
||
when="检查实际源码访问归属",
|
||
then=["仅持久化 metadata 自有定义，不导入业务内部或构造动态表名"],
|
||
contract="docs/系统架构/新版设计/接口契约/元数据投影与版本.md",
|
||
)
|
||
def test_元数据仅访问自有定义且无动态表名__a40020() -> None:
|
||
根 = Path(__file__).resolve().parents[2] / "src/muse/元数据"
|
||
数据表 = set()
|
||
for 文件 in 根.glob("*.py"):
|
||
问题, 表 = _检查元数据来源(文件.read_text(), 文件.name)
|
||
assert 问题 == [], f"{文件.name}: {问题}"
|
||
数据表.update(表)
|
||
assert "metadata.schema_version" in 数据表
|
||
assert "metadata.schema_binding" in 数据表
|
||
|
||
|
||
def _检查元数据来源(代码: str, 文件名: str) -> tuple[list[str], set[str]]:
|
||
树 = ast.parse(代码)
|
||
别名, 赋值 = {}, {}
|
||
问题, 数据表 = [], set()
|
||
for 节点 in ast.walk(树):
|
||
if isinstance(节点, ast.Import):
|
||
for 项 in 节点.names:
|
||
别名[项.asname or 项.name.split(".")[0]] = (
|
||
项.name if 项.asname else 项.name.split(".")[0]
|
||
)
|
||
if 项.name.startswith("muse.") and not 项.name.startswith(
|
||
("muse.共享.", "muse.元数据.")
|
||
):
|
||
问题.append("越界导入")
|
||
elif isinstance(节点, ast.ImportFrom):
|
||
来源 = (
|
||
resolve_name("." * 节点.level + (节点.module or ""), "muse.元数据")
|
||
if 节点.level
|
||
else (节点.module or "")
|
||
)
|
||
for 项 in 节点.names:
|
||
全名 = 来源 + "." + 项.name
|
||
别名[项.asname or 项.name] = 全名
|
||
if 全名.startswith("muse.") and not 全名.startswith(("muse.共享.", "muse.元数据.")):
|
||
问题.append("越界导入")
|
||
elif isinstance(节点, ast.Assign):
|
||
for 目标 in 节点.targets:
|
||
if isinstance(目标, ast.Name):
|
||
赋值[目标.id] = 节点.value
|
||
|
||
def 名称(节点, 已见=frozenset()):
|
||
if isinstance(节点, ast.Name):
|
||
if 节点.id in 赋值 and 节点.id not in 已见:
|
||
return 名称(赋值[节点.id], 已见 | {节点.id})
|
||
return 别名.get(节点.id, 节点.id)
|
||
if isinstance(节点, ast.Attribute):
|
||
return 名称(节点.value, 已见) + "." + 节点.attr
|
||
return ""
|
||
|
||
def 文本(节点, 已见=frozenset()):
|
||
if isinstance(节点, ast.Constant) and isinstance(节点.value, str):
|
||
return 节点.value
|
||
if isinstance(节点, ast.Name) and 节点.id in 赋值 and 节点.id not in 已见:
|
||
return 文本(赋值[节点.id], 已见 | {节点.id})
|
||
if isinstance(节点, ast.BinOp) and isinstance(节点.op, ast.Add):
|
||
左, 右 = 文本(节点.left, 已见), 文本(节点.right, 已见)
|
||
return 左 + 右 if 左 is not None and 右 is not None else None
|
||
if isinstance(节点, ast.JoinedStr):
|
||
return "".join(
|
||
值.value if isinstance(值, ast.Constant) else "{动态}" for 值 in 节点.values
|
||
)
|
||
return None
|
||
|
||
for 节点 in ast.walk(树):
|
||
if isinstance(节点, ast.Attribute):
|
||
来源 = 名称(节点)
|
||
if 来源.startswith("muse.") and not 来源.startswith(("muse.共享.", "muse.元数据.")):
|
||
问题.append("越界导入访问")
|
||
if isinstance(节点, ast.Call) and 名称(节点.func) == "psycopg.sql.Identifier":
|
||
问题.append("不得接受动态表列名")
|
||
值 = 文本(节点)
|
||
if 值 is None:
|
||
continue
|
||
表 = re.findall(r"\b(?:FROM|JOIN|INSERT\s+INTO|UPDATE(?!\s+SET))\s+([^\s(,;]+)", 值, re.I)
|
||
for 名 in 表:
|
||
if 文件名 != "存储.py":
|
||
问题.append("绕过S04存储归属")
|
||
if not re.fullmatch(r"metadata\.[a-z_][a-z0-9_]*", 名):
|
||
问题.append("越界或动态表名")
|
||
数据表.add(名)
|
||
return 问题, 数据表
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-o04-metadata-owner-counterexamples",
|
||
environment="离线，临时元数据Python源文件",
|
||
given="绝对/相对导入、SQL别名、拼接和动态表名的绕过形式",
|
||
when="使用与真实源码相同的AST归属检查",
|
||
then=["越界写与动态表名拒绝，合法metadata静态查询不误报"],
|
||
)
|
||
@pytest.mark.parametrize(
|
||
"代码",
|
||
[
|
||
"import muse.正文写作.存储 as business",
|
||
"from ..正文写作 import 存储 as business",
|
||
"from muse import 正文写作",
|
||
"import muse as m; m.正文写作.存储.写()",
|
||
"from psycopg.sql import Identifier as ID; ID('muse_work')",
|
||
"import psycopg.sql as s; s.Identifier('muse_work')",
|
||
"from psycopg import sql; ident=sql.Identifier; ident('muse_work')",
|
||
"sql = 'UP' + 'DATE muse_work SET payload=%s'",
|
||
"table='muse_work'; sql=f'UPDATE {table} SET payload=%s'",
|
||
"sql='DELETE FROM muse_work WHERE work_id=%s'",
|
||
],
|
||
ids=[
|
||
"import",
|
||
"relative",
|
||
"from-root",
|
||
"root-alias",
|
||
"identifier-import",
|
||
"identifier-alias",
|
||
"identifier-assignment",
|
||
"concat",
|
||
"fstring",
|
||
"delete",
|
||
],
|
||
)
|
||
def test_元数据归属门禁拒绝AST绕过(tmp_path, 代码):
|
||
文件 = tmp_path / "存储.py"
|
||
文件.write_text(代码)
|
||
assert _检查元数据来源(文件.read_text(), 文件.name)[0]
|
||
assert (
|
||
_检查元数据来源(
|
||
"sql='SELECT definition FROM metadata.schema_version WHERE type_id=%s'", "存储.py"
|
||
)[0]
|
||
== []
|
||
)
|
||
assert _检查元数据来源("sql='SELECT definition FROM metadata.schema_version'", "接口.py")[0]
|