1package app
2
3import (
4 "bufio"
5 "context"
6 "encoding/json"
7 "errors"
8 "io"
9 "os"
10 "path/filepath"
11 "strconv"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/bnema/vev/internal/ports"
17 portsmocks "github.com/bnema/vev/internal/ports/mocks"
18 "github.com/bnema/vev/internal/usecase/client"
19 "github.com/stretchr/testify/mock"
20)
21
22func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
23 path := filepath.Join(t.TempDir(), "process.jsonl")
24 t.Setenv("VEV_PERF_TRACE", path)
25 t.Setenv("VEV_PERF_PROCESS_ID", "idle-local-r001-daemon-01")
26 t.Setenv("VEV_PERF_SCENARIO", "1x4-idle-local")
27 t.Setenv("VEV_PERF_RUN", "1")
28
29 clock := portsmocks.NewMockClock(t)
30 clock.EXPECT().Now().Return(time.Unix(0, 1)).Once()
31 observer, closer, err := performanceTrace(clock)
32 if err != nil {
33 t.Fatal(err)
34 }
35 if observer == nil || closer == nil {
36 t.Fatal("performance trace was not configured")
37 }
38 // performanceTrace's serialized return type makes a raw JSONL sink
39 // unrepresentable to application transport wiring.
40 observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})
41 if err := closer.Close(); err != nil {
42 t.Fatal(err)
43 }
44 if err := closer.Close(); err != nil {
45 t.Fatal(err)
46 }
47 f, err := os.Open(path)
48 if err != nil {
49 t.Fatal(err)
50 }
51 defer func() {
52 if err := f.Close(); err != nil {
53 t.Errorf("close trace file: %v", err)
54 }
55 }()
56 var mark struct {
57 ProcessID string `json:"process_id"`
58 Scenario string `json:"scenario"`
59 Run uint64 `json:"run"`
60 Sequence uint64 `json:"sequence"`
61 RequestID uint64 `json:"request_id"`
62 Epoch uint64 `json:"epoch"`
63 }
64 if err := json.NewDecoder(bufio.NewReader(f)).Decode(&mark); err != nil {
65 t.Fatal(err)
66 }
67 if mark.ProcessID != "idle-local-r001-daemon-01" || mark.Scenario != "1x4-idle-local" || mark.Run != 1 || mark.Sequence == 0 || mark.RequestID == 0 || mark.Epoch == 0 {
68 t.Fatalf("mark does not match harness mapping: %+v", mark)
69 }
70}
71
72func TestPerformanceTraceWithFactoriesCreatesOneConfiguredObserver(t *testing.T) {
73 t.Setenv("VEV_PERF_TRACE", "trace.jsonl")
74 t.Setenv("VEV_PERF_PROCESS_ID", "client-01")
75 t.Setenv("VEV_PERF_SCENARIO", "lossy-remote")
76 t.Setenv("VEV_PERF_RUN", "7")
77
78 clock := portsmocks.NewMockClock(t)
79 sink := portsmocks.NewMockRuntimeObserver(t)
80 sink.EXPECT().ObserveRuntime(mock.MatchedBy(func(mark ports.RuntimeMark) bool {
81 return mark.Component == "client" && mark.Scenario == "lossy-remote" && mark.Run == 7 &&
82 mark.Sequence != 0 && mark.RequestID != 0 && mark.Epoch != 0 && mark.Kind == ports.RuntimeEmitEnd
83 })).Once()
84
85 var sinkCalls, correlationCalls int
86 observer, closer, err := performanceTraceWithFactories(clock,
87 func(path string, gotClock ports.Clock, processID string) (ports.RuntimeObserver, io.Closer, error) {
88 sinkCalls++
89 if path != "trace.jsonl" || gotClock != clock || processID != "client-01" {
90 t.Fatalf("trace sink config = (%q, %v, %q)", path, gotClock, processID)
91 }
92 return sink, errorCloser{}, nil
93 },
94 func(got ports.RuntimeObserver, inputs ports.RuntimeCorrelationInputs) (ports.RuntimeObserver, error) {
95 correlationCalls++
96 if got != sink || inputs != (ports.RuntimeCorrelationInputs{Scenario: "lossy-remote", Run: 7}) {
97 t.Fatalf("trace correlation config = (%v, %+v)", got, inputs)
98 }
99 return ports.NewRuntimeCorrelationObserver(got, inputs)
100 },
101 )
102 if err != nil {
103 t.Fatal(err)
104 }
105 if sinkCalls != 1 || correlationCalls != 1 || observer == nil || closer == nil {
106 t.Fatalf("trace setup = sink calls %d, correlation calls %d, observer %v, closer %v; want one configured observer", sinkCalls, correlationCalls, observer, closer)
107 }
108 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeEmitEnd, 0, true))
109 if err := closer.Close(); err != nil {
110 t.Fatal(err)
111 }
112}
113
114func TestPerformanceTraceSetupRollbackJoinsCloseError(t *testing.T) {
115 setupErr := errors.New("correlation setup failed")
116 closeErr := errors.New("rollback close failed")
117
118 tests := []struct {
119 name string
120 rawRun string
121 correlationErr error
122 wantPrimaryText string
123 wantPrimaryErr error
124 wantNumError bool
125 }{
126 {
127 name: "validation",
128 rawRun: "not-a-run",
129 wantPrimaryText: `invalid VEV_PERF_RUN "not-a-run"`,
130 wantNumError: true,
131 },
132 {
133 name: "correlation",
134 rawRun: "1",
135 correlationErr: setupErr,
136 wantPrimaryText: "runtime trace correlation",
137 wantPrimaryErr: setupErr,
138 },
139 }
140
141 for _, tc := range tests {
142 t.Run(tc.name, func(t *testing.T) {
143 t.Setenv("VEV_PERF_TRACE", "unused-by-test-sink")
144 t.Setenv("VEV_PERF_PROCESS_ID", "test-process")
145 t.Setenv("VEV_PERF_RUN", tc.rawRun)
146
147 newSink := func(string, ports.Clock, string) (ports.RuntimeObserver, io.Closer, error) {
148 return runtimeObserverFunc(func(ports.RuntimeMark) {}), errorCloser{err: closeErr}, nil
149 }
150 newCorrelation := func(observer ports.RuntimeObserver, inputs ports.RuntimeCorrelationInputs) (ports.RuntimeObserver, error) {
151 if tc.correlationErr != nil {
152 return nil, tc.correlationErr
153 }
154 return ports.NewRuntimeCorrelationObserver(observer, inputs)
155 }
156
157 observer, closer, err := performanceTraceWithFactories(nil, newSink, newCorrelation)
158 if observer != nil || closer != nil {
159 t.Fatalf("performanceTrace() returned observer=%v closer=%v after setup failure", observer, closer)
160 }
161 if !errors.Is(err, closeErr) {
162 t.Errorf("performanceTrace() error = %v, want joined rollback error %v", err, closeErr)
163 }
164 if tc.wantPrimaryErr != nil && !errors.Is(err, tc.wantPrimaryErr) {
165 t.Errorf("performanceTrace() error = %v, want primary setup error %v", err, tc.wantPrimaryErr)
166 }
167 if tc.wantNumError {
168 var numErr *strconv.NumError
169 if !errors.As(err, &numErr) {
170 t.Errorf("performanceTrace() error = %v, want discoverable parse error", err)
171 }
172 }
173 if !strings.Contains(err.Error(), tc.wantPrimaryText) {
174 t.Errorf("performanceTrace() error = %q, want context %q", err, tc.wantPrimaryText)
175 }
176 })
177 }
178}
179
180func TestRunAttachJoinsTraceCloseError(t *testing.T) {
181 traceCloseErr := errors.New("trace close failed")
182 original := newPerformanceTrace
183 newPerformanceTrace = func(ports.Clock) (ports.SerializedRuntimeObserver, io.Closer, error) {
184 return ports.NewSerializedRuntimeObserver(runtimeObserverFunc(func(ports.RuntimeMark) {}), 1), errorCloser{err: traceCloseErr}, nil
185 }
186 t.Cleanup(func() { newPerformanceTrace = original })
187 t.Setenv("VEV", "")
188 t.Setenv("XDG_STATE_HOME", t.TempDir())
189 t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
190 t.Setenv(envRemoteTransport, "invalid")
191
192 err := runAttach(context.Background(), ports.IntentEphemeral, "", "host")
193 if !strings.Contains(err.Error(), "invalid remote transport") {
194 t.Fatalf("runAttach() error = %v, want remote transport validation error", err)
195 }
196 if !errors.Is(err, traceCloseErr) {
197 t.Fatalf("runAttach() error = %v, want joined trace close error %v", err, traceCloseErr)
198 }
199}
200
201type errorCloser struct{ err error }
202
203func (c errorCloser) Close() error { return c.err }
204
205type runtimeObserverFunc func(ports.RuntimeMark)
206
207func (f runtimeObserverFunc) ObserveRuntime(mark ports.RuntimeMark) { f(mark) }
208
209func TestRunAttachPropagatesOneObserverToRemoteTransportFactory(t *testing.T) {
210 t.Setenv("VEV", "")
211 t.Setenv("XDG_STATE_HOME", t.TempDir())
212 t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
213 t.Setenv(envRemoteTransport, string(ports.RemoteTransportStdio))
214
215 observer := ports.NewSerializedRuntimeObserver(runtimeObserverFunc(func(ports.RuntimeMark) {}), 1)
216 originalTrace := newPerformanceTrace
217 traceCalls := 0
218 newPerformanceTrace = func(ports.Clock) (ports.SerializedRuntimeObserver, io.Closer, error) {
219 traceCalls++
220 return observer, &runtimeTraceCloser{reporter: observer}, nil
221 }
222 t.Cleanup(func() { newPerformanceTrace = originalTrace })
223
224 factory := portsmocks.NewMockRemoteDialerFactory(t)
225 factory.EXPECT().DialerForRemote("remote.example", "work", ports.RemoteTransportStdio, mock.Anything).Return(namedDialer{name: "remote"}, nil)
226 originalFactory := newRemoteDialerFactoryWithRuntimeObserver
227 factoryCalls := 0
228 newRemoteDialerFactoryWithRuntimeObserver = func(got ports.SerializedRuntimeObserver) ports.RemoteDialerFactory {
229 factoryCalls++
230 if got != observer {
231 t.Fatalf("remote transport observer = %v, want process observer %v", got, observer)
232 }
233 return factory
234 }
235 t.Cleanup(func() { newRemoteDialerFactoryWithRuntimeObserver = originalFactory })
236
237 originalRunClient := runClientWithDeps
238 runClientWithDeps = func(_ context.Context, deps client.Dependencies, _ client.AttachRequest) error {
239 if deps.RuntimeObserver != observer {
240 t.Fatalf("client transport observer = %v, want process observer %v", deps.RuntimeObserver, observer)
241 }
242 if deps.Dialer != (namedDialer{name: "remote"}) {
243 t.Fatalf("remote transport dialer = %v, want configured factory dialer", deps.Dialer)
244 }
245 return nil
246 }
247 t.Cleanup(func() { runClientWithDeps = originalRunClient })
248
249 if err := runAttach(context.Background(), ports.IntentAttach, "work", "remote.example"); err != nil {
250 t.Fatal(err)
251 }
252 if traceCalls != 1 || factoryCalls != 1 {
253 t.Fatalf("trace/factory calls = %d/%d, want exactly one observer and one transport factory", traceCalls, factoryCalls)
254 }
255}