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