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
16func pipelineRequest(t *testing.T, id uint64) ports.CaptureRequest {
17 t.Helper()
18 f, err := os.CreateTemp(t.TempDir(), "capture")
19 if err != nil {
20 t.Fatal(err)
21 }
22 t.Cleanup(func() { _ = f.Close() })
23 if err := f.Truncate(16); err != nil {
24 t.Fatal(err)
25 }
26 return ports.CaptureRequest{ID: id, Region: image.Rect(0, 0, 2, 2), Width: 2, Height: 2, Stride: 8, Format: 1, Dst: ports.SHMBuffer{File: f}}
27}
28
29func awaitCapture(t *testing.T, replies <-chan ports.CaptureDone) ports.CaptureDone {
30 t.Helper()
31 select {
32 case done := <-replies:
33 return done
34 case <-time.After(2 * time.Second):
35 t.Fatal("no capture reply")
36 return ports.CaptureDone{}
37 }
38}
39
40func TestPipelineCopiesOffOwnerAndRecycles(t *testing.T) {
41 r := portsmocks.NewMockRenderer(t)
42 frame := portsmocks.NewMockCaptureFrame(t)
43 frame.EXPECT().Done().Return(nil)
44 entered, release := make(chan struct{}), make(chan struct{})
45 frame.EXPECT().Read(image.Rect(0, 0, 2, 2), mock.Anything, 8).RunAndReturn(func(_ image.Rectangle, dst []byte, _ int) error {
46 close(entered)
47 <-release
48 copy(dst, []byte{1, 2, 3, 255})
49 return nil
50 }).Once()
51 r.EXPECT().BeginCapture().Return(frame, nil).Once()
52 r.EXPECT().EndCapture(frame).Return().Once()
53 replies := make(chan ports.CaptureDone, 1)
54 p := NewPipeline(context.Background(), replies)
55 q := pipelineRequest(t, 1)
56 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{q})
57 select {
58 case <-entered:
59 case <-time.After(time.Second):
60 t.Fatal("copy worker did not run")
61 }
62 // Submit returned while the worker is still inside Read.
63 select {
64 case <-replies:
65 t.Fatal("reply before copy completed")
66 default:
67 }
68 close(release)
69 done := awaitCapture(t, replies)
70 if done.Err != nil || done.ID != q.ID {
71 t.Fatalf("result %+v", done)
72 }
73 if _, err := q.Dst.File.Stat(); err == nil {
74 t.Fatal("descriptor open at completion")
75 }
76 p.Close(r)
77 // Mockery retains Read arguments, which point at an unmapped SHM slice.
78 frame.Calls = nil
79}
80
81func TestPipelineSaturationAndShutdownWithoutGPUCompletion(t *testing.T) {
82 r := portsmocks.NewMockRenderer(t)
83 replies := make(chan ports.CaptureDone, 4)
84 p := NewPipeline(context.Background(), replies)
85 for id := range uint64(2) {
86 read, write, err := os.Pipe()
87 if err != nil {
88 t.Fatal(err)
89 }
90 t.Cleanup(func() { _ = read.Close(); _ = write.Close() })
91 frame := portsmocks.NewMockCaptureFrame(t)
92 frame.EXPECT().Done().Return(read).Maybe()
93 r.EXPECT().BeginCapture().Return(frame, nil).Once()
94 r.EXPECT().EndCapture(frame).Return().Once()
95 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{pipelineRequest(t, id)})
96 }
97 q := pipelineRequest(t, 3)
98 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{q})
99 if done := awaitCapture(t, replies); done.ID != 3 || done.Err == nil {
100 t.Fatalf("saturation result %+v", done)
101 }
102 closed := make(chan struct{})
103 go func() { p.Close(r); close(closed) }()
104 select {
105 case <-closed:
106 case <-time.After(time.Second):
107 t.Fatal("shutdown waited for GPU fence")
108 }
109 for range 2 {
110 if done := awaitCapture(t, replies); done.Err == nil {
111 t.Fatalf("cancelled result %+v", done)
112 }
113 }
114}
115
116func TestPipelineBeginFailureClosesRequest(t *testing.T) {
117 r := portsmocks.NewMockRenderer(t)
118 boom := errors.New("submission failed")
119 r.EXPECT().BeginCapture().Return(nil, boom).Once()
120 replies := make(chan ports.CaptureDone, 1)
121 p := NewPipeline(context.Background(), replies)
122 q := pipelineRequest(t, 1)
123 p.SubmitScoped(ports.SecurityState{}, r, []ports.CaptureRequest{q})
124 if done := awaitCapture(t, replies); !errors.Is(done.Err, boom) {
125 t.Fatalf("result %+v", done)
126 }
127 if _, err := q.Dst.File.Stat(); err == nil {
128 t.Fatal("descriptor open")
129 }
130 p.Close(r)
131}
132
133func TestPipelineRejectsOversizedBatch(t *testing.T) {
134 r := portsmocks.NewMockRenderer(t)
135 replies := make(chan ports.CaptureDone, MaxRequests+1)
136 p := NewPipeline(context.Background(), replies)
137 requests := make([]ports.CaptureRequest, MaxRequests+1)
138 for i := range requests {
139 requests[i] = pipelineRequest(t, uint64(i))
140 }
141 p.SubmitScoped(ports.SecurityState{}, r, requests)
142 for range requests {
143 if done := awaitCapture(t, replies); done.Err == nil {
144 t.Fatal("oversized batch accepted")
145 }
146 }
147 p.Close(r)
148}