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