167 lines
6.8 KiB
Python
167 lines
6.8 KiB
Python
"""Kernel M0 骨架测试。
|
||
|
||
覆盖 rewrite-plan M0 验收的核心面：
|
||
- 事件序列完整且有序（run.queued → run.started → … → run.finished）；
|
||
- 命台席位：同人物第二个前台 Run 排队，其 run.started 晚于第一个的 run.finished；
|
||
- 瀑布否决：llm.pre_request 否决 → safety.ruled + run.interrupted（HJ-1301），无 llm.delta；
|
||
- 不变量断言：流重组不一致即 InvariantViolation；
|
||
- SSE 端到端：POST /runs → GET /runs/{id}/stream 读到 run.finished 终态。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from collections.abc import AsyncGenerator, AsyncIterator
|
||
|
||
import httpx
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.db.session import AsyncSessionLocal
|
||
from app.kernel import Kernel, KernelEventLog
|
||
from app.kernel.invariants import InvariantViolation, assert_stream_reconstruction
|
||
from app.modules.runtime.events import Event
|
||
from app.runtime.bus import Veto
|
||
from app.services.event_store import DbEventStore
|
||
from app.shenshi_client import EchoShenshi, ShenshiDelta
|
||
|
||
|
||
class SlowEchoShenshi(EchoShenshi):
|
||
"""带延迟的回声桩：让命台排队与 SSE 流时序可观测。"""
|
||
|
||
delay: float = 0.15
|
||
|
||
async def stream_chat(self, request) -> AsyncIterator[ShenshiDelta]: # noqa: ANN001
|
||
async for delta in super().stream_chat(request):
|
||
await asyncio.sleep(self.delay)
|
||
yield delta
|
||
|
||
|
||
@pytest.fixture
|
||
async def kernel_runtime(monkeypatch: pytest.MonkeyPatch) -> AsyncGenerator[Kernel, None]:
|
||
"""手工装配 Kernel 到 app.state（测试不经 lifespan，与既有测试做法一致）。"""
|
||
from app.main import app
|
||
|
||
event_store = DbEventStore(AsyncSessionLocal)
|
||
kernel = Kernel(
|
||
shenshi=SlowEchoShenshi(),
|
||
session_factory=AsyncSessionLocal,
|
||
event_log=KernelEventLog(event_store),
|
||
timeout_seconds=10.0,
|
||
)
|
||
app.state.kernel = kernel
|
||
yield kernel
|
||
app.state.kernel = None
|
||
|
||
|
||
@pytest.fixture
|
||
async def kernel_client(
|
||
kernel_runtime: Kernel, auth_headers: dict[str, str]
|
||
) -> AsyncGenerator[AsyncClient, None]:
|
||
from app.main import app
|
||
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test", headers=auth_headers) as c:
|
||
yield c
|
||
|
||
|
||
async def _wait_status(client: AsyncClient, run_id: str, target: set[str]) -> dict:
|
||
for _ in range(500):
|
||
resp = await client.get(f"/api/v1/runs/{run_id}")
|
||
body = resp.json()
|
||
if body["status"] in target:
|
||
return body
|
||
await asyncio.sleep(0.02)
|
||
raise AssertionError(f"Run 未到达终态：{run_id}")
|
||
|
||
|
||
async def _events(client: AsyncClient, run_id: str) -> list[dict]:
|
||
resp = await client.get(f"/api/v1/runs/{run_id}/events")
|
||
assert resp.status_code == 200
|
||
return resp.json()
|
||
|
||
|
||
async def test_placeholder_run_event_sequence(kernel_client: AsyncClient) -> None:
|
||
created = await kernel_client.post("/api/v1/runs", json={"prompt": "你好"})
|
||
assert created.status_code == 200
|
||
run_id = created.json()["id"]
|
||
final = await _wait_status(kernel_client, run_id, {"completed"})
|
||
assert final["origin"] == "kernel"
|
||
|
||
types = [e["type"] for e in await _events(kernel_client, run_id)]
|
||
assert types[0] == "run.queued"
|
||
assert "run.started" in types
|
||
order = {t: types.index(t) for t in ["run.started", "context.projected", "llm.requested", "llm.completed", "result.accepted", "run.finished"]}
|
||
assert order["run.started"] < order["context.projected"] < order["llm.requested"]
|
||
assert order["llm.completed"] < order["result.accepted"] < order["run.finished"]
|
||
assert types[-1] == "run.finished"
|
||
assert types.count("llm.delta") >= 1
|
||
|
||
# "模型可见即可重建"：llm.requested 事件携带完整 IR 消息
|
||
requested = next(e for e in await _events(kernel_client, run_id) if e["type"] == "llm.requested")
|
||
messages = requested["payload"]["request"]["messages"]
|
||
assert messages[-1]["content"] == "你好"
|
||
|
||
|
||
async def test_seat_serializes_foreground(kernel_client: AsyncClient) -> None:
|
||
first = (await kernel_client.post("/api/v1/runs", json={"prompt": "第一个"})).json()
|
||
second = (await kernel_client.post("/api/v1/runs", json={"prompt": "第二个"})).json()
|
||
await _wait_status(kernel_client, first["id"], {"completed"})
|
||
await _wait_status(kernel_client, second["id"], {"completed"})
|
||
|
||
def interval(events: list[dict]) -> tuple[str, str]:
|
||
started = next(e["created_at"] for e in events if e["type"] == "run.started")
|
||
finished = next(e["created_at"] for e in events if e["type"] == "run.finished")
|
||
return started, finished
|
||
|
||
start1, finish1 = interval(await _events(kernel_client, first["id"]))
|
||
start2, finish2 = interval(await _events(kernel_client, second["id"]))
|
||
# 同人物前台串行：两个执行区间不相交（谁先谁后由调度决定）
|
||
late_start, early_finish = max(start1, start2), min(finish1, finish2)
|
||
assert late_start >= early_finish
|
||
|
||
|
||
async def test_veto_interrupts_run(
|
||
kernel_runtime: Kernel, kernel_client: AsyncClient
|
||
) -> None:
|
||
async def veto_llm(node: str, context) -> None: # noqa: ANN001, ARG001
|
||
raise Veto("测试否决：禁止本次模型调用", rule="test_rule")
|
||
|
||
kernel_runtime.bus.on("llm.pre_request", veto_llm)
|
||
try:
|
||
created = await kernel_client.post("/api/v1/runs", json={"prompt": "被否决"})
|
||
run_id = created.json()["id"]
|
||
final = await _wait_status(kernel_client, run_id, {"interrupted"})
|
||
assert final["error_code"] == "HJ-1301"
|
||
types = [e["type"] for e in await _events(kernel_client, run_id)]
|
||
assert "safety.ruled" in types
|
||
assert "llm.delta" not in types
|
||
assert types[-1] == "run.interrupted"
|
||
finally:
|
||
kernel_runtime.bus._listeners.pop("llm.pre_request", None) # noqa: SLF001
|
||
|
||
|
||
def test_stream_reconstruction_invariant() -> None:
|
||
def ev(delta: str) -> Event:
|
||
return Event(type="llm.delta", source="kernel", payload={"delta": delta})
|
||
|
||
assert_stream_reconstruction([ev("你"), ev("好")], "你好")
|
||
with pytest.raises(InvariantViolation):
|
||
assert_stream_reconstruction([ev("你"), ev("好")], "你好吗")
|
||
|
||
|
||
async def test_sse_stream_end_to_end(kernel_client: AsyncClient) -> None:
|
||
created = await kernel_client.post("/api/v1/runs", json={"prompt": "流式验证"})
|
||
run_id = created.json()["id"]
|
||
seen_types: list[str] = []
|
||
async with kernel_client.stream("GET", f"/api/v1/runs/{run_id}/stream") as response:
|
||
assert response.status_code == 200
|
||
async for line in response.aiter_lines():
|
||
if line.startswith("event:"):
|
||
seen_types.append(line[len("event:") :].strip())
|
||
if line.startswith("data:") and '"run.finished"' in line:
|
||
break
|
||
assert "run.started" in seen_types
|
||
assert "llm.delta" in seen_types
|
||
assert "run.finished" in seen_types
|