client-go/kubernetes_test/clientset_test: intro TestUnSupportWatchListSemantics

Kubernetes-commit: f8868e9cace5903b425425605e4946817dd11770

kubernetes_test/clientset_test.go

@@ -18,17 +18,122 @@ package kubernetes_test
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
+	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/json"
+	"k8s.io/apimachinery/pkg/watch"
+	clientfeatures "k8s.io/client-go/features"
+	clientfeaturestesting "k8s.io/client-go/features/testing"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	fakeclientset "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/watchlist"
 )
 
+func TestDoesClientSupportWatchListSemanticsForKubeClient(t *testing.T) {
+	target, err := kubernetes.NewForConfig(&rest.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if watchlist.DoesClientNotSupportWatchListSemantics(target) {
+		t.Fatalf("Kubernetes client should support WatchList semantics")
+	}
+}
+
+func TestWatchListSemanticsSimple(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.WatchListClient, true)
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if _, ok := req.URL.Query()["watch"]; !ok {
+			t.Errorf("expected a watch request, params: %v", req.URL.Query())
+			http.Error(w, fmt.Errorf("unexpected request").Error(), http.StatusInternalServerError)
+			return
+		}
+
+		obj := &appsv1.Deployment{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "apps/v1",
+				Kind:       "Deployment",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					metav1.InitialEventsAnnotationKey: "true",
+				},
+			},
+		}
+		rawObj, err := json.Marshal(obj)
+		if err != nil {
+			t.Errorf("failed to marshal rawObj: %v", err)
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		watchEvent := &metav1.WatchEvent{
+			Type:   string(watch.Bookmark),
+			Object: runtime.RawExtension{Raw: rawObj},
+		}
+		rawRsp, err := json.Marshal(watchEvent)
+		if err != nil {
+			t.Errorf("failed to marshal watchEvent: %v", err)
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, err = w.Write(rawRsp)
+		if err != nil {
+			t.Fatalf("failed to write response: %v", err)
+		}
+	}))
+	defer server.Close()
+
+	cfg := &rest.Config{Host: server.URL}
+	client, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	factory := informers.NewSharedInformerFactory(client, 0)
+	target := factory.Apps().V1().Deployments().Informer()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	factory.Start(ctx.Done())
+
+	if !cache.WaitForCacheSync(ctx.Done(), target.HasSynced) {
+		t.Fatalf("failed to wait for caches to sync")
+	}
+}
+
+func TestUnSupportWatchListSemantics(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.WatchListClient, true)
+	// The fake client doesn’t support WatchList semantics,
+	// so we don’t need to prepare a response.
+	fakeClient := fakeclientset.NewClientset()
+	factory := informers.NewSharedInformerFactory(fakeClient, 0)
+
+	// register a deployment infm
+	target := factory.Apps().V1().Deployments().Informer()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	factory.Start(ctx.Done())
+
+	if !cache.WaitForCacheSync(ctx.Done(), target.HasSynced) {
+		t.Fatalf("failed to wait for caches to sync")
+	}
+}
+
 func TestClientUserAgent(t *testing.T) {
 	tests := []struct {
 		name      string