1package capture
2
3import (
4 "context"
5 "errors"
6 "image"
7 "os"
8 "testing"
9 "time"
10
11 portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
12 "github.com/bnema/neferwl/internal/ports"
13 "github.com/stretchr/testify/mock"
14)
15
16// fencePipe returns a pipe standing in for a sync file: readable is signalled.
17func fencePipe(t *testing.T) (read, write *os.File) {
18 t.Helper()
19 read, write, err := os.Pipe()
20 if err != nil {
21 t.Fatal(err)
22 }
23 t.Cleanup(func() { _ = read.Close(); _ = write.Close() })
24 return read, write
25}
26
27func TestPipelineWaitsForFenceBeforeRead(t *testing.T) {
28 r := portsmocks.NewMockRenderer(t)
29 frame := portsmocks.NewMockCaptureFrame(t)
30 fence, signal := fencePipe(t)
31 frame.EXPECT().Done().Return(fence)
32 entered := make(chan struct{})
33 frame.EXPECT().Read(image.Rect(0, 0, 2, 2), mock.Anything, 8).RunAndReturn(func(_ image.Rectangle, dst []byte, _ int) error {
34 close(entered)
35 copy(dst, []byte{1, 2, 3, 255})
36 return nil
37 }).Once()
38 r.EXPECT().BeginCapture().Return(frame, nil).Once()
39 r.EXPECT().EndCapture(frame).Return().Once()
40 replies := make(chan ports.CaptureDone, 1)
41 p := NewPipeline(context.Background(), replies)
42 q := pipelineRequest(t, 1)
43 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{q})
44 // Longer than one syncfile poll slice: the worker has polled at least once.
45 select {
46 case <-entered:
47 t.Fatal("Read before the fence signalled")
48 case <-replies:
49 t.Fatal("reply before the fence signalled")
50 case <-time.After(250 * time.Millisecond):
51 }
52 if _, err := signal.Write([]byte{1}); err != nil {
53 t.Fatal(err)
54 }
55 done := awaitCapture(t, replies)
56 if done.Err != nil || done.ID != q.ID {
57 t.Fatalf("result %+v", done)
58 }
59 select {
60 case <-entered:
61 default:
62 t.Fatal("Read never ran")
63 }
64 if _, err := q.Dst.File.Stat(); err == nil {
65 t.Fatal("descriptor open at completion")
66 }
67 p.Close(r)
68 frame.Calls = nil
69}
70
71func TestPipelinePropagatesReadError(t *testing.T) {
72 r := portsmocks.NewMockRenderer(t)
73 frame := portsmocks.NewMockCaptureFrame(t)
74 frame.EXPECT().Done().Return(nil)
75 boom := errors.New("readback failed")
76 frame.EXPECT().Read(image.Rect(0, 0, 2, 2), mock.Anything, 8).Return(boom).Once()
77 r.EXPECT().BeginCapture().Return(frame, nil).Once()
78 r.EXPECT().EndCapture(frame).Return().Once()
79 replies := make(chan ports.CaptureDone, 1)
80 p := NewPipeline(context.Background(), replies)
81 q := pipelineRequest(t, 1)
82 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{q})
83 done := awaitCapture(t, replies)
84 if !errors.Is(done.Err, boom) || done.ID != q.ID {
85 t.Fatalf("result %+v", done)
86 }
87 if _, err := q.Dst.File.Stat(); err == nil {
88 t.Fatal("descriptor open at completion")
89 }
90 // The lease returns even though the read failed.
91 p.Close(r)
92 frame.Calls = nil
93}