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 and pid
75// of every peer.
76type peers struct {
77 exe atomic.Value // string
78 exeNow func() string
79 pid atomic.Int64
80 openErr error
81 exePid func(open int) int // pid Exe reports; the opened one when nil
82
83 // exeFile, when set, tells the executable of one pidfd: the pidfds are
84 // numbered in the order Open handed them out (0 is the first peer).
85 exeFile func(n int) string
86 mu sync.Mutex
87 opened []*os.File
88}
89
90func (p *peers) number(f *os.File) int {
91 p.mu.Lock()
92 defer p.mu.Unlock()
93 for i, o := range p.opened {
94 if o == f {
95 return i
96 }
97 }
98 return -1
99}
100
101// svcHarness runs a Service on a private bus with mocked executables and
102// PipeWire.
103type svcHarness struct {
104 t *testing.T
105 addr string
106 svc *Service
107 cmds chan ports.CastCommand
108 peers *peers
109 core *mockvideoCore
110 run *mockrunner
111 log *logBuf
112 done chan struct{}
113
114 killBus func() // loses the bus: the daemon dies
115 cancel context.CancelFunc // stops the service, as the process exit does
116 closed chan struct{} // a PipeWire connection was closed
117}
118
119type harnessOpts struct {
120 allow *captureallow.Policy
121 noPW bool
122 // drain overrides how long the stopping service waits for frames.
123 drain time.Duration
124 // iterate runs inside every PipeWire loop iteration, on the owner.
125 iterate func()
126}
127
128func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
129 t.Helper()
130 h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
131 h.addr, h.killBus = startBus(t)
132 h.peers.exe.Store("/usr/bin/grim")
133 allow := o.allow
134 if allow == nil {
135 allow = captureallow.Builtin()
136 }
137 h.svc = New(func() *captureallow.Policy { return allow }, h.cmds, testLog(h.log))
138 h.svc.address = h.addr
139 if o.drain > 0 {
140 h.svc.drain = o.drain
141 }
142
143 res := newMockpeerResolver(t)
144 res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {
145 if h.peers.openErr != nil {
146 return nil, h.peers.openErr
147 }
148 h.peers.pid.Store(int64(pid))
149 f, err := os.Open(os.DevNull)
150 if err == nil {
151 h.peers.mu.Lock()
152 h.peers.opened = append(h.peers.opened, f)
153 h.peers.mu.Unlock()
154 }
155 return f, err
156 }).Maybe()
157 res.EXPECT().Exe(mock.Anything).RunAndReturn(func(f *os.File) (string, int, error) {
158 open := int(h.peers.pid.Load())
159 pid := open
160 if h.peers.exePid != nil {
161 pid = h.peers.exePid(open)
162 }
163 if h.peers.exeFile != nil {
164 return h.peers.exeFile(h.peers.number(f)), pid, nil
165 }
166 if h.peers.exeNow != nil {
167 return h.peers.exeNow(), pid, nil
168 }
169 return h.peers.exe.Load().(string), pid, nil
170 }).Maybe()
171 h.svc.peers = res
172 h.run = newMockrunner(t)
173 h.svc.runner = h.run
174
175 pwOpen := newMockpipewireOpener(t)
176 if o.noPW {
177 pwOpen.EXPECT().Open().Return(nil, os.ErrNotExist)
178 } else {
179 r, w, err := os.Pipe()
180 if err != nil {
181 t.Fatal(err)
182 }
183 t.Cleanup(func() { r.Close(); w.Close() })
184 h.core = newMockvideoCore(t)
185 h.core.EXPECT().Fd().Return(int(r.Fd()))
186 h.core.EXPECT().Iterate().Run(func() {
187 if o.iterate != nil {
188 o.iterate()
189 }
190 }).Maybe()
191 h.core.EXPECT().Close().Run(func() {
192 select {
193 case h.closed <- struct{}{}:
194 default:
195 }
196 }).Maybe()
197 pwOpen.EXPECT().Open().Return(h.core, nil)
198 }
199 h.svc.pw = pwOpen
200
201 ctx, cancel := context.WithCancel(context.Background())
202 h.cancel = cancel
203 go func() { defer close(h.done); h.svc.Run(ctx, 10*time.Millisecond) }()
204 t.Cleanup(func() {
205 cancel()
206 select {
207 case <-h.done:
208 case <-time.After(5 * time.Second):
209 t.Error("Run did not stop")
210 }
211 })
212 return h
213}
214
215// client is a bus connection of the test.
216func (h *svcHarness) client() *zerobus.Conn {
217 h.t.Helper()
218 c, err := zerobus.Dial(h.addr)
219 if err != nil {
220 h.t.Fatal(err)
221 }
222 h.t.Cleanup(func() { _ = c.Close() })
223 return c
224}
225
226// waitName waits until the service owns name.
227func (h *svcHarness) waitName(c *zerobus.Conn, name string) {
228 h.t.Helper()
229 deadline := time.Now().Add(5 * time.Second)
230 for time.Now().Before(deadline) {
231 e := c.NewCall("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "NameHasOwner", "s")
232 e.Str(name)
233 m, err := c.Call()
234 if err != nil {
235 h.t.Fatal(err)
236 }
237 if m.Body().Bool() {
238 return
239 }
240 time.Sleep(5 * time.Millisecond)
241 }
242 h.t.Fatalf("%s not owned", name)
243}
244
245// callErr calls and returns the D-Bus error name, "" on success.
246func callErr(c *zerobus.Conn, dest, path, iface, member, sig string, body func(*zerobus.Encoder)) (string, *zerobus.Message) {
247 e := c.NewCall(dest, path, iface, member, sig)
248 if body != nil {
249 body(e)
250 }
251 m, err := c.Call()
252 if err == nil {
253 return "", m
254 }
255 if de, ok := err.(*zerobus.Error); ok {
256 return de.Name, m
257 }
258 return err.Error(), m
259}