test: add tests for csv reader and writer

Co-authored-by: Dmitry Fedotov <dmitry@uint32.ru>
Co-committed-by: Dmitry Fedotov <dmitry@uint32.ru>

.gitignore

@@ -1,2 +1,3 @@
 cmd
 coverage.out
+example

LICENSE

@@ -0,0 +1,22 @@
+Copyright 2025 Dmitry Fedotov
+
+Permission is hereby granted, free of charge, to any person
+obtaining a copy of this software and associated documentation 
+files (the “Software”), to deal in the Software without 
+restriction, including without limitation the rights to use, 
+copy, modify, merge, publish, distribute, sublicense, and/or 
+sell copies of the Software, and to permit persons to whom 
+the Software is furnished to do so, 
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall 
+be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, 
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 
+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
+OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,137 @@
+# package script
+
+This package is intended as a collection of helper functions and tools fow quick construction of scripts that process .csv tables.
+
+Note that this is a work in progress. API is not guaranteed to be stable.
+
+## Example usage
+
+This demonstrates a very basic case: reading from stdin, checking whether array len is 2 and switching array elements.
+
+```go
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os/signal"
+	"syscall"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+
+	reader := script.NewStdinReader()
+	writer := script.NewStdoutWriter()
+
+	processor := func(_ context.Context, in []string) ([]string, error) {
+		if len(in) != 2 {
+			return nil, errors.New("incorrect input len")
+		}
+
+		return []string{in[1], in[0]}, nil
+	}
+
+	conf := script.RunConfig{
+		Input:     reader,
+		Output:    writer,
+		Processor: processor,
+	}
+
+	if err := script.Run(ctx, conf); err != nil {
+		fmt.Println(err)
+	}
+}
+```
+
+A more conplicated example featuring a processor which is itself a struct with external dependency.
+
+```go
+package main
+
+import (
+	"context"
+	"fmt"
+	"os/signal"
+	"syscall"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+// ExternalDep represent an external dependency
+// which your script utilizes. This can be a DB connection etc.
+type ExternalDep interface {
+	QueryID(string) ([]string, error)
+}
+
+type Processor struct {
+	dep ExternalDep
+}
+
+func New(e ExternalDep) *Processor {
+
+	return &Processor{
+		dep: e,
+	}
+}
+
+func (p *Processor) Process(ctx context.Context, in []string) ([]string, error) {
+	var out []string
+
+	result, err := p.dep.QueryID(in[0])
+	if err != nil {
+		// you can as well record the error and continue
+		// processing without yielding the error here
+		return nil, err
+	}
+
+	out = append(out, result...)
+	out = append(out, "processed")
+
+	return out, nil
+}
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+
+	r, err := script.NewCSVReader("/tmp/some_file.csv")
+	if err != nil {
+		panic(err)
+	}
+
+	defer func() {
+		if err := r.Close(); err != nil {
+			fmt.Println("err closing reader", err)
+		}
+	}()
+
+	w, err := script.NewCSVWriter("/tmp/some_output.csv")
+	if err != nil {
+		panic(err)
+	}
+
+	defer func() {
+		if err := w.Close(); err != nil {
+			fmt.Println("err closing writer", err)
+		}
+	}()
+
+    // intializing the processor
+	p := New(ExternalDep(nil))
+
+	conf := script.RunConfig{
+		Input:     r,
+		Output:    w,
+		Processor: p.Process, // Process implements script.Processor
+	}
+
+	if err := script.Run(ctx, conf); err != nil {
+		fmt.Println(err)
+	}
+}
+```

input_csv.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"encoding/csv"
 	"os"
 )
@@ -16,7 +15,7 @@ type CSVReader struct {
 	rdr *csv.Reader
 }
 
