func handleDownlinkTaskQueueTest(ctx context.Context, q DownlinkTaskQueue, consumerIDs []string) {
	t, a := test.MustNewTFromContext(ctx)
	pbs := [...]ttnpb.EndDeviceIdentifiers{
		{
			ApplicationIdentifiers: ttnpb.ApplicationIdentifiers{
				ApplicationId: "test-app",
			},
			DeviceId: "test-dev",
		},
		{
			ApplicationIdentifiers: ttnpb.ApplicationIdentifiers{
				ApplicationId: "test-app2",
			},
			DeviceId: "test-dev",
		},
		{
			ApplicationIdentifiers: ttnpb.ApplicationIdentifiers{
				ApplicationId: "test-app2",
			},
			DeviceId: "test-dev2",
		},
	}
	type slot struct {
		ctx        context.Context
		id         ttnpb.EndDeviceIdentifiers
		t          time.Time
		consumerID string
		respCh     chan<- TaskPopFuncResponse
	}
	popCtx, cancelPopCtx := context.WithCancel(context.WithValue(ctx, &struct{}{}, "pop"))
	defer cancelPopCtx()
	nextPop := make(map[string]chan struct{}, len(consumerIDs))
	for _, consumerID := range consumerIDs {
		nextPop[consumerID] = make(chan struct{})
	}
	slotCh := make(chan slot)
	errCh := make(chan error)
	for _, consumerID := range consumerIDs {
		go func(consumerID string) {
			for {
				select {
				case <-nextPop[consumerID]:
				case <-popCtx.Done():
					errCh <- popCtx.Err()
				}
				select {
				case <-ctx.Done():
					return
				case errCh <- q.Pop(popCtx, consumerID, func(ctx context.Context, id ttnpb.EndDeviceIdentifiers, t time.Time) (time.Time, error) {
					respCh := make(chan TaskPopFuncResponse)
					slotCh <- slot{
						ctx:        ctx,
						id:         id,
						t:          t,
						respCh:     respCh,
						consumerID: consumerID,
					}
					resp := <-respCh
					return resp.Time, resp.Error
				}):
				}
			}
		}(consumerID)
	}
	for _, consumerID := range consumerIDs {
		nextPop[consumerID] <- struct{}{}
	}
	select {
	case s := <-slotCh:
		t.Fatalf("Pop called f on empty schedule, slot: %+v", s)
	case err := <-errCh:
		t.Fatalf("Pop returned on empty schedule with error: %s", test.FormatError(err))
	case <-time.After(test.Delay):
	}
	err := q.Add(ctx, pbs[0], time.Unix(0, 0), false)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	select {
	case s := <-slotCh:
		a.So(s.id, should.Resemble, pbs[0])
		a.So(s.t, should.Equal, time.Unix(0, 0))
		if !a.So(s.ctx, should.HaveParentContextOrEqual, popCtx) {
			t.Fatal(s.ctx)
		}
		s.respCh <- TaskPopFuncResponse{}
		select {
		case err := <-errCh:
			if !a.So(err, should.BeNil) {
				t.FailNow()
			}
		case <-ctx.Done():
			t.Fatal("Timed out waiting for Pop to return")
		}
		nextPop[s.consumerID] <- struct{}{}
	case err := <-errCh:
		a.So(err, should.BeNil)
		t.Fatal("Pop returned without calling f on non-empty schedule")
	case <-ctx.Done():
		t.Fatal("Timed out waiting for Pop to call f")
	}
	err = q.Add(ctx, pbs[0], time.Unix(0, 42), false)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[1], time.Now().Add(time.Hour), false)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[1], time.Now().Add(2*time.Hour), true)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[1], time.Unix(13, 0), false)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[1], time.Unix(42, 0), true)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[2], time.Now().Add(42*time.Hour), true)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	err = q.Add(ctx, pbs[2], time.Unix(42, 42), true)
	if !a.So(err, should.BeNil) {
		t.FailNow()
	}
	receivedSlots := make(map[ttnpb.EndDeviceIdentifiers][]time.Time, 3)
	for i := 0; i < 3; i++ {
		t.Helper()
		a := assertions.New(t)
		select {
		case s := <-slotCh:
			receivedSlots[s.id] = append(receivedSlots[s.id], s.t)
			a.So(s.ctx, should.HaveParentContextOrEqual, popCtx)
			s.respCh <- TaskPopFuncResponse{}
			select {
			case err := <-errCh:
				if !a.So(err, should.BeNil) {
					t.FailNow()
				}
			case <-ctx.Done():
				t.Fatal("Timed out waiting for Pop to return")
			}
			nextPop[s.consumerID] <- struct{}{}
		case err := <-errCh:
			a.So(err, should.BeNil)
			t.Fatal("Pop returned without calling f on non-empty schedule")
		case <-ctx.Done():
			t.Fatal("Timed out waiting for Pop to call f")
		}
	}
	a.So(receivedSlots, should.Resemble, map[ttnpb.EndDeviceIdentifiers][]time.Time{
		pbs[0]: {time.Unix(0, 42).UTC()},
		pbs[1]: {time.Unix(42, 0).UTC()},
		pbs[2]: {time.Unix(42, 42).UTC()},
	})
	cancelPopCtx()
	for range consumerIDs {
		select {
		case err := <-errCh:
			a.So(errors.IsCanceled(err), should.BeTrue)
		case <-ctx.Done():
		}
	}
}
handleDownlinkTaskQueueTest runs a test suite on q.
package test
import (
	"context"
	"testing"
	"time"
	"github.com/smartystreets/assertions"
	"go.thethings.network/lorawan-stack/v3/pkg/errors"
	. "go.thethings.network/lorawan-stack/v3/pkg/networkserver"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
	"go.thethings.network/lorawan-stack/v3/pkg/util/test"
	"go.thethings.network/lorawan-stack/v3/pkg/util/test/assertions/should"
)