How do you test Go code that processes streaming data or works with channels?

Testing channel-based pipelines requires careful synchronisation. Common patterns: bounded channels with timeout assertions, channel-based test doubles that feed input and capture output, and the fan-out test harness.

// Pipeline under test
func processEvents(ctx context.Context, in <-chan Event) <-chan ProcessedEvent {
    out := make(chan ProcessedEvent)
    go func() {
        defer close(out)
        for event := range in {
            result := transform(event)
            select {
            case out <- result:
            case <-ctx.Done(): return
            }
        }
    }()
    return out
}

// Test helper: send N items and collect results with timeout
func drainChannel[T any](t *testing.T, ch <-chan T, timeout time.Duration) []T {
    t.Helper()
    var results []T
    timer := time.NewTimer(timeout)
    defer timer.Stop()
    for {
        select {
        case item, ok := <-ch:
            if !ok { return results } // channel closed
            results = append(results, item)
        case <-timer.C:
            t.Fatalf("timeout: channel did not close after %v", timeout)
            return results
        }
    }
}

// Test the pipeline
func TestProcessEvents(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    input := make(chan Event, 3)
    input <- Event{ID: 1, Type: "click"}
    input <- Event{ID: 2, Type: "view"}
    input <- Event{ID: 3, Type: "click"}
    close(input) // signal end of stream

    output := processEvents(ctx, input)
    results := drainChannel(t, output, 3*time.Second)

    if len(results) != 3 {
        t.Errorf("expected 3 results, got %d", len(results))
    }
    // Assert specific transformations
    for _, r := range results {
        if r.ProcessedAt.IsZero() {
            t.Errorf("ProcessedAt not set for event %d", r.EventID)
        }
    }
}

// Test cancellation: pipeline exits cleanly when ctx is cancelled
func TestProcessEvents_Cancellation(t *testing.T) {
    ctx, cancel := context.WithCancel(context.Background())
    input := make(chan Event) // never closes
    output := processEvents(ctx, input)

    cancel() // cancel immediately

    // Output should close promptly after cancellation
    select {
    case _, ok := <-output:
        if ok { t.Error("expected channel to be closed after cancellation") }
    case <-time.After(time.Second):
        t.Error("channel did not close after cancellation")
    }
}

Invest now in Acorns!!! 🚀
Join Acorns and get your $5 bonus!

Acorns is a micro-investing app that automatically invests your "spare change" from daily purchases into diversified, expert-built portfolios of ETFs. It is designed for beginners, allowing you to start investing with as little as $5. The service automates saving and investing. Disclosure: I may receive a referral bonus.

Invest now!!! Get Free equity stock (US, UK only)!

Use Robinhood app to invest in stocks. It is safe and secure. Use the Referral link to claim your free stock when you sign up!.

The Robinhood app makes it easy to trade stocks, crypto and more.

Webull! Receive free stock by signing up using the link: Webull signup.