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

// 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")
    }
}