test(gateway): 补 graph_update 的 WS 中继与 wire 契约测试（此前零覆盖）
`#/admin/flow` 报障调查发现：flow 链路每一环都是对的（loop 的 FlowPublisher 三入口
在真实回合里都被调用 / wire.go 的 allow 白名单正确 / parseGraph 与 Snapshot 字段逐条对得上），
但**「bus 的 ws.graph_update 经 ws_bridge 中继成 WS 帧」这一跳此前完全没有测试**——
逐环人工核对永远核不到它。补两个契约测试把它钉死：
- TestFlowGraphUpdateRelayedToWS：真 bus Publish("ws.graph_update", <真 trace.RunGraph.Snapshot()>)
→ 断言 WS 收到 graph_update 且数据逐字节未被改写
- TestFlowGraphUpdateWireFields：精确断言 graph/node/edge/log 四层的 key 集合
= 前端 parseGraph 读取的那一套（改名/多字段/少必填都 FAIL）
验证：go build/vet 绿；-race 网关 ok；全量 58 包 ok / 0 FAIL；交叉编译绿。
变异验证：① 改 trace.FlowNode.Label 的 json tag → 键集断言 FAIL；
② 让 ws_bridge 的 normalizeEvent 丢掉 ws.graph_update → 中继测试 i/o timeout FAIL。
This commit is contained in:
AskaEth committed
1 parent
5847705e35
commit
28b663231e
1 file changed
+281
@@ -0,0 +1,281 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// These tests cover the only leg of the design/19 flow chain that had no test:
|
||||
// the producer's bus event → hub relay → outbound WS frame. The loop side is
|
||||
// pinned by internal/runtime/loop/flow_test.go; this pins the gateway relay AND
|
||||
// the wire field names the frontend's parseGraph reads, so a rename on either side
|
||||
// fails here instead of silently blanking the 全链路流程 page.
|
||||
|
||||
// TestFlowGraphUpdateRelayedToWS publishes a real trace.RunGraph snapshot on the
|
||||
// producer subject (ws.graph_update) and asserts the client receives a
|
||||
// `graph_update` frame carrying the snapshot unchanged.
|
||||
func TestFlowGraphUpdateRelayedToWS(t *testing.T) {
|
||||
env := newEnv(t, nil, false)
|
||||
tok := env.token(t, auth.RoleOwner)
|
||||
wsURL := "ws" + strings.TrimPrefix(env.ts.URL, "http") + "/ws/chat?token=" + tok
|
||||
|
||||
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
|
||||
|
||||
// Round-trip a mode_switch first so the connection is registered before the
|
||||
// publish (same guard as TestWSUpgradeAndBroadcast).
|
||||
if err := conn.WriteJSON(wsInbound{Type: "mode_switch", Data: json.RawMessage(`{"mode":"chat"}`)}); err != nil {
|
||||
t.Fatalf("write mode_switch: %v", err)
|
||||
}
|
||||
if got := readWS(t, conn); got.Type != "mode_switch" {
|
||||
t.Fatalf("first frame type = %q, want mode_switch", got.Type)
|
||||
}
|
||||
|
||||
snap := sampleFlowSnapshot()
|
||||
if err := env.bus.Publish(context.Background(), "ws.graph_update", snap); err != nil {
|
||||
t.Fatalf("publish graph: %v", err)
|
||||
}
|
||||
raw := readRawWS(t, conn)
|
||||
|
||||
var frame struct {
|
||||
Type string `json:"type"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||||
t.Fatalf("decode frame %s: %v", raw, err)
|
||||
}
|
||||
if frame.Type != "graph_update" {
|
||||
t.Fatalf("frame type = %q, want graph_update (frame=%s)", frame.Type, raw)
|
||||
}
|
||||
|
||||
// The relayed payload must be the snapshot verbatim — same JSON, not re-shaped.
|
||||
want, err := json.Marshal(snap)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal snapshot: %v", err)
|
||||
}
|
||||
if string(frame.Data) != string(want) {
|
||||
t.Fatalf("relayed data differs from the published snapshot:\n got %s\nwant %s", frame.Data, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowGraphUpdateWireFields pins the JSON key names on every level of the
|
||||
// snapshot against the fields the frontend reader (parseGraph in lib/wsParse.ts)
|
||||
// requires. It is deliberately an exact key-set comparison: adding, renaming or
|
||||
// dropping a field the frontend reads must fail here.
|
||||
func TestFlowGraphUpdateWireFields(t *testing.T) {
|
||||
raw, err := json.Marshal(sampleFlowSnapshot())
|
||||
if err != nil {
|
||||
t.Fatalf("marshal snapshot: %v", err)
|
||||
}
|
||||
var graph map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &graph); err != nil {
|
||||
t.Fatalf("decode graph: %v", err)
|
||||
}
|
||||
|
||||
assertGraphKeys(t, graph)
|
||||
nodes := decodeObjects(t, graph["nodes"], "nodes")
|
||||
assertNodeKeys(t, nodes)
|
||||
assertEdgeKeys(t, decodeObjects(t, graph["edges"], "edges"))
|
||||
assertLogKeys(t, nodes)
|
||||
assertGraphMeta(t, raw)
|
||||
}
|
||||
|
||||
// assertGraphKeys pins the top-level FlowGraph keys read by parseGraph/Flow.tsx.
|
||||
func assertGraphKeys(t *testing.T, graph map[string]json.RawMessage) {
|
||||
t.Helper()
|
||||
want := []string{"id", "nodes", "edges", "sessionId", "startedAt", "endedAt"}
|
||||
if got := sortedKeys(graph); !equalKeys(got, want) {
|
||||
t.Fatalf("graph keys = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// assertNodeKeys checks every emitted node key is one the frontend reads, that each
|
||||
// node carries the always-read id/label, and that the node holding props+log has
|
||||
// the complete set (which pins the omitempty tags too).
|
||||
func assertNodeKeys(t *testing.T, nodes []map[string]json.RawMessage) {
|
||||
t.Helper()
|
||||
want := []string{"id", "label", "kind", "status", "startedAt", "endedAt", "props", "log"}
|
||||
fullSeen := false
|
||||
for i, n := range nodes {
|
||||
keys := sortedKeys(n)
|
||||
for _, k := range keys {
|
||||
if !containsKey(want, k) {
|
||||
t.Fatalf("node[%d] has unexpected key %q; frontend reads %v", i, k, want)
|
||||
}
|
||||
}
|
||||
if !containsKey(keys, "id") || !containsKey(keys, "label") {
|
||||
t.Fatalf("node[%d] keys = %v, must contain id and label", i, keys)
|
||||
}
|
||||
if containsKey(keys, "props") && containsKey(keys, "log") {
|
||||
fullSeen = true
|
||||
if !equalKeys(keys, want) {
|
||||
t.Fatalf("node[%d] keys = %v, want the full set %v", i, keys, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
if !fullSeen {
|
||||
t.Fatal("no node carried both props and log; the omitempty tags are unpinned")
|
||||
}
|
||||
}
|
||||
|
||||
// assertEdgeKeys pins the FlowEdge shape (parseGraph reads from/to).
|
||||
func assertEdgeKeys(t *testing.T, edges []map[string]json.RawMessage) {
|
||||
t.Helper()
|
||||
if len(edges) == 0 {
|
||||
t.Fatal("snapshot has no edges")
|
||||
}
|
||||
for i, e := range edges {
|
||||
if got := sortedKeys(e); !equalKeys(got, []string{"from", "to"}) {
|
||||
t.Fatalf("edge[%d] keys = %v, want [from to]", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertLogKeys pins a node's log entry shape (parseGraph reads time/type/props).
|
||||
func assertLogKeys(t *testing.T, nodes []map[string]json.RawMessage) {
|
||||
t.Helper()
|
||||
var logged []map[string]json.RawMessage
|
||||
for _, n := range nodes {
|
||||
rawLog, ok := n["log"]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if err := json.Unmarshal(rawLog, &logged); err != nil {
|
||||
t.Fatalf("decode log: %v", err)
|
||||
}
|
||||
}
|
||||
if len(logged) == 0 {
|
||||
t.Fatal("snapshot has no log entries to pin")
|
||||
}
|
||||
for i, ev := range logged {
|
||||
if got := sortedKeys(ev); !equalKeys(got, []string{"time", "type", "props"}) {
|
||||
t.Fatalf("log[%d] keys = %v, want [props time type]", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertGraphMeta pins the envelope values the frontend renders (graph id/session
|
||||
// pair) and that timestamps arrive as RFC3339Nano (what Date.parse expects).
|
||||
func assertGraphMeta(t *testing.T, raw []byte) {
|
||||
t.Helper()
|
||||
var meta struct {
|
||||
ID string `json:"id"`
|
||||
SessionID string `json:"sessionId"`
|
||||
StartedAt string `json:"startedAt"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &meta); err != nil {
|
||||
t.Fatalf("decode meta: %v", err)
|
||||
}
|
||||
if meta.ID != "web-1" || meta.SessionID != "web" {
|
||||
t.Fatalf("graph id/sessionId = %q/%q, want web-1/web", meta.ID, meta.SessionID)
|
||||
}
|
||||
if _, err := time.Parse(time.RFC3339Nano, meta.StartedAt); err != nil {
|
||||
t.Fatalf("startedAt %q is not RFC3339Nano: %v", meta.StartedAt, err)
|
||||
}
|
||||
}
|
||||
|
||||
// sampleFlowSnapshot builds a real producer-side snapshot (trace.RunGraph, the same
|
||||
// type the loop's FlowPublisher publishes) whose every optional wire field is
|
||||
// populated, so the key-set assertions above are meaningful.
|
||||
func sampleFlowSnapshot() *trace.FlowGraph {
|
||||
base := time.Date(2026, 10, 3, 21, 0, 0, 0, time.UTC)
|
||||
g := trace.NewRunGraph("web")
|
||||
g.Start(1, base)
|
||||
g.SeedInput("帮我查一下天气")
|
||||
g.AddSpanStart("span-llm", "llm.call", trace.SpanKindLLM, "", base.Add(time.Millisecond))
|
||||
g.AddSpan(trace.SpanRecord{
|
||||
ID: "span-llm",
|
||||
Name: "llm.call",
|
||||
Kind: trace.SpanKindLLM,
|
||||
Status: trace.SpanStatusOK,
|
||||
StartedAt: base.Add(time.Millisecond),
|
||||
EndedAt: base.Add(20 * time.Millisecond),
|
||||
Props: map[string]any{"stop_reason": "tool_calls", "step": 1},
|
||||
Events: []trace.Event{{Time: base.Add(2 * time.Millisecond), Type: "llm.done", Props: map[string]any{"tools": 1}}},
|
||||
})
|
||||
g.AddSpanStart("span-tool", "tool.get_weather", trace.SpanKindTool, "span-llm", base.Add(21*time.Millisecond))
|
||||
g.AddSpan(trace.SpanRecord{
|
||||
ID: "span-tool",
|
||||
ParentID: "span-llm",
|
||||
Name: "tool.get_weather",
|
||||
Kind: trace.SpanKindTool,
|
||||
Status: trace.SpanStatusOK,
|
||||
StartedAt: base.Add(21 * time.Millisecond),
|
||||
EndedAt: base.Add(30 * time.Millisecond),
|
||||
Props: map[string]any{"tool": "get_weather", "result": "晴 26℃"},
|
||||
Events: []trace.Event{{Time: base.Add(25 * time.Millisecond), Type: "tool.done", Props: map[string]any{"ok": true}}},
|
||||
})
|
||||
g.SeedOutput("今天晴，26 度。")
|
||||
return g.Snapshot()
|
||||
}
|
||||
|
||||
// decodeObjects decodes a raw JSON array of objects, failing on a malformed shape.
|
||||
func decodeObjects(t *testing.T, raw json.RawMessage, what string) []map[string]json.RawMessage {
|
||||
t.Helper()
|
||||
var out []map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
t.Fatalf("decode %s: %v", what, err)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
t.Fatalf("snapshot has no %s", what)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// readRawWS reads one text frame and returns its raw bytes.
|
||||
func readRawWS(t *testing.T, conn *websocket.Conn) []byte {
|
||||
t.Helper()
|
||||
_, msg, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
t.Fatalf("read ws: %v", err)
|
||||
}
|
||||
return msg
|
||||
}
|
||||
|
||||
// sortedKeys returns the object's JSON keys in sorted order.
|
||||
func sortedKeys(obj map[string]json.RawMessage) []string {
|
||||
out := make([]string, 0, len(obj))
|
||||
for k := range obj {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// equalKeys reports whether got has exactly the members of want (both sorted).
|
||||
func equalKeys(got []string, want []string) bool {
|
||||
want = append([]string(nil), want...)
|
||||
sort.Strings(want)
|
||||
if len(got) != len(want) {
|
||||
return false
|
||||
}
|
||||
for i := range got {
|
||||
if got[i] != want[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// containsKey reports whether the (sorted) key list holds k.
|
||||
func containsKey(keys []string, k string) bool {
|
||||
for _, key := range keys {
|
||||
if key == k {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
Reference in new issue
Block a user