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 router := newOutputRouter(context.Background(), replies, outputRoutes{})
15 f, err := os.CreateTemp(t.TempDir(), "capture")
16 if err != nil {
17 t.Fatal(err)
18 }
19 router.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 router := newOutputRouter(ctx, replies, outputRoutes{})
38 queued := make(chan ports.CaptureRequest, 1)
39 queued <- ports.CaptureRequest{ID: 1}
40 router.outs["test"] = &mailbox{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 router.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 router := newOutputRouter(ctx, replies, outputRoutes{})
75 router.outs["test"] = &mailbox{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 router.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 replies := make(chan ports.CaptureDone, 1)
97 queued := make(chan ports.CaptureRequest, 1)
98 f, err := os.CreateTemp(t.TempDir(), "capture")
99 if err != nil {
100 t.Fatal(err)
101 }
102 queued <- ports.CaptureRequest{ID: 8, Dst: ports.SHMBuffer{File: f}}
103 failQueued(ctx, &mailbox{captures: queued}, replies)
104 select {
105 case result := <-replies:
106 if result.ID != 8 || result.Err == nil {
107 t.Fatalf("drained capture reply: %+v", result)
108 }
109 default:
110 t.Fatal("drained capture lost completion")
111 }
112 if _, err := f.Stat(); err == nil {
113 t.Fatal("drained descriptor still open")
114 }
115}
116
117func TestDrainCapturesClosesQueuedFDs(t *testing.T) {
118 ctx, cancel := context.WithCancel(context.Background())
119 cancel()
120 pending := make(chan ports.CaptureRequest, 3)
121 for i := 0; i < 3; i++ {
122 f, err := os.CreateTemp(t.TempDir(), "queued")
123 if err != nil {
124 t.Fatal(err)
125 }
126 pending <- ports.CaptureRequest{Dst: ports.SHMBuffer{File: f}}
127 t.Cleanup(func() {
128 if _, err := f.Stat(); err == nil {
129 t.Error("queued fd not closed")
130 }
131 })
132 }
133 drainCaptures(ctx, pending, nil)
134 if len(pending) != 0 {
135 t.Fatal("requests still queued")
136 }
137}
138
139func TestOutputStopWithFullReplies(t *testing.T) {
140 ctx, cancel := context.WithCancel(context.Background())
141 replies := make(chan ports.CaptureDone, 1)
142 replies <- ports.CaptureDone{}
143 captures := make(chan ports.CaptureRequest)
144 set := newOutputSet(ctx, replies, outputRoutes{captures: captures})
145 set.start(ctx, "test", idleRun)
146 f, err := os.CreateTemp(t.TempDir(), "queued")
147 if err != nil {
148 t.Fatal(err)
149 }
150 captures <- ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}}
151 cancel()
152 done := make(chan struct{})
153 go func() { _ = set.wait(); close(done) }()
154 select {
155 case <-done:
156 case <-time.After(2 * time.Second):
157 t.Fatal("output teardown blocked on replies")
158 }
159 if _, err := f.Stat(); err == nil {
160 t.Fatal("queued fd still open")
161 }
162}