1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/bnema/vev/pkg/safedir"
18)
19
20func TestHarnessCanonicalLocalRolesAreIsolatedAcrossRepetitions(t *testing.T) {
21 if testing.Short() {
22 t.Skip("public CLI repetition lifecycle")
23 }
24 root, err := filepath.Abs(filepath.Join("..", ".."))
25 if err != nil {
26 t.Fatal(err)
27 }
28 bin := filepath.Join(t.TempDir(), "vev")
29 build := exec.Command("go", "build", "-o", bin, "./")
30 build.Dir = root
31 if output, err := build.CombinedOutput(); err != nil {
32 t.Fatalf("build public CLI: %v\n%s", err, output)
33 }
34 m, err := readManifest(filepath.Join(root, "testdata", "perf", "manifest.json"))
35 if err != nil {
36 t.Fatal(err)
37 }
38 var local scenario
39 for _, candidate := range m.Scenarios {
40 if candidate.ID == "1x4-idle-local" {
41 local = candidate
42 break
43 }
44 }
45 if local.ID == "" {
46 t.Fatal("canonical local scenario missing")
47 }
48 out, err := os.MkdirTemp("", "vev-ob-")
49 if err != nil {
50 t.Fatal(err)
51 }
52 removeTestTree(t, out)
53 raw, err := os.OpenFile(filepath.Join(out, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
54 if err != nil {
55 t.Fatal(err)
56 }
57 closeTestFile(t, raw)
58 h := defaultHarness()
59 h.clock = &fakeClock{}
60 h.launcher = &cliLauncher{bin: bin}
61 for run := 1; run <= 3; run++ {
62 result, err := h.runOne(options{vevBin: bin, out: out, duration: minimumDuration}, m, local, run, raw)
63 if err != nil {
64 t.Fatalf("run %d: %v", run, err)
65 }
66 if len(result.Processes) != 2 {
67 t.Fatalf("run %d process mappings = %+v", run, result.Processes)
68 }
69 runDir := filepath.Join(out, fmt.Sprintf("%s-run-%03d", safeName(local.ID), run))
70 manifestBytes, err := os.ReadFile(filepath.Join(runDir, "manifest.json"))
71 if err != nil {
72 t.Fatalf("run %d manifest: %v", run, err)
73 }
74 var persisted runManifest
75 if err := json.Unmarshal(manifestBytes, &persisted); err != nil {
76 t.Fatalf("run %d manifest JSON: %v", run, err)
77 }
78 if persisted.Run != run || len(persisted.Processes) != len(result.Processes) {
79 t.Fatalf("run %d manifest = %+v", run, persisted)
80 }
81 for _, process := range result.Processes {
82 if process.Role != "daemon" && process.Role != "client" {
83 t.Fatalf("run %d unexpected role %q", run, process.Role)
84 }
85 if filepath.Dir(process.TracePath) != runDir {
86 t.Fatalf("run %d %s trace outside its isolated run directory: %q", run, process.Role, process.TracePath)
87 }
88 trace, err := os.ReadFile(process.TracePath)
89 if err != nil {
90 t.Fatalf("run %d %s trace %q: %v", run, process.Role, process.TracePath, err)
91 }
92 if len(bytes.TrimSpace(trace)) == 0 {
93 t.Fatalf("run %d %s did not emit its declared trace", run, process.Role)
94 }
95 for _, line := range bytes.Split(bytes.TrimSpace(trace), []byte{'\n'}) {
96 var record traceRecord
97 if err := json.Unmarshal(line, &record); err != nil {
98 t.Fatalf("run %d %s trace JSON: %v", run, process.Role, err)
99 }
100 if record.ProcessID != process.ProcessID || record.Scenario != local.ID || record.Run != uint64(run) {
101 t.Fatalf("run %d %s trace escaped its manifest identity: %+v", run, process.Role, record)
102 }
103 }
104 }
105 if _, err := os.Stat(filepath.Join(runDir, "runtime", "vev", "daemon.sock")); !errors.Is(err, os.ErrNotExist) {
106 t.Fatalf("run %d left daemon socket for the next repetition: %v", run, err)
107 }
108 }
109}
110
111func TestHarnessResolvesCanonicalRelativePathsBeforeRoleCWDChanges(t *testing.T) {
112 if testing.Short() {
113 t.Skip("public CLI relative-path lifecycle")
114 }
115 root, err := filepath.Abs(filepath.Join("..", ".."))
116 if err != nil {
117 t.Fatal(err)
118 }
119 // Keep the real Unix-socket path below its platform limit while testing the
120 // exact documented relative output hierarchy.
121 workspace, err := os.MkdirTemp("", "vev-rel-")
122 if err != nil {
123 t.Fatal(err)
124 }
125 removeTestTree(t, workspace)
126 bin := filepath.Join(workspace, "vev")
127 build := exec.Command("go", "build", "-o", bin, "./")
128 build.Dir = root
129 if output, err := build.CombinedOutput(); err != nil {
130 t.Fatalf("build public CLI: %v\n%s", err, output)
131 }
132 manifestPath := filepath.Join(workspace, "testdata", "perf", "manifest.json")
133 if err := safedir.EnsurePrivate(filepath.Dir(manifestPath)); err != nil {
134 t.Fatal(err)
135 }
136 contents, err := os.ReadFile(filepath.Join(root, "testdata", "perf", "manifest.json"))
137 if err != nil {
138 t.Fatal(err)
139 }
140 if err := os.WriteFile(manifestPath, contents, 0o600); err != nil {
141 t.Fatal(err)
142 }
143 originalWD, err := os.Getwd()
144 if err != nil {
145 t.Fatal(err)
146 }
147 if err := os.Chdir(workspace); err != nil {
148 t.Fatal(err)
149 }
150 t.Cleanup(func() {
151 if err := os.Chdir(originalWD); err != nil {
152 t.Errorf("restore cwd: %v", err)
153 }
154 })
155
156 // These are the documented relative locations. cliLauncher changes every
157 // role to its run directory, so a failure to resolve them before launch
158 // would make VEV_PERF_TRACE point at a nonexistent nested directory.
159 h := defaultHarness()
160 h.clock = &fakeClock{}
161 if err := run([]string{
162 "--vev-bin", "./vev",
163 "--manifest", "testdata/perf/manifest.json",
164 "--out", "testdata/perf/results",
165 "--scenario", "1x4-idle-local",
166 "--warmup", "0s",
167 "--duration", "30s",
168 "--repetitions", "10",
169 }, h); err != nil {
170 t.Fatal(err)
171 }
172
173 out := filepath.Join(workspace, "testdata", "perf", "results")
174 for run := 1; run <= 10; run++ {
175 runDir := filepath.Join(out, fmt.Sprintf("1x4-idle-local-run-%03d", run))
176 manifestBytes, err := os.ReadFile(filepath.Join(runDir, "manifest.json"))
177 if err != nil {
178 t.Fatalf("run %d manifest: %v", run, err)
179 }
180 var persisted runManifest
181 if err := json.Unmarshal(manifestBytes, &persisted); err != nil {
182 t.Fatalf("run %d manifest JSON: %v", run, err)
183 }
184 if persisted.Run != run || len(persisted.Processes) != 2 {
185 t.Fatalf("run %d manifest=%+v", run, persisted)
186 }
187 for _, process := range persisted.Processes {
188 if !filepath.IsAbs(process.TracePath) || filepath.Dir(process.TracePath) != runDir {
189 t.Fatalf("run %d %s trace path=%q, want absolute path in %q", run, process.Role, process.TracePath, runDir)
190 }
191 trace, err := os.ReadFile(process.TracePath)
192 if err != nil || len(bytes.TrimSpace(trace)) == 0 {
193 t.Fatalf("run %d %s trace=%q err=%v", run, process.Role, process.TracePath, err)
194 }
195 }
196 }
197}
198
199func TestHarnessRejectsExistingRunDirectoryBeforeLaunchingRoles(t *testing.T) {
200 out := t.TempDir()
201 raw, err := os.OpenFile(filepath.Join(out, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
202 if err != nil {
203 t.Fatal(err)
204 }
205 closeTestFile(t, raw)
206 s := scenario{ID: "isolated", Roles: []string{"daemon", "client"}}
207 stale := filepath.Join(out, "isolated-run-001")
208 if err := os.Mkdir(stale, 0o700); err != nil {
209 t.Fatal(err)
210 }
211 if err := os.WriteFile(filepath.Join(stale, "preserve"), []byte("stale"), 0o600); err != nil {
212 t.Fatal(err)
213 }
214 launcher := &fakeLauncher{}
215 h := defaultHarness()
216 h.clock, h.launcher = &fakeClock{}, launcher
217 _, err = h.runOne(options{out: out, duration: minimumDuration}, manifest{}, s, 1, raw)
218 if err == nil || !strings.Contains(err.Error(), "create isolated run directory") {
219 t.Fatalf("existing run directory error = %v", err)
220 }
221 if len(launcher.mappings) != 0 {
222 t.Fatalf("roles launched against stale run directory: %+v", launcher.mappings)
223 }
224 if contents, readErr := os.ReadFile(filepath.Join(stale, "preserve")); readErr != nil || string(contents) != "stale" {
225 t.Fatalf("stale directory changed: contents=%q err=%v", contents, readErr)
226 }
227}
228
229func TestHarnessDoesNotInheritNestedSession(t *testing.T) {
230 env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")
231 if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {
232 t.Fatalf("environment=%q", env)
233 }
234}
235
236func TestHarnessRoutesEveryRemoteFixtureThroughItsDeclaredPeer(t *testing.T) {
237 cases := []struct {
238 name string
239 transport transport
240 peer string
241 wantRTT int
242 wantLoss int
243 }{
244 {"ssh", transport{ID: "ssh_stdio", Kind: "ssh_stdio"}, "ssh_stdio_peer", 0, 0},
245 {"udp baseline", transport{ID: "udp_baseline", Kind: "udp"}, "udp_peer", 0, 0},
246 {"udp 25ms", transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}, "udp_peer", 25, 0},
247 {"udp 100ms", transport{ID: "udp_100ms", Kind: "udp", RTTMS: 100}, "udp_peer", 100, 0},
248 {"udp loss zero", transport{ID: "udp_loss_0pct", Kind: "udp", LossPercent: 0}, "udp_peer", 0, 0},
249 {"udp loss", transport{ID: "udp_loss_1pct", Kind: "udp", LossPercent: 1}, "udp_peer", 0, 1},
250 }
251 for _, tc := range cases {
252 t.Run(tc.name, func(t *testing.T) {
253 s := scenario{ID: "route", Transport: tc.transport.ID}
254 client := routeRoleArgs(s, processMapping{Role: "client", Run: 2}, tc.transport)
255 peer := routeRoleArgs(s, processMapping{Role: tc.peer, Run: 2}, tc.transport)
256 if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) {
257 t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args)
258 }
259 if tc.peer == "ssh_stdio_peer" && !equalStrings(peer.Args, []string{"_stdio"}) {
260 t.Fatalf("ssh peer command=%q", peer.Args)
261 }
262 if tc.peer == "udp_peer" && !equalStrings(peer.Args, []string{"_udp-proxy"}) {
263 t.Fatalf("udp peer command=%q", peer.Args)
264 }
265 if peer.Transport.RTTMS != tc.wantRTT || peer.Transport.LossPercent != tc.wantLoss {
266 t.Fatalf("peer lost manifest network settings: %+v", peer.Transport)
267 }
268 })
269 }
270}
271
272func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
273 for _, tc := range []struct {
274 name, transport, peer, peerCommand string
275 }{
276 {"ssh", "ssh_stdio", "ssh_stdio_peer", "_stdio"},
277 {"udp baseline", "udp_baseline", "udp_peer", "_udp-proxy"},
278 {"udp 25ms", "udp_25ms", "udp_peer", "_udp-proxy"},
279 {"udp 100ms", "udp_100ms", "udp_peer", "_udp-proxy"},
280 {"udp loss zero", "udp_loss_0pct", "udp_peer", "_udp-proxy"},
281 {"udp loss", "udp_loss_1pct", "udp_peer", "_udp-proxy"},
282 } {
283 t.Run(tc.name, func(t *testing.T) {
284 dir := t.TempDir()
285 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
286 if err != nil {
287 t.Fatal(err)
288 }
289 closeTestFile(t, raw)
290 l := &fakeLauncher{}
291 h := defaultHarness()
292 h.clock, h.launcher = &fakeClock{}, l
293 s := scenario{ID: "routed", Transport: tc.transport, Roles: []string{"daemon", "client", tc.peer}}
294 if _, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, s, 1, raw); err != nil {
295 t.Fatal(err)
296 }
297 if len(l.mappings) != 3 || l.mappings[1].Role != tc.peer || l.mappings[2].Role != "client" {
298 t.Fatalf("dependency launch order=%+v", l.mappings)
299 }
300 if l.commands[1].Args[0] != tc.peerCommand || l.commands[2].Args[0] != "attach" {
301 t.Fatalf("commands do not connect client through declared peer: %+v", l.commands)
302 }
303 for i, p := range l.process {
304 if !p.closed {
305 t.Fatalf("role %s was not cleaned up", l.mappings[i].Role)
306 }
307 }
308 })
309 }
310}
311
312func TestWorkloadInputShapes(t *testing.T) {
313 tests := []struct {
314 workload string
315 contains []string
316 }{
317 {"active_output", []string{`printf '\033[10;20H1'`}},
318 {"all_output", []string{"120", "vev perf line"}},
319 {"inactive_output", []string{
320 `printf '\033[38;2;20;120;220mred'`,
321 `printf '\033[38;2;220;80;40mgreen'`,
322 `printf '\033[38;2;80;220;120ms'`,
323 }},
324 {"interactive_flood", []string{"printf 'vev perf output\\n'"}},
325 {"snapshot_output_resize", []string{`\033[2J`, "vev perf snapshot-output-resize"}},
326 {"attach_restore_tab_switch", []string{`"$VEV_PERF_BIN" cmd next-tab`, `"$VEV_PERF_BIN" cmd previous-tab`}},
327 }
328 for _, tt := range tests {
329 t.Run(tt.workload, func(t *testing.T) {
330 got := string(workloadInput(scenario{ID: "s", Workload: tt.workload}, 1, "measured-1"))
331 for _, want := range tt.contains {
332 if !strings.Contains(got, want) {
333 t.Errorf("workload input %q does not contain %q", got, want)
334 }
335 }
336 if !strings.Contains(got, "__VEV_HARNESS_s_r1_measured-1__") {
337 t.Errorf("workload input has no terminal marker: %q", got)
338 }
339 })
340 }
341
342 activeWarmup := string(workloadInput(scenario{ID: "s", Workload: "active_output"}, 1, "warmup"))
343 activeOne := string(workloadInput(scenario{ID: "s", Workload: "active_output"}, 1, "measured-1"))
344 activeTwo := string(workloadInput(scenario{ID: "s", Workload: "active_output"}, 1, "measured-2"))
345 for phase, got := range map[string]string{"warmup": activeWarmup, "measured-1": activeOne, "measured-2": activeTwo} {
346 marker := fmt.Sprintf("__VEV_HARNESS_s_r1_%s__", phase)
347 wantByte := map[string]byte{"warmup": 'X', "measured-1": '1', "measured-2": '0'}[phase]
348 want := fmt.Sprintf("printf '\\033[10;20H%c'; : 's'; printf '%s\\n'\n", wantByte, marker)
349 if got != want {
350 t.Errorf("active %s input = %q, want exactly one changing cell write %q", phase, got, want)
351 }
352 }
353 if activeWarmup == activeOne || activeOne == activeTwo {
354 t.Fatalf("active one-cell payload did not change by sequence: %q %q %q", activeWarmup, activeOne, activeTwo)
355 }
356 inactive := string(workloadInput(scenario{ID: "s", Workload: "inactive_output"}, 1, "measured-1"))
357 if count := strings.Count(inactive, `\033[38;2;`); count != 3 {
358 t.Fatalf("inactive output truecolor runs = %d, want 3: %q", count, inactive)
359 }
360
361 interactive := string(workloadInput(scenario{ID: "s", Workload: "interactive_flood"}, 1, "measured-1"))
362 if count := strings.Count(interactive, `printf 'vev perf output\n'`); count != 128 {
363 t.Fatalf("interactive flood printf commands = %d, want 128: %q", count, interactive)
364 }
365 if strings.Contains(interactive, "i=0") || strings.Contains(interactive, "while [") {
366 t.Fatalf("interactive flood uses shell-specific loop syntax: %q", interactive)
367 }
368
369 warmup := string(workloadInput(scenario{ID: "s", Workload: "attach_restore_tab_switch"}, 1, "warmup"))
370 if count := strings.Count(warmup, `"$VEV_PERF_BIN" cmd new-tab`); count != 1 {
371 t.Fatalf("attach warmup new-tab commands = %d, want 1: %q", count, warmup)
372 }
373 wantWarmup := `"$VEV_PERF_BIN" cmd new-tab && "$VEV_PERF_BIN" cmd previous-tab && printf 'vev perf attach-restore-tab-switch s\n' && printf '__VEV_HARNESS_s_r1_warmup__\n'` + "\n"
374 if warmup != wantWarmup {
375 t.Fatalf("attach warmup success marker is not gated by every command:\n got %q\nwant %q", warmup, wantWarmup)
376 }
377 measured := string(workloadInput(scenario{ID: "s", Workload: "attach_restore_tab_switch"}, 1, "measured-1"))
378 wantMeasured := `"$VEV_PERF_BIN" cmd next-tab && "$VEV_PERF_BIN" cmd previous-tab && printf 'vev perf attach-restore-tab-switch s\n' && printf '__VEV_HARNESS_s_r1_measured-1__\n'` + "\n"
379 if measured != wantMeasured {
380 t.Fatalf("attach measured success marker is not gated by every command:\n got %q\nwant %q", measured, wantMeasured)
381 }
382}
383
384func TestWorkloadInputResizeCommandsAreComplete(t *testing.T) {
385 for _, tt := range []struct {
386 workload string
387 want string
388 }{
389 {"resize_sweep", `printf 'vev perf resize-sweep s\n'; printf '__VEV_HARNESS_s_r1_measured-1__\n'` + "\n"},
390 {"snapshot_output_resize", `printf '\033[2J\033[Hvev perf snapshot-output-resize s\n'; printf '__VEV_HARNESS_s_r1_measured-1__\n'` + "\n"},
391 } {
392 t.Run(tt.workload, func(t *testing.T) {
393 got := string(workloadInput(scenario{ID: "s", Workload: tt.workload}, 1, "measured-1"))
394 if got != tt.want {
395 t.Fatalf("workload input = %q, want complete command %q", got, tt.want)
396 }
397 if strings.Contains(got, "%!") {
398 t.Fatalf("workload input contains fmt diagnostic: %q", got)
399 }
400 })
401 }
402}
403
404func TestWorkloadResize(t *testing.T) {
405 for _, tt := range []struct {
406 workload string
407 sequence uint64
408 wantCols uint16
409 wantOK bool
410 }{
411 {"active_output", 1, 0, false},
412 {"resize_sweep", 1, 119, true},
413 {"resize_sweep", 2, 120, true},
414 {"snapshot_output_resize", 1, 119, true},
415 } {
416 t.Run(fmt.Sprintf("%s-%d", tt.workload, tt.sequence), func(t *testing.T) {
417 cols, rows, ok := workloadResize(scenario{Workload: tt.workload}, tt.sequence)
418 if ok != tt.wantOK || cols != tt.wantCols {
419 t.Fatalf("workloadResize() = (%d, %d, %t), want cols=%d ok=%t", cols, rows, ok, tt.wantCols, tt.wantOK)
420 }
421 if ok && rows != 40 {
422 t.Fatalf("workloadResize() rows = %d, want 40", rows)
423 }
424 })
425 }
426}
427
428type resizeOrderingWriter struct {
429 bytes.Buffer
430 order *[]string
431}
432
433func (w *resizeOrderingWriter) Write(p []byte) (int, error) {
434 if bytes.Contains(p, []byte(`"sequence":2`)) {
435 switch {
436 case bytes.Contains(p, []byte(`"kind":"input_injected"`)):
437 *w.order = append(*w.order, "injected")
438 case bytes.Contains(p, []byte(`"kind":"terminal_flushed"`)):
439 *w.order = append(*w.order, "terminal_flushed")
440 }
441 }
442 return w.Buffer.Write(p)
443}
444
445type resizeOrderingProcess struct {
446 order *[]string
447 inputs [][]byte
448 resizes int
449 measures int
450}
451
452func (*resizeOrderingProcess) Warmup([]byte) error { return nil }
453
454func (p *resizeOrderingProcess) Resize(cols, rows uint16) error {
455 p.resizes++
456 if p.resizes == 1 {
457 *p.order = append(*p.order, fmt.Sprintf("resize:%dx%d", cols, rows))
458 }
459 return nil
460}
461
462func (p *resizeOrderingProcess) Measure(input []byte, injected, flushed func() error) error {
463 p.measures++
464 p.inputs = append(p.inputs, append([]byte(nil), input...))
465 if p.measures == 1 {
466 *p.order = append(*p.order, "workload_write")
467 }
468 if err := injected(); err != nil {
469 return err
470 }
471 if p.measures == 1 {
472 *p.order = append(*p.order, "workload_flush")
473 }
474 return flushed()
475}
476
477func (*resizeOrderingProcess) Close() error { return nil }
478
479type resizeOrderingLauncher struct{ client *resizeOrderingProcess }
480
481func (l resizeOrderingLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
482 if m.Role == "client" {
483 return l.client, nil
484 }
485 return &fakeProcess{}, nil
486}
487
488func TestResizeBoundaryPrecedesWorkloadWriteAndMarker(t *testing.T) {
489 dir := t.TempDir()
490 order := []string{}
491 client := &resizeOrderingProcess{order: &order}
492 h := defaultHarness()
493 h.clock, h.launcher = &fakeClock{}, resizeOrderingLauncher{client: client}
494 raw := &resizeOrderingWriter{order: &order}
495 _, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "resize", Workload: "resize_sweep", Roles: []string{"daemon", "client"}}, 1, raw)
496 if err != nil {
497 t.Fatal(err)
498 }
499 wantOrder := []string{"injected", "resize:120x40", "workload_write", "workload_flush", "terminal_flushed"}
500 if len(order) < len(wantOrder) {
501 t.Fatalf("resize event order = %q, want prefix %q", order, wantOrder)
502 }
503 if !equalStrings(order[:len(wantOrder)], wantOrder) {
504 t.Fatalf("first resize event order = %q, want %q", order[:len(wantOrder)], wantOrder)
505 }
506 first := string(client.inputs[0])
507 for _, want := range []string{`printf 'vev perf resize-sweep resize\n'`, `printf '__VEV_HARNESS_resize_r1_measured-2__\n'`} {
508 if !strings.Contains(first, want) {
509 t.Errorf("resize input has no observable workload or marker %q: %q", want, first)
510 }
511 }
512 if strings.Contains(first, "while [") || strings.Contains(first, "%!") {
513 t.Fatalf("resize input contains unsupported loop or fmt diagnostic: %q", first)
514 }
515}
516
517func TestHarnessAggregatesRawEventsSeparatelyFromRunDispersion(t *testing.T) {
518 first := append([]int64(nil), make([]int64, minimumInIntervalEventSamples+5)...)
519 for i := range first {
520 first[i] = 1
521 }
522 second := []int64{100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 1_000, 1_000}
523 all, _, _, runP50s := aggregateEventResults([]runResult{
524 {EndToEnd: first, Event: distribution{P50: 1}},
525 {EndToEnd: second, Event: distribution{P50: 100}},
526 })
527 if got, dispersion := percentiles(all), percentiles(runP50s); got.Count != 30 || got.P99 != 1_000 || dispersion.Count != 2 || dispersion.P99 != 1 {
528 t.Fatalf("event aggregation=%+v run dispersion=%+v", got, dispersion)
529 }
530}
531
532func TestHarnessRawRecordsContainEveryMarkWhileSamplesRemainFiltered(t *testing.T) {
533 dir := t.TempDir()
534 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
535 if err != nil {
536 t.Fatal(err)
537 }
538 h := defaultHarness()
539 h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
540 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "raw-complete", Roles: []string{"daemon", "client"}}, 1, raw)
541 if err != nil {
542 t.Fatal(err)
543 }
544 if err := raw.Close(); err != nil {
545 t.Fatal(err)
546 }
547 marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
548 if got, want := len(marks), 2*(result.Samples+1); got != want {
549 t.Fatalf("raw marks=%d, want warmup plus every measured boundary=%d", got, want)
550 }
551 for i, mark := range marks[:2] {
552 if mark.Sequence != 1 || !mark.Valid || []string{"input_injected", "terminal_flushed"}[i] != mark.Kind {
553 t.Fatalf("warmup mark %d=%+v", i, mark)
554 }
555 }
556 for sequence := uint64(2); sequence <= uint64(result.Samples+1); sequence++ {
557 if marks[2*(sequence-1)].Sequence != sequence || marks[2*(sequence-1)+1].Sequence != sequence {
558 t.Fatalf("raw marks did not retain both boundaries for sequence %d: %+v", sequence, marks)
559 }
560 }
561}
562
563func TestHarnessPersistsFailedFlushDiagnostics(t *testing.T) {
564 dir := t.TempDir()
565 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
566 if err != nil {
567 t.Fatal(err)
568 }
569 failure := errors.New("flush failed")
570 h := defaultHarness()
571 h.clock, h.launcher = &fakeClock{}, failedFlushLauncher{err: failure}
572 _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "failed-flush", Roles: []string{"daemon", "client"}}, 1, raw)
573 if !errors.Is(err, failure) {
574 t.Fatalf("run error=%v, want failed flush", err)
575 }
576 if err := raw.Close(); err != nil {
577 t.Fatal(err)
578 }
579 marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
580 if got, want := len(marks), 4; got != want {
581 t.Fatalf("raw marks=%d, want warmup and failed measured pair=%d: %+v", got, want, marks)
582 }
583 failed := marks[len(marks)-1]
584 if failed.Sequence != 2 || failed.Kind != "terminal_flushed" || failed.Valid {
585 t.Fatalf("failed flush diagnostic=%+v", failed)
586 }
587}
588
589func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
590 dir := t.TempDir()
591 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
592 if err != nil {
593 t.Fatal(err)
594 }
595 closeTestFile(t, raw)
596 h := defaultHarness()
597 h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
598 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)
599 if err != nil {
600 t.Fatal(err)
601 }
602 if result.Samples < minimumInIntervalEventSamples || result.Cadence.Count != result.Samples-1 || result.MaxGap <= 0 {
603 t.Fatalf("result did not include repeated measured events: %+v", result)
604 }
605 if err := raw.Close(); err != nil {
606 t.Fatal(err)
607 }
608 records, err := readJSONL(filepath.Join(dir, "raw.jsonl"))
609 if err != nil {
610 t.Fatal(err)
611 }
612 if want := result.Samples*2 + 2; len(records) != want {
613 t.Fatalf("raw records=%d, want %d warmup and measured boundaries", len(records), want)
614 }
615}
616
617func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
618 dir := t.TempDir()
619 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
620 if err != nil {
621 t.Fatal(err)
622 }
623 closeTestFile(t, raw)
624 primary := errors.New("primary workload failure")
625 cleanup := errors.New("daemon cleanup failure")
626 launcher := &fakeLauncher{
627 measureErr: map[string]error{"client": primary},
628 closeErr: map[string]error{"daemon": cleanup},
629 }
630 h := defaultHarness()
631 h.clock, h.launcher = &fakeClock{}, launcher
632 _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, scenario{ID: "errors", Roles: []string{"daemon", "client"}}, 1, raw)
633 if !errors.Is(err, primary) || !errors.Is(err, cleanup) {
634 t.Fatalf("workload and cleanup errors were not both preserved: %v", err)
635 }
636 if got, want := err.Error(), primary.Error()+"\nclose daemon role: "+cleanup.Error(); got != want {
637 t.Fatalf("error ordering=%q want %q", got, want)
638 }
639 for i, p := range launcher.process {
640 if !p.closed {
641 t.Fatalf("role %s was not closed", launcher.mappings[i].Role)
642 }
643 }
644}
645
646func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) {
647 dir := t.TempDir()
648 raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
649 if err != nil {
650 t.Fatal(err)
651 }
652 closeTestFile(t, raw)
653 launcher := &fakeLauncher{}
654 h := defaultHarness()
655 h.clock = &fakeClock{}
656 // The client is the second role and fails after the daemon has started.
657 h.launcher = &failingLauncher{fakeLauncher: launcher}
658 o := options{vevBin: "ignored", out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}
659 if _, err := h.runOne(o, manifest{}, scenario{ID: "cleanup", Roles: []string{"daemon", "client"}}, 1, raw); err == nil {
660 t.Fatal("failed process accepted")
661 }
662 if _, err := os.Stat(filepath.Join(dir, "cleanup-run-001")); !os.IsNotExist(err) {
663 t.Fatalf("failed run directory remains: %v", err)
664 }
665}