d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
"""真实驱动池的会话隔离与容量验证；只连接显式隔离测试库。"""
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Event
|
|
|
|
import pytest
|
|
|
|
from muse.共享.调用身份 import 用途
|
|
from muse.基础设施.数据库.连接 import 数据库工厂, 连接池耗尽
|
|
|
|
pytestmark = pytest.mark.数据库
|
|
|
|
|
|
@pytest.mark.case_id("NC-O06-POOL-PG-ISOLATION")
|
|
def test_真实池只读与自动提交失败后不串用(应用测试库):
|
|
provider = 数据库工厂(应用测试库[用途.生产].引用, 用途.生产, 常驻=True, 最大连接=1)
|
|
provider.打开()
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
with provider.连接(只读=True) as conn:
|
|
assert conn.execute("SHOW transaction_read_only").fetchone()[0] == "on"
|
|
assert conn.execute("SELECT current_user").fetchone()[0] == "muse_app"
|
|
raise ValueError("synthetic rollback")
|
|
with provider.连接(autocommit=True) as conn:
|
|
assert conn.autocommit
|
|
assert conn.execute("SHOW transaction_read_only").fetchone()[0] == "off"
|
|
assert (
|
|
conn.execute("SELECT current_setting('app.purpose')").fetchone()[0] == "production"
|
|
)
|
|
with provider.连接(只读=True, autocommit=True) as conn:
|
|
assert conn.execute("SHOW transaction_read_only").fetchone()[0] == "on"
|
|
finally:
|
|
provider.关闭()
|
|
|
|
|
|
@pytest.mark.case_id("NC-O06-POOL-PG-CAPACITY")
|
|
def test_真实池超容量按期限拒绝且释放后可再借(应用测试库):
|
|
provider = 数据库工厂(应用测试库[用途.生产].引用, 用途.生产, 常驻=True, 最大连接=1, 等待秒=1)
|
|
provider.打开()
|
|
持有 = Event()
|
|
释放 = Event()
|
|
|
|
def occupy():
|
|
with provider.连接() as conn:
|
|
conn.execute("SELECT 1")
|
|
持有.set()
|
|
assert 释放.wait(10)
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
future = executor.submit(occupy)
|
|
try:
|
|
assert 持有.wait(5)
|
|
with pytest.raises(连接池耗尽):
|
|
with provider.连接():
|
|
pytest.fail("容量一的池不得借出第二条")
|
|
finally:
|
|
释放.set()
|
|
future.result()
|
|
with provider.连接() as conn:
|
|
assert conn.execute("SELECT 1").fetchone()[0] == 1
|
|
finally:
|
|
provider.关闭()
|