Function TestQueueBuildUp

flytestdlib/cache/in_memory_auto_refresh_test.go:279–321

(t *testing.T)

}

func TestQueueBuildUp(t *testing.T) {
	testResyncPeriod := time.Hour
	rateLimiter := workqueue.DefaultTypedControllerRateLimiter[*Batch]()
	fakeClock := testingclock.NewFakeClock(time.Now())

	syncCount := atomic.NewInt32(0)
	m := sync.Map{}
	alwaysFailing := func(ctx context.Context, batch Batch) (
		updatedBatch []ItemSyncResponse, err error) {
		assert.Len(t, batch, 1)
		_, existing := m.LoadOrStore(batch[0].GetID(), 0)
		assert.False(t, existing, "Saw %v before", batch[0].GetID())
		if existing {
			t.FailNow()
		}

		syncCount.Inc()
		return nil, fmt.Errorf("expected error")
	}

	size := uint(100)
	cache, err := NewInMemoryAutoRefresh("fake2", alwaysFailing, rateLimiter, testResyncPeriod, 10, size, promutils.NewTestScope(), WithClock(fakeClock))
	assert.NoError(t, err)

	ctx, cancel := context.WithCancel(context.Background())
	assert.NoError(t, cache.Start(ctx))
	defer cancel()

	for i := uint(0); i < size; i++ {
		// #nosec G115
		_, err := cache.GetOrCreate(strconv.Itoa(int(i)), fakeCacheItem{val: 3})
		assert.NoError(t, err)
	}

	// wait for all workers to run
	assert.Eventually(t, func() bool {
		// trigger a sync and unlock the work queue
		fakeClock.Step(time.Millisecond)

		return syncCount.Load() == int32(size) // #nosec G115

	}, 5*time.Second, time.Millisecond)
}

func TestInProcessing(t *testing.T) {
	syncer := &panickingSyncer{}