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

test: update integration lifecycle fixtures

Browse Source
This commit is contained in:
kami
2026-07-26 20:49:30 +04:00
parent 18e35bcd45
commit 076c3de6dd
1 changed files with 25 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/integration/end_to_end_test.go
+25 -4
View File
@@ -34,6 +34,9 @@ func (h *harness) Release(context.Context, herdr.Session) (string, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.releases++
// A boundary is a one-shot turn transition. Once released, the next
// lease represents a new turn and must not be released again immediately.
h.boundary = false
return h.ref, nil
}
func (h *harness) Kill(context.Context, herdr.Session) error {
@@ -114,12 +117,21 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
go c.Monitor(ctx, .8, time.Millisecond)
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
if got, _ := s.Task(task.ID); got.State == domain.StateQueued {
if h.releases == 1 {
break
}
time.Sleep(time.Millisecond)
}
got, _ := s.Task(task.ID)
if got.State == domain.StateLeased && h.releases == 1 {
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{
"handoff_ref": h.ref,
"anchor_sha": "0123456789012345678901234567890123456789",
})}); err != nil {
t.Fatal(err)
}
got, _ = s.Task(task.ID)
}
if got.State != domain.StateQueued || h.releases != 1 {
t.Fatalf("rotation state=%s releases=%d", got.State, h.releases)
}
@@ -134,7 +146,10 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil {
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]any{
"report_ref": ref,
"receipt": map[string]any{"harness_id": "h1", "consumed": 1},
})}); err != nil {
t.Fatal(err)
}
got, _ = s.Task(task.ID)
@@ -159,7 +174,10 @@ func TestRestartReplayAndReconcileKillsOrphan(t *testing.T) {
}
// A fresh coordinator sees the durable session, then drops it once the lease is gone.
ref, _ := s.PutArtifact([]byte("handoff"))
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{"handoff_ref": ref})}); err != nil {
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{
"handoff_ref": ref,
"anchor_sha": "0123456789012345678901234567890123456789",
})}); err != nil {
t.Fatal(err)
}
s2, err := store.Open(dir)
@@ -200,7 +218,10 @@ func TestProviderRetryReflectionQuotaAndVersionConflict(t *testing.T) {
t.Fatal(err)
}
reflector := &fakeReflector{}
if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil {
if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]any{
"report_ref": ref,
"receipt": map[string]any{"harness_id": "h1", "consumed": 1},
})}); err != nil {
t.Fatal(err)
}
if reflector.events != 1 {