e2e/apimachinery/watchlist: test if WatchList is requested by client-go's List method when WatchListClient is enabled

test/e2e/apimachinery/watchlist.go

@@ -19,11 +19,13 @@ package apimachinery
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"sort"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,7 +34,9 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientfeatures "k8s.io/client-go/features"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/consistencydetector"
 	"k8s.io/component-base/featuregate"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/test/e2e/feature"
@@ -41,7 +45,7 @@ import (
 
 var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithSerial(), feature.WatchList, func() {
 	f := framework.NewDefaultFramework("watchlist")
-	ginkgo.It("should be requested when WatchListClient is enabled", func(ctx context.Context) {
+	ginkgo.It("should be requested by informers when WatchListClient is enabled", func(ctx context.Context) {
 		featuregatetesting.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), utilfeature.DefaultFeatureGate, featuregate.Feature(clientfeatures.WatchListClient), true)
 		stopCh := make(chan struct{})
 		defer close(stopCh)
@@ -85,8 +89,56 @@ var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithSerial(), fe
 		framework.ExpectNoError(err)
 		verifyStore(ctx, f, secretInformer.GetStore())
 	})
+	ginkgo.It("should be requested by client-go's List method when WatchListClient is enabled", func(ctx context.Context) {
+		featuregatetesting.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), utilfeature.DefaultFeatureGate, featuregate.Feature(clientfeatures.WatchListClient), true)
+
+		ginkgo.By(fmt.Sprintf("Adding 5 secrets to %s namespace", f.Namespace.Name))
+		var expectedSecrets []v1.Secret
+		for i := 1; i <= 5; i++ {
+			secret, err := f.ClientSet.CoreV1().Secrets(f.Namespace.Name).Create(ctx, newSecret(fmt.Sprintf("secret-%d", i)), metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+			expectedSecrets = append(expectedSecrets, *secret)
+		}
+
+		var actualRequestsMadeByKubeClient []string
+		clientConfig := f.ClientConfig()
+		clientConfig.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+			return roundTripFunc(func(req *http.Request) (*http.Response, error) {
+				actualRequestsMadeByKubeClient = append(actualRequestsMadeByKubeClient, req.URL.RawQuery)
+				return rt.RoundTrip(req)
+			})
+		})
+		wrappedKubeClient, err := kubernetes.NewForConfig(clientConfig)
+		framework.ExpectNoError(err)
+
+		ginkgo.By("Streaming secrets from the server")
+		secretList, err := wrappedKubeClient.CoreV1().Secrets(f.Namespace.Name).List(ctx, metav1.ListOptions{})
+		framework.ExpectNoError(err)
+
+		ginkgo.By("Verifying if the secret list was properly streamed")
+		streamedSecrets := secretList.Items
+		sort.Sort(byName(expectedSecrets))
+		gomega.Expect(cmp.Equal(expectedSecrets, streamedSecrets)).To(gomega.BeTrueBecause("data received via watchlist must match the added data"))
+
+		ginkgo.By("Verifying if expected requests were sent to the server")
+		expectedRequestMadeByKubeClient := []string{
+			// corresponds to a streaming request made by the kube client to stream the secrets
+			"allowWatchBookmarks=true&resourceVersionMatch=NotOlderThan&sendInitialEvents=true&watch=true",
+		}
+		if consistencydetector.IsDataConsistencyDetectionForWatchListEnabled() {
+			// corresponds to a standard list request made by the consistency detector build in into the kube client
+			expectedRequestMadeByKubeClient = append(expectedRequestMadeByKubeClient, fmt.Sprintf("resourceVersion=%s&resourceVersionMatch=Exact", secretList.ResourceVersion))
+		}
+		gomega.Expect(actualRequestsMadeByKubeClient).To(gomega.Equal(expectedRequestMadeByKubeClient))
+	})
 })
 
+type roundTripFunc func(req *http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
 func verifyStore(ctx context.Context, f *framework.Framework, store cache.Store) {
 	ginkgo.By(fmt.Sprintf("Listing secrets directly from the server from %s namespace", f.Namespace.Name))
 	expectedSecretsList, err := f.ClientSet.CoreV1().Secrets(f.Namespace.Name).List(ctx, metav1.ListOptions{})