debian-forge-composer/internal/jobqueue/fsjobqueue/fsjobqueue_test.go

package fsjobqueue_test

import (
	"context"
	"encoding/json"
	"io/ioutil"
	"os"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/stretchr/testify/require"

	"github.com/osbuild/osbuild-composer/internal/jobqueue"
	"github.com/osbuild/osbuild-composer/internal/jobqueue/fsjobqueue"
)

type testResult struct {
}

func cleanupTempDir(t *testing.T, dir string) {
	err := os.RemoveAll(dir)
	require.NoError(t, err)
}

func newTemporaryQueue(t *testing.T, jobTypes []string) (jobqueue.JobQueue, string) {
	dir, err := ioutil.TempDir("", "jobqueue-test-")
	require.NoError(t, err)

	q, err := fsjobqueue.New(dir, jobTypes)
	require.NoError(t, err)
	require.NotNil(t, q)

	return q, dir
}

func pushTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, args interface{}, dependencies []uuid.UUID) uuid.UUID {
	t.Helper()
	id, err := q.Enqueue(jobType, args, dependencies)
	require.NoError(t, err)
	require.NotEmpty(t, id)
	return id
}

func finishNextTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, result interface{}) uuid.UUID {
	id, err := q.Dequeue(context.Background(), []string{jobType}, &json.RawMessage{})
	require.NoError(t, err)
	require.NotEmpty(t, id)

	err = q.FinishJob(id, result)
	require.NoError(t, err)

	return id
}

func TestNonExistant(t *testing.T) {
	q, err := fsjobqueue.New("/non-existant-directory", []string{})
	require.Error(t, err)
	require.Nil(t, q)
}

func TestErrors(t *testing.T) {
	q, dir := newTemporaryQueue(t, []string{"test"})
	defer cleanupTempDir(t, dir)

	// not serializable to JSON
	id, err := q.Enqueue("test", make(chan string), nil)
	require.Error(t, err)
	require.Equal(t, uuid.Nil, id)

	// invalid dependency
	id, err = q.Enqueue("test", "arg0", []uuid.UUID{uuid.New()})
	require.Error(t, err)
	require.Equal(t, uuid.Nil, id)
}

func TestArgs(t *testing.T) {
	type argument struct {
		I int
		S string
	}

	q, dir := newTemporaryQueue(t, []string{"fish", "octopus"})
	defer cleanupTempDir(t, dir)

	oneargs := argument{7, "🐠"}
	one := pushTestJob(t, q, "fish", oneargs, nil)

	twoargs := argument{42, "🐙"}
	two := pushTestJob(t, q, "octopus", twoargs, nil)

	var args argument
	id, err := q.Dequeue(context.Background(), []string{"octopus"}, &args)
	require.NoError(t, err)
	require.Equal(t, two, id)
	require.Equal(t, twoargs, args)

	id, err = q.Dequeue(context.Background(), []string{"fish"}, &args)
	require.NoError(t, err)
	require.Equal(t, one, id)
	require.Equal(t, oneargs, args)
}

func TestJobTypes(t *testing.T) {
	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
	defer cleanupTempDir(t, dir)

	one := pushTestJob(t, q, "octopus", nil, nil)
	two := pushTestJob(t, q, "clownfish", nil, nil)

	require.Equal(t, two, finishNextTestJob(t, q, "clownfish", testResult{}))
	require.Equal(t, one, finishNextTestJob(t, q, "octopus", testResult{}))

	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	id, err := q.Dequeue(ctx, []string{"zebra"}, nil)
	require.Equal(t, err, context.Canceled)
	require.Equal(t, uuid.Nil, id)
}

