1package app
2
3import (
4 "context"
5 "os"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/mock"
11
12 portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
13 "github.com/bnema/neferwl/internal/ports"
14)
15
16// routeInputs are the router's input channels, as wired in production.
17type routeInputs struct {
18 scenes chan []ports.Scene
19 cursor chan ports.CursorChange
20 captures chan ports.CaptureRequest
21}
22
23func newRouteInputs() routeInputs {
24 return routeInputs{scenes: make(chan []ports.Scene), cursor: make(chan ports.CursorChange), captures: make(chan ports.CaptureRequest)}
25}
26
27func (in routeInputs) routes() outputRoutes {
28 return outputRoutes{scenes: in.scenes, cursor: in.cursor, captures: in.captures}
29}
30
31// idleRun is an output that runs until stopped.
32func idleRun(ctx context.Context, _ outputFeed) error {
33 <-ctx.Done()
34 return nil
35}
36
37func TestOutputRouting(t *testing.T) {
38 ctx, cancel := context.WithCancel(context.Background())
39 defer cancel()
40 replies := make(chan ports.CaptureDone, 2)
41 in := newRouteInputs()
42 set := newOutputSet(ctx, replies, in.routes())
43 for _, name := range []string{"A", "B"} {
44 set.start(ctx, name, idleRun)
45 <-set.outs[name].cursor // initial cursor
46 }
47 in.scenes <- []ports.Scene{{Output: "A", Seq: 1}, {Output: "B", Seq: 2}}
48 in.scenes <- []ports.Scene{{Output: "A", Seq: 3}}
49 in.cursor <- ports.CursorChange{} // the router took both scene sets
50 if got := <-set.outs["A"].scenes; got.Seq != 3 {
51 t.Fatalf("A scene: %+v", got)
52 }
53 if got := <-set.outs["B"].scenes; got.Seq != 2 {
54 t.Fatalf("B scene: %+v", got)
55 }
56 in.cursor <- ports.CursorChange{Shape: "text"}
57 in.cursor <- ports.CursorChange{Shape: "pointer"}
58 in.scenes <- nil // the router took both cursors
59 for _, name := range []string{"A", "B"} {
60 if got := <-set.outs[name].cursor; got.Shape != "pointer" {
61 t.Fatalf("%s cursor: %+v", name, got)
62 }
63 }
64 f, err := os.CreateTemp(t.TempDir(), "capture")
65 if err != nil {
66 t.Fatal(err)
67 }
68 in.captures <- ports.CaptureRequest{ID: 7, Output: "B", Dst: ports.SHMBuffer{File: f}}
69 if got := <-set.outs["B"].captures; got.ID != 7 {
70 t.Fatalf("B capture: %+v", got)
71 }
72 if err := f.Close(); err != nil {
73 t.Fatal(err)
74 }
75 a := set.outs["A"]
76 a.stop()
77 if err := set.finish("A"); err != nil {
78 t.Fatal(err)
79 }
80 in.scenes <- []ports.Scene{{Output: "A", Seq: 4}}
81 in.captures <- ports.CaptureRequest{ID: 8, Output: "A"}
82 if got := <-replies; got.ID != 8 || got.Err == nil {
83 t.Fatalf("removed output capture: %+v", got)
84 }
85 select {
86 case sc := <-a.scenes:
87 t.Fatalf("removed output fed a scene: %+v", sc)
88 default:
89 }
90 if err := set.wait(); err != nil {
91 t.Fatal(err)
92 }
93}
94
95// An output started after a scene turned its name off starts off; one
96// turned back on starts on.
97func TestOutputStartsOffAfterOffScene(t *testing.T) {
98 ctx, cancel := context.WithCancel(context.Background())
99 defer cancel()
100 in := newRouteInputs()
101 set := newOutputSet(ctx, nil, in.routes())
102 startOff := make(chan bool, 1)
103 run := func(ctx context.Context, feed outputFeed) error {
104 startOff <- feed.StartOff
105 <-ctx.Done()
106 return nil
107 }
108 in.scenes <- []ports.Scene{{Output: "A", Off: true}, {Output: "B"}}
109 for _, tc := range []struct {
110 name string
111 want bool
112 }{{"A", true}, {"B", false}} {
113 set.start(ctx, tc.name, run)
114 if got := <-startOff; got != tc.want {
115 t.Fatalf("%s start off %v, want %v", tc.name, got, tc.want)
116 }
117 }
118 in.scenes <- []ports.Scene{{Output: "A"}}
119 set.outs["A"].stop()
120 if err := set.finish("A"); err != nil {
121 t.Fatal(err)
122 }
123 set.start(ctx, "A", run)
124 if <-startOff {
125 t.Fatal("A starts off after a scene turned it on")
126 }
127 if err := set.wait(); err != nil {
128 t.Fatal(err)
129 }
130}
131
132// A late unregister of a previous lifetime never removes its successor.
133func TestRouterUnregisterKeepsNewerLifetime(t *testing.T) {
134 r := newOutputRouter(context.Background(), nil, outputRoutes{})
135 newer := &mailbox{instance: 2, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1)}
136 r.register("A", newer)
137 r.unregister("A", 1)
138 if r.outs["A"] != newer {
139 t.Fatal("stale unregister removed the newer lifetime")
140 }
141 r.unregister("A", 2)
142 if _, ok := r.outs["A"]; ok {
143 t.Fatal("unregister kept the lifetime")
144 }
145}
146
147// Lifecycle work on the backend loop never holds up routing: while a new
148// output's content attach blocks (it waits for the display goroutine), the
149// running output still gets its scenes and cursor.
150func TestBlockedStartDoesNotDelayRouting(t *testing.T) {
151 ctx, cancel := context.WithCancel(context.Background())
152 defer cancel()
153 in := newRouteInputs()
154 set := newOutputSet(ctx, nil, in.routes())
155 set.start(ctx, "A", idleRun)
156 <-set.outs["A"].cursor
157 attacher := newMockcontentAttacher(t)
158 set.attach = attacher
159 release := make(chan struct{})
160 attaching := make(chan struct{})
161 ret := portsmocks.NewMockContentReturn(t)
162 attacher.EXPECT().AttachOutput("B").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) {
163 close(attaching)
164 <-release
165 return nil, ret
166 }).Once()
167 attacher.EXPECT().DetachOutput(mock.Anything).Maybe()
168 releaseB := sync.OnceFunc(func() { close(release) })
169 wait := sync.OnceValue(set.wait)
170 started := make(chan bool, 1)
171 startDone := make(chan struct{})
172 go func() {
173 defer close(startDone)
174 started <- set.start(ctx, "B", idleRun)
175 }()
176 // A failed assertion still unblocks and joins B's start.
177 t.Cleanup(func() {
178 releaseB()
179 cancel()
180 <-startDone
181 _ = wait()
182 })
183 <-attaching
184 a := set.outs["A"]
185 deadline := time.After(2 * time.Second)
186 for seq := uint64(1); seq <= 3; seq++ {
187 select {
188 case in.scenes <- []ports.Scene{{Output: "A", Seq: seq}}:
189 case <-deadline:
190 t.Fatal("router blocked behind a starting output")
191 }
192 select {
193 case got := <-a.scenes:
194 if got.Seq != seq {
195 t.Fatalf("A scene %d, want %d", got.Seq, seq)
196 }
197 case <-deadline:
198 t.Fatal("scene not routed while another output starts")
199 }
200 }
201 select {
202 case in.cursor <- ports.CursorChange{Shape: "text"}:
203 case <-deadline:
204 t.Fatal("cursor blocked behind a starting output")
205 }
206 if got := <-a.cursor; got.Shape != "text" {
207 t.Fatalf("A cursor %+v", got)
208 }
209 releaseB()
210 if !<-started {
211 t.Fatal("B did not start")
212 }
213 if err := wait(); err != nil {
214 t.Fatal(err)
215 }
216}
217
218func TestOutputRoutingAllocations(t *testing.T) {
219 r := newOutputRouter(context.Background(), nil, outputRoutes{})
220 ctx, cancel := context.WithCancel(context.Background())
221 defer cancel()
222 m := newMailbox(1, ctx, cancel, make(chan struct{}))
223 r.register("A", m)
224 <-m.cursor
225 scenes := []ports.Scene{{Output: "A"}}
226 cursor := ports.CursorChange{Shape: "pointer"}
227 capture := ports.CaptureRequest{Output: "A"}
228 if got := testing.AllocsPerRun(100, func() {
229 r.scenes(scenes)
230 r.setCursor(cursor)
231 r.routeCapture(capture)
232 <-m.captures
233 }); got != 0 {
234 t.Fatalf("routing: %g allocations per run, want 0", got)
235 }
236}