1package screencast
2
3import (
4 "context"
5 "errors"
6 "sync/atomic"
7 "testing"
8 "time"
9
10 "github.com/bnema/neferwl/internal/adapters/captureallow"
11 "github.com/bnema/neferwl/internal/ports"
12 "github.com/bnema/purego-pipewire/pipewire"
13 "github.com/bnema/zerobus"
14 "github.com/stretchr/testify/mock"
15)
16
17// producer is a service whose owner state is driven directly by the test, on
18// the test goroutine (the owner): a generated mock bus, PipeWire core and
19// video source stand where the real ones would be.
20type producer struct {
21 t *testing.T
22 s *Service
23 st *stream
24 src *mockvideoSource
25 core *mockvideoCore
26 bus *mockbus
27 now time.Time
28 buf *pipewire.VideoBuffer
29 cmds chan ports.CastCommand
30}
31
32func newProducer(t *testing.T) *producer {
33 t.Helper()
34 cmds := make(chan ports.CastCommand, 16)
35 p := &producer{t: t, now: time.Unix(1000, 0), cmds: cmds}
36 p.s = New(func() *captureallow.Policy { return captureallow.Disabled() }, cmds, testLog(&logBuf{}))
37 t.Cleanup(func() { _ = p.s.poller.Close() })
38 p.bus = newMockbus(t)
39 p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
40 func(string, string, string, string, string) *zerobus.Encoder { return &zerobus.Encoder{} }).Maybe()
41 p.bus.EXPECT().Send().Return(1, nil).Maybe()
42 p.s.conn = p.bus
43 p.src = newMockvideoSource(t)
44 p.core = newMockvideoCore(t)
45 p.s.pwc = p.core
46 p.buf = &pipewire.VideoBuffer{Index: 3, Data: make([]byte, 64*32*4), Stride: 64 * 4, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
47
48 sess := &p.s.sessions[0]
49 *sess = session{used: true, id: 1, owner: ":1.7", path: prefixSession + "1", started: true}
50 st := &sess.streams[0]
51 *st = stream{used: true, sess: sess, id: 1, cast: 1, path: prefixStream + "1", output: "HEADLESS-1", maxFPS: 60, pwState: -2, needOpen: true}
52 st.h2 = streamHandler{st: st}
53 p.st = st
54 return p
55}
56
57func (p *producer) cmd() ports.CastCommand {
58 p.t.Helper()
59 select {
60 case c := <-p.cmds:
61 return c
62 default:
63 p.t.Fatal("no command")
64 return ports.CastCommand{}
65 }
66}
67
68func (p *producer) none() {
69 p.t.Helper()
70 select {
71 case c := <-p.cmds:
72 p.t.Fatalf("unexpected command %+v", c)
73 default:
74 }
75}
76
77// streaming opens the cast, sizes it, creates the source and brings it to
78// STREAMING with the negotiated format.
79func (p *producer) streaming() {
80 p.t.Helper()
81 p.s.pump(p.st, p.now)
82 if c := p.cmd(); c.Kind != ports.CastOpen || c.ID != 1 || c.Output != "HEADLESS-1" {
83 p.t.Fatalf("open %+v", c)
84 }
85 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
86 p.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
87 func(cfg pipewire.VideoConfig, h pipewire.VideoHandler) (videoSource, error) {
88 if cfg.Width != 64 || cfg.Height != 32 || cfg.MaxFPS != 60 {
89 p.t.Errorf("config %+v", cfg)
90 }
91 return p.src, nil
92 }).Once()
93 p.s.pump(p.st, p.now)
94 p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
95 p.st.h2.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
96 p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
97}
98
99// The node id goes to the session owner alone, once, as the stream's signal.
100func TestAnnouncesNodeToOwnerOnce(t *testing.T) {
101 p := newProducer(t)
102 var dest, path, member string
103 p.bus.ExpectedCalls = nil
104 p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
105 func(d, pa, iface, m, sig string) *zerobus.Encoder {
106 dest, path, member = d, pa, m
107 if iface != ifaceStream || sig != "u" {
108 t.Errorf("signal %s %s", iface, sig)
109 }
110 return &zerobus.Encoder{}
111 }).Once()
112 p.bus.EXPECT().Send().Return(1, nil).Once()
113 p.streaming()
114 p.src.EXPECT().Dequeue().Return(nil).Maybe()
115 p.s.pump(p.st, p.now)
116 p.s.pump(p.st, p.now)
117 if dest != ":1.7" || path != prefixStream+"1" || member != "PipeWireStreamAdded" {
118 t.Fatalf("signal to %q %q %q", dest, path, member)
119 }
120}
121
122// Nothing is requested without damage, except the first frame.
123func TestPacingForcesFirstFrameThenWaitsForDamage(t *testing.T) {
124 p := newProducer(t)
125 p.streaming()
126 p.src.EXPECT().Dequeue().Return(p.buf).Once()
127 p.s.pump(p.st, p.now)
128 c := p.cmd()
129 if c.Kind != ports.CastFrame || c.Buf != 3 || c.Stride != 256 || c.Width != 64 || c.Height != 32 || c.Format != shmXRGB8888 || len(c.Data) != 64*32*4 || &c.Data[0] != &p.buf.Data[0] {
130 t.Fatalf("frame %+v", c)
131 }
132 // Done: queued, then triggered, in that order.
133 var order []string
134 p.src.EXPECT().Queue(p.buf, uint64(1), mock.Anything, false).Run(func(*pipewire.VideoBuffer, uint64, time.Duration, bool) { order = append(order, "queue") }).Once()
135 p.src.EXPECT().Trigger().Run(func() { order = append(order, "trigger") }).Once()
136 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
137 if len(order) != 2 || order[0] != "queue" || order[1] != "trigger" {
138 t.Fatalf("order %v", order)
139 }
140 // No damage: no request however much time passes (Dequeue is not expected).
141 p.now = p.now.Add(time.Second)
142 p.s.pump(p.st, p.now)
143 p.none()
144}
145
146// At most one frame per interval, damage or not; the poll timeout is the
147// time left.
148func TestPacingOnePerInterval(t *testing.T) {
149 p := newProducer(t)
150 p.streaming()
151 p.src.EXPECT().Dequeue().Return(p.buf).Twice()
152 p.s.pump(p.st, p.now)
153 p.cmd()
154 p.src.EXPECT().Queue(mock.Anything, mock.Anything, mock.Anything, false).Return().Once()
155 p.src.EXPECT().Trigger().Return().Once()
156 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
157 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
158 p.now = p.now.Add(5 * time.Millisecond)
159 p.s.pumpAll(p.now)
160 p.none()
161 if got := p.s.timeout(); got < 11 || got > 12 {
162 t.Fatalf("timeout %d ms, want the ~11.7 ms left of 16.7", got)
163 }
164 p.now = p.now.Add(12 * time.Millisecond)
165 p.s.pumpAll(p.now)
166 if c := p.cmd(); c.Kind != ports.CastFrame {
167 t.Fatalf("%+v", c)
168 }
169 if got := p.s.timeout(); got != -1 {
170 t.Fatalf("idle timeout %d", got)
171 }
172}
173
174// A new size waits for the frame in flight, then renegotiates.
175func TestResizeWaitsForInflightFrame(t *testing.T) {
176 p := newProducer(t)
177 p.streaming()
178 p.src.EXPECT().Dequeue().Return(p.buf).Once()
179 p.s.pump(p.st, p.now)
180 p.cmd()
181 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 32, Height: 16})
182 p.s.pump(p.st, p.now) // Resize is not expected while the frame is in flight
183 p.src.EXPECT().Resize(32, 16).Return(nil).Once()
184 p.src.EXPECT().Release(p.buf).Return().Once()
185 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: ports.ErrCaptureTransient})
186 p.s.pump(p.st, p.now)
187 // No frame until the new format is negotiated.
188 p.now = p.now.Add(time.Second)
189 p.s.pump(p.st, p.now)
190 p.none()
191}
192
193// A failed frame gives the buffer back; three in a row stop the stream.
194func TestFrameErrorsReleaseAndStop(t *testing.T) {
195 p := newProducer(t)
196 p.streaming()
197 p.src.EXPECT().Dequeue().Return(p.buf)
198 p.src.EXPECT().Release(p.buf).Return()
199 for i := range maxFrameErrors {
200 p.s.pump(p.st, p.now)
201 if c := p.cmd(); c.Kind != ports.CastFrame {
202 t.Fatalf("frame %d: %+v", i, c)
203 }
204 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: errTest})
205 p.now = p.now.Add(time.Second)
206 }
207 if !p.st.closing || !p.s.sessions[0].dying {
208 t.Fatal("stream not stopped after three errors")
209 }
210 p.src.EXPECT().Destroy().Return().Once()
211 p.s.pump(p.st, p.now)
212 if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
213 t.Fatalf("%+v", c)
214 }
215 if p.st.used {
216 t.Fatal("stream slot not freed")
217 }
218}
219
220// Ending the session with a frame in flight closes the cast at once but
221// keeps the PipeWire node until the frame's CastFrameDone.
222func TestTeardownWaitsForFrameDone(t *testing.T) {
223 p := newProducer(t)
224 p.streaming()
225 p.src.EXPECT().Dequeue().Return(p.buf).Once()
226 p.s.pump(p.st, p.now)
227 p.cmd()
228 p.s.closeSession(p.st.sess, true)
229 p.s.pump(p.st, p.now)
230 if c := p.cmd(); c.Kind != ports.CastClose {
231 t.Fatalf("%+v", c)
232 }
233 p.s.pump(p.st, p.now)
234 p.s.reap()
235 if !p.st.used || !p.s.sessions[0].used {
236 t.Fatal("released while a frame is in flight") // Destroy is not expected yet
237 }
238 p.src.EXPECT().Release(p.buf).Return().Once()
239 p.src.EXPECT().Destroy().Return().Once()
240 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
241 p.s.pump(p.st, p.now)
242 p.s.reap()
243 if p.st.used || p.s.sessions[0].used {
244 t.Fatal("stream and session not freed after the frame")
245 }
246}
247
248// A cast that wayland ended closes the session and signals Closed.
249func TestCastStoppedClosesSession(t *testing.T) {
250 p := newProducer(t)
251 p.streaming()
252 p.bus.ExpectedCalls = nil
253 var member string
254 p.bus.EXPECT().NewSignalTo(":1.7", prefixSession+"1", ifaceSession, "Closed", "").RunAndReturn(
255 func(_, _, _, m, _ string) *zerobus.Encoder {
256 member = m
257 return &zerobus.Encoder{}
258 }).Once()
259 p.bus.EXPECT().Send().Return(1, nil).Once()
260 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastStopped, ID: 1, Reason: ports.CaptureReasonWindowGone})
261 if member != "Closed" {
262 t.Fatal("Closed not signalled")
263 }
264 p.src.EXPECT().Destroy().Return().Once()
265 p.s.pump(p.st, p.now)
266 // Wayland ended it itself: no CastClose.
267 p.none()
268}
269
270// A PipeWire stream that went to ERROR ends the session.
271func TestPipeWireErrorEndsSession(t *testing.T) {
272 p := newProducer(t)
273 p.streaming()
274 p.st.h2.StateChanged(pipewire.StreamStateError, 42, "gone")
275 p.s.pump(p.st, p.now)
276 if !p.s.sessions[0].dying {
277 t.Fatal("session still open")
278 }
279}
280
281var errTest = errors.New("frame failed")
282
283// The steady damage -> frame -> done cycle allocates nothing. Measured with
284// real values: the sink, the pacing and the settle logic. The PipeWire calls
285// of the cycle (Dequeue, Queue, Trigger) are the VideoSource's own,
286// measured in purego-pipewire (TestVideoSourceFrameAllocations); a mock in
287// the closure would record every call and grow.
288func TestCastProducerAllocations(t *testing.T) {
289 if raceEnabled {
290 t.Skip("the race detector makes the wake callback allocate")
291 }
292 p := newProducer(t)
293 p.cmds = make(chan ports.CastCommand, 4)
294 p.s.cmds = p.cmds
295 st := p.st
296 st.pwState, st.w, st.h, st.fw, st.fh, st.haveSize = pipewire.StreamStateStreaming, 64, 32, 64, 32, true
297 cmds := p.cmds
298 b := p.buf
299 damage := ports.CastEvent{Kind: ports.CastDamage, ID: 1}
300 done := ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: b.Index, Time: time.Unix(5, 0)}
301 now := p.now
302 cycle := func() {
303 p.s.Push(damage)
304 evs, _, _, _ := p.s.take()
305 for i := range evs {
306 p.s.onCastEvent(&evs[i])
307 }
308 now = now.Add(time.Second)
309 if !p.s.frameDue(st, now) || !p.s.issueFrame(st, b, now) {
310 t.Fatal("no frame")
311 }
312 <-cmds
313 p.s.Push(done)
314 evs, _, _, _ = p.s.take()
315 if p.s.settleFrame(st, &evs[0]) != b {
316 t.Fatal("frame not settled")
317 }
318 st.buf = nil
319 }
320 for range 8 {
321 cycle()
322 }
323 if n := testing.AllocsPerRun(200, cycle); n != 0 {
324 t.Fatalf("%v allocations per frame, want 0", n)
325 }
326}
327
328// The consumer leaves: the node goes to PAUSED and no frame is requested,
329// whatever damage and time bring, while the session stays open. A frame
330// already in flight is released, not queued. When the node streams again the
331// first frame is forced.
332func TestPausedNodeGetsNoFramesAndSessionStays(t *testing.T) {
333 p := newProducer(t)
334 p.streaming()
335 p.src.EXPECT().Dequeue().Return(p.buf).Twice()
336 p.s.pump(p.st, p.now)
337 if c := p.cmd(); c.Kind != ports.CastFrame {
338 t.Fatalf("%+v", c)
339 }
340 // Paused with the frame in flight: it is given back, not queued (no
341 // Queue or Trigger expectation).
342 p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
343 p.src.EXPECT().Release(p.buf).Return().Once()
344 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
345 for range 5 {
346 p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
347 p.now = p.now.Add(time.Second)
348 p.s.pumpAll(p.now) // Dequeue is not expected a second time while paused
349 p.none()
350 }
351 if p.s.sessions[0].dying || p.st.closing {
352 t.Fatal("the session ended with the consumer")
353 }
354 if got := p.s.timeout(); got != -1 {
355 t.Fatalf("paused stream asks for a timeout of %d ms", got)
356 }
357 // A consumer is back: the picture goes out at once, damage or not.
358 p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
359 p.s.pumpAll(p.now)
360 if c := p.cmd(); c.Kind != ports.CastFrame {
361 t.Fatalf("no frame after the node streams again: %+v", c)
362 }
363}
364
365// Shutdown pumps every stream: CastClose goes out, the node is destroyed and
366// PipeWire closed, with no frame in flight.
367func TestShutdownClosesStreams(t *testing.T) {
368 p := newProducer(t)
369 p.streaming()
370 var order []string
371 p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
372 p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
373 p.s.shutdown(context.Background())
374 if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
375 t.Fatalf("%+v", c)
376 }
377 if len(order) != 2 || order[0] != "destroy" || order[1] != "close" || p.s.pwc != nil {
378 t.Fatalf("order %v", order)
379 }
380 if p.st.used || p.s.sessions[0].used {
381 t.Fatal("slots not freed")
382 }
383}
384
385// A stream opened on the wayland side that has no node yet is closed too.
386func TestShutdownClosesStreamWithoutSource(t *testing.T) {
387 p := newProducer(t)
388 p.s.pump(p.st, p.now)
389 if c := p.cmd(); c.Kind != ports.CastOpen {
390 t.Fatalf("%+v", c)
391 }
392 p.core.EXPECT().Close().Return().Once()
393 p.s.shutdown(context.Background())
394 if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
395 t.Fatalf("%+v", c)
396 }
397 if p.st.used {
398 t.Fatal("stream not freed")
399 }
400}
401
402// A frame in flight at shutdown is waited for: nothing is released or
403// destroyed before its CastFrameDone, then the buffer is released, the node
404// destroyed and PipeWire closed, in that order. The wait has no deadline
405// while the service runs on (drain is far shorter than the wait here).
406func TestShutdownWaitsForFrameInFlight(t *testing.T) {
407 p := newProducer(t)
408 p.s.drain = time.Nanosecond
409 p.streaming()
410 p.src.EXPECT().Dequeue().Return(p.buf).Once()
411 p.s.pump(p.st, p.now)
412 p.cmd()
413 var delivered atomic.Bool
414 var order []string
415 step := func(name string) {
416 if !delivered.Load() {
417 t.Errorf("%s before the CastFrameDone", name)
418 }
419 order = append(order, name)
420 }
421 p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
422 p.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
423 p.core.EXPECT().Close().Run(func() { step("close") }).Once()
424 done := make(chan struct{})
425 go func() { defer close(done); p.s.shutdown(context.Background()) }()
426 select {
427 case c := <-p.cmds:
428 if c.Kind != ports.CastClose {
429 t.Errorf("%+v", c)
430 }
431 case <-time.After(5 * time.Second):
432 t.Fatal("no CastClose")
433 }
434 delivered.Store(true)
435 p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
436 select {
437 case <-done:
438 case <-time.After(5 * time.Second):
439 t.Fatal("shutdown did not return")
440 }
441 if len(order) != 3 || order[0] != "release" || order[1] != "destroy" || order[2] != "close" {
442 t.Fatalf("order %v", order)
443 }
444}
445
446// When the service itself is stopping and the frame never comes back, nothing is destroyed or closed (the
447// source's handler and its buffer are still in use), the storage the
448// handler points into is left untouched and abandoned, and the service has
449// a fresh table: repeated reconnects do not exhaust it.
450func TestShutdownTimeoutKeepsHandlerStateAndFreesTable(t *testing.T) {
451 p := newProducer(t)
452 p.s.drain = 20 * time.Millisecond
453 p.streaming()
454 p.src.EXPECT().Dequeue().Return(p.buf).Once()
455 p.s.pump(p.st, p.now)
456 p.cmd()
457 old := p.s.sessions
458 // The service itself is stopping: the wait is bounded.
459 stopped, stop := context.WithCancel(context.Background())
460 stop()
461 p.s.shutdown(stopped) // no Destroy, Release or Close is expected
462 if c := p.cmd(); c.Kind != ports.CastClose {
463 t.Fatalf("%+v", c)
464 }
465 if p.s.sessions == old {
466 t.Fatal("the table the leaked handler points into is still in use")
467 }
468 if !p.st.used || p.st.src == nil || !p.st.inflight || p.st.h2.st != p.st {
469 t.Fatal("the leaked stream's state was reset")
470 }
471 if p.s.streamsLive() || p.s.pwc != nil {
472 t.Fatal("new table not empty or PipeWire kept")
473 }
474 // The handler of the leaked source still works on its own storage.
475 p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
476 if p.st.pwState != pipewire.StreamStatePaused {
477 t.Fatal("handler lost its stream")
478 }
479 // A new attempt starts clean and can fill its table again.
480 p.s.reset(p.s.ctx, p.bus)
481 for i := range maxSessions {
482 p.s.sessions[i] = session{used: true, id: uint32(i + 1)}
483 }
484}