1package capture
2
3import (
4 "context"
5 "testing"
6
7 "github.com/bnema/neferwl/internal/ports"
8 "github.com/stretchr/testify/require"
9)
10
11func TestRequestsAdmitRefuseAndFull(t *testing.T) {
12 replies := make(chan ports.CaptureDone, MaxRequests+2)
13 var r Requests
14 r.Init(context.Background(), replies)
15 scene := ports.Scene{OutputWidth: 10, OutputHeight: 10}
16
17 require.False(t, r.Admit(ports.CaptureRequest{ID: 1}, ErrOutputOff, scene, true))
18 require.ErrorIs(t, (<-replies).Err, ErrOutputOff)
19 require.Zero(t, r.Len())
20
21 // A request without an indicator is ready at once on a known scene.
22 require.True(t, r.Admit(ports.CaptureRequest{ID: 2}, nil, scene, true))
23 require.False(t, r.Admit(ports.CaptureRequest{ID: 3}, nil, scene, false), "no scene yet")
24 require.False(t, r.List[0].Since.IsZero())
25 for i := r.Len(); i < MaxRequests; i++ {
26 r.Admit(ports.CaptureRequest{ID: uint64(10 + i)}, nil, scene, true)
27 }
28 require.False(t, r.Admit(ports.CaptureRequest{ID: 99}, nil, scene, true))
29 done := <-replies
30 require.Equal(t, uint64(99), done.ID)
31 require.ErrorIs(t, done.Err, ErrBatchFull)
32 require.Equal(t, MaxRequests, r.Len())
33}
34
35func TestRequestsFailPendingSkipsHanded(t *testing.T) {
36 replies := make(chan ports.CaptureDone, 4)
37 var r Requests
38 r.Init(context.Background(), replies)
39 r.Admit(ports.CaptureRequest{ID: 1}, nil, ports.Scene{}, false)
40 r.Admit(ports.CaptureRequest{ID: 2}, nil, ports.Scene{}, false)
41 r.List[0] = ports.CaptureRequest{} // handed to the worker
42 r.FailPending(ErrEpochChanged)
43 require.Zero(t, r.Len())
44 require.Len(t, replies, 1)
45 done := <-replies
46 require.Equal(t, uint64(2), done.ID)
47 require.ErrorIs(t, done.Err, ErrEpochChanged)
48}
49
50func TestRequestsStopDrainsIncoming(t *testing.T) {
51 replies := make(chan ports.CaptureDone, 4)
52 var r Requests
53 r.Init(context.Background(), replies)
54 r.Admit(ports.CaptureRequest{ID: 1}, nil, ports.Scene{}, false)
55 r.Admit(ports.CaptureRequest{ID: 3}, nil, ports.Scene{}, false)
56 r.List[1] = ports.CaptureRequest{} // handed to the pipeline
57 incoming := make(chan ports.CaptureRequest, 2)
58 incoming <- ports.CaptureRequest{ID: 2}
59 close(incoming)
60 r.Stop(incoming)
61 require.Len(t, replies, 2)
62 for _, id := range []uint64{1, 2} {
63 done := <-replies
64 require.Equal(t, id, done.ID)
65 require.ErrorIs(t, done.Err, ErrOutputStopped)
66 }
67}
68
69func TestRequestsStopReturnsOnOpenEmptyChannel(t *testing.T) {
70 replies := make(chan ports.CaptureDone, 1)
71 var r Requests
72 r.Init(context.Background(), replies)
73 r.Stop(make(chan ports.CaptureRequest))
74 require.Empty(t, replies)
75}