1package screencast
2
3import (
4 "sync/atomic"
5 "testing"
6 "time"
7
8 "github.com/bnema/neferwl/internal/ports"
9 "github.com/bnema/purego-pipewire/pipewire"
10 "github.com/stretchr/testify/mock"
11)
12
13// liveCast is a cast on a private bus, brought to STREAMING with a mocked
14// PipeWire source, so a test can lose the bus under it.
15type liveCast struct {
16 h *svcHarness
17 src *mockvideoSource
18 buf *pipewire.VideoBuffer
19 // onLoop runs functions inside the PipeWire loop iteration, where the
20 // real handler callbacks run.
21 onLoop chan func()
22 hdl chan pipewire.VideoHandler
23}
24
25func (l *liveCast) loop(f func()) {
26 l.onLoop <- f
27 l.h.svc.Push(ports.CastEvent{Kind: ports.CastDamage, ID: 1 << 40}) // wakes the owner
28}
29
30func startLiveCast(t *testing.T, withFrame bool, drain time.Duration) *liveCast {
31 t.Helper()
32 l := &liveCast{onLoop: make(chan func(), 8), hdl: make(chan pipewire.VideoHandler, 1)}
33 l.h, _ = ready(t, harnessOpts{drain: drain, iterate: func() {
34 select {
35 case f := <-l.onLoop:
36 f()
37 default:
38 }
39 }})
40 h := l.h
41 c := h.client()
42 l.src = newMockvideoSource(t)
43 l.buf = &pipewire.VideoBuffer{Index: 2, Data: make([]byte, 64*32*4), Stride: 256, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
44 h.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
45 func(_ pipewire.VideoConfig, hd pipewire.VideoHandler) (videoSource, error) {
46 l.hdl <- hd
47 return l.src, nil
48 }).Once()
49 if withFrame {
50 l.src.EXPECT().Dequeue().Return(l.buf).Once()
51 } else {
52 l.src.EXPECT().Dequeue().Return(nil).Maybe()
53 }
54
55 sess := createSession(t, c)
56 if name, _ := recordMonitor(c, sess, "HEADLESS-1"); name != "" {
57 t.Fatal(name)
58 }
59 if name, _ := callErr(c, nameScreenCast, sess, ifaceSession, "Start", "", nil); name != "" {
60 t.Fatal(name)
61 }
62 if cmd := recvCmd(t, h); cmd.Kind != ports.CastOpen {
63 t.Fatalf("%+v", cmd)
64 }
65 h.svc.Push(ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
66 var hd pipewire.VideoHandler
67 select {
68 case hd = <-l.hdl:
69 case <-time.After(5 * time.Second):
70 t.Fatal("no PipeWire source")
71 }
72 l.loop(func() {
73 hd.StateChanged(pipewire.StreamStatePaused, 42, "")
74 hd.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
75 hd.StateChanged(pipewire.StreamStateStreaming, 42, "")
76 })
77 if withFrame {
78 if cmd := recvCmd(t, h); cmd.Kind != ports.CastFrame || cmd.Buf != 2 {
79 t.Fatalf("%+v", cmd)
80 }
81 }
82 return l
83}
84
85func recvCmd(t *testing.T, h *svcHarness) ports.CastCommand {
86 t.Helper()
87 select {
88 case c := <-h.cmds:
89 return c
90 case <-time.After(5 * time.Second):
91 t.Fatal("no command\n" + h.log.String())
92 return ports.CastCommand{}
93 }
94}
95
96func waitClosed(t *testing.T, h *svcHarness) {
97 t.Helper()
98 select {
99 case <-h.closed:
100 case <-time.After(5 * time.Second):
101 t.Fatal("PipeWire not closed\n" + h.log.String())
102 }
103}
104
105// Losing the bus with a live stream and no frame in flight: the cast is
106// closed on the wayland side, the source destroyed, PipeWire closed.
107func TestBusLossTearsDownStream(t *testing.T) {
108 l := startLiveCast(t, false, 0)
109 l.src.EXPECT().Destroy().Return().Once()
110 l.h.killBus()
111 if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
112 t.Fatalf("%+v", cmd)
113 }
114 waitClosed(t, l.h)
115}
116
117// With a frame in flight: CastClose goes out at once, but the source is
118// destroyed only after the frame's CastFrameDone, even when that comes later
119// than drain: the service runs on, so nothing is left behind. The order is
120// synchronized by channels: the buffer must be released before the node is
121// destroyed, and PipeWire closed after both.
122func TestBusLossWaitsForFrameInFlight(t *testing.T) {
123 l := startLiveCast(t, true, time.Nanosecond)
124 l.h.killBus()
125 if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
126 t.Fatalf("%+v", cmd)
127 }
128 var delivered atomic.Bool
129 steps := make(chan string, 3)
130 step := func(name string) {
131 if !delivered.Load() {
132 t.Errorf("%s before the CastFrameDone", name)
133 }
134 steps <- name
135 }
136 l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
137 l.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
138 delivered.Store(true)
139 l.h.svc.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 2})
140 waitClosed(t, l.h)
141 if len(steps) != 2 || <-steps != "release" || <-steps != "destroy" {
142 t.Fatal("release and destroy did not run in order before PipeWire closed")
143 }
144}
145
146// The service itself stopping with a frame that is never answered returns
147// within the bounded wait, and leaves the source and PipeWire alone.
148func TestStopWithUnansweredFrameIsBounded(t *testing.T) {
149 l := startLiveCast(t, true, 20*time.Millisecond)
150 l.h.cancel()
151 if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
152 t.Fatalf("%+v", cmd)
153 }
154 select {
155 case <-l.h.done:
156 case <-time.After(5 * time.Second):
157 t.Fatal("Run did not return")
158 }
159 select {
160 case <-l.h.closed:
161 t.Fatal("PipeWire closed under a source with a frame in flight")
162 default:
163 }
164}