Skip to content

refactor(tests): replace duplicated wait-for helpers with testutil.Wa… #1039

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Fayupable wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(tests): replace duplicated wait-for helpers with testutil.Wa… #1039

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 6 additions & 18 deletions internal/agent/async_diagnostics_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"testing"
"time"

"github.com/Gitlawb/zero/internal/testutil"
"github.com/Gitlawb/zero/internal/tools"
"github.com/Gitlawb/zero/internal/zeroruntime"
)
Expand Down Expand Up @@ -111,7 +112,11 @@ func TestAsyncDiagnosticsReEditReplacesResult(t *testing.T) {
diagnostics := newAsyncDiagnostics(check, "/ws")

diagnostics.enqueue(context.Background(), []string{"a.go"})
waitForIdle(t, diagnostics)
testutil.WaitFor(t, "worker idle", func() bool {
diagnostics.mu.Lock()
defer diagnostics.mu.Unlock()
return diagnostics.working == nil
})
mu.Lock()
response = "ERR new"
mu.Unlock()
Expand All @@ -123,23 +128,6 @@ func TestAsyncDiagnosticsReEditReplacesResult(t *testing.T) {
}
}

func waitForIdle(t *testing.T, diagnostics *asyncDiagnostics) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for {
diagnostics.mu.Lock()
busy := diagnostics.working
diagnostics.mu.Unlock()
if busy == nil {
return
}
if time.Now().After(deadline) {
t.Fatal("worker never went idle")
}
time.Sleep(2 * time.Millisecond)
}
}

// changedFilesTool is a fake mutating tool reporting a changed file without
// touching the filesystem.
type changedFilesTool struct{}
Expand Down
18 changes: 4 additions & 14 deletions internal/daemon/pool_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ import (
"sync/atomic"
"testing"
"time"

"github.com/Gitlawb/zero/internal/testutil"
)

// --- test doubles ---------------------------------------------------------
Expand Down Expand Up @@ -191,7 +193,7 @@ func TestPoolQueuesWhenFull(t *testing.T) {
}()
<-started
// Wait until the first run holds the only slot.
waitFor(t, func() bool { return pool.QueueDepth() == 1 })
testutil.WaitFor(t, "queue depth 1", func() bool { return pool.QueueDepth() == 1 })

secondDone := make(chan struct{})
go func() {
Expand Down Expand Up @@ -224,7 +226,7 @@ func TestPoolDrainKillsStraggler(t *testing.T) {
_, _ = pool.Run(context.Background(), WorkerSpec{Session: "a"}, &collectSink{})
close(runDone)
}()
waitFor(t, func() bool { return pool.QueueDepth() == 1 })
testutil.WaitFor(t, "queue depth 1", func() bool { return pool.QueueDepth() == 1 })

pool.Drain() // KillTimeout elapses, straggler is force-killed
if atomic.LoadInt32(&straggler.killed) != 1 {
Expand All @@ -242,18 +244,6 @@ func TestPoolDrainKillsStraggler(t *testing.T) {
}
}

func waitFor(t *testing.T, cond func() bool) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if cond() {
return
}
time.Sleep(time.Millisecond)
}
t.Fatal("condition not met within timeout")
}

func TestPoolRunSurfacesStdoutReadError(t *testing.T) {
// A worker stdout read error must surface as a failure (not be swallowed and
// reported as clean success), and the worker is killed so it can't block on an
Expand Down
20 changes: 5 additions & 15 deletions internal/daemon/server_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ import (
"path/filepath"
"testing"
"time"

"github.com/Gitlawb/zero/internal/testutil"
)

func newTestServer(t *testing.T, launcher Launcher) (*Server, Paths) {
Expand All @@ -31,26 +33,14 @@ func newTestServer(t *testing.T, launcher Launcher) (*Server, Paths) {
return srv, paths
}

func waitForFile(t *testing.T, path string) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if _, err := os.Stat(path); err == nil {
return
}
time.Sleep(2 * time.Millisecond)
}
t.Fatalf("file %s did not appear within timeout", path)
}