func TestDependencies(t *testing.T) {
	q, dir := newTemporaryQueue(t, []string{"test"})
	defer cleanupTempDir(t, dir)

	t.Run("done-before-pushing-dependant", func(t *testing.T) {
		one := pushTestJob(t, q, "test", nil, nil)
		two := pushTestJob(t, q, "test", nil, nil)

		r := []uuid.UUID{}
		r = append(r, finishNextTestJob(t, q, "test", testResult{}))
		r = append(r, finishNextTestJob(t, q, "test", testResult{}))
		require.ElementsMatch(t, []uuid.UUID{one, two}, r)

		j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
		queued, started, finished, canceled, err := q.JobStatus(j, nil)
		require.NoError(t, err)
		require.True(t, !queued.IsZero())
		require.True(t, started.IsZero())
		require.True(t, finished.IsZero())
		require.False(t, canceled)

		require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))

		queued, started, finished, canceled, err = q.JobStatus(j, &testResult{})
		require.NoError(t, err)
		require.True(t, !queued.IsZero())
		require.True(t, !started.IsZero())
		require.True(t, !finished.IsZero())
		require.False(t, canceled)
	})

	t.Run("done-after-pushing-dependant", func(t *testing.T) {
		one := pushTestJob(t, q, "test", nil, nil)
		two := pushTestJob(t, q, "test", nil, nil)

		j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
		queued, started, finished, canceled, err := q.JobStatus(j, nil)
		require.NoError(t, err)
		require.True(t, !queued.IsZero())
		require.True(t, started.IsZero())
		require.True(t, finished.IsZero())
		require.False(t, canceled)

		r := []uuid.UUID{}
		r = append(r, finishNextTestJob(t, q, "test", testResult{}))
		r = append(r, finishNextTestJob(t, q, "test", testResult{}))
		require.ElementsMatch(t, []uuid.UUID{one, two}, r)

		require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))

		queued, started, finished, canceled, err = q.JobStatus(j, &testResult{})
		require.NoError(t, err)
		require.True(t, !queued.IsZero())
		require.True(t, !started.IsZero())
		require.True(t, !finished.IsZero())
		require.False(t, canceled)
	})
}

// Test that a job queue allows parallel access to multiple workers, mainly to
// verify the quirky unlocking in Dequeue().
func TestMultipleWorkers(t *testing.T) {
	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
	defer cleanupTempDir(t, dir)

	done := make(chan struct{})
	go func() {
		defer close(done)
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		id, err := q.Dequeue(ctx, []string{"octopus"}, &json.RawMessage{})
		require.NoError(t, err)
		require.NotEmpty(t, id)
	}()

	// Increase the likelihood that the above goroutine was scheduled and
	// is waiting in Dequeue().
	time.Sleep(10 * time.Millisecond)

	// This call to Dequeue() should not block on the one in the goroutine.
	id := pushTestJob(t, q, "clownfish", nil, nil)
	r, err := q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
	require.NoError(t, err)
	require.Equal(t, id, r)

	// Now wake up the Dequeue() in the goroutine and wait for it to finish.
	_ = pushTestJob(t, q, "octopus", nil, nil)
	<-done
}

func TestCancel(t *testing.T) {
	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
	defer cleanupTempDir(t, dir)

	// Cancel a non-existing job
	err := q.CancelJob(uuid.New())
	require.Error(t, err)

	// Cancel a pending job
	id := pushTestJob(t, q, "clownfish", nil, nil)
	require.NotEmpty(t, id)
	err = q.CancelJob(id)
	require.NoError(t, err)
	_, _, _, canceled, err := q.JobStatus(id, &testResult{})
	require.NoError(t, err)
	require.True(t, canceled)
	err = q.FinishJob(id, &testResult{})
	require.Error(t, err)

	// Cancel a running job, which should not dequeue the canceled job from above
	id = pushTestJob(t, q, "clownfish", nil, nil)
	require.NotEmpty(t, id)
	r, err := q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
	require.NoError(t, err)
	require.Equal(t, id, r)
	err = q.CancelJob(id)
	require.NoError(t, err)
	_, _, _, canceled, err = q.JobStatus(id, &testResult{})
	require.NoError(t, err)
	require.True(t, canceled)
	err = q.FinishJob(id, &testResult{})
	require.Error(t, err)

	// Cancel a finished job, which is a no-op
	id = pushTestJob(t, q, "clownfish", nil, nil)
	require.NotEmpty(t, id)
	r, err = q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
	require.NoError(t, err)
	require.Equal(t, id, r)
	err = q.FinishJob(id, &testResult{})
	require.NoError(t, err)
	err = q.CancelJob(id)
	require.NoError(t, err)
	_, _, _, canceled, err = q.JobStatus(id, &testResult{})
	require.NoError(t, err)
	require.False(t, canceled)
}