1package sshstdio
2
3import (
4 "bytes"
5 "encoding/json"
6 "errors"
7 "io"
8 "os"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/bnema/vev/internal/adapters/observability"
15 "github.com/bnema/vev/internal/ports"
16)
17
18func TestTransportObservabilitySSHStdioEOFEndsReceive(t *testing.T) {
19 tracePath := t.TempDir() + "/ssh-eof.jsonl"
20 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 309)}, "ssh-process")
21 if err != nil {
22 t.Fatalf("NewJSONL() error = %v", err)
23 }
24 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
25 defer reporter.Close()
26 if _, err := NewTransport(bytes.NewReader(nil), &bytes.Buffer{}, nil, WithRuntimeObserver(reporter)).Recv(); err == nil {
27 t.Fatal("Recv() error = nil at EOF")
28 }
29 reporter.Close()
30 if err := closer.Close(); err != nil {
31 t.Fatalf("trace Close() error = %v", err)
32 }
33 assertSSHReceiveFailurePair(t, tracePath)
34}
35
36func TestTransportObservabilitySSHStdioCloseEndsBlockedReceive(t *testing.T) {
37 tracePath := t.TempDir() + "/ssh-close.jsonl"
38 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 311)}, "ssh-process")
39 if err != nil {
40 t.Fatalf("NewJSONL() error = %v", err)
41 }
42 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
43 defer reporter.Close()
44 reader := &sshShutdownReader{entered: make(chan struct{}), done: make(chan struct{})}
45 shutdownErr := errors.New("ssh shutdown failed")
46 transport := NewTransport(reader, &bytes.Buffer{}, func() error { return shutdownErr }, WithRuntimeObserver(reporter))
47 recvDone := make(chan error, 1)
48 go func() { _, err := transport.Recv(); recvDone <- err }()
49 <-reader.entered
50 if err := transport.Close(); !errors.Is(err, shutdownErr) {
51 t.Fatalf("Close() error = %v, want %v", err, shutdownErr)
52 }
53 if err := <-recvDone; err == nil {
54 t.Fatal("Recv() error = nil after Close()")
55 }
56 reporter.Close()
57 if err := closer.Close(); err != nil {
58 t.Fatalf("trace Close() error = %v", err)
59 }
60 assertSSHReceiveFailurePair(t, tracePath)
61}
62
63type sshShutdownReader struct {
64 entered chan struct{}
65 done chan struct{}
66 once sync.Once
67}
68
69func (r *sshShutdownReader) Read([]byte) (int, error) {
70 r.once.Do(func() { close(r.entered) })
71 <-r.done
72 return 0, errors.New("reader closed")
73}
74
75func (r *sshShutdownReader) Close() error {
76 r.once.Do(func() { close(r.entered) })
77 close(r.done)
78 return nil
79}
80
81func assertSSHReceiveFailurePair(t *testing.T, path string) {
82 t.Helper()
83 data, err := os.ReadFile(path)
84 if err != nil {
85 t.Fatalf("ReadFile(trace): %v", err)
86 }
87 var starts, ends []struct {
88 ProcessID string `json:"process_id"`
89 Scenario string `json:"scenario"`
90 Run uint64 `json:"run"`
91 Sequence uint64 `json:"sequence"`
92 RequestID uint64 `json:"request_id"`
93 Epoch uint64 `json:"epoch"`
94 Kind string `json:"kind"`
95 Valid bool `json:"valid"`
96 }
97 for line := range strings.SplitSeq(strings.TrimSpace(string(data)), "\n") {
98 var mark struct {
99 ProcessID string `json:"process_id"`
100 Scenario string `json:"scenario"`
101 Run uint64 `json:"run"`
102 Sequence uint64 `json:"sequence"`
103 RequestID uint64 `json:"request_id"`
104 Epoch uint64 `json:"epoch"`
105 Kind string `json:"kind"`
106 Valid bool `json:"valid"`
107 }
108 if err := json.Unmarshal([]byte(line), &mark); err != nil {
109 t.Fatalf("Unmarshal(trace): %v", err)
110 }
111 switch mark.Kind {
112 case "adapter_receive_start":
113 starts = append(starts, mark)
114 case "adapter_receive_end":
115 ends = append(ends, mark)
116 }
117 }
118 if len(starts) != 1 || len(ends) != 1 {
119 t.Fatalf("receive spans start=%d end=%d, want exactly one pair", len(starts), len(ends))
120 }
121 start, end := starts[0], ends[0]
122 if start.ProcessID != end.ProcessID || start.Scenario != end.Scenario || start.Run != end.Run || start.Sequence != end.Sequence || start.RequestID != end.RequestID || start.Epoch != end.Epoch {
123 t.Fatalf("mismatched receive correlation: start=%+v end=%+v", start, end)
124 }
125 if !start.Valid || end.Valid {
126 t.Fatalf("receive validity start=%t end=%t, want true/false", start.Valid, end.Valid)
127 }
128}
129
130func TestProcessStdinPumpIsSingleton(t *testing.T) {
131 first := processStdinPumpFor(os.Stdin)
132 second := processStdinPumpFor(os.Stdin)
133 if first != second {
134 t.Fatal("process stdin created more than one lifetime pump")
135 }
136}
137
138func TestTransportCloseLeavesProcessStdinOpen(t *testing.T) {
139 transport := NewTransport(os.Stdin, io.Discard, nil)
140 if err := transport.Close(); err != nil {
141 t.Fatalf("Close() error = %v", err)
142 }
143 if _, err := os.Stdin.Stat(); err != nil {
144 t.Fatalf("process stdin was closed: %v", err)
145 }
146}
147
148func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *testing.T) {
149 tracePath := t.TempDir() + "/ssh-unowned-close.jsonl"
150 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 313)}, "ssh-process")
151 if err != nil {
152 t.Fatalf("NewJSONL() error = %v", err)
153 }
154 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
155 defer reporter.Close()
156 reader := &sshUnownedBlockedReader{entered: make(chan struct{}), release: make(chan struct{})}
157 t.Cleanup(reader.unblock)
158 transport := NewTransport(reader, &bytes.Buffer{}, nil, WithRuntimeObserver(reporter))
159 recvDone := make(chan error, 1)
160 go func() { _, err := transport.Recv(); recvDone <- err }()
161 <-reader.entered
162
163 closeDone := make(chan error, 1)
164 go func() { closeDone <- transport.Close() }()
165 select {
166 case err := <-closeDone:
167 if err != nil {
168 t.Fatalf("Close() error = %v", err)
169 }
170 case <-time.After(time.Second):
171 t.Fatal("Close() waited for an unowned blocked reader")
172 }
173
174 select {
175 case err := <-recvDone:
176 if err == nil {
177 t.Fatal("Recv() error = nil after Close()")
178 }
179 case <-time.After(time.Second):
180 t.Fatal("Recv() did not finish after Close()")
181 }
182 reporter.Close()
183 if err := closer.Close(); err != nil {
184 t.Fatalf("trace Close() error = %v", err)
185 }
186 assertSSHReceiveFailurePair(t, tracePath)
187}
188
189type sshUnownedBlockedReader struct {
190 entered chan struct{}
191 release chan struct{}
192 enteredOnce sync.Once
193 releaseOnce sync.Once
194}
195
196func (r *sshUnownedBlockedReader) Read([]byte) (int, error) {
197 r.enteredOnce.Do(func() { close(r.entered) })
198 <-r.release
199 return 0, errors.New("reader released")
200}
201
202func (r *sshUnownedBlockedReader) unblock() {
203 r.releaseOnce.Do(func() { close(r.release) })
204}
205
206func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
207 tracePath := t.TempDir() + "/ssh.jsonl"
208 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")
209 if err != nil {
210 t.Fatalf("NewJSONL() error = %v", err)
211 }
212 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
213 defer reporter.Close()
214
215 frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte("stdio bytes stay exact")}
216 var observed, baseline bytes.Buffer
217 if err := NewTransport(bytes.NewReader(nil), &observed, nil, WithRuntimeObserver(reporter)).Send(frame); err != nil {
218 t.Fatalf("observed Send() error = %v", err)
219 }
220 if err := NewTransport(bytes.NewReader(nil), &baseline, nil).Send(frame); err != nil {
221 t.Fatalf("baseline Send() error = %v", err)
222 }
223 if !bytes.Equal(observed.Bytes(), baseline.Bytes()) {
224 t.Fatalf("observer changed SSH stdio wire bytes: got %x, want %x", observed.Bytes(), baseline.Bytes())
225 }
226 received, err := NewTransport(bytes.NewReader(baseline.Bytes()), &bytes.Buffer{}, nil, WithRuntimeObserver(reporter)).Recv()
227 if err != nil {
228 t.Fatalf("observed Recv() error = %v", err)
229 }
230 if received.Type != frame.Type || !bytes.Equal(received.Payload, frame.Payload) {
231 t.Fatalf("observed Recv() = %#v, want %#v", received, frame)
232 }
233 reporter.Close()
234 if err := closer.Close(); err != nil {
235 t.Fatalf("trace Close() error = %v", err)
236 }
237 trace, err := os.ReadFile(tracePath)
238 if err != nil {
239 t.Fatalf("ReadFile(trace): %v", err)
240 }
241 for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
242 if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
243 t.Errorf("trace missing %s", kind)
244 }
245 }
246}
247
248type sshRuntimeClock struct{ now time.Time }
249
250func (c sshRuntimeClock) Now() time.Time { return c.now }
251func (c sshRuntimeClock) NewTimer(d time.Duration) ports.Timer {
252 return sshRuntimeTimer{timer: time.NewTimer(d)}
253}
254
255type sshRuntimeTimer struct{ timer *time.Timer }
256
257func (t sshRuntimeTimer) C() <-chan time.Time { return t.timer.C }
258func (t sshRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
259func (t sshRuntimeTimer) Stop() bool { return t.timer.Stop() }