worker: fix tests for new factory interface

cmd/worker/worker_test.go

@@ -24,7 +24,7 @@ func TestPersistentPreRun(t *testing.T) {
 	defer cleanup()
 
 	c, h := newCommandHooks()
-	h.Run = func(ctx context.Context, conf config.Config, log *logger.Logger, taskID string) error {
+	h.Run = func(ctx context.Context, conf config.Config, log *logger.Logger, opts *Options) error {
 		if conf.Server.HostName != host {
 			t.Fatal("unexpected Server.HostName in config", conf.Server.HostName)
 		}

tests/core/basic_test.go

@@ -1125,26 +1125,34 @@ func TestConcurrentStateUpdate(t *testing.T) {
 	f := tests.NewFunnel(c)
 	f.StartServer()
 
-	w, err := workerCmd.NewWorker(ctx, c, log)
-	if err != nil {
-		t.Fatal(err)
-	}
-	e := w.EventWriter
-
 	ids := []string{}
 	for i := 0; i < 10; i++ {
 		id := f.Run(`--sh 'echo hello'`)
 		ids = append(ids, id)
+
 		go func(id string) {
+			opts := &workerCmd.Options{TaskID: id}
+			w, err := workerCmd.NewWorker(ctx, c, log, opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+
 			log.Info("writing state initializing event", "taskID", id)
-			err := e.WriteEvent(ctx, events.NewState(id, tes.Initializing))
+			err := w.EventWriter.WriteEvent(ctx, events.NewState(id, tes.Initializing))
 			if err != nil {
 				log.Error("error writing event", err)
 			}
 		}(id)
+
 		go func(id string) {
+			opts := &workerCmd.Options{TaskID: id}
+			w, err := workerCmd.NewWorker(ctx, c, log, opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+
 			log.Info("writing state canceled event", "taskID", id)
-			err := e.WriteEvent(ctx, events.NewState(id, tes.Canceled))
+			err := w.EventWriter.WriteEvent(ctx, events.NewState(id, tes.Canceled))
 			if err != nil {
 				log.Error("error writing event", "error", err, "taskID", id)
 			}
@@ -1169,12 +1177,13 @@ func TestMetadataEvent(t *testing.T) {
 	f := tests.NewFunnel(c)
 	f.StartServer()
 
-	w, err := workerCmd.NewWorker(ctx, c, log)
+	id := f.Run(`--sh 'echo hello'`)
+
+	w, err := workerCmd.NewWorker(ctx, c, log, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal(err)
 	}
 	e := w.EventWriter
-	id := f.Run(`--sh 'echo hello'`)
 
 	err = e.WriteEvent(ctx, events.NewMetadata(id, 0, map[string]string{"one": "two"}))
 	if err != nil {

tests/core/worker_test.go

@@ -32,7 +32,7 @@ func TestWorkerRun(t *testing.T) {
   `)
 
 	ctx := context.Background()
-	err := workerCmd.Run(ctx, c, log, id)
+	err := workerCmd.Run(ctx, c, log, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}
@@ -70,7 +70,7 @@ func TestWorkDirCleanup(t *testing.T) {
 	workdir := path.Join(c.Worker.WorkDir, id)
 
 	ctx := context.Background()
-	err := workerCmd.Run(ctx, c, log, id)
+	err := workerCmd.Run(ctx, c, log, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}
@@ -99,7 +99,7 @@ func TestWorkDirCleanup(t *testing.T) {
 	c.Worker.LeaveWorkDir = true
 	workdir = path.Join(c.Worker.WorkDir, id)
 
-	err = workerCmd.Run(ctx, c, log, id)
+	err = workerCmd.Run(ctx, c, log, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}
@@ -180,7 +180,7 @@ func TestLargeLogRate(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
-	w.Run(ctx, task.Id)
+	w.Run(ctx)
 
 	// Given the difficulty of timing how long it task a task + docker container to start,
 	// we just check that a small amount of events were generated.
@@ -219,7 +219,7 @@ func TestZeroLogRate(t *testing.T) {
 		EventWriter: m,
 	}
 
-	w.Run(context.Background(), task.Id)
+	w.Run(context.Background())
 
 	time.Sleep(time.Second)
 
@@ -260,7 +260,7 @@ func TestZeroLogTailSize(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
-	w.Run(ctx, task.Id)
+	w.Run(ctx)
 
 	// we expect zero events to be generated
 	if counts.stdout != 0 {
@@ -295,7 +295,7 @@ func TestLogTailContent(t *testing.T) {
 		EventWriter: m,
 	}
 
-	err := w.Run(context.Background(), task.Id)
+	err := w.Run(context.Background())
 	if err != nil {
 		t.Error("unexpected worker.Run error", err)
 	}
@@ -339,7 +339,7 @@ func TestDockerContainerMetadata(t *testing.T) {
 		EventWriter: m,
 	}
 
-	err := w.Run(context.Background(), task.Id)
+	err := w.Run(context.Background())
 	if err != nil {
 		t.Error("unexpected worker.Run error", err)
 	}

tests/kafka/worker_test.go

@@ -61,7 +61,7 @@ func TestKafkaWorkerRun(t *testing.T) {
     --sh 'echo hello world'
   `)
 
-	err = workerCmd.Run(ctx, conf, log, id)
+	err = workerCmd.Run(ctx, conf, log, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}

tests/pubsub/pubsub_test.go

@@ -62,7 +62,7 @@ func TestPubSubWorkerRun(t *testing.T) {
 	// compute backend is in use
 	id := fun.Run(`'echo hello world'`)
 
-	err := workerCmd.Run(ctx, conf, nil, id)
+	err := workerCmd.Run(ctx, conf, nil, &workerCmd.Options{TaskID: id})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}

tests/scheduler/node_test.go

@@ -86,11 +86,16 @@ func TestManualBackend(t *testing.T) {
 	srv.Conf.Node.ID = "test-node-manual"
 	// create a node
 	srv.Conf.Node.ID = "test-node-manual"
-	w, err := workercmd.NewWorker(ctx, conf, log)
-	if err != nil {
-		t.Fatal("failed to create worker factory", err)
+
+	factory := func(ctx context.Context, taskID string) error {
+		w, err := workercmd.NewWorker(ctx, conf, log, &workercmd.Options{TaskID: taskID})
+		if err != nil {
+			return err
+		}
+		return w.Run(ctx)
 	}
-	n, err := scheduler.NewNodeProcess(ctx, srv.Conf, w.Run, log)
+
+	n, err := scheduler.NewNodeProcess(ctx, srv.Conf, factory, log)
 	if err != nil {
 		t.Fatal("failed to create node", err)
 	}

worker/base64_taskreader.go

@@ -10,10 +10,12 @@ import (
 	"github.com/ohsu-comp-bio/funnel/tes"
 )
 
+// Base64TaskReader reads a task from a base64 encoded string.
 type Base64TaskReader struct {
 	task *tes.Task
 }
 
+// NewBase64TaskReader creates a new Base64TaskReader.
 func NewBase64TaskReader(raw string) (*Base64TaskReader, error) {
 	data, err := base64.StdEncoding.DecodeString(raw)
 	if err != nil {

worker/file_taskreader.go

@@ -14,6 +14,7 @@ type FileTaskReader struct {
 	task *tes.Task
 }
 
+// NewFileTaskReader creates a new FileTaskReader.
 func NewFileTaskReader(path string) (*FileTaskReader, error) {
 	fh, err := os.Open(path)
 	if err != nil {