kami/orchestra
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

eliminate test harness and SSE race conditions

Browse Source
This commit is contained in:
kami
2026-07-26 20:51:58 +04:00
parent 076c3de6dd
commit 325c684eb0
2 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/admin/admin_test.go
+2 -2
View File
@@ -10,6 +10,7 @@ import (
"orchestra/internal/store" "orchestra/internal/store"
"strings" "strings"
"testing" "testing"
"time"
) )
func TestReadinessRequiresRouterAndReturnsJSON(t *testing.T) { func TestReadinessRequiresRouterAndReturnsJSON(t *testing.T) {
@@ -82,8 +83,7 @@ func TestSubscribeEmitsCursorAndEvent(t *testing.T) {
w := httptest.NewRecorder() w := httptest.NewRecorder()
done := make(chan struct{}) done := make(chan struct{})
go func() { h(w, r); close(done) }() go func() { h(w, r); close(done) }()
for !strings.Contains(w.Body.String(), "id: 1") { time.Sleep(150 * time.Millisecond)
}
cancel() cancel()
<-done <-done
if !strings.Contains(w.Body.String(), "id: 1") { if !strings.Contains(w.Body.String(), "id: 1") {
internal/integration/end_to_end_test.go
+14 -7
View File
@@ -13,6 +13,7 @@ import (
"orchestra/internal/store" "orchestra/internal/store"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"testing" "testing"
"time" "time"
) )
@@ -117,7 +118,10 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
go c.Monitor(ctx, .8, time.Millisecond) go c.Monitor(ctx, .8, time.Millisecond)
deadline := time.Now().Add(time.Second) deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) { for time.Now().Before(deadline) {
if h.releases == 1 { h.mu.Lock()
released := h.releases == 1
h.mu.Unlock()
if released {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@@ -132,8 +136,11 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
} }
got, _ = s.Task(task.ID) got, _ = s.Task(task.ID)
} }
if got.State != domain.StateQueued || h.releases != 1 { h.mu.Lock()
t.Fatalf("rotation state=%s releases=%d", got.State, h.releases) releases := h.releases
h.mu.Unlock()
if got.State != domain.StateQueued || releases != 1 {
t.Fatalf("rotation state=%s releases=%d", got.State, releases)
} }
if _, err := rt.AssignPending(); err != nil { if _, err := rt.AssignPending(); err != nil {
t.Fatal(err) t.Fatal(err)
@@ -227,14 +234,14 @@ func TestProviderRetryReflectionQuotaAndVersionConflict(t *testing.T) {
if reflector.events != 1 { if reflector.events != 1 {
t.Fatalf("reflections=%d", reflector.events) t.Fatalf("reflections=%d", reflector.events)
} }
var calls int var calls atomic.Int32
sup := provider.Supervisor{Name: "fake", Backoff: time.Millisecond, Run: func(context.Context) error { calls++; return errors.New("retry") }} sup := provider.Supervisor{Name: "fake", Backoff: time.Millisecond, Run: func(context.Context) error { calls.Add(1); return errors.New("retry") }}
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
sup.Start(ctx) sup.Start(ctx)
time.Sleep(5 * time.Millisecond) time.Sleep(5 * time.Millisecond)
cancel() cancel()
if calls < 2 || sup.Health().LastError != "retry" { if calls.Load() < 2 || sup.Health().LastError != "retry" {
t.Fatalf("calls=%d health=%+v", calls, sup.Health()) t.Fatalf("calls=%d health=%+v", calls.Load(), sup.Health())
} }
} }