Use gomock to test command executor listener

This also requires that `TaskID` is no longer a custom type, because that
would cause import cycles. The alternative would be to put the generated
mocks directly into the `worker` package, but I didn't think that was
particularly nice. Maybe this'll be reconsidered later.

internal/worker/command_executor.go

@@ -31,11 +31,15 @@ import (
 	"gitlab.com/blender/flamenco-ng-poc/pkg/api"
 )
 
+// Generate mock implementation of this interface.
+//go:generate go run github.com/golang/mock/mockgen -destination mocks/command_listener.gen.go -package mocks gitlab.com/blender/flamenco-ng-poc/internal/worker CommandListener
+
+// CommandListener sends the result of commands (log, output files) to the Manager.
 type CommandListener interface {
 	// LogProduced sends any logging to whatever service for storing logging.
-	LogProduced(taskID TaskID, logLines ...string) error
+	LogProduced(ctx context.Context, taskID string, logLines ...string) error
 	// OutputProduced tells the Manager there has been some output (most commonly a rendered frame or video).
-	OutputProduced(taskID TaskID, outputLocation string) error
+	OutputProduced(ctx context.Context, taskID string, outputLocation string) error
 }
 
 type CommandExecutor struct {
@@ -48,7 +52,7 @@ type CommandExecutor struct {
 
 var _ CommandRunner = (*CommandExecutor)(nil)
 
-type commandCallable func(ctx context.Context, logger zerolog.Logger, taskID TaskID, cmd api.Command) error
+type commandCallable func(ctx context.Context, logger zerolog.Logger, taskID string, cmd api.Command) error
 
 // TimeService is a service that operates on time.
 type TimeService interface {
@@ -72,7 +76,7 @@ func NewCommandExecutor(listener CommandListener, timeService TimeService) *Comm
 	return ce
 }
 
-func (ce *CommandExecutor) Run(ctx context.Context, taskID TaskID, cmd api.Command) error {
+func (ce *CommandExecutor) Run(ctx context.Context, taskID string, cmd api.Command) error {
 	logger := log.With().Str("task", string(taskID)).Str("command", cmd.Name).Logger()
 	logger.Info().Interface("settings", cmd.Settings).Msg("running command")
 
@@ -85,7 +89,7 @@ func (ce *CommandExecutor) Run(ctx context.Context, taskID TaskID, cmd api.Comma
 }
 
 // cmdEcho executes the "echo" command.
-func (ce *CommandExecutor) cmdEcho(ctx context.Context, logger zerolog.Logger, taskID TaskID, cmd api.Command) error {
+func (ce *CommandExecutor) cmdEcho(ctx context.Context, logger zerolog.Logger, taskID string, cmd api.Command) error {
 	message, ok := cmd.Settings["message"]
 	if !ok {
 		return fmt.Errorf("missing 'message' setting")
@@ -93,14 +97,14 @@ func (ce *CommandExecutor) cmdEcho(ctx context.Context, logger zerolog.Logger, t
 	messageStr := fmt.Sprintf("%v", message)
 
 	logger.Info().Str("message", messageStr).Msg("echo")
-	if err := ce.listener.LogProduced(taskID, fmt.Sprintf("echo: %q", messageStr)); err != nil {
+	if err := ce.listener.LogProduced(ctx, taskID, fmt.Sprintf("echo: %q", messageStr)); err != nil {
 		return err
 	}
 	return nil
 }
 
 // cmdSleep executes the "sleep" command.
-func (ce *CommandExecutor) cmdSleep(ctx context.Context, logger zerolog.Logger, taskID TaskID, cmd api.Command) error {
+func (ce *CommandExecutor) cmdSleep(ctx context.Context, logger zerolog.Logger, taskID string, cmd api.Command) error {
 
 	sleepTime, ok := cmd.Settings["duration_in_seconds"]
 	if !ok {
@@ -127,7 +131,7 @@ func (ce *CommandExecutor) cmdSleep(ctx context.Context, logger zerolog.Logger,
 		log.Debug().Msg("sleeping done")
 	}
 
-	if err := ce.listener.LogProduced(taskID, fmt.Sprintf("slept %v", duration)); err != nil {
+	if err := ce.listener.LogProduced(ctx, taskID, fmt.Sprintf("slept %v", duration)); err != nil {
 		return err
 	}
 

internal/worker/command_executor_test.go

@@ -25,78 +25,51 @@ import (
 	"testing"
 	"time"
 
-	"github.com/benbjohnson/clock"
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"gitlab.com/blender/flamenco-ng-poc/internal/worker/mocks"
 	"gitlab.com/blender/flamenco-ng-poc/pkg/api"
 )
 
-type mockCommandListener struct {
-	log    []loggedLines
-	output []producedOutput
-}
-type loggedLines struct {
-	taskID   TaskID
-	logLines []string
-}
-type producedOutput struct {
-	taskID         TaskID
-	outputLocation string
-}
-
-// LogProduced sends any logging to whatever service for storing logging.
-func (ml *mockCommandListener) LogProduced(taskID TaskID, logLines ...string) error {
-	ml.log = append(ml.log, loggedLines{taskID, logLines})
-	return nil
-}
-
-// OutputProduced tells the Manager there has been some output (most commonly a rendered frame or video).
-func (ml *mockCommandListener) OutputProduced(taskID TaskID, outputLocation string) error {
-	ml.output = append(ml.output, producedOutput{taskID, outputLocation})
-	return nil
-}
-
-func mockedClock(t *testing.T) *clock.Mock {
-	c := clock.NewMock()
-	now, err := time.Parse(time.RFC3339, "2006-01-02T15:04:05+07:00")
-	assert.NoError(t, err)
-	c.Set(now)
-	return c
-}
-
 func TestCommandEcho(t *testing.T) {
-	l := mockCommandListener{}
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	listener := mocks.NewMockCommandListener(mockCtrl)
 	clock := mockedClock(t)
-	ce := NewCommandExecutor(&l, clock)
+	ce := NewCommandExecutor(listener, clock)
 
 	ctx := context.Background()
 	message := "понављај за мном"
-	taskID := TaskID("90e9d656-e201-4ef0-b6b0-c80684fafa27")
+	taskID := "90e9d656-e201-4ef0-b6b0-c80684fafa27"
 	cmd := api.Command{
 		Name:     "echo",
 		Settings: map[string]interface{}{"message": message},
 	}
 
+	listener.EXPECT().LogProduced(gomock.Any(), taskID, "echo: \"понављај за мном\"")
+
 	err := ce.Run(ctx, taskID, cmd)
 	assert.NoError(t, err)
-
-	assert.Len(t, l.log, 1)
-	assert.Equal(t, taskID, l.log[0].taskID)
-	assert.Equal(t, "echo: \"понављај за мном\"", l.log[0].logLines[0])
-	assert.Len(t, l.output, 0)
 }
 
 func TestCommandSleep(t *testing.T) {
-	l := mockCommandListener{}
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	listener := mocks.NewMockCommandListener(mockCtrl)
 	clock := mockedClock(t)
-	ce := NewCommandExecutor(&l, clock)
+	ce := NewCommandExecutor(listener, clock)
 
 	ctx := context.Background()
-	taskID := TaskID("90e9d656-e201-4ef0-b6b0-c80684fafa27")
+	taskID := "90e9d656-e201-4ef0-b6b0-c80684fafa27"
 	cmd := api.Command{
 		Name:     "sleep",
 		Settings: map[string]interface{}{"duration_in_seconds": 47},
 	}
 
+	listener.EXPECT().LogProduced(gomock.Any(), taskID, "slept 47s")
+
 	timeBefore := clock.Now()
 
 	// Run the test in a goroutine, as we also need to actually increase the
@@ -124,9 +97,4 @@ loop:
 	timeAfter := clock.Now()
 	// Within the step size is precise enough. We're testing our implementation, not the precision of `time.After()`.
 	assert.WithinDuration(t, timeBefore.Add(47*time.Second), timeAfter, timeStepSize)
-
-	assert.Len(t, l.log, 1)
-	assert.Equal(t, taskID, l.log[0].taskID)
-	assert.Equal(t, "slept 47s", l.log[0].logLines[0])
-	assert.Len(t, l.output, 0)
 }

internal/worker/common_test.go

@@ -0,0 +1,37 @@
+package worker
+
+/* ***** BEGIN GPL LICENSE BLOCK *****
+ *
+ * Original Code Copyright (C) 2022 Blender Foundation.
+ *
+ * This file is part of Flamenco.
+ *
+ * Flamenco is free software: you can redistribute it and/or modify it under
+ * the terms of the GNU General Public License as published by the Free Software
+ * Foundation, either version 3 of the License, or (at your option) any later
+ * version.
+ *
+ * Flamenco is distributed in the hope that it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ * A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * Flamenco.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ * ***** END GPL LICENSE BLOCK ***** */
+
+import (
+	"testing"
+	"time"
+
+	"github.com/benbjohnson/clock"
+	"github.com/stretchr/testify/assert"
+)
+
+func mockedClock(t *testing.T) *clock.Mock {
+	c := clock.NewMock()
+	now, err := time.Parse(time.RFC3339, "2006-01-02T15:04:05+07:00")
+	assert.NoError(t, err)
+	c.Set(now)
+	return c
+}

internal/worker/listener.go

@@ -2,6 +2,9 @@ package worker
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"net/http"
 	"sync"
 	"time"
 
@@ -32,6 +35,10 @@ import (
 var _ CommandListener = (*Listener)(nil)
 var _ TaskExecutionListener = (*Listener)(nil)
 
+var (
+	ErrTaskReassigned = errors.New("task was reassigned to other worker")
+)
+
 // Listener listens to the result of task and command execution, and sends it to the Manager.
 type Listener struct {
 	doneWg *sync.WaitGroup
@@ -71,26 +78,45 @@ func (l *Listener) Wait() {
 }
 
 // TaskStarted tells the Manager that task execution has started.
-func (l *Listener) TaskStarted(taskID TaskID) error {
+func (l *Listener) TaskStarted(ctx context.Context, taskID string) error {
-	return nil
+	activity := "Started"
+	status := api.TaskStatusActive
+	update := api.TaskUpdateJSONRequestBody{
+		Activity:   &activity,
+		TaskStatus: &status,
+	}
+
+	resp, err := l.client.TaskUpdateWithResponse(ctx, string(taskID), update)
+	if err != nil {
+		return fmt.Errorf("error notifying Manager of task start: %w", err)
+	}
+
+	switch resp.StatusCode() {
+	case http.StatusNoContent:
+		return nil
+	case http.StatusConflict:
+		return ErrTaskReassigned
+	default:
+		return fmt.Errorf("unknown error from Manager: %v", resp.JSONDefault)
+	}
 }
 
 // TaskFailed tells the Manager the task failed for some reason.
-func (l *Listener) TaskFailed(taskID TaskID, reason string) error {
+func (l *Listener) TaskFailed(ctx context.Context, taskID string, reason string) error {
 	return nil
 }
 
 // TaskCompleted tells the Manager the task has been completed.
-func (l *Listener) TaskCompleted(taskID TaskID) error {
+func (l *Listener) TaskCompleted(ctx context.Context, taskID string) error {
 	return nil
 }
 
 // LogProduced sends any logging to whatever service for storing logging.
-func (l *Listener) LogProduced(taskID TaskID, logLines ...string) error {
+func (l *Listener) LogProduced(ctx context.Context, taskID string, logLines ...string) error {
 	return nil
 }
 
 // OutputProduced tells the Manager there has been some output (most commonly a rendered frame or video).
-func (l *Listener) OutputProduced(taskID TaskID, outputLocation string) error {
+func (l *Listener) OutputProduced(ctx context.Context, taskID string, outputLocation string) error {
 	return nil
 }

internal/worker/mocks/command_listener.gen.go

@@ -0,0 +1,68 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: gitlab.com/blender/flamenco-ng-poc/internal/worker (interfaces: CommandListener)
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	context "context"
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockCommandListener is a mock of CommandListener interface.
+type MockCommandListener struct {
+	ctrl     *gomock.Controller
+	recorder *MockCommandListenerMockRecorder
+}
+
+// MockCommandListenerMockRecorder is the mock recorder for MockCommandListener.
+type MockCommandListenerMockRecorder struct {
+	mock *MockCommandListener
+}
+
+// NewMockCommandListener creates a new mock instance.
+func NewMockCommandListener(ctrl *gomock.Controller) *MockCommandListener {
+	mock := &MockCommandListener{ctrl: ctrl}
+	mock.recorder = &MockCommandListenerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockCommandListener) EXPECT() *MockCommandListenerMockRecorder {
+	return m.recorder
+}
+
+// LogProduced mocks base method.
+func (m *MockCommandListener) LogProduced(arg0 context.Context, arg1 string, arg2 ...string) error {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "LogProduced", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// LogProduced indicates an expected call of LogProduced.
+func (mr *MockCommandListenerMockRecorder) LogProduced(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{arg0, arg1}, arg2...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LogProduced", reflect.TypeOf((*MockCommandListener)(nil).LogProduced), varargs...)
+}
+
+// OutputProduced mocks base method.
+func (m *MockCommandListener) OutputProduced(arg0 context.Context, arg1, arg2 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "OutputProduced", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// OutputProduced indicates an expected call of OutputProduced.
+func (mr *MockCommandListenerMockRecorder) OutputProduced(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OutputProduced", reflect.TypeOf((*MockCommandListener)(nil).OutputProduced), arg0, arg1, arg2)
+}

internal/worker/mocks/task_exe_listener.gen.go

@@ -0,0 +1,77 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: gitlab.com/blender/flamenco-ng-poc/internal/worker (interfaces: TaskExecutionListener)
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	context "context"
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockTaskExecutionListener is a mock of TaskExecutionListener interface.
+type MockTaskExecutionListener struct {
+	ctrl     *gomock.Controller
+	recorder *MockTaskExecutionListenerMockRecorder
+}
+
+// MockTaskExecutionListenerMockRecorder is the mock recorder for MockTaskExecutionListener.
+type MockTaskExecutionListenerMockRecorder struct {
+	mock *MockTaskExecutionListener
+}
+
+// NewMockTaskExecutionListener creates a new mock instance.
+func NewMockTaskExecutionListener(ctrl *gomock.Controller) *MockTaskExecutionListener {
+	mock := &MockTaskExecutionListener{ctrl: ctrl}
+	mock.recorder = &MockTaskExecutionListenerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTaskExecutionListener) EXPECT() *MockTaskExecutionListenerMockRecorder {
+	return m.recorder
+}
+
+// TaskCompleted mocks base method.
+func (m *MockTaskExecutionListener) TaskCompleted(arg0 context.Context, arg1 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TaskCompleted", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TaskCompleted indicates an expected call of TaskCompleted.
+func (mr *MockTaskExecutionListenerMockRecorder) TaskCompleted(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TaskCompleted", reflect.TypeOf((*MockTaskExecutionListener)(nil).TaskCompleted), arg0, arg1)
+}
+
+// TaskFailed mocks base method.
+func (m *MockTaskExecutionListener) TaskFailed(arg0 context.Context, arg1, arg2 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TaskFailed", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TaskFailed indicates an expected call of TaskFailed.
+func (mr *MockTaskExecutionListenerMockRecorder) TaskFailed(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TaskFailed", reflect.TypeOf((*MockTaskExecutionListener)(nil).TaskFailed), arg0, arg1, arg2)
+}
+
+// TaskStarted mocks base method.
+func (m *MockTaskExecutionListener) TaskStarted(arg0 context.Context, arg1 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TaskStarted", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// TaskStarted indicates an expected call of TaskStarted.
+func (mr *MockTaskExecutionListenerMockRecorder) TaskStarted(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TaskStarted", reflect.TypeOf((*MockTaskExecutionListener)(nil).TaskStarted), arg0, arg1)
+}

internal/worker/task_executor.go

@@ -29,23 +29,24 @@ import (
 )
 
 type CommandRunner interface {
-	Run(ctx context.Context, taskID TaskID, cmd api.Command) error
+	Run(ctx context.Context, taskID string, cmd api.Command) error
 }
 
+// Generate mock implementation of this interface.
+//go:generate go run github.com/golang/mock/mockgen -destination mocks/task_exe_listener.gen.go -package mocks gitlab.com/blender/flamenco-ng-poc/internal/worker TaskExecutionListener
+
+// TaskExecutionListener sends task lifecycle events (start/fail/complete) to the Manager.
 type TaskExecutionListener interface {
 	// TaskStarted tells the Manager that task execution has started.
-	TaskStarted(taskID TaskID) error
+	TaskStarted(ctx context.Context, taskID string) error
 
 	// TaskFailed tells the Manager the task failed for some reason.
-	TaskFailed(taskID TaskID, reason string) error
+	TaskFailed(ctx context.Context, taskID string, reason string) error
 
 	// TaskCompleted tells the Manager the task has been completed.
-	TaskCompleted(taskID TaskID) error
+	TaskCompleted(ctx context.Context, taskID string) error
 }
 
-// TODO: move me to a more appropriate place.
-type TaskID string
-
 type TaskExecutor struct {
 	cmdRunner CommandRunner
 	listener  TaskExecutionListener
@@ -64,9 +65,7 @@ func (te *TaskExecutor) Run(ctx context.Context, task api.AssignedTask) error {
 	logger := log.With().Str("task", task.Uuid).Logger()
 	logger.Info().Str("taskType", task.TaskType).Msg("starting task")
 
-	taskID := TaskID(task.Uuid)
+	if err := te.listener.TaskStarted(ctx, task.Uuid); err != nil {
-
-	if err := te.listener.TaskStarted(taskID); err != nil {
 		return fmt.Errorf("error sending notification to manager: %w", err)
 	}
 
@@ -80,17 +79,17 @@ func (te *TaskExecutor) Run(ctx context.Context, task api.AssignedTask) error {
 		default:
 		}
 
-		err := te.cmdRunner.Run(ctx, taskID, cmd)
+		err := te.cmdRunner.Run(ctx, task.Uuid, cmd)
 
 		if err != nil {
-			if err := te.listener.TaskFailed(taskID, err.Error()); err != nil {
+			if err := te.listener.TaskFailed(ctx, task.Uuid, err.Error()); err != nil {
 				return fmt.Errorf("error sending notification to manager: %w", err)
 			}
 			return err
 		}
 	}
 
-	if err := te.listener.TaskCompleted(taskID); err != nil {
+	if err := te.listener.TaskCompleted(ctx, task.Uuid); err != nil {
 		return fmt.Errorf("error sending notification to manager: %w", err)
 	}