From ae201951a8a39316d4154a07592f464293764a83 Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Sat, 20 Sep 2025 22:57:14 +0200 Subject: [PATCH] k8s.io/client-go/tools/cache/test: wrap the LW with toListWatcherWithUnSupportedWatchListSemantics --- .../client-go/tools/cache/controller_test.go | 18 +++++++- .../tools/cache/mutation_detector_test.go | 4 +- .../client-go/tools/cache/reflector_test.go | 44 +++++++++---------- 3 files changed, 40 insertions(+), 26 deletions(-) diff --git a/staging/src/k8s.io/client-go/tools/cache/controller_test.go b/staging/src/k8s.io/client-go/tools/cache/controller_test.go index 23904006f68..a44537e08b7 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller_test.go @@ -363,7 +363,7 @@ func TestUpdate(t *testing.T) { // everything we've added has been deleted. watchCh := make(chan struct{}) _, controller := NewInformer( - &ListWatch{ + toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { watch, err := source.Watch(options) close(watchCh) @@ -372,7 +372,7 @@ func TestUpdate(t *testing.T) { ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return source.List(options) }, - }, + }), &v1.Pod{}, 0, ResourceEventHandlerFuncs{ @@ -733,3 +733,17 @@ func TestDeletionHandlingObjectToName(t *testing.T) { t.Errorf("Expected %#v, got %#v", expected, actual) } } + +type listWatchWithUnSupportedWatchListSemanticsWrapper struct { + *ListWatch +} + +func (lw listWatchWithUnSupportedWatchListSemanticsWrapper) IsWatchListSemanticsUnSupported() bool { + return true +} + +func toListWatcherWithUnSupportedWatchListSemantics(lw *ListWatch) ListerWatcher { + return listWatchWithUnSupportedWatchListSemanticsWrapper{ + lw, + } +} diff --git a/staging/src/k8s.io/client-go/tools/cache/mutation_detector_test.go b/staging/src/k8s.io/client-go/tools/cache/mutation_detector_test.go index fab9f4dd07e..29f74676444 100644 --- a/staging/src/k8s.io/client-go/tools/cache/mutation_detector_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/mutation_detector_test.go @@ -29,14 +29,14 @@ import ( func TestMutationDetector(t *testing.T) { fakeWatch := watch.NewFake() - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return fakeWatch, nil }, ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return &v1.PodList{}, nil }, - } + }) pod := &v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "anything", diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go index 8c20ae0b02a..6a7dd120e8c 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go @@ -686,7 +686,7 @@ func TestReflectorListAndWatchWithErrors(t *testing.T) { watchRet, watchErr := item.events, item.watchErr _, ctx := ktesting.NewTestContext(t) ctx, cancel := context.WithCancelCause(ctx) - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { if watchErr != nil { return nil, watchErr @@ -710,7 +710,7 @@ func TestReflectorListAndWatchWithErrors(t *testing.T) { ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return item.list, item.listErr }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) err := r.ListAndWatchWithContext(ctx) if item.listErr != nil && !errors.Is(err, item.listErr) { @@ -991,7 +991,7 @@ func TestReflectorResync(t *testing.T) { }, } - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { fw := watch.NewFake() return fw, nil @@ -999,7 +999,7 @@ func TestReflectorResync(t *testing.T) { ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "0"}}, nil }, - } + }) resyncPeriod := 1 * time.Millisecond r := NewReflector(lw, &v1.Pod{}, s, resyncPeriod) if err := r.ListAndWatchWithContext(ctx); err != nil { @@ -1016,7 +1016,7 @@ func TestReflectorWatchListPageSize(t *testing.T) { ctx, cancel := context.WithCancelCause(ctx) s := NewStore(MetaNamespaceKeyFunc) - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1043,7 +1043,7 @@ func TestReflectorWatchListPageSize(t *testing.T) { } return nil, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) // Set resource version to test pagination also for not consistent reads. r.setLastSyncResourceVersion("10") @@ -1062,7 +1062,7 @@ func TestReflectorNotPaginatingNotConsistentReads(t *testing.T) { ctx, cancel := context.WithCancelCause(ctx) s := NewStore(MetaNamespaceKeyFunc) - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1082,7 +1082,7 @@ func TestReflectorNotPaginatingNotConsistentReads(t *testing.T) { } return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "10"}, Items: pods}, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) r.setLastSyncResourceVersion("10") require.NoError(t, r.ListAndWatchWithContext(ctx)) @@ -1098,7 +1098,7 @@ func TestReflectorPaginatingNonConsistentReadsIfWatchCacheDisabled(t *testing.T) var cancel func(error) s := NewStore(MetaNamespaceKeyFunc) - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1126,7 +1126,7 @@ func TestReflectorPaginatingNonConsistentReadsIfWatchCacheDisabled(t *testing.T) } return nil, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) // Initial list should initialize paginatedResult in the reflector. @@ -1155,7 +1155,7 @@ func TestReflectorResyncWithResourceVersion(t *testing.T) { s := NewStore(MetaNamespaceKeyFunc) listCallRVs := []string{} - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1178,7 +1178,7 @@ func TestReflectorResyncWithResourceVersion(t *testing.T) { } return nil, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) // Initial list should use RV=0 @@ -1216,7 +1216,7 @@ func TestReflectorExpiredExactResourceVersion(t *testing.T) { s := NewStore(MetaNamespaceKeyFunc) listCallRVs := []string{} - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1242,7 +1242,7 @@ func TestReflectorExpiredExactResourceVersion(t *testing.T) { } return nil, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) // Initial list should use RV=0 @@ -1276,7 +1276,7 @@ func TestReflectorFullListIfExpired(t *testing.T) { s := NewStore(MetaNamespaceKeyFunc) listCallRVs := []string{} - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1311,7 +1311,7 @@ func TestReflectorFullListIfExpired(t *testing.T) { return nil, err } }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) r.WatchListPageSize = 4 @@ -1351,7 +1351,7 @@ func TestReflectorFullListIfTooLarge(t *testing.T) { listCallRVs := []string{} version := 30 - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { // Stop once the reflector begins watching since we're only interested in the list. cancel(errors.New("done")) @@ -1388,7 +1388,7 @@ func TestReflectorFullListIfTooLarge(t *testing.T) { return nil, fmt.Errorf("unexpected List call: %s", options.ResourceVersion) } }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) // Initial list should use RV=0 @@ -1581,14 +1581,14 @@ func TestReflectorResourceVersionUpdate(t *testing.T) { ctx, cancel := context.WithCancelCause(ctx) fw := watch.NewFake() - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return fw, nil }, ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "10"}}, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, s, 0) makePod := func(rv string) *v1.Pod { @@ -1997,7 +1997,7 @@ func TestReflectorReplacesStoreOnUnsafeDelete(t *testing.T) { } var once sync.Once - lw := &ListWatch{ + lw := toListWatcherWithUnSupportedWatchListSemantics(&ListWatch{ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { fw := watch.NewFake() go func() { @@ -2017,7 +2017,7 @@ func TestReflectorReplacesStoreOnUnsafeDelete(t *testing.T) { } return list, nil }, - } + }) r := NewReflector(lw, &v1.Pod{}, store, 0) doneCh, stopCh := make(chan struct{}), make(chan struct{})