1package app
2
3import (
4 "context"
5 "encoding/json"
6 "strings"
7 "testing"
8 "testing/synctest"
9 "time"
10
11 "github.com/bnema/zerowrap"
12
13 "github.com/bnema/neferwl/internal/ports"
14)
15
16// statsRegistered returns a router with output A registered and its initial
17// cursor drained.
18func statsRegistered(ctx context.Context, in outputRoutes) (*outputRouter, *mailbox) {
19 r := newOutputRouter(ctx, nil, in)
20 mctx, stop := context.WithCancel(ctx)
21 m := newMailbox(1, mctx, stop, make(chan struct{}))
22 r.register("A", m)
23 <-m.cursor
24 return r, m
25}
26
27// A scene or cursor sent before the output read the previous one counts as
28// replaced: the output is behind.
29func TestRouterStatsCountReplaced(t *testing.T) {
30 r, m := statsRegistered(t.Context(), outputRoutes{})
31 r.scenes([]ports.Scene{{Output: "A", Seq: 1}})
32 if r.stats.scenes != 1 || r.stats.scenesReplaced != 0 {
33 t.Fatalf("one scene: %+v", r.stats)
34 }
35 r.scenes([]ports.Scene{{Output: "A", Seq: 2}, {Output: "missing"}})
36 if r.stats.scenes != 2 || r.stats.scenesReplaced != 1 {
37 t.Fatalf("two scenes unread: %+v", r.stats)
38 }
39 <-m.scenes
40 r.scenes([]ports.Scene{{Output: "A", Seq: 3}})
41 if r.stats.scenesReplaced != 1 {
42 t.Fatalf("a drained mailbox is not behind: %+v", r.stats)
43 }
44
45 r.setCursor(ports.CursorChange{Shape: "text"})
46 r.setCursor(ports.CursorChange{Shape: "pointer"})
47 if r.stats.cursor != 2 || r.stats.cursorReplaced != 1 {
48 t.Fatalf("cursor: %+v", r.stats)
49 }
50 if got := <-m.cursor; got.Shape != "pointer" {
51 t.Fatalf("cursor %+v", got)
52 }
53}
54
55// A full capture queue fails the request fast and counts it.
56func TestRouterStatsCapturesFull(t *testing.T) {
57 replies := make(chan ports.CaptureDone, 2)
58 r, m := statsRegistered(t.Context(), outputRoutes{})
59 r.captured = replies
60 for id := range uint64(cap(m.captures)) {
61 r.routeCapture(ports.CaptureRequest{ID: id, Output: "A"})
62 }
63 if r.stats.capturesFull != 0 {
64 t.Fatalf("queue not full yet: %+v", r.stats)
65 }
66 r.routeCapture(ports.CaptureRequest{ID: 99, Output: "A"})
67 r.routeCapture(ports.CaptureRequest{ID: 100, Output: "missing"})
68 if r.stats.capturesFull != 1 {
69 t.Fatalf("captures full: %+v", r.stats)
70 }
71 if got := <-replies; got.ID != 99 || got.Err == nil {
72 t.Fatalf("rejected capture: %+v", got)
73 }
74}
75
76// The backend loop reports its longest wait on core; shorter ones keep it.
77func TestRouterNoteEventSendKeepsMax(t *testing.T) {
78 r := newOutputRouter(t.Context(), nil, outputRoutes{})
79 for _, d := range []time.Duration{3 * time.Millisecond, 12 * time.Millisecond, time.Millisecond} {
80 r.noteEventSend(d)
81 }
82 if got := time.Duration(r.eventWait.Load()); got != 12*time.Millisecond {
83 t.Fatalf("max wait %v, want 12ms", got)
84 }
85}
86
87// Every 10 s the router logs and resets its counters, and stays quiet when
88// nothing happened.
89func TestRouterLogsStatsEvery10s(t *testing.T) {
90 synctest.Test(t, func(t *testing.T) {
91 var out lockedBuffer
92 log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out})
93 ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), log))
94 in := newRouteInputs()
95 r := newOutputRouter(ctx, nil, in.routes())
96 go r.run(ctx)
97 m := newMailbox(1, ctx, func() {}, make(chan struct{}))
98 if _, ok := r.add("A", m); !ok {
99 t.Fatal("router stopped")
100 }
101 <-m.cursor
102
103 time.Sleep(routerStatsEvery)
104 synctest.Wait()
105 if got := out.String(); got != "" {
106 t.Fatalf("idle router logged: %s", got)
107 }
108
109 in.scenes <- []ports.Scene{{Output: "A", Seq: 1}}
110 in.scenes <- []ports.Scene{{Output: "A", Seq: 2}}
111 in.cursor <- ports.CursorChange{Shape: "text"}
112 in.cursor <- ports.CursorChange{Shape: "pointer"}
113 in.scenes <- nil // the router took both cursors
114 r.noteEventSend(1500 * time.Microsecond)
115 time.Sleep(routerStatsEvery)
116 synctest.Wait()
117
118 lines := strings.Split(strings.TrimSpace(out.String()), "\n")
119 if len(lines) != 1 {
120 t.Fatalf("log lines: %q", lines)
121 }
122 var entry map[string]any
123 if err := json.Unmarshal([]byte(lines[0]), &entry); err != nil {
124 t.Fatal(err)
125 }
126 want := map[string]any{"message": "routing stats", "component": "router", "scenes": 2.0, "scenes_replaced": 1.0, "cursor": 2.0, "cursor_replaced": 1.0, "captures_full": 0.0, "max_event_send_ms": 1.5}
127 for k, v := range want {
128 if entry[k] != v {
129 t.Errorf("%s = %v, want %v (%s)", k, entry[k], v, lines[0])
130 }
131 }
132
133 time.Sleep(routerStatsEvery)
134 synctest.Wait()
135 if n := strings.Count(out.String(), "routing stats"); n != 1 {
136 t.Fatalf("counters were not reset: %d entries", n)
137 }
138 cancel()
139 <-r.done
140 })
141}