d909d1bd1b
实现侧： - 上下文：任务范围拆分为 范围校验/范围授权；索引按可发现口径重建、索引新鲜度改对称差；依赖校验统一快照漂移说明。 - 知识方法：方法与材料读取口径统一；超限方法材料按可选省略，核对路径不再二次计费；删除无合同的读时重算。 - 任务运行：新增 context.usage/tool.denied 事件类型；连接池常驻并在装配生命周期内开关；调用结算与核对分列。 - 效果评测/审校修订/交付连载/作者经验/作品规划：凭据冻结、标定消费、导出补证、事实引文核对等收尾修复。 - 资源加载：能力正文不再夹带索引用的导航注记（该注记此前进入角色与技能的模型提示）。 - 元数据：受保护骨架与代码保护属性对齐；字段校验与内置结构口径同步。 - 基础设施：环境预检进入装配生命周期；数据库连接运行期字段不参与相等比较；索引指纹归一化 jsonb 浮点。 - 删除被替代实现：7 份旧提示词模板与空壳 资料来源 读取器。 用例侧： - 用例身份与导航元信息迁移；夹具补生命周期、同库暴露与模板封存； - 本轮定向修复：方法材料省略、事实引文、迁移回执、额度与暂停用例、慢用例超时预算等。
216 lines
7.9 KiB
Python
216 lines
7.9 KiB
Python
"""合成持久事件页驱动真实SSE迭代器；不建立网络、数据库或真实浏览器。"""
|
||
|
||
import asyncio
|
||
import json
|
||
from datetime import UTC, datetime
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
from muse.任务运行.模型 import 事件类型, 事件续接, 运行事件
|
||
from muse.接入.http import 事件订阅
|
||
from muse.接入.http.作者会话 import 要求作者
|
||
from muse.接入.http.错误响应 import 安装错误响应
|
||
|
||
|
||
def 事件(序号):
|
||
return 运行事件(
|
||
f"event-{序号}",
|
||
"task",
|
||
None,
|
||
None,
|
||
序号,
|
||
事件类型.步骤完成,
|
||
datetime(2026, 9, 17, tzinfo=UTC),
|
||
1,
|
||
{"step": f"step-{序号}"},
|
||
)
|
||
|
||
|
||
class 合成事件服务:
|
||
def __init__(self, 页集):
|
||
self.页集 = 页集
|
||
self.调用 = []
|
||
|
||
def 续接事件(self, task_id, cursor, *, 数量=100):
|
||
self.调用.append((task_id, cursor, 数量))
|
||
assert cursor in self.页集, f"意外续接游标：{cursor}"
|
||
return self.页集[cursor]
|
||
|
||
|
||
def 装配替身(monkeypatch, 服务):
|
||
授权 = []
|
||
monkeypatch.setattr(事件订阅, "取得任务服务", lambda _: 服务)
|
||
monkeypatch.setattr(
|
||
事件订阅,
|
||
"读取作者任务",
|
||
lambda request, task_id, work_id=None: 授权.append((task_id, work_id)),
|
||
)
|
||
return 授权
|
||
|
||
|
||
def 读流(
|
||
monkeypatch, 服务, *, protocol="bounded-v2", cursor=0, last_event_id=None, 断开检查次数=10
|
||
):
|
||
授权 = 装配替身(monkeypatch, 服务)
|
||
检查 = []
|
||
等待 = []
|
||
|
||
async def 已断开():
|
||
检查.append(True)
|
||
return len(检查) > 断开检查次数
|
||
|
||
async def 不等待(秒):
|
||
等待.append(秒)
|
||
|
||
monkeypatch.setattr(事件订阅.asyncio, "sleep", 不等待)
|
||
|
||
async def 收集():
|
||
响应 = await 事件订阅.订阅事件(
|
||
"task",
|
||
SimpleNamespace(is_disconnected=已断开),
|
||
"author",
|
||
cursor=cursor,
|
||
work_id="work",
|
||
last_event_id=last_event_id,
|
||
protocol=protocol,
|
||
)
|
||
assert 响应.media_type == "text/event-stream"
|
||
assert 响应.headers["cache-control"] == "no-cache"
|
||
return [块 async for 块 in 响应.body_iterator]
|
||
|
||
return asyncio.run(收集()), 授权, 等待, 检查
|
||
|
||
|
||
def 拆块(块):
|
||
return dict(行.split(": ", 1) for 行 in 块.strip().splitlines() if ": " in 行)
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-001")
|
||
def test_v2逐页发完最终序号才收尾__o08101(monkeypatch):
|
||
服务 = 合成事件服务(
|
||
{
|
||
0: 事件续接((事件(1), 事件(2)), False, 1, 3, 3),
|
||
2: 事件续接((事件(3),), False, 1, 3, 3),
|
||
}
|
||
)
|
||
块, 授权, 等待, 检查 = 读流(monkeypatch, 服务)
|
||
数据 = [拆块(项) for 项 in 块]
|
||
assert [项["event"] for 项 in 数据] == ["task-event"] * 3 + ["stream-complete"]
|
||
assert [项["id"] for 项 in 数据[:3]] == ["1", "2", "3"]
|
||
assert [json.loads(项["data"])["sequence"] for 项 in 数据[:3]] == [1, 2, 3]
|
||
assert json.loads(数据[-1]["data"]) == {"execution_final_sequence": 3}
|
||
assert 服务.调用 == [("task", 0, 100), ("task", 2, 100)]
|
||
assert 授权 == [("task", "work")]
|
||
assert 等待 == [1] and len(检查) == 2
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-002")
|
||
@pytest.mark.parametrize("cursor,last_event_id", [(3, None), (1, 3), (3, 1)])
|
||
def test_v2最终游标重连直接结束__o08102(monkeypatch, cursor, last_event_id):
|
||
服务 = 合成事件服务({3: 事件续接((), False, 1, 3, 3)})
|
||
块, _, 等待, _ = 读流(monkeypatch, 服务, cursor=cursor, last_event_id=last_event_id)
|
||
assert len(块) == 1
|
||
assert 拆块(块[0])["event"] == "stream-complete"
|
||
assert 服务.调用 == [("task", 3, 100)]
|
||
assert 等待 == []
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-003")
|
||
def test_legacy终态继续心跳保持原协议__o08103(monkeypatch):
|
||
服务 = 合成事件服务(
|
||
{
|
||
0: 事件续接((事件(1),), False, 1, 1, 1),
|
||
1: 事件续接((), False, 1, 1, 1),
|
||
}
|
||
)
|
||
块, _, 等待, 检查 = 读流(monkeypatch, 服务, protocol="legacy", 断开检查次数=2)
|
||
assert 拆块(块[0])["event"] == "task-event"
|
||
assert 块[1:] == [": keep-alive\n\n"]
|
||
assert "stream-complete" not in "".join(块)
|
||
assert 服务.调用 == [("task", 0, 100), ("task", 1, 100)]
|
||
assert 等待 == [1, 1] and len(检查) == 3
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-004")
|
||
@pytest.mark.parametrize("protocol", ["legacy", "bounded-v2"])
|
||
def test_reset要求快照后立即结束__o08104(monkeypatch, protocol):
|
||
服务 = 合成事件服务({0: 事件续接((), True, 5, 8, 8)})
|
||
块, _, 等待, 检查 = 读流(monkeypatch, 服务, protocol=protocol)
|
||
assert len(块) == 1
|
||
数据 = 拆块(块[0])
|
||
assert 数据["event"] == "reset"
|
||
assert json.loads(数据["data"]) == {
|
||
"items": [],
|
||
"reset_required": True,
|
||
"first_sequence": 5,
|
||
"last_sequence": 8,
|
||
"execution_final_sequence": 8,
|
||
}
|
||
assert len(服务.调用) == len(检查) == 1
|
||
assert 等待 == []
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-005")
|
||
def test_未终态保留心跳且断开不再查询__o08105(monkeypatch):
|
||
服务 = 合成事件服务({0: 事件续接((), False, 1, 0)})
|
||
块, _, 等待, 检查 = 读流(monkeypatch, 服务, 断开检查次数=1)
|
||
assert 块 == [": keep-alive\n\n"]
|
||
assert len(服务.调用) == 1 and len(检查) == 2
|
||
assert 等待 == [1]
|
||
先断 = 合成事件服务({})
|
||
assert 读流(monkeypatch, 先断, 断开检查次数=0)[0] == []
|
||
assert 先断.调用 == []
|
||
|
||
|
||
def 合成应用(monkeypatch, 服务):
|
||
装配替身(monkeypatch, 服务)
|
||
app = FastAPI()
|
||
app.include_router(事件订阅.路由)
|
||
app.dependency_overrides[要求作者] = lambda: "author"
|
||
安装错误响应(app)
|
||
return app
|
||
|
||
|
||
@pytest.mark.case_id("TC-O08-SSE-006")
|
||
def test_HTTP事件页边界与v2终态合同__o08106(monkeypatch):
|
||
服务 = 合成事件服务({0: 事件续接((事件(1),), False, 1, 1, 1)})
|
||
with TestClient(合成应用(monkeypatch, 服务)) as 客户端:
|
||
页 = 客户端.get("/api/v1/tasks/task/events", params={"limit": 1})
|
||
assert 页.status_code == 200
|
||
assert 页.json()["execution_final_sequence"] == 1
|
||
assert 服务.调用[-1] == ("task", 0, 1)
|
||
assert 客户端.get("/api/v1/tasks/task/events", params={"limit": 100}).status_code == 200
|
||
assert 服务.调用[-1] == ("task", 0, 100)
|
||
响应 = 客户端.get("/api/v1/tasks/task/events/stream", params={"protocol": "bounded-v2"})
|
||
assert 响应.status_code == 200
|
||
assert 响应.headers["content-type"].startswith("text/event-stream")
|
||
帧 = [拆块(块) for 块 in 响应.text.strip().split("\n\n")]
|
||
assert [项["event"] for 项 in 帧] == ["task-event", "stream-complete"]
|
||
assert json.loads(帧[-1]["data"])["execution_final_sequence"] == 1
|
||
schema = 合成应用(monkeypatch, 服务).openapi()
|
||
参数 = schema["paths"]["/api/v1/tasks/{task_id}/events/stream"]["get"]["parameters"]
|
||
assert next(p["schema"] for p in 参数 if p["name"] == "protocol")["default"] == "legacy"
|
||
|
||
|
||
@pytest.mark.case_id("NC-O08-SSE-007")
|
||
@pytest.mark.parametrize(
|
||
"路径,参数,头",
|
||
[
|
||
("events", {"limit": 0}, {}),
|
||
("events", {"limit": 101}, {}),
|
||
("events", {"cursor": -1}, {}),
|
||
("events/stream", {"protocol": "v3"}, {}),
|
||
("events/stream", {"cursor": -1}, {}),
|
||
("events/stream", {}, {"Last-Event-ID": "-1"}),
|
||
],
|
||
)
|
||
def test_HTTP无效范围不触发服务__o08107(monkeypatch, 路径, 参数, 头):
|
||
服务 = 合成事件服务({})
|
||
with TestClient(合成应用(monkeypatch, 服务)) as 客户端:
|
||
响应 = 客户端.get(f"/api/v1/tasks/task/{路径}", params=参数, headers=头)
|
||
assert 响应.status_code == 422
|
||
assert 服务.调用 == []
|