713cc45c63
- 51 个增量迁移文件压扁为 V0001__基线.sql 完整快照（结构+种子+授权）， 等价门禁：旧链全量执行库与基线库 pg_dump 逐字节一致 - 剔除 pg_dump 固化的 public schema 超级用户归属断言（muse_maint 无权执行） - 测试删减至保留集：备份往返 2 + 预算 2 + 租约 2 + 基线建库 1 - 租约/预算夹具改共享库，消除按例克隆建库 - 修复共享库三类既有污染：账本注入残留（系统管理）、失败触发器残留（评测）、 建表残留（正式变更事务），发布包迁移文件名硬编码改动态核对 - 全量数据库验收 722 passed / 0 failed / 0 errors（main 基线为 24F+291E） - 运行手册登记基线模式改表流程与账本校验和同步
142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
"""预算的核心保护：并发不超额、重复调用不重复计费。
|
||
|
||
按简化决策删除其余变体用例；夹具从按例克隆改为共享库以消除建库开销。
|
||
"""
|
||
|
||
import uuid
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import UTC, datetime, timedelta
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from muse.任务运行.接口 import (
|
||
任务服务,
|
||
任务请求,
|
||
步骤处理器,
|
||
步骤结果,
|
||
步骤计划,
|
||
)
|
||
from muse.任务运行.预算管理 import (
|
||
任务预算计划,
|
||
角色预算,
|
||
预算不足,
|
||
预算状态冲突,
|
||
预算管理,
|
||
额度策略,
|
||
)
|
||
from muse.共享.调用身份 import 内容用途, 用途
|
||
from muse.基础设施.数据库.连接 import 数据库工厂
|
||
from muse.编排.接口 import 流程定义, 流程服务, 流程登记
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
|
||
|
||
@pytest.fixture
|
||
def 预算环境(数据库底座, monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
|
||
with 数据库底座.共享库(tmp_path / "连接引用") as 工厂:
|
||
for 声明, 当前 in 工厂.items():
|
||
monkeypatch.setenv(f"MUSE_BUDGET_{声明.name}_URL", Path(当前.引用.位置).read_text())
|
||
yield 工厂
|
||
|
||
|
||
def 新任务(工厂: 数据库工厂):
|
||
登记 = 流程登记()
|
||
登记.登记处理器(步骤处理器("call", "1", lambda _: 步骤结果({}), "v1", "v1"))
|
||
登记.登记类型("budget-test", 必需保护=())
|
||
运行 = 任务服务(工厂, 登记)
|
||
编排 = 流程服务(运行, 登记)
|
||
编排.发布(流程定义("budget-test", "1", (步骤计划("call", "call", "1"),)))
|
||
请求 = 任务请求(
|
||
"budget-test",
|
||
uuid.uuid4().hex,
|
||
"author",
|
||
工厂.用途,
|
||
内容用途.生成,
|
||
{},
|
||
"policy-1",
|
||
"release-1",
|
||
{
|
||
"source_scope": {},
|
||
"schema_versions": {},
|
||
"authorization": "grant",
|
||
"budget": {"approved": True},
|
||
"stop_conditions": ["cancelled"],
|
||
},
|
||
)
|
||
身份 = 运行.创建任务(请求, "budget-test", "1")
|
||
领取 = 运行.领取步骤("worker", ["call"])
|
||
assert 领取 is not None and 领取.任务ID == 身份
|
||
return 身份, 领取
|
||
|
||
|
||
def 建预算(工厂: 数据库工厂, 任务ID: str, *, 窗口金额="24", 单次="6", 次数=3, 窗口次数=6000):
|
||
管理 = 预算管理(工厂, "quota")
|
||
管理.登记策略(额度策略("quota", "1", Decimal(窗口金额), 窗口次数))
|
||
管理.登记任务预算(
|
||
任务ID,
|
||
任务预算计划(
|
||
Decimal(单次) * 次数,
|
||
(角色预算("writer", 次数, 次数, Decimal(单次)),),
|
||
"approval",
|
||
datetime.now(UTC) + timedelta(minutes=10),
|
||
),
|
||
)
|
||
return 管理
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-budget-concurrent-reserve",
|
||
environment="隔离 PostgreSQL；配置验证器为显式协议替身",
|
||
given="两个真实任务共用10额度账户",
|
||
when="两个独立连接同时预留6",
|
||
then=["只有一笔成功且余额为4"],
|
||
contract="docs/系统架构/新版设计/模块设计/S02-任务运行.md",
|
||
)
|
||
def test_并发预留共享窗口且不超额__a61001(预算环境) -> None:
|
||
from threading import Barrier
|
||
|
||
A, 领取A = 新任务(预算环境[用途.生产])
|
||
B, 领取B = 新任务(预算环境[用途.生产])
|
||
预算A = 建预算(预算环境[用途.生产], A, 窗口金额="10")
|
||
预算B = 建预算(预算环境[用途.生产], B, 窗口金额="10")
|
||
栅栏 = Barrier(2)
|
||
|
||
def 预留(输入):
|
||
管理, 领取 = 输入
|
||
栅栏.wait()
|
||
try:
|
||
return 管理.预留(领取, 领取.任务ID, "writer")
|
||
except 预算不足:
|
||
return None
|
||
|
||
with ThreadPoolExecutor(max_workers=2) as 线程:
|
||
结果 = list(线程.map(预留, [(预算A, 领取A), (预算B, 领取B)]))
|
||
assert sum(r is not None for r in 结果) == 1
|
||
assert 预算A.窗口余额()["在途预留"] == Decimal("6")
|
||
assert 预算A.窗口余额()["可用金额"] == Decimal("4")
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-budget-idempotent-settlement",
|
||
environment="隔离 PostgreSQL；配置验证器为显式协议替身",
|
||
given="同调用身份和失败调用可信成本回执",
|
||
when="重复预留、外发申请与结算",
|
||
then=["只授权外发一次、只计费一次，冲突成本拒绝"],
|
||
contract="docs/系统架构/新版设计/模块设计/S02-任务运行.md",
|
||
)
|
||
def test_重复预留发送和结算不重复计费__a61002(预算环境) -> None:
|
||
身份, 领取 = 新任务(预算环境[用途.生产])
|
||
预算 = 建预算(预算环境[用途.生产], 身份)
|
||
初次 = 预算.预留(领取, "call-1", "writer")
|
||
assert 预算.预留(领取, "call-1", "writer") == 初次
|
||
assert 预算.标记已发送(领取, "call-1", 最长秒=30).允许外发
|
||
assert not 预算.标记已发送(领取, "call-1", 最长秒=30).允许外发
|
||
结果 = 预算.结算("call-1", Decimal("0.125"), 回执ID="failed-call-receipt")
|
||
assert 预算.结算("call-1", Decimal("0.125"), 回执ID="failed-call-receipt") == 结果
|
||
with pytest.raises(预算状态冲突):
|
||
预算.结算("call-1", Decimal("0.2"), 回执ID="changed")
|
||
assert 预算.窗口余额()["已知成本"] == Decimal("0.125")
|
||
assert 预算.窗口余额()["已占次数"] == 1
|