1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "context"
8 "fmt"
9 "net"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "strings"
14 "sync"
15 "testing"
16 "time"
17
18 "github.com/bnema/vev/pkg/safedir"
19)
20
21type discardTerminalOutput struct{}
22
23func (discardTerminalOutput) Write(p []byte) (int, error) { return len(p), nil }
24func (discardTerminalOutput) Sync() error { return nil }
25func (discardTerminalOutput) Close() error { return nil }
26
27func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) {
28 harnessPTY, clientPTY := net.Pipe()
29 t.Cleanup(func() { _ = clientPTY.Close() })
30 waitErr := make(chan error, 1)
31 secondRead := make(chan struct{})
32 p := &cliProcess{
33 pty: harnessPTY,
34 output: discardTerminalOutput{},
35 chunks: make(chan []byte, 1),
36 done: make(chan struct{}),
37 waitErr: waitErr,
38 waitTimeout: func() <-chan time.Time { return time.After(time.Second) },
39 forceCleanup: func() { waitErr <- nil },
40 }
41 go p.copyTerminal()
42 go func() {
43 _, _ = clientPTY.Write([]byte("first"))
44 _, _ = clientPTY.Write([]byte("second"))
45 close(secondRead)
46 }()
47 select {
48 case <-secondRead:
49 case <-time.After(time.Second):
50 t.Fatal("terminal reader did not queue output")
51 }
52 closed := make(chan error, 1)
53 go func() { closed <- p.Close() }()
54 select {
55 case err := <-closed:
56 if err != nil {
57 t.Fatal(err)
58 }
59 case <-time.After(time.Second):
60 t.Fatal("Close did not drain queued terminal output before waiting for client exit")
61 }
62}
63
64func TestCLIProcessCloseKeepsTerminalDrainActiveUntilClientExit(t *testing.T) {
65 pty, err := os.CreateTemp(t.TempDir(), "pty")
66 if err != nil {
67 t.Fatal(err)
68 }
69 waiting := make(chan struct{})
70 deadline := make(chan time.Time)
71 waitErr := make(chan error)
72 done := make(chan struct{})
73 closed := make(chan error, 1)
74 var waitingOnce sync.Once
75 p := &cliProcess{
76 pty: pty,
77 done: done,
78 waitErr: waitErr,
79 waitTimeout: func() <-chan time.Time {
80 waitingOnce.Do(func() { close(waiting) })
81 return deadline
82 },
83 }
84 go func() { closed <- p.Close() }()
85
86 select {
87 case <-waiting:
88 case <-time.After(time.Second):
89 t.Fatal("Close did not begin waiting for graceful client exit")
90 }
91 select {
92 case <-done:
93 t.Fatal("terminal drain stopped before the client exited")
94 default:
95 }
96 waitErr <- nil
97 select {
98 case err := <-closed:
99 if err != nil {
100 t.Fatal(err)
101 }
102 case <-time.After(time.Second):
103 t.Fatal("Close did not finish after the client exited")
104 }
105 select {
106 case <-done:
107 default:
108 t.Fatal("terminal drain remained active after client exit")
109 }
110}
111
112func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
113 deadline := make(chan time.Time)
114 close(deadline)
115 var term, kill int
116 p := &cliProcess{
117 waitErr: make(chan error),
118 waitTimeout: func() <-chan time.Time { return deadline },
119 forceCleanup: func() { term++ },
120 forceKill: func() { kill++ },
121 }
122 if err := p.Close(); err == nil {
123 t.Fatal("stuck public role close succeeded")
124 }
125 if term != 1 || kill != 1 {
126 t.Fatalf("TERM/KILL calls = %d/%d, want 1/1", term, kill)
127 }
128}
129
130func TestCLIProcessCloseEscalatesHungClientAfterGracefulDrainTimeout(t *testing.T) {
131 deadline := make(chan time.Time)
132 close(deadline)
133 var graceful, term, kill int
134 p := &cliProcess{
135 pty: &orderedPTY{close: func() {}},
136 waitErr: make(chan error),
137 waitTimeout: func() <-chan time.Time { return deadline },
138 gracefulShutdown: func() { graceful++ },
139 forceCleanup: func() { term++ },
140 forceKill: func() { kill++ },
141 }
142 if err := p.Close(); err == nil {
143 t.Fatal("hung client close succeeded")
144 }
145 if graceful != 1 || term != 1 || kill != 1 {
146 t.Fatalf("graceful/TERM/KILL calls = %d/%d/%d, want 1/1/1", graceful, term, kill)
147 }
148}
149
150func TestCLIProcessCloseEscalationOutcome(t *testing.T) {
151 tests := []struct {
152 name string
153 reapsAt string // "term" or "kill": which rung's callback reaps the process
154 waitTimeout func() func() <-chan time.Time
155 wantErrSub string // "" means Close must return nil
156 wantTerm int
157 wantKill int
158 }{
159 {
160 // A role reaped during the SIGTERM wait stopped gracefully: no SIGKILL,
161 // and Close must not report an error for a stage that can still flush a
162 // trace. The deadline channel never fires, so only the SIGTERM callback
163 // reaping the process can end the wait.
164 name: "SIGTERM stop is graceful",
165 reapsAt: "term",
166 waitTimeout: func() func() <-chan time.Time {
167 never := make(chan time.Time)
168 return func() <-chan time.Time { return never }
169 },
170 wantErrSub: "",
171 wantTerm: 1,
172 wantKill: 0,
173 },
174 {
175 // A role that only dies to SIGKILL may have truncated its trace, so
176 // Close must surface a named escalation error rather than a silent
177 // success. One ready deadline releases the SIGTERM wait so the ladder
178 // escalates to SIGKILL, whose callback then reaps the process.
179 name: "errors after kill escalation",
180 reapsAt: "kill",
181 waitTimeout: func() func() <-chan time.Time {
182 deadline := make(chan time.Time, 1)
183 deadline <- time.Now()
184 return func() <-chan time.Time { return deadline }
185 },
186 wantErrSub: "killed",
187 wantTerm: 1,
188 wantKill: 1,
189 },
190 }
191 for _, tt := range tests {
192 t.Run(tt.name, func(t *testing.T) {
193 waitErr := make(chan error, 1)
194 var term, kill int
195 p := &cliProcess{
196 waitErr: waitErr,
197 waitTimeout: tt.waitTimeout(),
198 forceCleanup: func() {
199 term++
200 if tt.reapsAt == "term" {
201 waitErr <- nil
202 }
203 },
204 forceKill: func() {
205 kill++
206 if tt.reapsAt == "kill" {
207 waitErr <- nil
208 }
209 },
210 }
211 err := p.Close()
212 if tt.wantErrSub == "" {
213 if err != nil {
214 t.Fatalf("Close() error = %v, want nil", err)
215 }
216 } else if err == nil || !strings.Contains(err.Error(), tt.wantErrSub) {
217 t.Fatalf("Close() error = %v, want an error containing %q", err, tt.wantErrSub)
218 }
219 if term != tt.wantTerm || kill != tt.wantKill {
220 t.Fatalf("TERM/KILL calls = %d/%d, want %d/%d", term, kill, tt.wantTerm, tt.wantKill)
221 }
222 })
223 }
224}
225
226func TestCLIProcessWaitReadyRequiresDaemonSocket(t *testing.T) {
227 path := filepath.Join(t.TempDir(), "daemon.sock")
228 var lc net.ListenConfig
229 listener, err := lc.Listen(context.Background(), "unix", path)
230 if err != nil {
231 t.Fatal(err)
232 }
233 t.Cleanup(func() {
234 if err := listener.Close(); err != nil {
235 t.Error(err)
236 }
237 })
238 if err := (&cliProcess{readyPath: path, waitErr: make(chan error, 1)}).WaitReady(); err != nil {
239 t.Fatal(err)
240 }
241}
242
243func TestCLILauncherIsolatesRoleXDGEnvironmentAcrossRepetitions(t *testing.T) {
244 captureDir := t.TempDir()
245 t.Setenv("CAPTURE_DIR", captureDir)
246 t.Setenv("XDG_RUNTIME_DIR", filepath.Join(t.TempDir(), "inherited-runtime"))
247 t.Setenv("XDG_STATE_HOME", filepath.Join(t.TempDir(), "inherited-state"))
248
249 bin := filepath.Join(t.TempDir(), "capture-env")
250 if err := os.WriteFile(bin, []byte("#!/bin/sh\nenv > \"$CAPTURE_DIR/$VEV_PERF_PROCESS_ID\"\n"), 0o700); err != nil {
251 t.Fatal(err)
252 }
253 launcher := &cliLauncher{bin: bin}
254 for run := 1; run <= 3; run++ {
255 runDir := filepath.Join(t.TempDir(), fmt.Sprintf("run-%d", run))
256 if err := safedir.EnsurePrivate(runDir); err != nil {
257 t.Fatal(err)
258 }
259 roles := []string{"daemon", "client"}
260 for _, role := range roles {
261 id := fmt.Sprintf("run-%d-%s", run, role)
262 p, err := launcher.Launch(processMapping{
263 Role: role, TracePath: filepath.Join(runDir, role+".jsonl"), ProcessID: id,
264 }, roleCommand{Args: []string{"test"}})
265 if err != nil {
266 t.Fatalf("run %d %s launch: %v", run, role, err)
267 }
268 // The fixture exits on its own. Synchronize with that exit instead of
269 // letting Close race its SIGTERM against the environment capture.
270 proc, ok := p.(*cliProcess)
271 if !ok {
272 t.Fatalf("run %d %s process type = %T, want *cliProcess", run, role, p)
273 }
274 exitErr := <-proc.waitErr
275 proc.waitErr <- exitErr
276 runtimeDir := launcher.runtimes[runDir]
277 if err := p.Close(); err != nil {
278 t.Fatalf("run %d %s close: %v", run, role, err)
279 }
280
281 env, err := os.ReadFile(filepath.Join(captureDir, id))
282 if err != nil {
283 t.Fatalf("run %d %s capture: %v", run, role, err)
284 }
285 for _, want := range []string{
286 "XDG_RUNTIME_DIR=" + runtimeDir,
287 "XDG_STATE_HOME=" + filepath.Join(runDir, "state"),
288 } {
289 name := strings.SplitN(want, "=", 2)[0]
290 if count := bytes.Count(env, []byte(name+"=")); count != 1 {
291 t.Fatalf("run %d %s %s entries = %d, want exactly one; env:\n%s", run, role, name, count, env)
292 }
293 if !bytes.Contains(env, []byte(want+"\n")) {
294 t.Fatalf("run %d %s missing %q; env:\n%s", run, role, want, env)
295 }
296 }
297 }
298 if err := launcher.releaseRuntime(runDir); err != nil {
299 t.Fatalf("run %d release runtime: %v", run, err)
300 }
301 }
302}
303
304func TestPreparePeerReleasesOwnedRuntimeOnPreparationFailure(t *testing.T) {
305 t.Setenv("TMPDIR", t.TempDir())
306 runDir := filepath.Join(t.TempDir(), "run")
307 if err := safedir.EnsurePrivate(runDir); err != nil {
308 t.Fatal(err)
309 }
310 launcher := &cliLauncher{}
311 _, err := launcher.preparePeer(processMapping{
312 Role: "ssh_stdio_peer",
313 TracePath: filepath.Join(runDir, "peer.jsonl"),
314 }, roleCommand{Args: []string{"invalid"}})
315 if err == nil || !strings.Contains(err.Error(), "unsupported public peer command") {
316 t.Fatalf("preparePeer error = %v", err)
317 }
318 if runtimeDir := launcher.runtimes[runDir]; runtimeDir != "" {
319 t.Fatalf("failed peer preparation leaked runtime directory %q", runtimeDir)
320 }
321 if paths, globErr := filepath.Glob(filepath.Join(os.TempDir(), "vev-harness-runtime-*")); globErr != nil || len(paths) != 0 {
322 t.Fatalf("failed peer preparation left runtime directories %q: %v", paths, globErr)
323 }
324}
325
326func TestCLIProcessCloseUsesPublicDaemonShutdown(t *testing.T) {
327 wait := make(chan error, 1)
328 wait <- nil
329 var shutdown int
330 p := &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; return nil }}
331 if err := p.Close(); err != nil {
332 t.Fatal(err)
333 }
334 if shutdown != 0 {
335 t.Fatalf("shutdown calls = %d, want 0 for an exited daemon", shutdown)
336 }
337
338 wait = make(chan error, 1)
339 p = &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; wait <- nil; return nil }}
340 if err := p.Close(); err != nil {
341 t.Fatal(err)
342 }
343 if shutdown != 1 {
344 t.Fatalf("shutdown calls = %d, want 1", shutdown)
345 }
346}
347
348func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
349 if testing.Short() {
350 t.Skip("public CLI smoke")
351 }
352 root, err := filepath.Abs(filepath.Join("..", ".."))
353 if err != nil {
354 t.Fatal(err)
355 }
356 bin := filepath.Join(t.TempDir(), "vev")
357 build := exec.Command("go", "build", "-o", bin, "./")
358 build.Dir = root
359 if output, err := build.CombinedOutput(); err != nil {
360 t.Fatalf("build public CLI: %v\n%s", err, output)
361 }
362 m, err := readManifest(filepath.Join(root, "testdata", "perf", "manifest.json"))
363 if err != nil {
364 t.Fatal(err)
365 }
366 if err := validateManifest(m); err != nil {
367 t.Fatal(err)
368 }
369 var local scenario
370 for _, s := range m.Scenarios {
371 if s.Topology == "1x4" && s.Workload == "idle" && s.Transport == "local" {
372 local = s
373 break
374 }
375 }
376 if local.ID == "" {
377 t.Fatal("canonical local scenario missing")
378 }
379 dir := t.TempDir()
380 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
381 if err != nil {
382 t.Fatal(err)
383 }
384 closeTestFile(t, raw)
385 h := defaultHarness()
386 h.clock = &fakeClock{} // injected duration keeps the public 30s contract bounded in test
387 h.launcher = &cliLauncher{bin: bin}
388 result, err := h.runOne(options{vevBin: bin, out: dir, duration: minimumDuration}, m, local, 1, raw)
389 if err != nil {
390 t.Fatal(err)
391 }
392 if result.Samples < minimumInIntervalEventSamples || len(result.Processes) != 2 {
393 t.Fatalf("canonical local smoke result=%+v", result)
394 }
395}
396
397func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {
398 mapping := processMapping{TracePath: "/tmp/trace.jsonl", ProcessID: "p-01", Scenario: "1x4-idle-local", Run: 3}
399 got := strings.Join(traceEnvironment(mapping), "\n")
400 for _, want := range []string{"VEV_PERF_TRACE=/tmp/trace.jsonl", "VEV_PERF_PROCESS_ID=p-01", "VEV_PERF_SCENARIO=1x4-idle-local", "VEV_PERF_RUN=3"} {
401 if !strings.Contains(got, want) {
402 t.Errorf("trace environment missing %q: %s", want, got)
403 }
404 }
405}
406
407func TestCLIProcessCloseStopsShellBeforeProcessGroupCleanup(t *testing.T) {
408 var order []string
409 waitErr := make(chan error, 1)
410 p := &cliProcess{
411 pty: &orderedPTY{close: func() { order = append(order, "pty_closed") }},
412 output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
413 waitErr: waitErr,
414 gracefulShutdown: func() {
415 order = append(order, "shell_exit")
416 // The client closes and waits for its transport descendant before it
417 // reports its own exit to the launcher.
418 order = append(order, "descendant_receive_end")
419 waitErr <- nil
420 },
421 forceCleanup: func() {
422 t.Fatal("normal shutdown sent SIGTERM to the traced descendant process group")
423 },
424 }
425
426 if err := p.Close(); err != nil {
427 t.Fatal(err)
428 }
429 if !equalStrings(order, []string{"shell_exit", "descendant_receive_end", "pty_closed", "output_closed"}) {
430 t.Fatalf("cleanup order=%q", order)
431 }
432}
433
434func TestCLIProcessWarmupWaitsForDelayedTerminalReadiness(t *testing.T) {
435 allowWrite := make(chan struct{})
436 pty, output := &stagedPTY{allow: allowWrite, writes: make(chan []byte, 1)}, &fakeOutput{}
437 p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte)}
438 input := workloadInput(scenario{ID: "s"}, 1, "warmup")
439 done := make(chan error, 1)
440 go func() { done <- p.Warmup(input) }()
441
442 // This is public client output from the initial application prompt/state,
443 // emitted only after the client has entered raw mode. A premature PTY write
444 // blocks on allowWrite and therefore cannot consume this delayed readiness.
445 go func() {
446 p.chunks <- []byte("shell prompt$ ")
447 close(allowWrite)
448 }()
449 select {
450 case got := <-pty.writes:
451 if !bytes.Equal(got, input) {
452 t.Fatalf("warmup write=%q want %q", got, input)
453 }
454 case <-time.After(time.Second):
455 t.Fatal("warmup did not wait for and then proceed from client readiness")
456 }
457 go func() { p.chunks <- append([]byte("application "), inputMarker(input)...); close(p.chunks) }()
458 if err := <-done; err != nil {
459 t.Fatal(err)
460 }
461 if output.syncs != 1 {
462 t.Fatalf("readiness output was not flushed: syncs=%d", output.syncs)
463 }
464}
465
466func TestCLIProcessRejectsPTYLocalEchoAsFlushEvidence(t *testing.T) {
467 pty, output := &fakePTY{}, &fakeOutput{}
468 p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)}
469 input := workloadInput(scenario{ID: "s"}, 1, "measured-1")
470 p.chunks <- newPTYLocalEcho(input).expected
471 close(p.chunks)
472 flushed := false
473 if err := p.Measure(input, func() error { return nil }, func() error { flushed = true; return nil }); err == nil {
474 t.Fatal("local PTY echo satisfied terminal flush boundary")
475 }
476 if flushed || output.syncs != 0 {
477 t.Fatalf("local echo stamped a flush: flushed=%t syncs=%d", flushed, output.syncs)
478 }
479}
480
481func TestCLIProcessPairsApplicationOutputWithSuccessfulFlush(t *testing.T) {
482 pty, output := &fakePTY{}, &fakeOutput{}
483 p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)}
484 input := workloadInput(scenario{ID: "s"}, 1, "measured-1")
485 echo := newPTYLocalEcho(input).expected
486 p.chunks <- append(append([]byte(nil), echo...), append([]byte("application "), inputMarker(input)...)...)
487 order := []string{}
488 if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil {
489 t.Fatal(err)
490 }
491 if len(pty.writes) != 1 || output.syncs != 1 || !equalStrings(order, []string{"injected", "flushed"}) {
492 t.Fatalf("writes=%q syncs=%d boundary order=%q", pty.writes, output.syncs, order)
493 }
494}
495
496func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) {
497 for _, tc := range []struct {
498 role string
499 want []string
500 }{
501 {"daemon", []string{"--daemon"}},
502 {"client", []string{"new", "perf-s-001"}},
503 {"ssh_stdio_peer", []string{"_stdio", "perf-s-001"}},
504 {"udp_peer", []string{"_udp-proxy", "perf-s-001"}},
505 } {
506 t.Run(tc.role, func(t *testing.T) {
507 got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1})
508 if !equalStrings(got.Args, tc.want) {
509 t.Fatalf("args=%q want %q", got.Args, tc.want)
510 }
511 })
512 }
513 input := string(workloadInput(scenario{ID: "s", Workload: "interactive_flood"}, 1, "measured-1"))
514 if string(inputMarker([]byte(input))) != "__VEV_HARNESS_s_r1_measured-1__" || !bytes.Contains([]byte(input), []byte("while [ $i -lt 128 ]")) || !strings.HasSuffix(input, "printf '__VEV_HARNESS_s_r1_measured-1__\\n'\n") {
515 t.Fatalf("workload is not real PTY shell input with an observable marker: %q", input)
516 }
517}