Containr/app/backend/internal/deployqueue/queue_test.go

package deployqueue

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"github.com/google/uuid"
)

func TestSerializesPerService(t *testing.T) {
	q := New()
	svc := uuid.New()

	var active, maxActive int32
	finish := make(chan struct{})
	for i := 0; i < 3; i++ {
		q.Enqueue(svc, Job{Run: func(ctx context.Context) {
			if n := atomic.AddInt32(&active, 1); n > atomic.LoadInt32(&maxActive) {
				atomic.StoreInt32(&maxActive, n)
			}
			<-finish // hold until released
			atomic.AddInt32(&active, -1)
		}})
	}

	deadline := time.After(2 * time.Second)
	for atomic.LoadInt32(&active) < 1 {
		select {
		case <-deadline:
			t.Fatal("first job never started")
		default:
			time.Sleep(time.Millisecond)
		}
	}
	close(finish)
	deadline = time.After(2 * time.Second)
	for atomic.LoadInt32(&active) > 0 {
		select {
		case <-deadline:
			t.Fatal("jobs did not drain")
		default:
			time.Sleep(time.Millisecond)
		}
	}
	if got := atomic.LoadInt32(&maxActive); got != 1 {
		t.Fatalf("expected max concurrency 1, got %d", got)
	}
}

func TestDifferentServicesRunInParallel(t *testing.T) {
	q := New()
	started := make(chan struct{}, 2)
	release := make(chan struct{})
	for i := 0; i < 2; i++ {
		q.Enqueue(uuid.New(), Job{Run: func(ctx context.Context) {
			started <- struct{}{}
			<-release
		}})
	}
	for i := 0; i < 2; i++ {
		select {
		case <-started:
		case <-time.After(2 * time.Second):
			t.Fatal("parallel services did not both start")
		}
	}
	close(release)
}

func TestCancelQueuedRemovesJob(t *testing.T) {
	q := New()
	svc := uuid.New()
	release := make(chan struct{})
	q.Enqueue(svc, Job{DeploymentID: uuid.New(), Run: func(ctx context.Context) { <-release }})

	queuedID := uuid.New()
	ran := make(chan struct{})
	q.Enqueue(svc, Job{DeploymentID: queuedID, Run: func(ctx context.Context) { close(ran) }})

	if got := q.Cancel(svc, queuedID); got != WasQueued {
		t.Fatalf("expected WasQueued, got %v", got)
	}
	close(release)
	select {
	case <-ran:
		t.Fatal("cancelled queued job still ran")
	case <-time.After(200 * time.Millisecond):
	}
}

func TestCancelActiveCancelsContext(t *testing.T) {
	q := New()
	svc := uuid.New()
	id := uuid.New()
	ctxDone := make(chan struct{})
	q.Enqueue(svc, Job{DeploymentID: id, Run: func(ctx context.Context) {
		<-ctx.Done()
		close(ctxDone)
	}})

	deadline := time.After(2 * time.Second)
	for {
		q.mu.Lock()
		_, ok := q.cancels[id]
		q.mu.Unlock()
		if ok {
			break
		}
		select {
		case <-deadline:
			t.Fatal("job never registered")
		default:
			time.Sleep(time.Millisecond)
		}
	}
	if got := q.Cancel(svc, id); got != WasActive {
		t.Fatalf("expected WasActive, got %v", got)
	}
	select {
	case <-ctxDone:
	case <-time.After(2 * time.Second):
		t.Fatal("active job context not cancelled")
	}
}

func TestEnqueuePosition(t *testing.T) {
	q := New()
	svc := uuid.New()
	release := make(chan struct{})
	if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) { <-release }}); pos != 0 {
		t.Fatalf("first job position = %d, want 0", pos)
	}
	if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 1 {
		t.Fatalf("second job position = %d, want 1", pos)
	}
	if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 2 {
		t.Fatalf("third job position = %d, want 2", pos)
	}
	close(release)
}