muse-agent-example/tests/集成/test_数据库底座隔离.py
zizi d909d1bd1b 后端实现与用例身份:19 包集成落地并修复收尾缺陷
实现侧:
- 上下文:任务范围拆分为 范围校验/范围授权;索引按可发现口径重建、索引新鲜度改对称差;依赖校验统一快照漂移说明。
- 知识方法:方法与材料读取口径统一;超限方法材料按可选省略,核对路径不再二次计费;删除无合同的读时重算。
- 任务运行:新增 context.usage/tool.denied 事件类型;连接池常驻并在装配生命周期内开关;调用结算与核对分列。
- 效果评测/审校修订/交付连载/作者经验/作品规划:凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。
- 资源加载:能力正文不再夹带索引用的导航注记(该注记此前进入角色与技能的模型提示)。
- 元数据:受保护骨架与代码保护属性对齐;字段校验与内置结构口径同步。
- 基础设施:环境预检进入装配生命周期;数据库连接运行期字段不参与相等比较;索引指纹归一化 jsonb 浮点。
- 删除被替代实现:7 份旧提示词模板与空壳 资料来源 读取器。

用例侧:
- 用例身份与导航元信息迁移;夹具补生命周期、同库暴露与模板封存;
- 本轮定向修复:方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
2026-09-18 01:15:00 +08:00

56 lines
2.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""获准隔离 PG 后验证模板保护、独立提交与角色,不使用共享事务替身。"""
import psycopg
import pytest
from psycopg.conninfo import conninfo_to_dict, make_conninfo
from muse.共享.调用身份 import 用途
pytestmark = pytest.mark.数据库
@pytest.mark.case_id("NC-database-baseline-pg-01")
def test_克隆保留真实提交且后例没有前例数据(数据库底座, tmp_path):
with 数据库底座.借库(tmp_path / "first", (1,)) as 一:
with 一[用途.维护].连接() as 连:
连.execute("CREATE TABLE public.clone_probe (id serial PRIMARY KEY, value text)")
连.execute("INSERT INTO public.clone_probe(value) VALUES ('first')")
with 一[用途.维护].连接() as 连:
assert 连.execute("SELECT id, value FROM public.clone_probe").fetchone() == (1, "first")
第一库 = 连.info.dbname
with 数据库底座.借库(tmp_path / "second", (1,)) as 二:
with 二[用途.维护].连接() as 连:
assert 连.info.dbname != 第一库
assert 连.execute("SELECT to_regclass('public.clone_probe')").fetchone() == (None,)
连.execute("CREATE TABLE public.clone_probe (id serial PRIMARY KEY, value text)")
assert 连.execute(
"INSERT INTO public.clone_probe(value) VALUES ('second') RETURNING id"
).fetchone() == (1,)
for 用途值, 角色 in [(用途.生产, "muse_app"), (用途.评测, "muse_eval")]:
with 二[用途值].连接() as 连:
assert 连.execute("SELECT current_user").fetchone() == (角色,)
# 此基线本 session 无论消费多少例都只执行一轮,clone 不再执行迁移。
模板 = 数据库底座.基线((1,))
assert (
len([e for e in 数据库底座.事件 if e["action"] == "ddl" and e["database"] == 模板.名称])
== 1
)
@pytest.mark.case_id("NC-database-baseline-pg-02")
def test_封存模板拒绝直接连接且失败用例精确清理(数据库底座, 隔离数据库URL, tmp_path):
模板 = 数据库底座.基线((1,))
参数 = conninfo_to_dict(隔离数据库URL)
参数.update(dbname=模板.名称, connect_timeout="2")
with pytest.raises(psycopg.OperationalError):
psycopg.connect(make_conninfo(**参数))
库名 = None
with pytest.raises(ValueError), 数据库底座.借库(tmp_path / "failed", (1,)) as 工厂:
with 工厂[用途.维护].连接() as 连:
库名 = 连.info.dbname
raise ValueError("模拟测试失败")
with psycopg.connect(隔离数据库URL) as 连:
assert (
连.execute("SELECT 1 FROM pg_database WHERE datname = %s", (库名,)).fetchone() is None
)