package uploads_test import ( "context" "errors" "strconv" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "git.d-ma.be/mathias/tapir/internal/uploads" ) // The queue was append-only: 100,000 submits with the worker idle all succeeded, and each // queued job holds up to the upload cap on disk until its turn (adversarial review F6). func queueWorker(t *testing.T, total, perUser int) *uploads.Worker { t.Helper() r := newWRig(t, time.Minute) return uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute, uploads.WithQueueLimits(total, perUser)) } func submit(w *uploads.Worker, user string, n int) error { return w.Submit(uploads.Job{UserID: user, UploadID: user + "-" + strconv.Itoa(n), Path: "/nonexistent"}) } func TestWorkerSubmit_RefusesBeyondTheTotalQueueCap(t *testing.T) { w := queueWorker(t, 3, 10) for i := 0; i < 3; i++ { require.NoError(t, submit(w, "u"+strconv.Itoa(i), i)) } err := submit(w, "u9", 9) assert.ErrorIs(t, err, uploads.ErrQueueFull) } func TestWorkerSubmit_OneUserCannotFillTheQueueForEveryoneElse(t *testing.T) { w := queueWorker(t, 100, 2) require.NoError(t, submit(w, "alice", 1)) require.NoError(t, submit(w, "alice", 2)) errAlice := submit(w, "alice", 3) errBob := submit(w, "bob", 1) assert.True(t, errors.Is(errAlice, uploads.ErrQueueFull), "a third waiting upload from one user is refused") assert.NoError(t, errBob, "another user is not affected by alice's queue") } func TestWorkerSubmit_ThereIsADefaultCapSoANeglectedWorkerCannotBeFilledForever(t *testing.T) { r := newWRig(t, time.Minute) w := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute) refused := 0 for i := 0; i < 1000; i++ { if err := submit(w, "u"+strconv.Itoa(i), i); errors.Is(err, uploads.ErrQueueFull) { refused++ } } assert.Positive(t, refused, "with the defaults, 1,000 submits to an idle worker must not all be accepted") } func TestWorkerSubmit_AJobThatHasLeftTheQueueFreesItsPlace(t *testing.T) { r := newWRig(t, time.Minute) w := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute, uploads.WithQueueLimits(1, 1)) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { defer close(done); w.Run(ctx) }() defer func() { cancel(); <-done }() job := r.spool(t, "first") require.NoError(t, w.Submit(job)) require.Eventually(t, func() bool { c, _ := r.st.snapshot(); _, ok := c["first"]; return ok }, 5*time.Second, 5*time.Millisecond, "the first job never completed") assert.NoError(t, w.Submit(r.spool(t, "second")), "the cap counts jobs waiting, not jobs ever submitted") } func TestWorkerHasRoom_SaysWhetherASubmitWouldBeAccepted(t *testing.T) { w := queueWorker(t, 3, 2) assert.True(t, w.HasRoom("alice")) require.NoError(t, submit(w, "alice", 1)) require.NoError(t, submit(w, "alice", 2)) assert.False(t, w.HasRoom("alice"), "alice is at her share") assert.True(t, w.HasRoom("bob"), "bob is not") require.NoError(t, submit(w, "bob", 1)) assert.False(t, w.HasRoom("carol"), "the queue as a whole is at its cap") } func TestWorkerHasRoom_AgreesWithSubmitAtEveryStepAndIsFalseOnceStopped(t *testing.T) { w := queueWorker(t, 2, 1) for i, user := range []string{"a", "a", "b", "c"} { room := w.HasRoom(user) err := submit(w, user, i) assert.Equal(t, room, err == nil, "step %d (%s): HasRoom=%v but Submit returned %v", i, user, room, err) } r := newWRig(t, time.Minute) stopped := uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, time.Minute) ctx, cancel := context.WithCancel(context.Background()) cancel() stopped.Run(ctx) assert.False(t, stopped.HasRoom("a"), "a worker that has stopped has no room") }