From 4f184a546fa5116c1910d210174fa4d7587226c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sybren=20A=2E=20St=C3=BCvel?= Date: Thu, 17 Feb 2022 10:47:52 +0100 Subject: [PATCH] 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 | 20 +++-- internal/worker/command_executor_test.go | 68 +++++----------- internal/worker/common_test.go | 37 +++++++++ internal/worker/listener.go | 38 +++++++-- internal/worker/mocks/command_listener.gen.go | 68 ++++++++++++++++ .../worker/mocks/task_exe_listener.gen.go | 77 +++++++++++++++++++ internal/worker/task_executor.go | 25 +++--- 7 files changed, 256 insertions(+), 77 deletions(-) create mode 100644 internal/worker/common_test.go create mode 100644 internal/worker/mocks/command_listener.gen.go create mode 100644 internal/worker/mocks/task_exe_listener.gen.go diff --git a/internal/worker/command_executor.go b/internal/worker/command_executor.go index 02fe85bc..f1750ea3 100644 --- a/internal/worker/command_executor.go +++ b/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 } diff --git a/internal/worker/command_executor_test.go b/internal/worker/command_executor_test.go index 2d154524..63035d21 100644 --- a/internal/worker/command_executor_test.go +++ b/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) } diff --git a/internal/worker/common_test.go b/internal/worker/common_test.go new file mode 100644 index 00000000..cf16d1e0 --- /dev/null +++ b/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 . + * + * ***** 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 +} diff --git a/internal/worker/listener.go b/internal/worker/listener.go index b6abddb0..700e6087 100644 --- a/internal/worker/listener.go +++ b/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 { - return nil +func (l *Listener) TaskStarted(ctx context.Context, taskID string) error { + 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 } diff --git a/internal/worker/mocks/command_listener.gen.go b/internal/worker/mocks/command_listener.gen.go new file mode 100644 index 00000000..a1bc44d9 --- /dev/null +++ b/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) +} diff --git a/internal/worker/mocks/task_exe_listener.gen.go b/internal/worker/mocks/task_exe_listener.gen.go new file mode 100644 index 00000000..aad9c53c --- /dev/null +++ b/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) +} diff --git a/internal/worker/task_executor.go b/internal/worker/task_executor.go index c91c8777..7ce12732 100644 --- a/internal/worker/task_executor.go +++ b/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(taskID); err != nil { + if err := te.listener.TaskStarted(ctx, task.Uuid); 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) }