Verified Moving all legacy quacker source code from github to cgitHEADmain

1 files changed, 106 insertions, 0 deletions

diff --git a/pipeline/batch_test.go b/pipeline/batch_test.go
new file mode 100644
index 0000000..4d79ddc
--- /dev/null
+++ b/pipeline/batch_test.go
@@ -0,0 +1,106 @@
+package pipeline
+
+import (
+        "context"
+        "testing"
+        "time"
+)
+
+func TestBatch_Full(t *testing.T) {
+        t.Parallel()
+
+        input := make(chan int, 10)
+        for i := 1; i <= 10; i++ {
+                input <- i
+        }
+        close(input)
+
+        out := Batch(t.Context(), input, 3, time.Second)
+
+        var batches [][]int
+        for b := range out {
+                batches = append(batches, b)
+        }
+
+        if len(batches) != 4 {
+                t.Fatalf("expected 4 batches, got %d: %v", len(batches), batches)
+        }
+        if len(batches[0]) != 3 {
+                t.Fatalf("expected first batch size 3, got %d", len(batches[0]))
+        }
+        if len(batches[3]) != 1 {
+                t.Fatalf("expected last batch size 1, got %d", len(batches[3]))
+        }
+}
+
+func TestBatch_TimeoutFlush(t *testing.T) {
+        t.Parallel()
+
+        input := make(chan int)
+
+        out := Batch(t.Context(), input, 100, 50*time.Millisecond)
+
+        input <- 1
+        input <- 2
+
+        select {
+        case batch := <-out:
+                if len(batch) != 2 {
+                        t.Fatalf("expected batch of 2 from timeout flush, got %d", len(batch))
+                }
+        case <-time.After(time.Second):
+                t.Fatal("timeout waiting for batch flush")
+        }
+
+        close(input)
+        for range out {
+        }
+}
+
+func TestBatch_ContextCancel(t *testing.T) {
+        t.Parallel()
+        ctx, cancel := context.WithCancel(t.Context())
+
+        input := make(chan int)
+        out := Batch(ctx, input, 100, time.Second)
+
+        cancel()
+
+        done := make(chan struct{})
+        go func() {
+                for range out {
+                }
+                close(done)
+        }()
+
+        select {
+        case <-done:
+        case <-time.After(time.Second):
+                t.Fatal("timeout waiting for shutdown")
+        }
+}
+
+func TestBatch_NilInput(t *testing.T) {
+        t.Parallel()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic for nil input")
+                }
+        }()
+
+        Batch[int](t.Context(), nil, 10, time.Second)
+}
+
+func TestBatch_PanicOnZeroSize(t *testing.T) {
+        t.Parallel()
+
+        defer func() {
+                if r := recover(); r == nil {
+                        t.Error("expected panic for batch size < 1")
+                }
+        }()
+
+        input := make(chan int)
+        Batch(t.Context(), input, 0, time.Second)
+}