From 28b663231eb306cd585108a21738d8dd961bceec Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 21:39:03 +0800 Subject: [PATCH] =?UTF-8?q?test(gateway):=20=E8=A1=A5=20graph=5Fupdate=20?= =?UTF-8?q?=E7=9A=84=20WS=20=E4=B8=AD=E7=BB=A7=E4=B8=8E=20wire=20=E5=A5=91?= =?UTF-8?q?=E7=BA=A6=E6=B5=8B=E8=AF=95=EF=BC=88=E6=AD=A4=E5=89=8D=E9=9B=B6?= =?UTF-8?q?=E8=A6=86=E7=9B=96=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `#/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。 --- .../services/gateway/flow_update_ws_test.go | 281 ++++++++++++++++++ 1 file changed, 281 insertions(+) create mode 100644 internal/services/gateway/flow_update_ws_test.go diff --git a/internal/services/gateway/flow_update_ws_test.go b/internal/services/gateway/flow_update_ws_test.go new file mode 100644 index 0000000..dbfb721 --- /dev/null +++ b/internal/services/gateway/flow_update_ws_test.go @@ -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 +}