Enhance test coverage

Author: markdingo

common_test.go

@@ -33,12 +33,66 @@ func (tbw *testBufWriter) Len() int {
 	return tbw.buf.Len()
 }
 
-// A Writer which can be configured to return a truncated Write() and/or an error.
+type testTTWresult struct {
+	desc string
+	len  int // 0: return 0, -1: return len(p), otherwise use len
+	err  error
+}
+
+// A Writer which can be configured to return a truncated Write() and/or an error. Each
+// call to Write pops a result off the stack and returns the results. If the stack is
+// empty the Write returns success. All successful writes are stored in buf.
 type testTruncateWriter struct {
-	len int
-	err error
+	index   int
+	results []testTTWresult
+	buf     bytes.Buffer
+}
+
+func (ttw *testTruncateWriter) String() string {
+	return ttw.buf.String()
 }
 
+// Place this result at the bottom of the stack
+func (ttw *testTruncateWriter) append(desc string, len int, err error) {
+	ttw.results = append(ttw.results, testTTWresult{desc, len, err})
+}
+
+// Pop the next results and return them to caller. A zero length returns zero, a positive
+// length means return that value and -1 means return the length of the slice.
 func (ttw *testTruncateWriter) Write(p []byte) (n int, err error) {
-	return ttw.len, ttw.err
+	ttw.index++
+	if len(ttw.results) == 0 { // If stack is empty, return complete success
+		return ttw.buf.Write(p)
+	}
+
+	r := ttw.results[0] // Pop off stack
+	ttw.results = ttw.results[1:]
+	err = r.err
+	switch r.len {
+	case 0:
+		n = 0
+	case -1:
+		n = len(p)
+	default:
+		n = r.len
+		if n > len(p) {
+			n = len(p)
+		}
+	}
+
+	if n > 0 {
+		ttw.buf.Write(p[:n])
+	}
+
+	return
+}
+
+func (ttw *testTruncateWriter) close() {
+}
+
+func (ttw *testTruncateWriter) getNext() writer {
+	return nil
+}
+
+func (ttw *testTruncateWriter) setNext(w writer) {
 }

queue_test.go

@@ -1,6 +1,7 @@
 package parallel
 
 import (
+	"errors"
 	"testing"
 	"time"
 )
@@ -149,3 +150,55 @@ func TestQueueBlock(t *testing.T) {
 	outQ.close()
 	errQ.close()
 }
+
+// First error should be returned and all subsequent errors ignored. Once a write fails,
+// no further writes to the io.Writer should occur.
+func TestQueueTransferOut(t *testing.T) {
+	ob := &testTruncateWriter{}
+	eb := &testTruncateWriter{}
+	outQ, errQ := newQueue(false, 0, ob, eb)
+	outQ.Write([]byte{'a', 'b', 'c', '\n'})
+	outQ.Write([]byte{'x', 'y', 'z', '\n'})
+	errQ.Write([]byte{'A', 'B', 'C', '\n'})
+	errQ.Write([]byte{'X', 'Y', 'Z', '\n'})
+	ob.append("stdout Write Fail", 1, errors.New("Stdout write failed"))
+	eb.append("stderr should not be visible", 2, errors.New("Stderr write Failed"))
+	e := outQ.cq.buf.transfer(ob, eb)
+	if e == nil {
+		t.Error("Expected error return from transfer")
+	}
+	res := ob.String()
+	if res != "a" { // Should be one byte
+		t.Error("Expected stdout to be 'a', not", res)
+	}
+	res = eb.String()
+	if res != "AB" { // Should be two bytes
+		t.Error("Expected stderr to be 'AB', not", res)
+	}
+}
+
+// All of stdout should be written, but stderr should be truncated and return an error.
+func TestQueueTransferErr(t *testing.T) {
+	ob := &testTruncateWriter{}
+	eb := &testTruncateWriter{}
+	outQ, errQ := newQueue(false, 0, ob, eb)
+	outQ.Write([]byte{'a'})
+	outQ.Write([]byte{'b'})
+	outQ.Write([]byte{'c'})
+	outQ.Write([]byte{'d'})
+	errQ.Write([]byte{'A', 'B', 'C', '\n'})
+	errQ.Write([]byte{'X', 'Y', 'Z', '\n'})
+	eb.append("stderr Write Fail", 3, errors.New("Stderr write failed"))
+	e := outQ.cq.buf.transfer(ob, eb)
+	if e == nil {
+		t.Error("Expected error return from transfer")
+	}
+	res := ob.String()
+	if res != "abcd" { // Should be one byte
+		t.Error("Expected stdout to be 'abcd', not", res)
+	}
+	res = eb.String()
+	if res != "ABC" { // Should be three bytes
+		t.Error("Expected stderr to be 'ABC', not", res)
+	}
+}

tagger_test.go

@@ -1,6 +1,7 @@
 package parallel
 
 import (
+	"errors"
 	"testing"
 )
 
@@ -24,7 +25,7 @@ func TestTaggerEmpty(t *testing.T) {
 	}
 }
 
