1package app
2
3import (
4 "context"
5 "os"
6 "testing"
7 "time"
8
9 "github.com/bnema/neferwl/internal/ports"
10)
11
12func TestCaptureUnknownOutputClosesFile(t *testing.T) {
13 replies := make(chan ports.CaptureDone, 1)
14 set := newOutputSet(context.Background(), replies)
15 f, err := os.CreateTemp(t.TempDir(), "capture")
16 if err != nil {
17 t.Fatal(err)
18 }
19 set.routeCapture(ports.CaptureRequest{ID: 9, Output: "missing", Dst: ports.SHMBuffer{File: f}})
20 select {
21 case result := <-replies:
22 if result.ID != 9 || result.Err == nil {
23 t.Fatal(result)
24 }
25 if _, err := f.Stat(); err == nil {
26 t.Fatal("request descriptor still open")
27 }
28 case <-time.After(time.Second):
29 t.Fatal("no reply")
30 }
31}
32
33func TestCaptureFullOutputQueueDoesNotBlockRouting(t *testing.T) {
34 ctx, cancel := context.WithCancel(context.Background())
35 defer cancel()
36 replies := make(chan ports.CaptureDone, 1)
37 set := newOutputSet(ctx, replies)
38 queued := make(chan ports.CaptureRequest, 1)
39 queued <- ports.CaptureRequest{ID: 1}
40 set.outs["test"] = &runningOutput{ctx: ctx, captures: queued, done: make(chan struct{}), stop: func() {}}
41 f, err := os.CreateTemp(t.TempDir(), "capture")
42 if err != nil {
43 t.Fatal(err)
44 }
45 routed := make(chan struct{})
46 go func() {
47 set.routeCapture(ports.CaptureRequest{ID: 2, Output: "test", Dst: ports.SHMBuffer{File: f}})
48 close(routed)
49 }()
50 select {
51 case <-routed:
52 case <-time.After(time.Second):
53 cancel()
54 <-routed
55 t.Fatal("full capture queue blocked output routing")
56 }
57 result := <-replies
58 if result.ID != 2 || result.Err == nil {
59 t.Fatalf("rejected capture: %+v", result)
60 }
61 if _, err := f.Stat(); err == nil {
62 t.Fatal("rejected descriptor still open")
63 }
64 if got := <-queued; got.ID != 1 {
65 t.Fatalf("queued request replaced: %+v", got)
66 }
67}
68
69func TestStoppedOutputCaptureStillReplies(t *testing.T) {
70 ctx := context.Background()
71 outputCtx, stop := context.WithCancel(ctx)
72 stop()
73 replies := make(chan ports.CaptureDone, 1)
74 set := newOutputSet(ctx, replies)
75 set.outs["test"] = &runningOutput{ctx: outputCtx, captures: make(chan ports.CaptureRequest)}
76 f, err := os.CreateTemp(t.TempDir(), "capture")
77 if err != nil {
78 t.Fatal(err)
79 }
80 set.routeCapture(ports.CaptureRequest{ID: 7, Output: "test", Dst: ports.SHMBuffer{File: f}})
81 select {
82 case result := <-replies:
83 if result.ID != 7 || result.Err == nil {
84 t.Fatalf("stopped output reply: %+v", result)
85 }
86 default:
87 t.Fatal("stopped output lost capture completion")
88 }
89 if _, err := f.Stat(); err == nil {
90 t.Fatal("stopped output kept descriptor open")
91 }
92}
93
94func TestFinishedOutputDrainsCapturesWithReplies(t *testing.T) {
95 ctx := context.Background()
96 outputCtx, stop := context.WithCancel(ctx)
97 stop()
98 replies := make(chan ports.CaptureDone, 1)
99 set := newOutputSet(ctx, replies)
100 queued := make(chan ports.CaptureRequest, 1)
101 f, err := os.CreateTemp(t.TempDir(), "capture")
102 if err != nil {
103 t.Fatal(err)
104 }
105 queued <- ports.CaptureRequest{ID: 8, Dst: ports.SHMBuffer{File: f}}
106 set.failQueued(&runningOutput{ctx: outputCtx, captures: queued})
107 select {
108 case result := <-replies:
109 if result.ID != 8 || result.Err == nil {
110 t.Fatalf("drained capture reply: %+v", result)
111 }
112 default:
113 t.Fatal("drained capture lost completion")
114 }
115 if _, err := f.Stat(); err == nil {
116 t.Fatal("drained descriptor still open")
117 }
118}
119
120func TestDrainCapturesClosesQueuedFDs(t *testing.T) {
121 ctx, cancel := context.WithCancel(context.Background())
122 cancel()
123 pending := make(chan ports.CaptureRequest, 3)
124 for i := 0; i < 3; i++ {
125 f, err := os.CreateTemp(t.TempDir(), "queued")
126 if err != nil {
127 t.Fatal(err)
128 }
129 pending <- ports.CaptureRequest{Dst: ports.SHMBuffer{File: f}}
130 t.Cleanup(func() {
131 if _, err := f.Stat(); err == nil {
132 t.Error("queued fd not closed")
133 }
134 })
135 }
136 drainCaptures(ctx, pending, nil)
137 if len(pending) != 0 {
138 t.Fatal("requests still queued")
139 }
140}
141
142func TestOutputStopWithFullReplies(t *testing.T) {
143 ctx, cancel := context.WithCancel(context.Background())
144 replies := make(chan ports.CaptureDone, 1)
145 replies <- ports.CaptureDone{}
146 set := newOutputSet(ctx, replies)
147 set.start(ctx, "test", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, requests <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
148 <-ctx.Done()
149 return nil
150 })
151 f, err := os.CreateTemp(t.TempDir(), "queued")
152 if err != nil {
153 t.Fatal(err)
154 }
155 set.routeCapture(ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}})
156 cancel()
157 done := make(chan struct{})
158 go func() { _ = set.wait(); close(done) }()
159 select {
160 case <-done:
161 case <-time.After(2 * time.Second):
162 t.Fatal("output teardown blocked on replies")
163 }
164 if _, err := f.Stat(); err == nil {
165 t.Fatal("queued fd still open")
166 }
167}