531a0184f7
Replace old used-to-be-GORM datastructures (#104305) with sqlc-generated structs. This also makes it possible to use more specific structs that are more taylored to the specific queries, increasing efficiency. This commit deals with the remaining areas, like the job deleter, task timeout checker, and task state machine. And anything else to get things running again. Functional changes are kept to a minimum, as the API still serves the same data. Because this work covers so much of Flamenco's code, it's been split up into different commits. Each commit brings Flamenco to a state where it compiles and unit tests pass. Only the result of the final commit has actually been tested properly. Ref: #104343
765 lines
24 KiB
Go
765 lines
24 KiB
Go
package api_impl
|
|
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"io"
|
|
"net/http"
|
|
"testing"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/labstack/echo/v4"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"projects.blender.org/studio/flamenco/internal/manager/config"
|
|
"projects.blender.org/studio/flamenco/internal/manager/last_rendered"
|
|
"projects.blender.org/studio/flamenco/internal/manager/persistence"
|
|
"projects.blender.org/studio/flamenco/pkg/api"
|
|
)
|
|
|
|
func TestTaskScheduleHappy(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
|
|
// Expect a call into the persistence layer, which should return a scheduled task.
|
|
job := persistence.Job{
|
|
ID: 1234,
|
|
UUID: "583a7d59-887a-4c6c-b3e4-a753018f71b0",
|
|
Priority: 47,
|
|
JobType: "simple-blender-test",
|
|
}
|
|
task := persistence.Task{
|
|
UUID: "4107c7aa-e86d-4244-858b-6c4fce2af503",
|
|
JobID: job.ID,
|
|
Priority: 327,
|
|
Commands: []byte(`[{"name": "test", "parameters": {"param": "prefix-{variable}-suffix"}}]`),
|
|
}
|
|
scheduledTask := persistence.ScheduledTask{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
JobPriority: job.Priority,
|
|
JobType: job.JobType,
|
|
}
|
|
|
|
ctx := echo.Request().Context()
|
|
bgCtx := gomock.Not(ctx)
|
|
mf.persistence.EXPECT().ScheduleTask(ctx, &worker).Return(&scheduledTask, nil)
|
|
mf.persistence.EXPECT().TaskTouchedByWorker(bgCtx, task.UUID)
|
|
mf.persistence.EXPECT().WorkerSeen(bgCtx, &worker)
|
|
mf.expectExpandVariables(t,
|
|
config.VariableAudienceWorkers,
|
|
config.VariablePlatform(worker.Platform),
|
|
map[string]string{"variable": "value"},
|
|
)
|
|
|
|
mf.logStorage.EXPECT().WriteTimestamped(bgCtx, job.UUID, task.UUID,
|
|
"Task assigned to worker дрон (e7632d62-c3b8-4af0-9e78-01752928952c)")
|
|
|
|
mf.stateMachine.EXPECT().TaskStatusChange(bgCtx, &task, api.TaskStatusActive)
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(gomock.Any())
|
|
|
|
err := mf.flamenco.ScheduleTask(echo)
|
|
require.NoError(t, err)
|
|
|
|
// Check the response
|
|
assignedTask := api.AssignedTask{
|
|
Uuid: task.UUID,
|
|
Job: job.UUID,
|
|
JobType: "simple-blender-test",
|
|
JobPriority: 47,
|
|
Priority: 327,
|
|
Commands: []api.Command{
|
|
{Name: "test", Parameters: map[string]interface{}{
|
|
"param": "prefix-value-suffix",
|
|
}},
|
|
},
|
|
}
|
|
assertResponseJSON(t, echo, http.StatusOK, assignedTask)
|
|
resp := getRecordedResponse(echo)
|
|
assert.Equal(t, http.StatusOK, resp.StatusCode)
|
|
}
|
|
|
|
func TestTaskScheduleNoTaskAvailable(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
|
|
// Expect a call into the persistence layer, which should return nil.
|
|
ctx := echo.Request().Context()
|
|
bgCtx := gomock.Not(ctx)
|
|
mf.persistence.EXPECT().ScheduleTask(ctx, &worker).Return(nil, nil)
|
|
|
|
// This call should still trigger a "worker seen" call, as the worker is
|
|
// actively asking for tasks.
|
|
mf.persistence.EXPECT().WorkerSeen(bgCtx, &worker)
|
|
|
|
err := mf.flamenco.ScheduleTask(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
|
|
func TestTaskScheduleNonActiveStatus(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusAsleep
|
|
|
|
echoCtx := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echoCtx, &worker)
|
|
|
|
// The worker should be marked as 'seen', even when it's in a state that
|
|
// doesn't allow task execution.
|
|
bgCtx := gomock.Not(echoCtx.Request().Context())
|
|
mf.persistence.EXPECT().WorkerSeen(bgCtx, &worker)
|
|
|
|
err := mf.flamenco.ScheduleTask(echoCtx)
|
|
require.NoError(t, err)
|
|
|
|
resp := getRecordedResponse(echoCtx)
|
|
assert.Equal(t, http.StatusConflict, resp.StatusCode)
|
|
}
|
|
|
|
func TestTaskScheduleOtherStatusRequested(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
worker.StatusChangeRequest(api.WorkerStatusAsleep, false)
|
|
|
|
echoCtx := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echoCtx, &worker)
|
|
|
|
// The worker should be marked as 'seen', even when it's in a state that
|
|
// doesn't allow task execution.
|
|
bgCtx := gomock.Not(echoCtx.Request().Context())
|
|
mf.persistence.EXPECT().WorkerSeen(bgCtx, &worker)
|
|
|
|
err := mf.flamenco.ScheduleTask(echoCtx)
|
|
require.NoError(t, err)
|
|
|
|
expectBody := api.WorkerStateChange{StatusRequested: api.WorkerStatusAsleep}
|
|
assertResponseJSON(t, echoCtx, http.StatusLocked, expectBody)
|
|
}
|
|
|
|
func TestTaskScheduleOtherStatusRequestedAndBadState(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
// Even when the worker is in a state that doesn't allow execution, if there
|
|
// is a status change requested, this should be communicated to the worker.
|
|
worker.Status = api.WorkerStatusError
|
|
worker.StatusChangeRequest(api.WorkerStatusAwake, false)
|
|
|
|
echoCtx := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echoCtx, &worker)
|
|
|
|
// The worker should be marked as 'seen', even when it's in a state that
|
|
// doesn't allow task execution.
|
|
bgCtx := gomock.Not(echoCtx.Request().Context())
|
|
mf.persistence.EXPECT().WorkerSeen(bgCtx, &worker)
|
|
|
|
err := mf.flamenco.ScheduleTask(echoCtx)
|
|
require.NoError(t, err)
|
|
|
|
expectBody := api.WorkerStateChange{StatusRequested: api.WorkerStatusAwake}
|
|
assertResponseJSON(t, echoCtx, http.StatusLocked, expectBody)
|
|
}
|
|
|
|
func TestWorkerSignOn(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusOffline
|
|
prevStatus := worker.Status
|
|
|
|
mf.sleepScheduler.EXPECT().WorkerStatus(gomock.Any(), worker.UUID).
|
|
Return(api.WorkerStatusAsleep, nil)
|
|
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: "Lazy Boi",
|
|
PreviousStatus: &prevStatus,
|
|
Status: api.WorkerStatusStarting,
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: "3.0-testing",
|
|
})
|
|
|
|
mf.persistence.EXPECT().SaveWorker(gomock.Any(), &worker).Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
echo := mf.prepareMockedJSONRequest(api.WorkerSignOn{
|
|
Name: "Lazy Boi",
|
|
SoftwareVersion: "3.0-testing",
|
|
SupportedTaskTypes: []string{"testing", "sleeping", "snoozing"},
|
|
})
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.SignOn(echo)
|
|
require.NoError(t, err)
|
|
|
|
assertResponseJSON(t, echo, http.StatusOK, api.WorkerStateChange{
|
|
StatusRequested: api.WorkerStatusAsleep,
|
|
})
|
|
}
|
|
|
|
func TestWorkerSignoffTaskRequeue(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
// Signing off should be handled completely, even when the HTTP connection
|
|
// breaks. This means using a different context than the one passed by Echo.
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
expectCtx := gomock.Not(gomock.Eq(echo.Request().Context()))
|
|
|
|
// Expect worker's tasks to be re-queued.
|
|
mf.stateMachine.EXPECT().RequeueActiveTasksOfWorker(expectCtx, &worker, "worker signed off").Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(expectCtx, &worker)
|
|
|
|
// Expect worker to be saved as 'offline'.
|
|
mf.persistence.EXPECT().
|
|
SaveWorkerStatus(expectCtx, &worker).
|
|
Do(func(ctx context.Context, w *persistence.Worker) error {
|
|
assert.Equal(t, api.WorkerStatusOffline, w.Status)
|
|
return nil
|
|
})
|
|
|
|
prevStatus := api.WorkerStatusAwake
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: &prevStatus,
|
|
Status: api.WorkerStatusOffline,
|
|
StatusChange: &api.WorkerStatusChangeRequest{
|
|
IsLazy: false,
|
|
Status: api.WorkerStatusAwake,
|
|
},
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
})
|
|
|
|
err := mf.flamenco.SignOff(echo)
|
|
require.NoError(t, err)
|
|
|
|
resp := getRecordedResponse(echo)
|
|
assert.Equal(t, http.StatusNoContent, resp.StatusCode)
|
|
}
|
|
|
|
func TestWorkerRememberPreviousStatus(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusAwake
|
|
worker.StatusChangeRequest(api.WorkerStatusOffline, true)
|
|
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: ptr(api.WorkerStatusAwake),
|
|
Status: api.WorkerStatusOffline,
|
|
StatusChange: &api.WorkerStatusChangeRequest{
|
|
IsLazy: false,
|
|
Status: api.WorkerStatusAwake,
|
|
},
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
})
|
|
|
|
savedWorker := worker
|
|
savedWorker.Status = api.WorkerStatusOffline
|
|
savedWorker.StatusRequested = api.WorkerStatusAwake
|
|
savedWorker.LazyStatusRequest = false
|
|
mf.persistence.EXPECT().SaveWorkerStatus(gomock.Any(), &savedWorker).Return(nil)
|
|
mf.stateMachine.EXPECT().RequeueActiveTasksOfWorker(gomock.Any(), &worker, "worker signed off").Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.SignOff(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
|
|
assert.Equal(t, api.WorkerStatusAwake, worker.StatusRequested)
|
|
assert.Equal(t, false, worker.LazyStatusRequest)
|
|
}
|
|
|
|
// Test that certain statuses (like 'starting') are not remembered.
|
|
func TestWorkerDontRememberPreviousStatus(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusError
|
|
worker.StatusChangeRequest(api.WorkerStatusOffline, true)
|
|
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: ptr(api.WorkerStatusError),
|
|
Status: api.WorkerStatusOffline,
|
|
StatusChange: nil,
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
})
|
|
|
|
savedWorker := worker
|
|
savedWorker.Status = api.WorkerStatusOffline
|
|
savedWorker.StatusChangeClear() // No status change should be queued.
|
|
mf.persistence.EXPECT().SaveWorkerStatus(gomock.Any(), &savedWorker).Return(nil)
|
|
mf.stateMachine.EXPECT().RequeueActiveTasksOfWorker(gomock.Any(), &worker, "worker signed off").Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.SignOff(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
|
|
func TestWorkerState(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker).Times(2)
|
|
|
|
// No state change requested.
|
|
{
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.WorkerState(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
|
|
// State change requested.
|
|
{
|
|
requestStatus := api.WorkerStatusAsleep
|
|
worker.StatusChangeRequest(requestStatus, false)
|
|
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
|
|
err := mf.flamenco.WorkerState(echo)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.WorkerStateChange{
|
|
StatusRequested: requestStatus,
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestWorkerStateChanged(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusAwake
|
|
prevStatus := worker.Status
|
|
|
|
// Expect a broadcast of the change
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: &prevStatus,
|
|
Status: api.WorkerStatusAsleep,
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
})
|
|
|
|
// Expect the Worker to be saved with the new status
|
|
savedWorker := worker
|
|
savedWorker.Status = api.WorkerStatusAsleep
|
|
mf.persistence.EXPECT().SaveWorkerStatus(gomock.Any(), &savedWorker).Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
mf.stateMachine.EXPECT().RequeueActiveTasksOfWorker(gomock.Any(), &worker,
|
|
"worker дрон (e7632d62-c3b8-4af0-9e78-01752928952c) changed status to 'asleep'")
|
|
|
|
// Perform the request
|
|
echo := mf.prepareMockedJSONRequest(api.WorkerStateChanged{
|
|
Status: api.WorkerStatusAsleep,
|
|
})
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.WorkerStateChanged(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
|
|
func TestWorkerStateChangedAfterChangeRequest(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
worker.Status = api.WorkerStatusOffline
|
|
worker.StatusChangeRequest(api.WorkerStatusAsleep, false)
|
|
|
|
{
|
|
// Expect a broadcast of the change, even though it's not the state that was requested.
|
|
// This is to allow some flexibility, for example when a worker has to go
|
|
// asleep but would do so via `offline → starting → asleep`.
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: ptr(api.WorkerStatusOffline),
|
|
Status: api.WorkerStatusStarting,
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
StatusChange: &api.WorkerStatusChangeRequest{
|
|
Status: api.WorkerStatusAsleep,
|
|
IsLazy: false,
|
|
},
|
|
})
|
|
|
|
// Expect the Worker to be saved with the new status, but with the status
|
|
// request still in place as it hasn't been met yet.
|
|
savedWorker := worker
|
|
savedWorker.Status = api.WorkerStatusStarting
|
|
mf.persistence.EXPECT().SaveWorkerStatus(gomock.Any(), &savedWorker).Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
// Perform the request
|
|
echo := mf.prepareMockedJSONRequest(api.WorkerStateChanged{
|
|
Status: api.WorkerStatusStarting,
|
|
})
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.WorkerStateChanged(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
|
|
// Do another status change, which does meet the requested state.
|
|
{
|
|
// Expect a broadcast.
|
|
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
|
|
Id: worker.UUID,
|
|
Name: worker.Name,
|
|
PreviousStatus: ptr(api.WorkerStatusStarting),
|
|
Status: api.WorkerStatusAsleep,
|
|
Updated: worker.UpdatedAt.Time,
|
|
Version: worker.Software,
|
|
})
|
|
|
|
// Expect the Worker to be saved with the new status and the status request
|
|
// erased.
|
|
savedWorker := worker
|
|
savedWorker.Status = api.WorkerStatusAsleep
|
|
savedWorker.StatusChangeClear()
|
|
mf.persistence.EXPECT().SaveWorkerStatus(gomock.Any(), &savedWorker).Return(nil)
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
// Perform the request
|
|
echo := mf.prepareMockedJSONRequest(api.WorkerStateChanged{
|
|
Status: api.WorkerStatusAsleep,
|
|
})
|
|
requestWorkerStore(echo, &worker)
|
|
err := mf.flamenco.WorkerStateChanged(echo)
|
|
require.NoError(t, err)
|
|
assertResponseNoContent(t, echo)
|
|
}
|
|
}
|
|
|
|
func TestMayWorkerRun(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
prepareRequest := func() echo.Context {
|
|
echo := mf.prepareMockedRequest(nil)
|
|
requestWorkerStore(echo, &worker)
|
|
return echo
|
|
}
|
|
|
|
job := persistence.Job{
|
|
ID: 1234,
|
|
UUID: "583a7d59-887a-4c6c-b3e4-a753018f71b0",
|
|
}
|
|
|
|
task := persistence.Task{
|
|
UUID: "4107c7aa-e86d-4244-858b-6c4fce2af503",
|
|
JobID: job.ID,
|
|
Status: api.TaskStatusActive,
|
|
}
|
|
|
|
// Expect the worker to be marked as 'seen' regardless of whether it may run
|
|
// its current task or not, so equal to the number of calls to
|
|
// `MayWorkerRun()` below.
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker).Times(5)
|
|
|
|
// Test: unhappy, task unassigned
|
|
{
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: "",
|
|
}
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
echo := prepareRequest()
|
|
err := mf.flamenco.MayWorkerRun(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.MayKeepRunning{
|
|
MayKeepRunning: false,
|
|
Reason: "task not assigned to this worker",
|
|
})
|
|
}
|
|
|
|
// Test: happy, task assigned to this worker.
|
|
{
|
|
task.WorkerID = sql.NullInt64{Int64: worker.ID, Valid: true}
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
}
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
// Expect a 'touch' of the task.
|
|
mf.persistence.EXPECT().TaskTouchedByWorker(gomock.Any(), task.UUID).Return(nil)
|
|
|
|
echo := prepareRequest()
|
|
err := mf.flamenco.MayWorkerRun(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.MayKeepRunning{
|
|
MayKeepRunning: true,
|
|
})
|
|
}
|
|
|
|
// Test: unhappy, assigned but cancelled.
|
|
{
|
|
task.WorkerID = sql.NullInt64{Int64: worker.ID, Valid: true}
|
|
task.Status = api.TaskStatusCanceled
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
}
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
echo := prepareRequest()
|
|
err := mf.flamenco.MayWorkerRun(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.MayKeepRunning{
|
|
MayKeepRunning: false,
|
|
Reason: "task is in non-runnable status \"canceled\"",
|
|
})
|
|
}
|
|
|
|
// Test: unhappy, assigned and runnable but worker should go to bed.
|
|
{
|
|
worker.StatusChangeRequest(api.WorkerStatusAsleep, false)
|
|
task.WorkerID = sql.NullInt64{Int64: worker.ID, Valid: true}
|
|
task.Status = api.TaskStatusActive
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
}
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
echo := prepareRequest()
|
|
err := mf.flamenco.MayWorkerRun(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.MayKeepRunning{
|
|
MayKeepRunning: false,
|
|
Reason: "worker status change requested",
|
|
StatusChangeRequested: true,
|
|
})
|
|
}
|
|
|
|
// Test: happy, assigned and runnable; worker should go to bed after task is finished.
|
|
{
|
|
worker.StatusChangeRequest(api.WorkerStatusAsleep, true)
|
|
task.WorkerID = sql.NullInt64{Int64: worker.ID, Valid: true}
|
|
task.Status = api.TaskStatusActive
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
}
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
// Expect a 'touch' of the task.
|
|
mf.persistence.EXPECT().TaskTouchedByWorker(gomock.Any(), task.UUID).Return(nil)
|
|
|
|
echo := prepareRequest()
|
|
err := mf.flamenco.MayWorkerRun(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseJSON(t, echo, http.StatusOK, api.MayKeepRunning{
|
|
MayKeepRunning: true,
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTaskOutputProduced(t *testing.T) {
|
|
mockCtrl := gomock.NewController(t)
|
|
defer mockCtrl.Finish()
|
|
|
|
mf := newMockedFlamenco(mockCtrl)
|
|
worker := testWorker()
|
|
|
|
prepareRequest := func(body io.Reader) echo.Context {
|
|
echo := mf.prepareMockedRequest(body)
|
|
requestWorkerStore(echo, &worker)
|
|
return echo
|
|
}
|
|
|
|
job := persistence.Job{
|
|
ID: 1234,
|
|
UUID: "583a7d59-887a-4c6c-b3e4-a753018f71b0",
|
|
}
|
|
task := persistence.Task{
|
|
UUID: "4107c7aa-e86d-4244-858b-6c4fce2af503",
|
|
JobID: job.ID,
|
|
Status: api.TaskStatusActive,
|
|
}
|
|
taskJobWorker := persistence.TaskJobWorker{
|
|
Task: task,
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
}
|
|
|
|
// Mock body to use in the request.
|
|
bodyBytes := []byte("JPEG file contents")
|
|
|
|
mf.lastRender.EXPECT().ThumbSpecs().Return([]last_rendered.Thumbspec{
|
|
{Filename: "big'un.jpg", MaxWidth: 1920, MaxHeight: 1080},
|
|
{Filename: "tiny.jpg", MaxWidth: 320, MaxHeight: 240},
|
|
}).AnyTimes()
|
|
mf.lastRender.EXPECT().PathForJob(job.UUID).Return("/path/to/job").AnyTimes()
|
|
mf.localStorage.EXPECT().RelPath("/path/to/job").Return("relative/path/to/job", nil).AnyTimes()
|
|
|
|
// Test: unhappy, missing Content-Length header.
|
|
{
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
echo := prepareRequest(nil)
|
|
err := mf.flamenco.TaskOutputProduced(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseAPIError(t, echo, http.StatusLengthRequired, "Content-Length header required")
|
|
}
|
|
|
|
// Test: unhappy, too large Content-Length header.
|
|
{
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
|
|
bodyBytes := bytes.Repeat([]byte("x"), int(last_rendered.MaxImageSizeBytes+1))
|
|
if int64(len(bodyBytes)) != last_rendered.MaxImageSizeBytes+1 {
|
|
panic("cannot generate enough bytes")
|
|
}
|
|
|
|
echo := prepareRequest(bytes.NewReader(bodyBytes))
|
|
err := mf.flamenco.TaskOutputProduced(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseAPIError(t, echo, http.StatusRequestEntityTooLarge,
|
|
"image too large; should be max %v bytes", last_rendered.MaxImageSizeBytes)
|
|
}
|
|
|
|
// Test: unhappy, wrong mime type
|
|
{
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
echo := prepareRequest(bytes.NewReader(bodyBytes))
|
|
echo.Request().Header.Set("Content-Type", "image/openexr")
|
|
mf.lastRender.EXPECT().QueueImage(gomock.Any()).Return(last_rendered.ErrMimeTypeUnsupported)
|
|
|
|
err := mf.flamenco.TaskOutputProduced(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseAPIError(t, echo, http.StatusUnsupportedMediaType, `unsupported mime type "image/openexr"`)
|
|
}
|
|
|
|
// Test: unhappy, queue full
|
|
{
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
|
|
echo := prepareRequest(bytes.NewReader(bodyBytes))
|
|
mf.lastRender.EXPECT().QueueImage(gomock.Any()).Return(last_rendered.ErrQueueFull)
|
|
|
|
err := mf.flamenco.TaskOutputProduced(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseAPIError(t, echo, http.StatusTooManyRequests, "image processing queue is full")
|
|
}
|
|
|
|
// Test: happy
|
|
{
|
|
mf.persistence.EXPECT().WorkerSeen(gomock.Any(), &worker)
|
|
mf.persistence.EXPECT().FetchTask(gomock.Any(), task.UUID).Return(taskJobWorker, nil)
|
|
// Don't expect persistence.SetLastRendered(...) quite yet. That should be
|
|
// called after the image processing is done.
|
|
|
|
echo := prepareRequest(bytes.NewReader(bodyBytes))
|
|
echo.Request().Header.Set("Content-Type", "image/jpeg")
|
|
expectPayload := last_rendered.Payload{
|
|
JobUUID: job.UUID,
|
|
WorkerUUID: worker.UUID,
|
|
MimeType: "image/jpeg",
|
|
Image: bodyBytes,
|
|
}
|
|
var actualPayload *last_rendered.Payload
|
|
mf.lastRender.EXPECT().QueueImage(gomock.Any()).DoAndReturn(func(payload last_rendered.Payload) error {
|
|
actualPayload = &payload
|
|
return nil
|
|
})
|
|
|
|
err := mf.flamenco.TaskOutputProduced(echo, task.UUID)
|
|
require.NoError(t, err)
|
|
assertResponseNoBody(t, echo, http.StatusAccepted)
|
|
|
|
if assert.NotNil(t, actualPayload) {
|
|
ctx := context.Background()
|
|
mf.persistence.EXPECT().SetLastRendered(ctx, job.UUID)
|
|
|
|
expectBroadcast := api.EventLastRenderedUpdate{
|
|
JobId: job.UUID,
|
|
Thumbnail: api.JobLastRenderedImageInfo{
|
|
Base: "/job-files/relative/path/to/job",
|
|
Suffixes: []string{"big'un.jpg", "tiny.jpg"},
|
|
},
|
|
}
|
|
mf.broadcaster.EXPECT().BroadcastLastRenderedImage(expectBroadcast)
|
|
|
|
// Calling the callback function is normally done by the last-rendered image processor.
|
|
actualPayload.Callback(ctx)
|
|
|
|
// Compare the parameter to `QueueImage()` in a way that ignores the callback function.
|
|
actualPayload.Callback = nil
|
|
assert.Equal(t, expectPayload, *actualPayload)
|
|
}
|
|
}
|
|
|
|
}
|