-func (c *CSVReader) Read(context.Context) ([]string, error) {
+func (c *CSVReader) Read() ([]string, error) {
 	return c.rdr.Read()
 }
 

input_csv_test.go

@@ -0,0 +1,29 @@
+package script_test
+
+import (
+	"slices"
+	"testing"
+
+	"code.uint32.ru/dmitry/script"
+)
+
+func TestCSVReader(t *testing.T) {
+	t.Parallel()
+	r, err := script.NewCSVReader("testdata/sample_csv.csv")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := []string{"one", "two", "three"}
+
+	for range 2 {
+		row, err := r.Read()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !slices.Equal(row, want) {
+			t.Fatalf("rows not equal, want: %v, have: %v", want, row)
+		}
+	}
+}

input_mem.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"io"
 )
 
@@ -16,7 +15,7 @@ func NewMemReader(records [][]string) *MemReader {
 	}
 }
 
-func (m *MemReader) Read(context.Context) ([]string, error) {
+func (m *MemReader) Read() ([]string, error) {
 	if m.curr == len(m.rows) || len(m.rows) == 0 {
 		return nil, io.EOF
 	}

input_stdin.go

@@ -2,7 +2,6 @@ package script
 
 import (
 	"bufio"
-	"context"
 	"io"
 	"os"
 	"strings"
@@ -23,7 +22,7 @@ func NewStdinReader() *StdinReader {
 // then splits result at "," separator and returns
 // the resulting slice.
 // It returns EOF when nothing left to read.
-func (s *StdinReader) Read(_ context.Context) ([]string, error) {
+func (s *StdinReader) Read() ([]string, error) {
 	if !s.scanner.Scan() {
 		err := s.scanner.Err()
 		if err == nil {

output_csv.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"encoding/csv"
 	"errors"
 	"os"
@@ -20,7 +19,7 @@ type CSVWriter struct {
 }
 
 // Write writer row to csv. Writes are buffered
-func (c *CSVWriter) Write(_ context.Context, record []string) error {
+func (c *CSVWriter) Write(record []string) error {
 	err := c.wr.Write(record)
 	c.err = err
 

output_csv_test.go

@@ -0,0 +1,42 @@
+package script
+
+import (
+	"bytes"
+	"os"
+	"testing"
+)
+
+func TestCSVWriter(t *testing.T) {
+	path := "testdata/output_csv.csv"
+
+	os.Remove(path)
+	defer os.Remove(path)
+
+	w, err := newCSVwriter(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := []byte("one,two,three\none,two,three\n")
+
+	row := []string{"one", "two", "three"}
+
+	for range 2 {
+		if err := w.Write(row); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("err reading output file: %s", err)
+	}
+
+	if !bytes.Equal(want, b) {
+		t.Errorf("incorrect result, want: %s, have: %s", string(want), string(b))
+	}
+}

output_mem.go

@@ -1,7 +1,5 @@
 package script
 
-import "context"
-
 type MemWriter struct {
 	rows [][]string
 }
@@ -10,7 +8,7 @@ func NewMemWriter() *MemWriter {
 	return new(MemWriter)
 }
 
-func (m *MemWriter) Write(_ context.Context, record []string) error {
+func (m *MemWriter) Write(record []string) error {
 	m.rows = append(m.rows, record)
 
 	return nil

output_nop.go

@@ -1,13 +1,11 @@
 package script
 
-import "context"
-
 type NopWriter struct{}
 
 func NewNopWriter() *NopWriter {
 	return new(NopWriter)
 }
 
-func (d *NopWriter) Write(context.Context, []string) error {
+func (d *NopWriter) Write([]string) error {
 	return nil
 }

output_stdout.go

@@ -1,7 +1,6 @@
 package script
 
 import (
-	"context"
 	"os"
 	"strings"
 )
@@ -12,7 +11,7 @@ func NewStdoutWriter() *StdoutWriter {
 	return new(StdoutWriter)
 }
 
-func (s *StdoutWriter) Write(_ context.Context, in []string) error {
+func (s *StdoutWriter) Write(in []string) error {
 	_, err := os.Stdout.WriteString(strings.Join(in, ",") + "\n")
 	return err
 }

runner.go

@@ -28,18 +28,18 @@ type Reader interface {
 	// Read is called by Processor to obtain input
 	// for processing. Read must return EOF/io.EOF to indicate
 	// that there is no more input.
-	Read(context.Context) ([]string, error)
+	Read() ([]string, error)
 }
 
 type Writer interface {
 	// Write is called by Processor to record
 	// output.
-	Write(context.Context, []string) error
+	Write([]string) error
 }
 
-// Process accepts string slice, does what it should
+// Processor accepts string slice, does what it should
 // and returns output. Non-nil error returned
-// by Processor terminates stops further processing.
+// by Processor stops further processing.
 type Processor func(context.Context, []string) ([]string, error)
 
 type RunConfig struct {
@@ -51,13 +51,15 @@ type RunConfig struct {
 	Concurrency int
 }
 
-// Run starts concurrency threads (goroutines), reads from provided Reader,
+// Run starts the script described by r.
-// executes each Processor in the order they were provide and records result
+// First Read is called offset times with output of Read being discarded.
-// with provided Writer.
-// At first Read is called offset times with output of Read being discarded.
 // Then limit Reads are made and processor is called for each portion
-// of data. If limit is 0 then Runner keep processing input until it receives
+// of data. If limit is 0 then Run keep processing input until it receives
 // EOF from Reader.
+// Run fails on any error including Reader error, Writer error and Processor error.
+// If an error is encountered the writer operation will be attampted anyway so that
+// the output is left in consistent state, recording what has been actually done
+// by Processor.
 func Run(ctx context.Context, r RunConfig) error {
 	if r.Concurrency == 0 {
 		r.Concurrency = 1
@@ -74,7 +76,7 @@ func Run(ctx context.Context, r RunConfig) error {
 		defer close(rdch)
 
 		for range r.Offset {
-			_, err := r.Input.Read(ctx)
+			_, err := r.Input.Read()
 			if err != nil {
 				return fmt.Errorf("could not advance to required offset: %w", err)
 			}
@@ -82,7 +84,7 @@ func Run(ctx context.Context, r RunConfig) error {
 
 		count := 0
 		for {
-			inp, err := r.Input.Read(ctx)
+			inp, err := r.Input.Read()
 			if err != nil && errors.Is(err, EOF) {
 				return nil
 			} else if err != nil {
@@ -109,8 +111,10 @@ func Run(ctx context.Context, r RunConfig) error {
 	grp.Go(func() error {
 		// not paying attention to context here
 		// because we must complete writes
+		// this is run within group so that write
+		// error would cancel group context
 		for outp := range wrch {
-			if err := r.Output.Write(ctx, outp); err != nil {
+			if err := r.Output.Write(outp); err != nil {
 				return err
 			}
 		}
@@ -118,6 +122,7 @@ func Run(ctx context.Context, r RunConfig) error {
 		return nil
 	})
 
+	// run processing routines
 	grp.Go(func() error {
 		// will close write chan once
 		// all workers are done
@@ -136,12 +141,16 @@ func Run(ctx context.Context, r RunConfig) error {
 					}
 
 					wrch <- result
+
+					// if one of workers died or parent context expired
+					// we should die too
+					if err := innrctx.Err(); err != nil {
+						return nil
+					}
 				}
 
 				return nil
-
 			})
-
 		}
 
 		if err := workergrp.Wait(); err != nil {

runner_test.go

@@ -41,7 +41,7 @@ func TestBasicRun(t *testing.T) {
 
 type infiniteReader struct{}
 
-func (ir *infiniteReader) Read(_ context.Context) ([]string, error) {
+func (ir *infiniteReader) Read() ([]string, error) {
 	return []string{"infinity", "looks", "like", "this"}, nil
 }
 

testdata/sample_csv.csv

@@ -0,0 +1,2 @@
+one,two,three
+one,two,three