1package screencast
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "os"
8 "os/exec"
9 "strings"
10 "sync"
11 "sync/atomic"
12 "testing"
13 "time"
14
15 "github.com/bnema/neferwl/internal/adapters/captureallow"
16 "github.com/bnema/neferwl/internal/ports"
17 "github.com/bnema/zerobus"
18 "github.com/bnema/zerowrap"
19 "github.com/stretchr/testify/mock"
20)
21
22// startBus starts a dbus-daemon of its own, so tests never touch the
23// desktop's session bus, and returns its address and a function killing it,
24// which is how a test loses the bus. CI must run these tests: there, a
25// missing daemon fails.
26func startBus(t *testing.T) (string, func()) {
27 t.Helper()
28 bin, err := exec.LookPath("dbus-daemon")
29 if err != nil {
30 if os.Getenv("CI") != "" {
31 t.Fatal("dbus-daemon not installed")
32 }
33 t.Skip("dbus-daemon not installed")
34 }
35 cmd := exec.Command(bin, "--session", "--nofork", "--nopidfile", "--print-address=1")
36 out, err := cmd.StdoutPipe()
37 if err != nil {
38 t.Fatal(err)
39 }
40 if err := cmd.Start(); err != nil {
41 t.Fatal(err)
42 }
43 kill := func() { _ = cmd.Process.Kill() }
44 t.Cleanup(func() { kill(); _ = cmd.Wait() })
45 line, err := bufio.NewReader(out).ReadString('\n')
46 if err != nil {
47 t.Fatal(err)
48 }
49 return strings.TrimSpace(line), kill
50}
51
52// logBuf is a locked bytes.Buffer, a real io.Writer.
53type logBuf struct {
54 mu sync.Mutex
55 b bytes.Buffer
56}
57
58func (l *logBuf) Write(p []byte) (int, error) {
59 l.mu.Lock()
60 defer l.mu.Unlock()
61 return l.b.Write(p)
62}
63
64func (l *logBuf) String() string {
65 l.mu.Lock()
66 defer l.mu.Unlock()
67 return l.b.String()
68}
69
70func testLog(out *logBuf) zerowrap.Logger {
71 return zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: out}).WithField("component", "screencast")
72}
73
74// peers is what the mocked peerResolver answers: the executable of every
75// peer, and a hook to hold Open.
76type peers struct {
77 exe atomic.Value // string
78 exeNow func() string
79 openGo chan struct{} // when set, Open waits for it
80 pid atomic.Int64
81 openErr error
82 exePid func(open int) int // pid Exe reports; the opened one when nil
83
84 // exeFile, when set, tells the executable of one pidfd: the pidfds are
85 // numbered in the order Open handed them out (0 is the first peer).
86 exeFile func(n int) string
87 mu sync.Mutex
88 opened []*os.File
89}
90
91func (p *peers) number(f *os.File) int {
92 p.mu.Lock()
93 defer p.mu.Unlock()
94 for i, o := range p.opened {
95 if o == f {
96 return i
97 }
98 }
99 return -1
100}
101
102// svcHarness runs a Service on a private bus with mocked executables and
103// PipeWire.
104type svcHarness struct {
105 t *testing.T
106 addr string
107 svc *Service
108 cmds chan ports.CastCommand
109 peers *peers
110 core *mockvideoCore
111 run *mockrunner
112 log *logBuf
113 done chan struct{}
114
115 killBus func() // loses the bus: the daemon dies
116 cancel context.CancelFunc // stops the service, as the process exit does
117 closed chan struct{} // a PipeWire connection was closed
118}
119
120type harnessOpts struct {
121 allow *captureallow.Policy
122 noPW bool
123 holdOp bool
124 // drain overrides how long the stopping service waits for frames.
125 drain time.Duration
126 // iterate runs inside every PipeWire loop iteration, on the owner.
127 iterate func()
128}
129
130func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
131 t.Helper()
132 h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
133 h.addr, h.killBus = startBus(t)
134 h.peers.exe.Store("/usr/bin/grim")
135 if o.holdOp {
136 h.peers.openGo = make(chan struct{})
137 }
138 allow := o.allow
139 if allow == nil {
140 allow = captureallow.Builtin()
141 }
142 h.svc = New(func() *captureallow.Policy { return allow }, h.cmds, testLog(h.log))
143 h.svc.address = h.addr
144 if o.drain > 0 {
145 h.svc.drain = o.drain
146 }
147
148 res := newMockpeerResolver(t)
149 res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {
150 if h.peers.openGo != nil {
151 <-h.peers.openGo
152 }
153 if h.peers.openErr != nil {
154 return nil, h.peers.openErr
155 }
156 h.peers.pid.Store(int64(pid))
157 f, err := os.Open(os.DevNull)
158 if err == nil {
159 h.peers.mu.Lock()
160 h.peers.opened = append(h.peers.opened, f)
161 h.peers.mu.Unlock()
162 }
163 return f, err
164 }).Maybe()
165 res.EXPECT().Exe(mock.Anything).RunAndReturn(func(f *os.File) (string, int, error) {
166 open := int(h.peers.pid.Load())
167 pid := open
168 if h.peers.exePid != nil {
169 pid = h.peers.exePid(open)
170 }
171 if h.peers.exeFile != nil {
172 return h.peers.exeFile(h.peers.number(f)), pid, nil
173 }
174 if h.peers.exeNow != nil {
175 return h.peers.exeNow(), pid, nil
176 }
177 return h.peers.exe.Load().(string), pid, nil
178 }).Maybe()
179 h.svc.peers = res
180 h.run = newMockrunner(t)
181 h.svc.runner = h.run
182
183 pwOpen := newMockpipewireOpener(t)
184 if o.noPW {
185 pwOpen.EXPECT().Open().Return(nil, os.ErrNotExist)
186 } else {
187 r, w, err := os.Pipe()
188 if err != nil {
189 t.Fatal(err)
190 }
191 t.Cleanup(func() { r.Close(); w.Close() })
192 h.core = newMockvideoCore(t)
193 h.core.EXPECT().Fd().Return(int(r.Fd()))
194 h.core.EXPECT().Iterate().Run(func() {
195 if o.iterate != nil {
196 o.iterate()
197 }
198 }).Maybe()
199 h.core.EXPECT().Close().Run(func() {
200 select {
201 case h.closed <- struct{}{}:
202 default:
203 }
204 }).Maybe()
205 pwOpen.EXPECT().Open().Return(h.core, nil)
206 }
207 h.svc.pw = pwOpen
208
209 ctx, cancel := context.WithCancel(context.Background())
210 h.cancel = cancel
211 go func() { defer close(h.done); h.svc.Run(ctx, 10*time.Millisecond) }()
212 t.Cleanup(func() {
213 cancel()
214 select {
215 case <-h.done:
216 case <-time.After(5 * time.Second):
217 t.Error("Run did not stop")
218 }
219 })
220 return h
221}
222
223// client is a bus connection of the test.
224func (h *svcHarness) client() *zerobus.Conn {
225 h.t.Helper()
226 c, err := zerobus.Dial(h.addr)
227 if err != nil {
228 h.t.Fatal(err)
229 }
230 h.t.Cleanup(func() { _ = c.Close() })
231 return c
232}
233
234// waitName waits until the service owns name.
235func (h *svcHarness) waitName(c *zerobus.Conn, name string) {
236 h.t.Helper()
237 deadline := time.Now().Add(5 * time.Second)
238 for time.Now().Before(deadline) {
239 e := c.NewCall("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "NameHasOwner", "s")
240 e.Str(name)
241 m, err := c.Call()
242 if err != nil {
243 h.t.Fatal(err)
244 }
245 if m.Body().Bool() {
246 return
247 }
248 time.Sleep(5 * time.Millisecond)
249 }
250 h.t.Fatalf("%s not owned", name)
251}
252
253// callErr calls and returns the D-Bus error name, "" on success.
254func callErr(c *zerobus.Conn, dest, path, iface, member, sig string, body func(*zerobus.Encoder)) (string, *zerobus.Message) {
255 e := c.NewCall(dest, path, iface, member, sig)
256 if body != nil {
257 body(e)
258 }
259 m, err := c.Call()
260 if err == nil {
261 return "", m
262 }
263 if de, ok := err.(*zerobus.Error); ok {
264 return de.Name, m
265 }
266 return err.Error(), m
267}