Merge pull request #101 from apenella/fix-parsing-of-long-lines-in-ou…

…tput

Fix parsing of long lines in output

mocks/mockIOReader.go

@@ -0,0 +1,18 @@
+package mocks
+
+import (
+	"github.com/stretchr/testify/mock"
+)
+
+type MockIOReader struct {
+	mock.Mock
+}
+
+func NewMockIOReader() *MockIOReader {
+	return &MockIOReader{}
+}
+
+func (m *MockIOReader) Read(p []byte) (n int, err error) {
+	args := m.Called(p)
+	return args.Get(0).(int), args.Get(1).(error)
+}

mocks/mockIOWriter.go

@@ -0,0 +1,18 @@
+package mocks
+
+import (
+	"github.com/stretchr/testify/mock"
+)
+
+type MockIOWriter struct {
+	mock.Mock
+}
+
+func NewMockIOWriter() *MockIOWriter {
+	return &MockIOWriter{}
+}
+
+func (m *MockIOWriter) Write(p []byte) (n int, err error) {
+	args := m.Called(p)
+	return args.Get(0).(int), args.Get(1).(error)
+}

pkg/stdoutcallback/results/DefaultResults.go

@@ -12,31 +12,34 @@ import (
 
 // DefaultStdoutCallbackResults is the default method to print ansible-playbook results
 func DefaultStdoutCallbackResults(ctx context.Context, r io.Reader, w io.Writer, trans ...TransformerFunc) error {
+	var transformers []TransformerFunc
 
-	tranformers := []TransformerFunc{}
+	errContext := "(results::DefaultStdoutCallbackResults)"
 
 	if len(trans) > 0 {
-		tranformers = append(tranformers, Prepend(PrefixTokenSeparator))
+		transformers = append(transformers, Prepend(PrefixTokenSeparator))
 	}
 
-	tranformers = append(tranformers, trans...)
+	transformers = append(transformers, trans...)
 
-	err := output(ctx, r, w, tranformers...)
+	err := output(ctx, r, w, transformers...)
 	if err != nil {
-		return errors.New("(results::DefaultStdoutCallbackResults)", "Error processing execution output", err)
+		return errors.New(errContext, "Error processing execution output", err)
 	}
 
 	return nil
 }
 
 // output process the output data with the transformers comming from the execution an writes it to the input writer
 func output(ctx context.Context, r io.Reader, w io.Writer, trans ...TransformerFunc) error {
-
 	printChan := make(chan string)
+	errChan := make(chan error)
 	done := make(chan struct{})
 
+	errContext := "(results::output)"
+
 	if r == nil {
-		return errors.New("(results::DefaultStdoutCallbackResults)", "Reader is not defined")
+		return errors.New(errContext, "Reader is not defined")
 	}
 
 	if w == nil {
@@ -49,11 +52,19 @@ func output(ctx context.Context, r io.Reader, w io.Writer, trans ...TransformerF
 
 	go func() {
 		defer close(done)
+		defer close(errChan)
 		defer close(printChan)
 
-		scanner := bufio.NewScanner(r)
-		for scanner.Scan() {
-			line := scanner.Text()
+		reader := bufio.NewReader(r)
+		for {
+			line, err := readLine(reader)
+			if err != nil {
+				if err != io.EOF {
+					errChan <- err
+				}
+
+				break
+			}
 
 			for _, t := range trans {
 				line = t(line)
@@ -67,11 +78,38 @@ func output(ctx context.Context, r io.Reader, w io.Writer, trans ...TransformerF
 	for {
 		select {
 		case line := <-printChan:
-			fmt.Fprintf(w, "%s\n", line)
+			_, err := fmt.Fprintln(w, line)
+			if err != nil {
+				return err
+			}
+		case err := <-errChan:
+			return err
 		case <-done:
 			return nil
 		case <-ctx.Done():
 			return nil
 		}
 	}
 }
+
+func readLine(r *bufio.Reader) (string, error) {
+	var line []byte
+	for {
+		l, more, err := r.ReadLine()
+		if err != nil {
+			return "", err
+		}
+
+		// Avoid the copy if the first call produced a full line.
+		if line == nil && !more {
+			return string(l), nil
+		}
+
+		line = append(line, l...)
+		if !more {
+			break
+		}
+	}
+
+	return string(line), nil
+}

pkg/stdoutcallback/results/DefaultResults_test.go

@@ -4,14 +4,25 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"fmt"
 	"io"
+	"math/rand"
 	"strings"
 	"testing"
 
+	"github.com/apenella/go-ansible/mocks"
+	errors "github.com/apenella/go-common-utils/error"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 )
 
+const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
 func TestDefaultStdoutCallbackResults(t *testing.T) {
+
+	longMessageLine := randStringBytes(512_000)
+	// errContext := "(results::DefaultStdoutCallbackResults)"
+
 	tests := []struct {
 		desc  string
 		input string
@@ -45,21 +56,142 @@ PLAY RECAP *********************************************************************
 
 Playbook run took 0 days, 0 hours, 0 minutes, 0 seconds
 `,
-			err: nil,
+			err: &errors.Error{},
+		},
+		{
+			desc: "Testing very long lines reading",
+			input: fmt.Sprintf(`
+PLAY [local] *********************************************************************************************************************************************************************************
+
+TASK [Print test message] ********************************************************************************************************************************************************************
+ok: [127.0.0.1] => 
+	msg: That's a message to test: %s
+
+PLAY RECAP ***********************************************************************************************************************************************************************************
+127.0.0.1                  : ok=1    changed=0    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0   			
+
+Playbook run took 0 days, 0 hours, 0 minutes, 0 seconds
+`, longMessageLine),
+			res: fmt.Sprintf(`
+PLAY [local] *********************************************************************************************************************************************************************************
+
+TASK [Print test message] ********************************************************************************************************************************************************************
+ok: [127.0.0.1] => 
+	msg: That's a message to test: %s
+
+PLAY RECAP ***********************************************************************************************************************************************************************************
+127.0.0.1                  : ok=1    changed=0    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0   			
+
+Playbook run took 0 days, 0 hours, 0 minutes, 0 seconds
+`, longMessageLine),
+			err: &errors.Error{},
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.desc, func(t *testing.T) {
+			t.Log(test.desc)
+
+			var reader io.Reader
+
 			wbuff := bytes.Buffer{}
 			writer := io.Writer(&wbuff)
-			reader := bufio.NewReader(strings.NewReader(test.input))
+
+			reader = bufio.NewReader(strings.NewReader(test.input))
+
 			err := DefaultStdoutCallbackResults(context.TODO(), reader, writer, test.trans...)
-			if err != nil && assert.Error(t, err) {
-				assert.Equal(t, test.err, err)
+			if err != nil {
+				assert.Equal(t, test.err.Error(), err.Error())
 			} else {
 				assert.Equal(t, test.res, wbuff.String(), "Unexpected value")
 			}
 		})
 	}
 }
+
+func TestOutput(t *testing.T) {
+
+	buff := bytes.Buffer{}
+	longMessageLine := randStringBytes(512_000) + "\n"
+	errContext := "(results::output)"
+
+	tests := []struct {
+		desc              string
+		reader            io.Reader
+		writer            io.Writer
+		err               error
+		res               string
+		trans             []TransformerFunc
+		prepareAssertFunc func(io.Reader, io.Writer)
+	}{
+		{
+			desc:   "Testing process an output message",
+			reader: strings.NewReader("output message"),
+			writer: io.Writer(&buff),
+			res:    "output message\n",
+		},
+		{
+			desc:   "Testing process a long output message",
+			reader: bytes.NewReader([]byte(longMessageLine)),
+			writer: io.Writer(&buff),
+			res:    longMessageLine,
+		},
+		{
+			desc:   "Testing error reading output message",
+			reader: mocks.NewMockIOReader(),
+			writer: io.Writer(&buff),
+			res:    longMessageLine,
+			prepareAssertFunc: func(reader io.Reader, writer io.Writer) {
+				if reader != nil {
+					reader.(*mocks.MockIOReader).On(
+						"Read",
+						mock.Anything,
+					).Return(0, errors.New(errContext, "error while reading"))
+				}
+			},
+			err: errors.New(errContext, "error while reading"),
+		},
+		{
+			desc:   "Testing error writing output message",
+			reader: bytes.NewReader([]byte(longMessageLine)),
+			writer: mocks.NewMockIOWriter(),
+			res:    longMessageLine,
+			prepareAssertFunc: func(reader io.Reader, writer io.Writer) {
+				if writer != nil {
+					writer.(*mocks.MockIOWriter).On(
+						"Write",
+						mock.Anything,
+					).Return(0, errors.New(errContext, "error while writing"))
+				}
+			},
+			err: errors.New(errContext, "error while writing"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			t.Log(test.desc)
+
+			buff.Reset()
+
+			if test.prepareAssertFunc != nil {
+				test.prepareAssertFunc(test.reader, test.writer)
+			}
+
+			err := output(context.TODO(), test.reader, test.writer, test.trans...)
+			if err != nil {
+				assert.Equal(t, test.err.Error(), err.Error())
+			} else {
+				assert.Equal(t, test.res, buff.String(), "Unexpected value")
+			}
+		})
+	}
+}
+
+func randStringBytes(n int) string {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = letterBytes[rand.Intn(len(letterBytes))]
+	}
+	return string(b)
+}

pkg/stdoutcallback/results/JSONResults_test.go

@@ -4,6 +4,8 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"encoding/json"
+	"fmt"
 	"io"
 	"strings"
 	"testing"
@@ -13,9 +15,7 @@ import (
 )
 
 func TestStdoutCallbackJSONResults(t *testing.T) {
-
-	t.Skip()
-
+	longMessageLine := randStringBytes(512_000)
 	tests := []struct {
 		desc           string
 		inputResult    string
@@ -104,6 +104,39 @@ func TestStdoutCallbackJSONResults(t *testing.T) {
 			expectedResult: `{				"custom_stats": {},				"global_custom_stats": {},				"plays": [					{						"play": {							"duration": {								"end": "2020-08-07T20:51:30.942955Z",								"start": "2020-08-07T20:51:30.607525Z"							},							"id": "a0a4c5d1-62fd-b6f1-98ea-000000000006",							"name": "local"						},						"tasks": [							{								"hosts": {									"127.0.0.1": {										"_ansible_no_log": false,										"_ansible_verbose_always": true,										"action": "debug",										"changed": false,										"msg": "That's a message to debug"									}								},								"task": {									"duration": {										"end": "2020-08-07T20:51:30.942955Z",										"start": "2020-08-07T20:51:30.908539Z"									},									"id": "a0a4c5d1-62fd-b6f1-98ea-000000000008",									"name": "Print line"								}							},							{								"hosts": {									"192.198.1.1": {										"_ansible_no_log": false,										"_ansible_verbose_always": true,										"action": "debug",										"changed": false,										"msg": "That's a message to debug"									}								},								"task": {									"duration": {										"end": "2020-08-07T20:51:30.942955Z",										"start": "2020-08-07T20:51:30.908539Z"									},									"id": "a0a4c5d1-62fd-b6f1-98ea-000000000008",									"name": "Print line"								}							}						]					}				],				"stats": {					"127.0.0.1": {						"changed": 0,						"failures": 0,						"ignored": 0,						"ok": 1,						"rescued": 0,						"skipped": 0,						"unreachable": 0					},					"192.168.1.1": {						"changed": 0,						"failures": 0,						"ignored": 0,						"ok": 1,						"rescued": 0,						"skipped": 0,						"unreachable": 0					}				}			}`,
 			err: nil,
 		},
+		{
+			desc: "Testing very long file in json output",
+			inputResult: fmt.Sprintf(`{
+				"custom_stats": {},
+				"global_custom_stats": {},
+				"plays": [
+					{
+						"play": {
+							"duration": {
+								"end": "2020-08-07T20:51:30.942955Z",
+								"start": "2020-08-07T20:51:30.607525Z"
+							},
+							"id": "a0a4c5d1-62fd-b6f1-98ea-000000000006",
+							"name": "%s"
+						},
+						"tasks": []
+					}
+				],
+				"stats": {
+					"127.0.0.1": {
+						"changed": 0,
+						"failures": 0,
+						"ignored": 0,
+						"ok": 1,
+						"rescued": 0,
+						"skipped": 0,
+						"unreachable": 0
+					}
+				}
+			}`, longMessageLine),
+			expectedResult: fmt.Sprintf(`{"custom_stats": {},"global_custom_stats": {},"plays": [{"play": {"duration": {"end": "2020-08-07T20:51:30.942955Z","start": "2020-08-07T20:51:30.607525Z"},"id": "a0a4c5d1-62fd-b6f1-98ea-000000000006","name": "%s"},"tasks": []}],"stats": {"127.0.0.1": {"changed": 0,"failures": 0,"ignored": 0,"ok": 1,"rescued": 0,"skipped": 0,"unreachable": 0}}}`, longMessageLine),
+			err:            nil,
+		},
 		{
 			desc: "Testing stdout callback json result skipping lines",
 			inputResult: `{
@@ -275,7 +308,19 @@ func TestStdoutCallbackJSONResults(t *testing.T) {
 			if err != nil && assert.Error(t, err) {
 				assert.Equal(t, test.err, err)
 			} else {
-				assert.Equal(t, test.expectedResult, wbuff.String(), "Unexpected value")
+				var expectedResult, actualResult interface{}
+				err := json.Unmarshal([]byte(test.expectedResult), &expectedResult)
+				if err != nil {
+					assert.Fail(t, "Failed to unmarshal json", err)
+					return
+				}
+
+				err = json.Unmarshal(wbuff.Bytes(), &actualResult)
+				if err != nil {
+					assert.Fail(t, "Failed to unmarshal json", err)
+					return
+				}
+				assert.Equal(t, expectedResult, actualResult, "Unexpected value")
 			}
 		})
 	}