How to test channel pipelines in Go

Question

I use the "channel pipeline" pattern quite a lot in Go, which looks something like this:

// getSomeNums spits out ints onto a channel. Temperatures, pressures, doesn't matter
func getSomeNums(ch chan<- int) {
    // Imagination goes here
}

// double takes numbers from in, doubles them, and pushes them into out
func double(in <-chan int, out chan<- int) {
    for v := range in {
        out <- v * 2
    close(out)
}

source := make(chan int)
go getSomeNums(source)

doubles := make(chan int)
double(source, doubles)

The problem I repeatedly run into, is I have to test a number of different features of these pipeline functions:

• Puts a value on the output channel when you put one on the input channel
• Doesn't put a value on the output channel when you don't put one on the input channel
• Times out if the output channel takes too long after a value appears on the input channel
• Closes the output channel when input channel closes
• Doesn't close the output channel before input channel is closed
• Performs the correct transformation on the input

Moreover, this is only a really simple example. More typical cases look something like this example, whereby we're trying to use redundant temperature sensors to find errors in output:

// Provided we have channels for sensorA, sensorB, and sensorC
import "math"

LIMIT = 0.1   // Set acceptable variation limit between sensors to 10%

type SafeTemp struct {
    Temp float64
    isSafe bool
}

// variation returns relative error between inputs. Unfortunately, "error" was taken
func variation(a, b float64) float64 {
    return math.Abs((a - b) / (a + b))
}

// safify zips together temperatures so long as error is below LIMIT
func safify(chA, chB, chC <-chan float64, chOut chan<- SafeTemp) {
    for {
        a, aOk := <-chA
        b, bOk := <-chB
        c, cOk := <-chC

        if !(aOk && bOk && cOk) {
            close(chOut)
            return
        }

        if variation(a, b) < LIMIT && variation(b, c) < LIMIT &&
                variation(c, a) < LIMIT {
            chOut <- SafeTemp{ (a + b + c) / 3, true }
        } else {
            chOut <- SafeTemp{ 0.0, false }
        }

    }
}

Now the number of things I have to test for the pipeline function (safify) increases significantly:

• Puts a value on the output channel when you get one on all input channels
• Doesn't put a value on the output channel when you don't get one on all input channels
• Times out if the output channel takes too long after inputs on all three input channels, but only all three
• Closes the output channel when any input channel closes
• Doesn't close the output channel if no input channel is closed
• Flags as not isSafe if first channel varies significantly from others, with timeouts
• Flags as not isSafe if second channel varies significantly from others, with timeouts
• Flags as not isSafe if third channel varies significantly from others, with timeouts
• Flags as not isSafe if all channels vary significantly from others, with timeouts

Further, the three input channels can go out of sync with each other, which adds significant complexity still beyond that shown above.

It seems that a lot of these checks (except specifically those having to do with correct computations) are common to basically any fan-in-style channel pipeline function in Go, and the Halting Problem guarantees that we've got to use timeouts for all of these operations, unless we want the halting of our tests to depend on the halting and eventual channel-pushing behavior of the functions being tested.

Given how similar these types of tests are across the board, and how I end up writing quite similar tests essentially testing how well these channel pipeline functions conform to basic channel pipeline function guarantees, instead of behavior of the functions, over and over and over again, is there either:

1. A standard set of practices around these kinds of channel pipeline function reliability tests OR
2. A standard or well-hardened framework or set of frameworks for testing channel-native functions?

Answer 1

You are mixing two different concerns. If you made a separate abstraction for the pipeline, you could test that once. Something like (forgive syntax, I don't know go):

func double(v int) int {
    return v * 2
}

pipeline(in, out, double)

or

func safe(v [3]float64) SafeTemp {
    if variation(v[0], v[1]) < LIMIT && variation(v[1], v[2]) < LIMIT &&
            variation(v[2], v[0]) < LIMIT {
        return SafeTemp{ (v[0] + v[1] + v[2]) / 3, true }
    } else {
        return SafeTemp{ 0.0, false }
    }
}

pipeline(in, out, safe)

Without parametric polymorphism, you can't really make a fully generic pipeline abstraction, so you have to accept a certain amount of duplication. However, you should be able to at least separate the concerns of the pipeline pattern from the more application-specific logic.