1package sshstdio
2
3import (
4 "bytes"
5 "encoding/json"
6 "errors"
7 "os"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/bnema/vev/internal/adapters/observability"
14 "github.com/bnema/vev/internal/ports"
15)
16
17func TestTransportObservabilitySSHStdioEOFEndsReceive(t *testing.T) {
18 tracePath := t.TempDir() + "/ssh-eof.jsonl"
19 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 309)}, "ssh-process")
20 if err != nil {
21 t.Fatalf("NewJSONL() error = %v", err)
22 }
23 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
24 defer reporter.Close()
25 if _, err := NewTransport(bytes.NewReader(nil), &bytes.Buffer{}, nil, WithRuntimeObserver(reporter)).Recv(); err == nil {
26 t.Fatal("Recv() error = nil at EOF")
27 }
28 reporter.Close()
29 if err := closer.Close(); err != nil {
30 t.Fatalf("trace Close() error = %v", err)
31 }
32 assertSSHReceiveFailurePair(t, tracePath)
33}
34
35func TestTransportObservabilitySSHStdioCloseEndsBlockedReceive(t *testing.T) {
36 tracePath := t.TempDir() + "/ssh-close.jsonl"
37 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 311)}, "ssh-process")
38 if err != nil {
39 t.Fatalf("NewJSONL() error = %v", err)
40 }
41 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
42 defer reporter.Close()
43 reader := &sshShutdownReader{entered: make(chan struct{}), done: make(chan struct{})}
44 shutdownErr := errors.New("ssh shutdown failed")
45 transport := NewTransport(reader, &bytes.Buffer{}, func() error { return shutdownErr }, WithRuntimeObserver(reporter))
46 recvDone := make(chan error, 1)
47 go func() { _, err := transport.Recv(); recvDone <- err }()
48 <-reader.entered
49 if err := transport.Close(); !errors.Is(err, shutdownErr) {
50 t.Fatalf("Close() error = %v, want %v", err, shutdownErr)
51 }
52 if err := <-recvDone; err == nil {
53 t.Fatal("Recv() error = nil after Close()")
54 }
55 reporter.Close()
56 if err := closer.Close(); err != nil {
57 t.Fatalf("trace Close() error = %v", err)
58 }
59 assertSSHReceiveFailurePair(t, tracePath)
60}
61
62type sshShutdownReader struct {
63 entered chan struct{}
64 done chan struct{}
65 once sync.Once
66}
67
68func (r *sshShutdownReader) Read([]byte) (int, error) {
69 r.once.Do(func() { close(r.entered) })
70 <-r.done
71 return 0, errors.New("reader closed")
72}
73
74func (r *sshShutdownReader) Close() error {
75 r.once.Do(func() { close(r.entered) })
76 close(r.done)
77 return nil
78}
79
80func assertSSHReceiveFailurePair(t *testing.T, path string) {
81 t.Helper()
82 data, err := os.ReadFile(path)
83 if err != nil {
84 t.Fatalf("ReadFile(trace): %v", err)
85 }
86 var starts, ends []struct {
87 ProcessID string `json:"process_id"`
88 Scenario string `json:"scenario"`
89 Run uint64 `json:"run"`
90 Sequence uint64 `json:"sequence"`
91 RequestID uint64 `json:"request_id"`
92 Epoch uint64 `json:"epoch"`
93 Kind string `json:"kind"`
94 Valid bool `json:"valid"`
95 }
96 for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
97 var mark struct {
98 ProcessID string `json:"process_id"`
99 Scenario string `json:"scenario"`
100 Run uint64 `json:"run"`
101 Sequence uint64 `json:"sequence"`
102 RequestID uint64 `json:"request_id"`
103 Epoch uint64 `json:"epoch"`
104 Kind string `json:"kind"`
105 Valid bool `json:"valid"`
106 }
107 if err := json.Unmarshal([]byte(line), &mark); err != nil {
108 t.Fatalf("Unmarshal(trace): %v", err)
109 }
110 switch mark.Kind {
111 case "adapter_receive_start":
112 starts = append(starts, mark)
113 case "adapter_receive_end":
114 ends = append(ends, mark)
115 }
116 }
117 if len(starts) != 1 || len(ends) != 1 {
118 t.Fatalf("receive spans start=%d end=%d, want exactly one pair", len(starts), len(ends))
119 }
120 start, end := starts[0], ends[0]
121 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 {
122 t.Fatalf("mismatched receive correlation: start=%+v end=%+v", start, end)
123 }
124 if !start.Valid || end.Valid {
125 t.Fatalf("receive validity start=%t end=%t, want true/false", start.Valid, end.Valid)
126 }
127}
128
129func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
130 tracePath := t.TempDir() + "/ssh.jsonl"
131 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")
132 if err != nil {
133 t.Fatalf("NewJSONL() error = %v", err)
134 }
135 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
136 defer reporter.Close()
137
138 frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte("stdio bytes stay exact")}
139 var observed, baseline bytes.Buffer
140 if err := NewTransport(bytes.NewReader(nil), &observed, nil, WithRuntimeObserver(reporter)).Send(frame); err != nil {
141 t.Fatalf("observed Send() error = %v", err)
142 }
143 if err := NewTransport(bytes.NewReader(nil), &baseline, nil).Send(frame); err != nil {
144 t.Fatalf("baseline Send() error = %v", err)
145 }
146 if !bytes.Equal(observed.Bytes(), baseline.Bytes()) {
147 t.Fatalf("observer changed SSH stdio wire bytes: got %x, want %x", observed.Bytes(), baseline.Bytes())
148 }
149 received, err := NewTransport(bytes.NewReader(baseline.Bytes()), &bytes.Buffer{}, nil, WithRuntimeObserver(reporter)).Recv()
150 if err != nil {
151 t.Fatalf("observed Recv() error = %v", err)
152 }
153 if received.Type != frame.Type || !bytes.Equal(received.Payload, frame.Payload) {
154 t.Fatalf("observed Recv() = %#v, want %#v", received, frame)
155 }
156 reporter.Close()
157 if err := closer.Close(); err != nil {
158 t.Fatalf("trace Close() error = %v", err)
159 }
160 trace, err := os.ReadFile(tracePath)
161 if err != nil {
162 t.Fatalf("ReadFile(trace): %v", err)
163 }
164 for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
165 if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
166 t.Errorf("trace missing %s", kind)
167 }
168 }
169}
170
171type sshRuntimeClock struct{ now time.Time }
172
173func (c sshRuntimeClock) Now() time.Time { return c.now }
174func (c sshRuntimeClock) NewTimer(d time.Duration) ports.Timer {
175 return sshRuntimeTimer{timer: time.NewTimer(d)}
176}
177
178type sshRuntimeTimer struct{ timer *time.Timer }
179
180func (t sshRuntimeTimer) C() <-chan time.Time { return t.timer.C }
181func (t sshRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
182func (t sshRuntimeTimer) Stop() bool { return t.timer.Stop() }