From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Thu, 3 Sep 2026 23:10:33 +0700
Subject: Moving all legacy quacker source code from github to cgit

---
 pipeline/pipeline_test.go | 255 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 255 insertions(+)
 create mode 100644 pipeline/pipeline_test.go

(limited to 'pipeline/pipeline_test.go')

diff --git a/pipeline/pipeline_test.go b/pipeline/pipeline_test.go
new file mode 100644
index 0000000..9593356
--- /dev/null
+++ b/pipeline/pipeline_test.go
@@ -0,0 +1,255 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestRun_Passthrough(t *testing.T) {
+	t.Parallel()
+
+	input := make(chan int, 3)
+	input <- 1
+	input <- 2
+	input <- 3
+	close(input)
+
+	out := Run(t.Context(), input)
+
+	var got []int
+	for v := range out {
+		got = append(got, v)
+	}
+
+	if len(got) != 3 {
+		t.Fatalf("expected 3 items, got %d", len(got))
+	}
+}
+
+func TestRun_NilInput(t *testing.T) {
+	t.Parallel()
+
+	defer func() {
+		if r := recover(); r == nil {
+			t.Error("expected panic for nil input")
+		}
+	}()
+
+	Run[int](t.Context(), nil)
+}
+
+func TestRun_Composition(t *testing.T) {
+	t.Parallel()
+
+	input := make(chan int, 10)
+	for i := 1; i <= 10; i++ {
+		input <- i
+	}
+	close(input)
+
+	out := Run(t.Context(), input,
+		Filter(func(v int) bool { return v%2 == 0 }),
+		Map(1, func(_ context.Context, v int) (int, bool) {
+			return v * 2, true
+		}),
+	)
+
+	var got []int
+	for v := range out {
+		got = append(got, v)
+	}
+
+	if len(got) != 5 {
+		t.Fatalf("expected 5 items, got %d: %v", len(got), got)
+	}
+
+	sum := 0
+	for _, v := range got {
+		sum += v
+	}
+	if sum != 60 {
+		t.Fatalf("expected sum 60, got %d", sum)
+	}
+}
+
+func TestRun_Backpressure(t *testing.T) {
+	t.Parallel()
+
+	input := make(chan int)
+	consumed := make(chan int, 10)
+
+	out := Run(t.Context(), input,
+		Map(1, func(_ context.Context, v int) (int, bool) {
+			consumed <- v
+			return v, true
+		}),
+	)
+
+	input <- 42
+	v := <-consumed
+	if v != 42 {
+		t.Fatalf("expected 42, got %d", v)
+	}
+
+	select {
+	case input <- 99:
+		t.Fatal("producer was not blocked")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	go func() {
+		for range out {
+		}
+	}()
+	close(input)
+}
+
+func TestRun_GracefulDrain(t *testing.T) {
+	t.Parallel()
+
+	input := make(chan int, 5)
+	for i := 1; i <= 5; i++ {
+		input <- i
+	}
+	close(input)
+
+	var count atomic.Int32
+	out := Run(t.Context(), input,
+		Sink(2, func(_ context.Context, v int) {
+			count.Add(1)
+		}),
+	)
+
+	for range out {
+	}
+
+	if count.Load() != 5 {
+		t.Fatalf("expected all 5 items drained, got %d", count.Load())
+	}
+}
+
+func TestRun_Cancellation(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithCancel(t.Context())
+
+	input := make(chan int)
+	out := Run(ctx, input,
+		Map(1, func(_ context.Context, v int) (int, bool) {
+			return v, true
+		}),
+	)
+
+	go func() {
+		select {
+		case input <- 1:
+		case <-ctx.Done():
+		}
+	}()
+
+	cancel()
+
+	done := make(chan struct{})
+	go func() {
+		for range out {
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("pipeline did not shut down after cancellation")
+	}
+}
+
+func TestRun_ErrorPropagation(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithCancelCause(t.Context())
+	defer cancel(nil)
+
+	input := make(chan int, 5)
+	for i := 1; i <= 5; i++ {
+		input <- i
+	}
+	close(input)
+
+	sentinel := fmt.Errorf("bad item")
+
+	out := Run(ctx, input,
+		Map(1, func(_ context.Context, v int) (int, bool) {
+			if v == 3 {
+				cancel(sentinel)
+				return v, false
+			}
+			return v * 2, true
+		}),
+	)
+
+	for range out {
+	}
+
+	if cause := context.Cause(ctx); cause != sentinel {
+		t.Fatalf("expected sentinel error, got: %v", cause)
+	}
+}
+
+func TestObserver(t *testing.T) {
+	t.Parallel()
+
+	var items, drops atomic.Int32
+	doneCh := make(chan struct{})
+
+	obs := &testObserver{
+		onItem: func() { items.Add(1) },
+		onDrop: func() { drops.Add(1) },
+		onDone: func() { close(doneCh) },
+	}
+
+	input := make(chan int, 6)
+	for i := 1; i <= 6; i++ {
+		input <- i
+	}
+	close(input)
+
+	out := Run(t.Context(), input,
+		Filter(func(v int) bool { return v%2 == 0 }, WithObserver(obs)),
+	)
+
+	for range out {
+	}
+	<-doneCh
+
+	if items.Load() != 3 {
+		t.Errorf("expected 3 items, got %d", items.Load())
+	}
+	if drops.Load() != 3 {
+		t.Errorf("expected 3 drops, got %d", drops.Load())
+	}
+}
+
+type testObserver struct {
+	onItem func()
+	onDrop func()
+	onDone func()
+}
+
+func (o *testObserver) OnItem() {
+	if o.onItem != nil {
+		o.onItem()
+	}
+}
+
+func (o *testObserver) OnDrop() {
+	if o.onDrop != nil {
+		o.onDrop()
+	}
+}
+
+func (o *testObserver) OnDone() {
+	if o.onDone != nil {
+		o.onDone()
+	}
+}
-- 
cgit v1.3.1