AskaEth
99a644d545
state ∈ 有效 / 被拒（需重扫）/ 未知；🔴 -14 是**被拒不是过期**，且**未知原因绝不写成「过期」** （面板文案与错误码都如实）。三态判定为纯函数 + 三态各有用例。 ⚠️ 未做（如实登记）：每账号逐条三态（accounts[] 数组上的三态）本片未做。
338 lines
13 KiB
Go
338 lines
13 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"sort"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/gorilla/websocket"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 这些测试锁住 design/11 §7.1 的帧门：viewer 连接可以跟随会话状态机，
|
||
// 但绝不可收到内容，而 owner 与 admin 两类都保留。frameScopeCases 是
|
||
// design/11 §7.1 表在代码中的体现 —— 表中点名的每种帧类型，都发布在
|
||
// 真正产生它的主题上 —— 因此分类错误（内容帧被公开，或状态信号被
|
||
// 私有化）会在这里失败，而不是静默泄露。
|
||
|
||
// frameScopeCase 是 design/11 §7.1 表的一行。
|
||
type frameScopeCase struct {
|
||
typ string // 下行帧类型
|
||
subject string // 产生它的 bus 主题
|
||
payload any // 发布在该主题上的载荷
|
||
public bool // 「状态机信号」：不含内容 → 每个已认证连接
|
||
}
|
||
|
||
// scopeSentinel 最后发布：bus 每个订阅跑一个 worker 且按序处理，因此
|
||
// 已经看到哨兵的连接必定已经被投递过它之前发布的每一个帧。这把
|
||
// 「viewer 没有收到别的」从猜测的空闲窗口变成精确计数。
|
||
const scopeSentinel = EventSessionIdle
|
||
|
||
// publishFrameScopeCases 使用的批次节流，远低于 hub 的 wsSendBuffer
|
||
// 深度，因此一次突发永不触发慢客户端丢弃。
|
||
const (
|
||
scopePaceFrames = 4
|
||
scopePacePause = 50 * time.Millisecond
|
||
)
|
||
|
||
// scopeRead 是一个连接读取 goroutine 的结果。
|
||
type scopeRead struct {
|
||
role auth.Role
|
||
types []string
|
||
err error
|
||
}
|
||
|
||
// frameScopeCases 返回 design/11 §7.1 表。转发帧使用它们真实的 ws.<type>
|
||
// 主题；流帧发布 loop 在 ws.stream 上发出的真实 model.StreamEvent，因此
|
||
// 测试驱动的是生产路径 bus → relay → broadcast，而不是手工构造的帧。
|
||
func frameScopeCases() []frameScopeCase {
|
||
relayed := func(typ string, public bool) frameScopeCase {
|
||
return frameScopeCase{typ: typ, subject: wsPrefix + typ, payload: map[string]any{"sessionId": "web"}, public: public}
|
||
}
|
||
streamed := func(kind model.StreamKind, typ string) frameScopeCase {
|
||
return frameScopeCase{
|
||
typ: typ,
|
||
subject: wsStreamSubject,
|
||
payload: model.StreamEvent{Kind: kind, Delta: "x", SessionID: "web", TurnID: "turn-scope"},
|
||
}
|
||
}
|
||
return []frameScopeCase{
|
||
// 状态机信号（白名单，不含任何内容）→ 所有已认证连接。
|
||
relayed(EventSessionBusy, true),
|
||
relayed(EventPendingAck, true),
|
||
relayed(EventPendingDrain, true),
|
||
relayed(EventSessionIdle, true),
|
||
relayed(EventSessionInterrupt, true),
|
||
relayed(EventPendingFull, true),
|
||
relayed(wsSessionRead, true),
|
||
|
||
// 流式内容（含角色回复正文与推理内容）→ 仅 owner/admin。
|
||
streamed(model.StreamKindTextStart, "message.start"),
|
||
streamed(model.StreamKindTextDelta, "message.delta"),
|
||
streamed(model.StreamKindTextEnd, "message.end"),
|
||
streamed(model.StreamKindThinkingStart, "thinking.start"),
|
||
streamed(model.StreamKindThinkingDelta, "thinking.delta"),
|
||
streamed(model.StreamKindThinkingEnd, "thinking.end"),
|
||
streamed(model.StreamKindToolCallStart, "tool_call.start"),
|
||
streamed(model.StreamKindToolCallDelta, "tool_call.delta"),
|
||
streamed(model.StreamKindToolCallEnd, "tool_call.end"),
|
||
streamed(model.StreamKindToolExecutionStart, "tool_execution.start"),
|
||
streamed(model.StreamKindToolExecutionUpdate, "tool_execution.update"),
|
||
streamed(model.StreamKindToolExecutionEnd, "tool_execution.end"),
|
||
streamed(model.StreamKindTurnStart, "turn.start"),
|
||
streamed(model.StreamKindTurnEnd, "turn.end"),
|
||
relayed("stream_chunk", false),
|
||
relayed("stream_end", false),
|
||
relayed("thinking", false),
|
||
|
||
// 持久化内容 → 仅 owner/admin。
|
||
relayed("message_persisted", false),
|
||
relayed("message_edited", false),
|
||
relayed("message_deleted", false),
|
||
relayed("channel_persisted", false),
|
||
// 新帧 turn_begin（design/15 §2.5.9）：载荷带 messageIds ⇒ 携带具体消息
|
||
// 身份，落 fail-closed 默认档（只发 owner + admin，与 steer_ack /
|
||
// chat_duplicate 同档），**不**进那 7 个状态机信号白名单。
|
||
relayed(EventTurnBegin, false),
|
||
|
||
// 系统内部 → 仅 owner/admin。
|
||
relayed("graph_update", false),
|
||
relayed("subagent_event", false),
|
||
// 后台子代理批次结果：这里断言的是**帧名**（常量，不是字面量，避免下次
|
||
// 再漂移）；它真实的产生主题是 orchestration.SubagentResultSubject
|
||
// （"subagent.result"，带点，那是 bus topic 的拼法），该端点对端路径
|
||
// 由 ws_bridge_subagent_result_test.go 单独驱动。
|
||
relayed(wsSubagentResult, false),
|
||
relayed("task_output", false),
|
||
relayed("task_done", false),
|
||
relayed("automation_result", false),
|
||
relayed("tool_used", false),
|
||
relayed("capacity_awaken", false),
|
||
relayed("mode_switch", false),
|
||
|
||
// 私有状态 → 仅 owner/admin。
|
||
relayed("mood", false),
|
||
relayed("device_update", false),
|
||
relayed("reminder", false),
|
||
// 渠道事件实时帧（design/26 §2.6 / §4.5；切片 E7）：载荷是 `channel_events`
|
||
// 的一行（平台 / 账号 / 类别 / 摘要…），属**管理面元数据** —— 不进白名单，
|
||
// 按 fail-closed 默认档只发 owner + admin。它真实的生产主题是
|
||
// `events.SubjectChannelEvent`（"channel.event"，带点），端到端路径由
|
||
// ws_channel_events_test.go 单独驱动；这一行钉的是**帧名**的分类。
|
||
relayed(wsChannelEvent, false),
|
||
}
|
||
}
|
||
|
||
// TestWSFrameScopePerRole 用真实角色 token 经真实 WebSocket 连接驱动
|
||
// 整张 design/11 §7.1 表：owner 与 admin 收到表中的每一帧，而 viewer
|
||
// 恰好收到白名单状态信号，且按发布顺序。设计点名的 viewer 断言
|
||
// （message.delta / thinking.delta / message_persisted / graph_update /
|
||
// mood 被拒，session_idle 被投递）在此显式复述，使契约在测试中可读。
|
||
func TestWSFrameScopePerRole(t *testing.T) {
|
||
t.Parallel()
|
||
cases := frameScopeCases()
|
||
env := newEnv(t, nil, false)
|
||
|
||
conns := map[auth.Role]*websocket.Conn{}
|
||
for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} {
|
||
conn := dialWS(t, env.ts.URL, env.token(t, role))
|
||
defer conn.Close()
|
||
conns[role] = conn
|
||
}
|
||
waitConnCount(t, env.srv, len(conns))
|
||
|
||
all := append(frameTypes(cases, false), scopeSentinel)
|
||
public := append(frameTypes(cases, true), scopeSentinel)
|
||
wants := map[auth.Role]int{
|
||
auth.RoleOwner: len(all),
|
||
auth.RoleAdmin: len(all),
|
||
auth.RoleViewer: len(public),
|
||
}
|
||
|
||
// 各客户端在批次发布期间于自己的 goroutine 上排空：hub 会丢弃
|
||
// 发送缓冲已满的客户端的帧（其慢客户端策略），因此之后才开始
|
||
// 读取的客户端会丢掉尾部。
|
||
results := make(chan scopeRead, len(conns))
|
||
for role, want := range wants {
|
||
role, want := role, want
|
||
go func() {
|
||
types, err := readFrameTypes(conns[role], want)
|
||
results <- scopeRead{role: role, types: types, err: err}
|
||
}()
|
||
}
|
||
publishFrameScopeCases(t, env, cases)
|
||
|
||
got := make(map[auth.Role][]string, len(conns))
|
||
for range conns {
|
||
res := <-results
|
||
if res.err != nil {
|
||
t.Fatalf("%s reader: %v", res.role, res.err)
|
||
}
|
||
got[res.role] = res.types
|
||
}
|
||
|
||
for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} {
|
||
assertFrameTypes(t, string(role), got[role], all)
|
||
}
|
||
gotViewer := got[auth.RoleViewer]
|
||
assertFrameTypes(t, string(auth.RoleViewer), gotViewer, public)
|
||
for _, denied := range []string{"message.delta", "thinking.delta", "message_persisted", "graph_update", "mood"} {
|
||
if containsFrameType(gotViewer, denied) {
|
||
t.Fatalf("viewer received %q; it must not (design/11 §7.1)", denied)
|
||
}
|
||
}
|
||
if !containsFrameType(gotViewer, EventSessionIdle) {
|
||
t.Fatalf("viewer missing %q, a whitelisted state signal (got %v)", EventSessionIdle, gotViewer)
|
||
}
|
||
}
|
||
|
||
// TestHubBroadcastScopeMatrix 是「每种帧类型 × 每种角色」矩阵，直接经
|
||
// 投递入口驱动，使表中每一行都被孤立断言（否则泄露的帧难以归因）。
|
||
// claims 按角色显式设置，因此默认拒绝的回归会表现为 viewer 丢掉它
|
||
// 本应收到的帧。
|
||
func TestHubBroadcastScopeMatrix(t *testing.T) {
|
||
t.Parallel()
|
||
hb := newTestHub(nil, nil)
|
||
owner := newFakeConn(hb, 4)
|
||
owner.claims = &auth.Claims{Role: auth.RoleOwner}
|
||
admin := newFakeConn(hb, 4)
|
||
admin.claims = &auth.Claims{Role: auth.RoleAdmin}
|
||
viewer := newFakeConn(hb, 4)
|
||
viewer.claims = &auth.Claims{Role: auth.RoleViewer}
|
||
for _, c := range []*wsConn{owner, admin, viewer} {
|
||
hb.register(c)
|
||
}
|
||
|
||
for _, tc := range frameScopeCases() {
|
||
tc := tc
|
||
t.Run(tc.typ, func(t *testing.T) {
|
||
hb.broadcast(wsEvent{Type: tc.typ, SessionID: "web"})
|
||
for _, c := range []*wsConn{owner, admin} {
|
||
if ev := readFakeFrame(t, c); ev.Type != tc.typ {
|
||
t.Fatalf("owner/admin frame = %q, want %q", ev.Type, tc.typ)
|
||
}
|
||
}
|
||
if tc.public {
|
||
if ev := readFakeFrame(t, viewer); ev.Type != tc.typ {
|
||
t.Fatalf("viewer frame = %q, want %q (whitelisted state signal)", ev.Type, tc.typ)
|
||
}
|
||
return
|
||
}
|
||
if err := connHasNoFrame(viewer); err != nil {
|
||
t.Fatalf("viewer received the restricted frame %q: %v", tc.typ, err)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestFrameScopeAllowsExactlyTheStateSignals 钉住两条使帧门不漂移的
|
||
// 性质：未分类的帧类型永不是公开的（因此后续里程碑新增的帧不会因
|
||
// 漏分类而泄露），且公开集合恰好是 design/11 §7.1 登记的状态信号
|
||
// —— 扩大它意味着要同时改设计、帧门和这张表。
|
||
func TestFrameScopeAllowsExactlyTheStateSignals(t *testing.T) {
|
||
t.Parallel()
|
||
if frameReachesAll("frame_type_nobody_classified") {
|
||
t.Fatal("an unclassified frame type reached every connection; the default must be owner/admin only")
|
||
}
|
||
public := make([]string, 0, len(frameScopeCases()))
|
||
for _, tc := range frameScopeCases() {
|
||
if got := frameReachesAll(tc.typ); got != tc.public {
|
||
t.Fatalf("frame %q: gate says public=%v, design/11 §7.1 says public=%v", tc.typ, got, tc.public)
|
||
}
|
||
if tc.public {
|
||
public = append(public, tc.typ)
|
||
}
|
||
}
|
||
want := []string{
|
||
EventSessionBusy,
|
||
EventPendingAck,
|
||
EventPendingDrain,
|
||
EventSessionIdle,
|
||
EventSessionInterrupt,
|
||
EventPendingFull,
|
||
wsSessionRead,
|
||
}
|
||
sort.Strings(public)
|
||
sort.Strings(want)
|
||
assertFrameTypes(t, "design/11 §7.1 whitelist", public, want)
|
||
}
|
||
|
||
// frameTypes 按表顺序返回表中的帧类型（全部，或仅公开的那些）。
|
||
func frameTypes(cases []frameScopeCase, publicOnly bool) []string {
|
||
out := make([]string, 0, len(cases))
|
||
for _, tc := range cases {
|
||
if publicOnly && !tc.public {
|
||
continue
|
||
}
|
||
out = append(out, tc.typ)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// publishFrameScopeCases 在每行各自的生产主题上发布它，随后发布关闭
|
||
// 批次的哨兵。批次是节流的：发送缓冲已满的客户端其帧会被丢弃
|
||
// （hub 有文档的慢客户端策略），因此 37 帧的突发放送会去考验丢弃
|
||
// 策略而不是帧门。
|
||
func publishFrameScopeCases(t *testing.T, env *testEnv, cases []frameScopeCase) {
|
||
t.Helper()
|
||
for i, tc := range cases {
|
||
if err := env.bus.Publish(context.Background(), tc.subject, tc.payload); err != nil {
|
||
t.Fatalf("publish %s: %v", tc.subject, err)
|
||
}
|
||
if (i+1)%scopePaceFrames == 0 {
|
||
time.Sleep(scopePacePause)
|
||
}
|
||
}
|
||
if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil {
|
||
t.Fatalf("publish sentinel: %v", err)
|
||
}
|
||
}
|
||
|
||
// readFrameTypes 恰好读取 want 个帧并按到达顺序返回其类型。它返回错误
|
||
// 而不是让测试失败，以便能跑在批次发布期间排空各客户端的读取
|
||
// goroutine 上。
|
||
func readFrameTypes(conn *websocket.Conn, want int) ([]string, error) {
|
||
got := make([]string, 0, want)
|
||
for len(got) < want {
|
||
_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||
_, msg, err := conn.ReadMessage()
|
||
if err != nil {
|
||
return got, fmt.Errorf("read frame %d/%d: %w (got %v)", len(got)+1, want, err, got)
|
||
}
|
||
var ev wsEvent
|
||
if err := json.Unmarshal(msg, &ev); err != nil {
|
||
return got, fmt.Errorf("decode frame %s: %w", msg, err)
|
||
}
|
||
got = append(got, ev.Type)
|
||
}
|
||
return got, nil
|
||
}
|
||
|
||
// assertFrameTypes 比较收到的帧类型与期望序列。
|
||
func assertFrameTypes(t *testing.T, what string, got, want []string) {
|
||
t.Helper()
|
||
if len(got) != len(want) {
|
||
t.Fatalf("%s frame count = %d, want %d (got %v)", what, len(got), len(want), got)
|
||
}
|
||
for i := range want {
|
||
if got[i] != want[i] {
|
||
t.Fatalf("%s frame[%d] = %q, want %q (got %v)", what, i, got[i], want[i], got)
|
||
}
|
||
}
|
||
}
|
||
|
||
// containsFrameType 报告 typ 是否在收到的帧类型之中。
|
||
func containsFrameType(types []string, typ string) bool {
|
||
for _, got := range types {
|
||
if got == typ {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|