Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go (Go)

Code before:

func TestListenerResyncPeriods(t *testing.T) {
	// source simulates an apiserver object endpoint.
	source := newFakeControllerSource(t)
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})

	// create the shared informer and resync every 1s
	informer := NewSharedInformer(source, &v1.Pod{}, 1*time.Second).(*sharedIndexInformer)

	clock := testingclock.NewFakeClock(time.Now())
	informer.clock = clock
	informer.processor.clock = clock

	// listener 1, never resync
	listener1 := newTestListener("listener1", 0, "pod1", "pod2")
	informer.AddEventHandlerWithResyncPeriod(listener1, listener1.resyncPeriod)

	// listener 2, resync every 2s
	listener2 := newTestListener("listener2", 2*time.Second, "pod1", "pod2")
	informer.AddEventHandlerWithResyncPeriod(listener2, listener2.resyncPeriod)

	// listener 3, resync every 3s
	listener3 := newTestListener("listener3", 3*time.Second, "pod1", "pod2")
	informer.AddEventHandlerWithResyncPeriod(listener3, listener3.resyncPeriod)
	listeners := []*testListener{listener1, listener2, listener3}

	var wg wait.Group
	stop := make(chan struct{})
	wg.StartWithChannel(stop, informer.Run)
	defer func() {
		close(stop)
		wg.Wait()
	}()

	// ensure all listeners got the initial List
	for _, listener := range listeners {
		if !listener.ok() {
			t.Errorf("%s: expected %v, got %v", listener.name, listener.expectedItemNames, listener.receivedItemNames)
		}
	}

	// reset
	for _, listener := range listeners {
		listener.receivedItemNames = []string{}
	}

	// advance so listener2 gets a resync
	clock.Step(2 * time.Second)

	// make sure listener2 got the resync

Review comment (suggestion):
```suggestion
	// This artificial delay simulates that situation.
```

Code after:
		},
	})
	assert.NoError(err)

	// Should be immediately available
	res, err = informer.GetIndexer().ByIndex("labels-again", "a")
	assert.NoError(err)
	if diff := cmp.Diff([]any{pod1, pod3}, res, cmpOps); diff != "" {
		t.Fatal(diff)
	}
	if got := informer.GetIndexer().ListIndexFuncValues("labels"); !sets.New(got...).Equal(sets.New("a", "b")) {
		t.Fatalf("got %v", got)
	}
	if got := informer.GetIndexer().ListIndexFuncValues("labels-again"); !sets.New(got...).Equal(sets.New("a", "b")) {
		t.Fatalf("got %v", got)
	}
}

func TestListenerResyncPeriods(t *testing.T) {
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, 0) })
}

func TestListenerResyncPeriodsDelayed(t *testing.T) {
	// sharedProcessor used to deadlock when sharedProcessor.run happened
	// to be reached after the first sharedProcessor.distribute.
	// This artififical delay simulates that situation.
	//
	// Must not run in parallel to other tests, but it doesn't need to:
	// it doesn't sleep in real-world time and therefore completes quickly.
	synctest.Test(t, func(t *testing.T) { testListenerResyncPeriods(t, time.Second) })
}

func testListenerResyncPeriods(t *testing.T, startupDelay time.Duration) {
	start := time.Now()
	println := func(msg string) {
		delta := time.Since(start)
		t.Logf("%s: %s", delta, msg)
	}

	if startupDelay > 0 {
		hook := func() {
			time.Sleep(startupDelay)
		}
		sharedProcessorRunHook.Store(&hook)
		defer sharedProcessorRunHook.Store(nil)
	}

	// source simulates an apiserver object endpoint.
	source := newFakeControllerSource(t)
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
	source.Add(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})