func TestServerEndToEnd(t *testing.T) {
out := []string{`{"type":"event","seq":1}`, `{"type":"event","seq":2}`}
launcher, _ := seqLauncher(&fakeWorker{pid: 1, out: out, exitCode: 0})
srv, paths := newTestServer(t, launcher)

serveErr := make(chan error, 1)
go func() { serveErr <- srv.Serve() }()
waitForFile(t, paths.Status)
testutil.WaitFor(t, "file "+paths.Status, func() bool { _, err := os.Stat(paths.Status); return err == nil })

// --- run a session and collect its stream-json output ---
runClient, err := Dial(paths.Socket)
Expand Down Expand Up @@ -144,7 +134,7 @@ func TestServerSecondInstanceFails(t *testing.T) {

serveErr := make(chan error, 1)
go func() { serveErr <- srv1.Serve() }()
waitForFile(t, paths.Status)
testutil.WaitFor(t, "file "+paths.Status, func() bool { _, err := os.Stat(paths.Status); return err == nil })

// A second server on the same paths must refuse to start (single instance).
pool2, _ := NewPool(PoolOptions{Size: 1, Launcher: launcher})
Expand All @@ -165,7 +155,7 @@ func TestServerRejectsUnknownCommand(t *testing.T) {
srv, paths := newTestServer(t, launcher)
go func() { _ = srv.Serve() }()
defer srv.Shutdown()
waitForFile(t, paths.Status)
testutil.WaitFor(t, "file "+paths.Status, func() bool { _, err := os.Stat(paths.Status); return err == nil })

client, err := Dial(paths.Socket)
if err != nil {
Expand Down
6 changes: 4 additions & 2 deletions internal/daemon/session_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ import (
"sync/atomic"
"testing"
"time"

"github.com/Gitlawb/zero/internal/testutil"
)

func drain(t *testing.T, buffered []string, live <-chan string) []string {
Expand Down Expand Up @@ -111,7 +113,7 @@ func TestSessionLeaseQueuesWhenPoolFull(t *testing.T) {
mgr, _ := NewSessionManager(SessionManagerOptions{Pool: pool})

sa, _ := mgr.Start(context.Background(), WorkerSpec{Session: "a"})
waitFor(t, func() bool { return sa.State() == SessionRunning })
testutil.WaitFor(t, "session running", func() bool { return sa.State() == SessionRunning })

sb, _ := mgr.Start(context.Background(), WorkerSpec{Session: "b"})
// b must stay queued (its worker not yet launched) while a holds the slot.
Expand Down Expand Up @@ -197,7 +199,7 @@ func TestSessionManagerKeepsRunningOverCap(t *testing.T) {

s1, _ := mgr.Start(context.Background(), WorkerSpec{Session: "r1"})
s2, _ := mgr.Start(context.Background(), WorkerSpec{Session: "r2"})
waitFor(t, func() bool { return s1.State() == SessionRunning && s2.State() == SessionRunning })
testutil.WaitFor(t, "both sessions running", func() bool { return s1.State() == SessionRunning && s2.State() == SessionRunning })
if _, ok := mgr.Get("r1"); !ok {
t.Fatal("a running session must never be evicted, even past the cap")
}
Expand Down
48 changes: 19 additions & 29 deletions internal/swarm/lifecycle_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ import (
"sync"
"testing"
"time"

"github.com/Gitlawb/zero/internal/testutil"
)

// controllableLauncher records every launched spec and lets a test control each
Expand Down Expand Up @@ -80,18 +82,6 @@ func newSwarmFor(t *testing.T, l MemberLauncher) *Swarm {
return sw
}

func waitFor(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if cond() {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("timed out waiting for %s", what)
}

func okFor(spec MemberSpec, _ int) (MemberResult, error) {
return MemberResult{Result: "ok:" + spec.Task, SessionID: "sess-" + spec.ID}, nil
}
Expand All @@ -103,7 +93,7 @@ func TestSpawnCompletes(t *testing.T) {
if err != nil {
t.Fatalf("Spawn: %v", err)
}
waitFor(t, "task done", func() bool {
testutil.WaitFor(t, "task done", func() bool {
task, ok := sw.Coordinator().Get(id)
return ok && task.Status == StatusDone
})
Expand All @@ -120,7 +110,7 @@ func TestSpawnInheritsPolicy(t *testing.T) {
if err != nil {
t.Fatalf("Spawn: %v", err)
}
waitFor(t, "spec recorded", func() bool { return len(l.recorded()) == 1 })
testutil.WaitFor(t, "spec recorded", func() bool { return len(l.recorded()) == 1 })
spec := l.recorded()[0]
if spec.Model != "orch-model" {
t.Fatalf("member model = %q, want inherited orch-model", spec.Model)
Expand Down Expand Up @@ -153,7 +143,7 @@ func TestConcurrencyCapAndQueueDrains(t *testing.T) {
}
// Release everyone; the queue should drain one-per-slot until all are done.
close(gate)
waitFor(t, "all tasks done", func() bool { return sw.Coordinator().Summarize().Done == 5 })
testutil.WaitFor(t, "all tasks done", func() bool { return sw.Coordinator().Summarize().Done == 5 })
if team.Running() != 0 || team.QueueDepth() != 0 {
t.Fatalf("after drain running=%d queue=%d, want 0/0", team.Running(), team.QueueDepth())
}
Expand All @@ -171,7 +161,7 @@ func TestRetryOnTemporaryError(t *testing.T) {
})
sw := newSwarmFor(t, l)
id, _ := sw.Spawn(Policy{}, "team", "teammate", "task", "")
waitFor(t, "task recovered", func() bool {
testutil.WaitFor(t, "task recovered", func() bool {
task, ok := sw.Coordinator().Get(id)
return ok && task.Status == StatusDone
})
Expand All @@ -190,7 +180,7 @@ func TestRetryExhaustionFails(t *testing.T) {
})
sw := newSwarmFor(t, l)
id, _ := sw.Spawn(Policy{}, "team", "teammate", "task", "")
waitFor(t, "task failed", func() bool {
testutil.WaitFor(t, "task failed", func() bool {
task, ok := sw.Coordinator().Get(id)
return ok && task.Status == StatusFailed
})
Expand All @@ -205,7 +195,7 @@ func TestPermanentErrorNoRetry(t *testing.T) {
})
sw := newSwarmFor(t, l)
id, _ := sw.Spawn(Policy{}, "team", "teammate", "task", "")
waitFor(t, "task failed", func() bool {
testutil.WaitFor(t, "task failed", func() bool {
task, ok := sw.Coordinator().Get(id)
return ok && task.Status == StatusFailed
})
Expand Down Expand Up @@ -291,7 +281,7 @@ func TestClosePreventsMemberRetry(t *testing.T) {
sw.Close()
close(closed)
}()
waitFor(t, "swarm closed state", func() bool {
testutil.WaitFor(t, "swarm closed state", func() bool {
sw.lifecycleMu.RLock()
defer sw.lifecycleMu.RUnlock()
return sw.closed
Expand Down Expand Up @@ -320,7 +310,7 @@ func TestCloseWaitsForMemberWatchers(t *testing.T) {
if err != nil {
t.Fatalf("Spawn: %v", err)
}
waitFor(t, "task running", func() bool {
testutil.WaitFor(t, "task running", func() bool {
task, ok := sw.Coordinator().Get(id)
return ok && task.Status == StatusRunning
})
Expand Down Expand Up @@ -360,7 +350,7 @@ func TestHandoffDeliversNoteAndRetiresOriginal(t *testing.T) {
sw := newSwarmFor(t, l)
pol := Policy{Model: "m"}
origID, _ := sw.Spawn(pol, "team", "teammate", "original task", "/w")
waitFor(t, "original running", func() bool {
testutil.WaitFor(t, "original running", func() bool {
task, ok := sw.Coordinator().Get(origID)
return ok && task.Status == StatusRunning
})
Expand All @@ -384,7 +374,7 @@ func TestHandoffDeliversNoteAndRetiresOriginal(t *testing.T) {
}
// The new member carries the handoff note in its task and preserves cwd.
close(gate)
waitFor(t, "spec for new member", func() bool {
testutil.WaitFor(t, "spec for new member", func() bool {
for _, s := range l.recorded() {
if s.ID == newID {
return true
Expand All @@ -400,7 +390,7 @@ func TestHandoffDeliversNoteAndRetiresOriginal(t *testing.T) {
}
}
// A handoff of an already-terminal task is rejected.
waitFor(t, "new task done", func() bool {
testutil.WaitFor(t, "new task done", func() bool {
task, ok := sw.Coordinator().Get(newID)
return ok && task.Status == StatusDone
})
Expand All @@ -427,7 +417,7 @@ func TestAdoptOrphans(t *testing.T) {
t.Fatalf("adopted = %v, want [orphan-1]", adopted)
}
// The orphan is relaunched under a fresh agent and completes.
waitFor(t, "orphan completed", func() bool {
testutil.WaitFor(t, "orphan completed", func() bool {
task, ok := sw.Coordinator().Get("orphan-1")
return ok && task.Status == StatusDone
})
Expand Down Expand Up @@ -468,7 +458,7 @@ func TestCollectScopesToTeam(t *testing.T) {
sw := newSwarmFor(t, l)
a, _ := sw.Spawn(Policy{}, "alpha", "teammate", "ta", "")
_, _ = sw.Spawn(Policy{}, "beta", "teammate", "tb", "")
waitFor(t, "alpha done", func() bool {
testutil.WaitFor(t, "alpha done", func() bool {
task, ok := sw.Coordinator().Get(a)
return ok && task.Status == StatusDone
})
Expand Down Expand Up @@ -614,7 +604,7 @@ func TestAdmittedDispatchDoesNotLaunchAfterClose(t *testing.T) {
sw.Close()
close(closed)
}()
waitFor(t, "swarm closed state", func() bool {
testutil.WaitFor(t, "swarm closed state", func() bool {
sw.lifecycleMu.RLock()
defer sw.lifecycleMu.RUnlock()
return sw.closed
Expand Down Expand Up @@ -674,7 +664,7 @@ func TestCloseBetweenLaunchPrecheckAndReturn(t *testing.T) {
sw.Close()
close(closed)
}()
waitFor(t, "swarm closed state", func() bool {
testutil.WaitFor(t, "swarm closed state", func() bool {
sw.lifecycleMu.RLock()
defer sw.lifecycleMu.RUnlock()
return sw.closed
Expand Down Expand Up @@ -772,7 +762,7 @@ func TestCloseBetweenRetryLaunchPrecheckAndReturn(t *testing.T) {
sw.Close()
close(closed)
}()
waitFor(t, "swarm closed state", func() bool {
testutil.WaitFor(t, "swarm closed state", func() bool {
sw.lifecycleMu.RLock()
defer sw.lifecycleMu.RUnlock()
return sw.closed
Expand Down Expand Up @@ -964,7 +954,7 @@ func TestCloseFailsQueuedSpecOnLateCreatedTeam(t *testing.T) {
sw.Close()
close(closed)
}()
waitFor(t, "swarm closed state", func() bool {
testutil.WaitFor(t, "swarm closed state", func() bool {
sw.lifecycleMu.RLock()
defer sw.lifecycleMu.RUnlock()
return sw.closed
Expand Down
Loading