GH-85: DataSource basic implementations

core/coretest/source.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2018 Yandex LLC. All rights reserved.
+// Use of this source code is governed by a MPL 2.0
+// license that can be found in the LICENSE file.
+// Author: Vladimir Skipor <[email protected]>
+
+package coretest
+
+import (
+	"io"
+	"io/ioutil"
+	"os"
+	"testing"
+
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/yandex/pandora/core"
+	"github.com/yandex/pandora/lib/testutil2"
+)
+
+func AssertSourceEqualStdStream(t *testing.T, expectedPtr **os.File, getSource func() core.DataSource) {
+	temp, err := ioutil.TempFile("", "")
+	require.NoError(t, err)
+
+	backup := *expectedPtr
+	defer func() {
+		*expectedPtr = backup
+	}()
+	*expectedPtr = temp
+	const testdata = "abcd"
+	_, err = io.WriteString(temp, testdata)
+	require.NoError(t, err)
+
+	rc, err := getSource().OpenSource()
+	require.NoError(t, err)
+
+	err = rc.Close()
+	require.NoError(t, err, "std stream should not be closed")
+
+	temp.Seek(0, io.SeekStart)
+	data, err := ioutil.ReadAll(temp)
+	assert.Equal(t, testdata, string(data))
+}
+
+func AssertSourceEqualFile(t *testing.T, fs afero.Fs, filename string, source core.DataSource) {
+	const testdata = "abcd"
+	afero.WriteFile(fs, filename, []byte(testdata), 644)
+
+	rc, err := source.OpenSource()
+	require.NoError(t, err)
+
+	data := testutil2.ReadString(t, rc)
+	err = rc.Close()
+	require.NoError(t, err)
+
+	assert.Equal(t, testdata, data)
+}

core/datasink/buffer.go → core/datasink/std.go

@@ -10,10 +10,12 @@ import (
 	"io"
 
 	"github.com/yandex/pandora/core"
+	"github.com/yandex/pandora/lib/ioutil2"
 )
 
 type Buffer struct {
 	bytes.Buffer
+	ioutil2.NopCloser
 }
 
 var _ core.DataSink = &Buffer{}
@@ -22,8 +24,6 @@ func (b *Buffer) OpenSink() (wc io.WriteCloser, err error) {
 	return b, nil
 }
 
-func (b *Buffer) Close() error { return nil }
-
 func NewBuffer() *Buffer {
 	return &Buffer{}
 }

core/datasource/file.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2018 Yandex LLC. All rights reserved.
+// Use of this source code is governed by a MPL 2.0
+// license that can be found in the LICENSE file.
+// Author: Vladimir Skipor <[email protected]>
+
+package datasource
+
+import (
+	"io"
+	"os"
+
+	"github.com/spf13/afero"
+
+	"github.com/yandex/pandora/core"
+)
+
+// TODO(skipor): auto unzip
+type FileConfig struct {
+	Path string `config:"path" validate:"required"`
+}
+
+func NewFile(fs afero.Fs, conf FileConfig) core.DataSource {
+	return &fileSource{afero.Afero{fs}, conf}
+}
+
+type fileSource struct {
+	fs   afero.Afero
+	conf FileConfig
+}
+
+func (s *fileSource) OpenSource() (wc io.ReadCloser, err error) {
+	return s.fs.Open(s.conf.Path)
+}
+
+func NewStdin() core.DataSource {
+	return hideCloseFileSource{os.Stdin}
+}
+
+type hideCloseFileSource struct{ afero.File }
+
+func (f hideCloseFileSource) OpenSource() (wc io.ReadCloser, err error) {
+	return f, nil
+}
+
+func (f hideCloseFileSource) Close() error { return nil }

core/datasource/file_test.go

@@ -0,0 +1,25 @@
+// Copyright (c) 2018 Yandex LLC. All rights reserved.
+// Use of this source code is governed by a MPL 2.0
+// license that can be found in the LICENSE file.
+// Author: Vladimir Skipor <[email protected]>
+
+package datasource
+
+import (
+	"os"
+	"testing"
+
+	"github.com/spf13/afero"
+	"github.com/yandex/pandora/core/coretest"
+)
+
+func TestFileSource(t *testing.T) {
+	const filename = "/xxx/yyy"
+	fs := afero.NewMemMapFs()
+	source := NewFile(fs, FileConfig{Path: filename})
+	coretest.AssertSourceEqualFile(t, fs, filename, source)
+}
+
+func TestStdin(t *testing.T) {
+	coretest.AssertSourceEqualStdStream(t, &os.Stdout, NewStdin)
+}

core/datasource/std.go

@@ -0,0 +1,46 @@
+// Copyright (c) 2018 Yandex LLC. All rights reserved.
+// Use of this source code is governed by a MPL 2.0
+// license that can be found in the LICENSE file.
+// Author: Vladimir Skipor <[email protected]>
+
+package datasource
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+
+	"github.com/yandex/pandora/core"
+	"github.com/yandex/pandora/lib/ioutil2"
+)
+
+func NewBuffer(buf *bytes.Buffer) core.DataSource {
+	return buffer{Buffer: buf}
+}
+
+type buffer struct {
+	*bytes.Buffer
+	ioutil2.NopCloser
+}
+
+func (b buffer) OpenSource() (wc io.ReadCloser, err error) {
+	return b, nil
+}
+
+// NewReader returns dummy core.DataSource that returns it on OpenSource call, wrapping it
+// ioutil.NopCloser if r is not io.Closer.
+// NONE(skipor): such wrapping hide
+func NewReader(r io.Reader) core.DataSource {
+	return &reader{r}
+}
+
+type reader struct {
+	source io.Reader
+}
+
+func (r *reader) OpenSource() (rc io.ReadCloser, err error) {
+	if rc, ok := r.source.(io.ReadCloser); ok {
+		return rc, nil
+	}
+	return ioutil.NopCloser(r.source), nil
+}

lib/ioutil2/closer.go

@@ -0,0 +1,11 @@
+// Copyright (c) 2018 Yandex LLC. All rights reserved.
+// Use of this source code is governed by a MPL 2.0
+// license that can be found in the LICENSE file.
+// Author: Vladimir Skipor <[email protected]>
+
+package ioutil2
+
+// NopCloser may be embedded to any struct to implement io.Closer doing nothing on closer.
+type NopCloser struct{}
+
+func (NopCloser) Close() error { return nil }

lib/testutil2/afero.go

@@ -6,19 +6,30 @@
 package testutil2
 
 import (
+	"io"
+	"io/ioutil"
+
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func ReadString(t TestingT, r io.Reader) string {
+	data, err := ioutil.ReadAll(r)
+	require.NoError(t, err)
+	return string(data)
+}
+
 func ReadFileString(t TestingT, fs afero.Fs, name string) string {
+	getHelper(t).Helper()
 	data, err := afero.ReadFile(fs, name)
 	require.NoError(t, err)
 	return string(data)
 
 }
 
 func AssertFileEqual(t TestingT, fs afero.Fs, name string, expected string) {
+	getHelper(t).Helper()
 	actual := ReadFileString(t, fs, name)
 	assert.Equal(t, expected, actual)
 }

lib/testutil2/matchers.go

@@ -71,7 +71,7 @@ func RunFlaky(t *testing.T, test func(t TestingT)) {
 }
 
 // getHelper allows to call t.Helper() without breaking compatibility with go version < 1.9
-func getHelper(t *testing.T) helper {
+func getHelper(t TestingT) helper {
 	var tInterface interface{} = t
 	if h, ok := tInterface.(helper); ok {
 		return h