-// Test tagger prepend logic with a whole line data
+// Prepend with a whole line data
 func TestTaggerSimple(t *testing.T) {
 	var buf testBufWriter
 	wtr := newTagger(&buf, []byte("host1: "))
@@ -66,7 +67,7 @@ func TestTaggerNoTrailingNL(t *testing.T) {
 	}
 }
 
-// Test tagger prepend logic with partial lines
+// Prepend with partial lines
 func TestTaggerPartialWrites(t *testing.T) {
 	var buf testBufWriter
 	wtr := newTagger(&buf, []byte("host1: "))
@@ -89,7 +90,7 @@ func TestTaggerPartialWrites(t *testing.T) {
 	}
 }
 
-// Test tagger with zero-length writes
+// Zero-length writes
 func TestTaggerWriteZero(t *testing.T) {
 	var buf testBufWriter
 	wtr := newTagger(&buf, []byte("host1: "))
@@ -109,3 +110,106 @@ func TestTaggerWriteZero(t *testing.T) {
 		t.Error("Unexpected modification. \nExp", exp, "\nactual", act)
 	}
 }
+
+// Data is zero length
+func TestTaggerIOErrorsW0(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte(""))
+	b, err := wtr.Write([]byte{})
+	if b != 0 || err != nil {
+		t.Error("Expected zero and nil, not", b, err)
+	}
+}
+
+// Tag is zero length so return downstream write results
+func TestTaggerIOErrorsW1(t *testing.T) {
+	buf := &testTruncateWriter{}
+	buf.append("Error on first write", -1, errors.New("W1 error"))
+	wtr := newTagger(buf, []byte(""))
+	b, err := wtr.Write([]byte{'a'})
+	if b != 1 || err == nil {
+		t.Error("Expected one and error, not", b)
+	}
+}
+
+// Error with tag pending on first line
+func TestTaggerIOErrorsW2(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte("W2Tag"))
+	buf.append("Error on first write", 0, errors.New("Tag Error"))
+	b, err := wtr.Write([]byte{'a', '\n', 'b', '\n'})
+	if b != 4 { // User bytes written is unaffected
+		t.Error("Expected 4 bytes written, not", b)
+	}
+	if err == nil || err.Error() != "Tag Error" { // But error should be present
+		t.Error("Expected Tag Error return, not", err)
+	}
+}
+
+// Error on writing of first line
+func TestTaggerIOErrorsW3(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte("W3Tag"))
+	buf.append("W1 is good", -1, nil)
+	buf.append("W3 is error", -1, errors.New("W3 error"))
+
+	data := []byte{'a', 'b', 'c', 'd', '\n', 'A', 'B', '\n'}
+	b, err := wtr.Write(data)
+	if b != len(data) {
+		t.Error("Expected W3 to return", len(data), "not", b)
+	}
+	if err == nil || err.Error() != "W3 error" {
+		t.Error("Expected 'W3 error', not", err)
+	}
+}
+
+// Error on writing NL between split lines
+func TestTaggerIOErrorsW4(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte("W4Tag"))
+	buf.append("W2Tag", -1, nil)
+	buf.append("W3Line 1 'abcd'", -1, nil)
+	buf.append("W4 NL fails", 0, errors.New("W4 NL failed"))
+
+	data := []byte{'a', 'b', 'c', 'd', '\n', 'A', 'B', '\n'}
+	b, err := wtr.Write(data)
+	if b != len(data)-1 { // Minus the NL
+		t.Error("Expected W4 to return", len(data)-1, "not", b)
+	}
+	if err == nil || err.Error() != "W4 NL failed" {
+		t.Error("Expected 'W4 NL failed', not", err)
+	}
+}
+
+// Error on writing tag on last line
+func TestTaggerIOErrorsW5(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte("W5Tag"))
+	buf.append("W5 Tag fails", 2, errors.New("W5 Tag failed"))
+
+	data := []byte{'a', 'b', 'c'} // Last line
+	b, err := wtr.Write(data)
+	if b != 3 { // Tag write failed, but data write succeeds
+		t.Error("Expected W5 to return", 3, "not", b)
+	}
+	if err == nil || err.Error() != "W5 Tag failed" {
+		t.Error("Expected 'W5 Tag failed', not", err)
+	}
+}
+
+// Error on writing data on last line
+func TestTaggerIOErrorsW6(t *testing.T) {
+	buf := &testTruncateWriter{}
+	wtr := newTagger(buf, []byte("W5Tag"))
+	buf.append("W5 Tag ok", -1, nil)
+	buf.append("W6 LL fails", 2, errors.New("W6 LL failed"))
+
+	data := []byte{'a', 'b', 'c'} // Last line
+	b, err := wtr.Write(data)
+	if b != 2 { // Tag write ok, but data write failed
+		t.Error("Expected W6 to return", 2, "not", b)
+	}
+	if err == nil || err.Error() != "W6 LL failed" {
+		t.Error("Expected 'W6 LL failed', not", err)
+	}
+}