informer_test.go

Documentation: k8s.io/client-go/dynamic/dynamicinformer

     1  /*
     2  Copyright 2018 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package dynamicinformer_test
    18  
    19  import (
    20  	"context"
    21  	"testing"
    22  	"time"
    23  
    24  	"github.com/google/go-cmp/cmp"
    25  	"k8s.io/apimachinery/pkg/api/equality"
    26  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    27  	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
    28  	"k8s.io/apimachinery/pkg/runtime"
    29  	"k8s.io/apimachinery/pkg/runtime/schema"
    30  	"k8s.io/client-go/dynamic/dynamicinformer"
    31  	"k8s.io/client-go/dynamic/fake"
    32  	"k8s.io/client-go/tools/cache"
    33  )
    34  
    35  type triggerFunc func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured
    36  
    37  func triggerFactory(t *testing.T) triggerFunc {
    38  	return func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, _ *unstructured.Unstructured) *unstructured.Unstructured {
    39  		testObject := newUnstructured("apps/v1", "Deployment", "ns-foo", "name-foo")
    40  		createdObj, err := fakeClient.Resource(gvr).Namespace(ns).Create(context.TODO(), testObject, metav1.CreateOptions{})
    41  		if err != nil {
    42  			t.Error(err)
    43  		}
    44  		return createdObj
    45  	}
    46  }
    47  
    48  func handler(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs {
    49  	return &cache.ResourceEventHandlerFuncs{
    50  		AddFunc: func(obj interface{}) {
    51  			rcvCh <- obj.(*unstructured.Unstructured)
    52  		},
    53  	}
    54  }
    55  
    56  func TestFilteredDynamicSharedInformerFactory(t *testing.T) {
    57  	scenarios := []struct {
    58  		name        string
    59  		existingObj *unstructured.Unstructured
    60  		gvr         schema.GroupVersionResource
    61  		informNS    string
    62  		ns          string
    63  		trigger     func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured
    64  		handler     func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs
    65  	}{
    66  		// scenario 1
    67  		{
    68  			name:     "scenario 1: test adding an object in different namespace should not trigger AddFunc",
    69  			informNS: "ns-bar",
    70  			ns:       "ns-foo",
    71  			gvr:      schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
    72  			trigger:  triggerFactory(t),
    73  			handler:  handler,
    74  		},
    75  		// scenario 2
    76  		{
    77  			name:     "scenario 2: test adding an object should trigger AddFunc",
    78  			informNS: "ns-foo",
    79  			ns:       "ns-foo",
    80  			gvr:      schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
    81  			trigger:  triggerFactory(t),
    82  			handler:  handler,
    83  		},
    84  	}
    85  
    86  	for _, ts := range scenarios {
    87  		t.Run(ts.name, func(t *testing.T) {
    88  			// test data
    89  			timeout := time.Duration(3 * time.Second)
    90  			ctx, cancel := context.WithTimeout(context.Background(), timeout)
    91  			defer cancel()
    92  			scheme := runtime.NewScheme()
    93  			informerReciveObjectCh := make(chan *unstructured.Unstructured, 1)
    94  			objs := []runtime.Object{}
    95  			if ts.existingObj != nil {
    96  				objs = append(objs, ts.existingObj)
    97  			}
    98  			// don't adjust the scheme to include deploymentlist. This is testing whether an informer can be created against using
    99  			// a client that doesn't have a type registered in the scheme.
   100  			gvrToListKind := map[schema.GroupVersionResource]string{
   101  				{Group: "apps", Version: "v1", Resource: "deployments"}: "DeploymentList",
   102  			}
   103  			fakeClient := fake.NewSimpleDynamicClientWithCustomListKinds(scheme, gvrToListKind, objs...)
   104  			target := dynamicinformer.NewFilteredDynamicSharedInformerFactory(fakeClient, 0, ts.informNS, nil)
   105  
   106  			// act
   107  			informerListerForGvr := target.ForResource(ts.gvr)
   108  			informerListerForGvr.Informer().AddEventHandler(ts.handler(informerReciveObjectCh))
   109  			target.Start(ctx.Done())
   110  			if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] {
   111  				t.Errorf("informer for %s hasn't synced", ts.gvr)
   112  			}
   113  
   114  			testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj)
   115  			select {
   116  			case objFromInformer := <-informerReciveObjectCh:
   117  				if ts.ns != ts.informNS {
   118  					t.Errorf("informer received an object for namespace %s when watching namespace %s", ts.ns, ts.informNS)
   119  				}
   120  				if !equality.Semantic.DeepEqual(testObject, objFromInformer) {
   121  					t.Fatalf("%v", cmp.Diff(testObject, objFromInformer))
   122  				}
   123  			case <-ctx.Done():
   124  				if ts.ns == ts.informNS {
   125  					t.Errorf("tested informer haven't received an object, waited %v", timeout)
   126  				}
   127  			}
   128  		})
   129  	}
   130  
   131  }
   132  
   133  func TestDynamicSharedInformerFactory(t *testing.T) {
   134  	scenarios := []struct {
   135  		name        string
   136  		existingObj *unstructured.Unstructured
   137  		gvr         schema.GroupVersionResource
   138  		ns          string
   139  		trigger     func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured
   140  		handler     func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs
   141  	}{
   142  		// scenario 1
   143  		{
   144  			name: "scenario 1: test if adding an object triggers AddFunc",
   145  			ns:   "ns-foo",
   146  			gvr:  schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
   147  			trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, _ *unstructured.Unstructured) *unstructured.Unstructured {
   148  				testObject := newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo")
   149  				createdObj, err := fakeClient.Resource(gvr).Namespace(ns).Create(context.TODO(), testObject, metav1.CreateOptions{})
   150  				if err != nil {
   151  					t.Error(err)
   152  				}
   153  				return createdObj
   154  			},
   155  			handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs {
   156  				return &cache.ResourceEventHandlerFuncs{
   157  					AddFunc: func(obj interface{}) {
   158  						rcvCh <- obj.(*unstructured.Unstructured)
   159  					},
   160  				}
   161  			},
   162  		},
   163  
   164  		// scenario 2
   165  		{
   166  			name:        "scenario 2: tests if updating an object triggers UpdateFunc",
   167  			ns:          "ns-foo",
   168  			gvr:         schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
   169  			existingObj: newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"),
   170  			trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured {
   171  				testObject.Object["spec"] = "updatedName"
   172  				updatedObj, err := fakeClient.Resource(gvr).Namespace(ns).Update(context.TODO(), testObject, metav1.UpdateOptions{})
   173  				if err != nil {
   174  					t.Error(err)
   175  				}
   176  				return updatedObj
   177  			},
   178  			handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs {
   179  				return &cache.ResourceEventHandlerFuncs{
   180  					UpdateFunc: func(old, updated interface{}) {
   181  						rcvCh <- updated.(*unstructured.Unstructured)
   182  					},
   183  				}
   184  			},
   185  		},
   186  
   187  		// scenario 3
   188  		{
   189  			name:        "scenario 3: test if deleting an object triggers DeleteFunc",
   190  			ns:          "ns-foo",
   191  			gvr:         schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"},
   192  			existingObj: newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"),
   193  			trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured {
   194  				err := fakeClient.Resource(gvr).Namespace(ns).Delete(context.TODO(), testObject.GetName(), metav1.DeleteOptions{})
   195  				if err != nil {
   196  					t.Error(err)
   197  				}
   198  				return testObject
   199  			},
   200  			handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs {
   201  				return &cache.ResourceEventHandlerFuncs{
   202  					DeleteFunc: func(obj interface{}) {
   203  						rcvCh <- obj.(*unstructured.Unstructured)
   204  					},
   205  				}
   206  			},
   207  		},
   208  	}
   209  
   210  	for _, ts := range scenarios {
   211  		t.Run(ts.name, func(t *testing.T) {
   212  			// test data
   213  			timeout := time.Duration(3 * time.Second)
   214  			ctx, cancel := context.WithTimeout(context.Background(), timeout)
   215  			defer cancel()
   216  			scheme := runtime.NewScheme()
   217  			informerReciveObjectCh := make(chan *unstructured.Unstructured, 1)
   218  			objs := []runtime.Object{}
   219  			if ts.existingObj != nil {
   220  				objs = append(objs, ts.existingObj)
   221  			}
   222  			// don't adjust the scheme to include deploymentlist. This is testing whether an informer can be created against using
   223  			// a client that doesn't have a type registered in the scheme.
   224  			gvrToListKind := map[schema.GroupVersionResource]string{
   225  				{Group: "extensions", Version: "v1beta1", Resource: "deployments"}: "DeploymentList",
   226  			}
   227  			fakeClient := fake.NewSimpleDynamicClientWithCustomListKinds(scheme, gvrToListKind, objs...)
   228  			target := dynamicinformer.NewDynamicSharedInformerFactory(fakeClient, 0)
   229  
   230  			// act
   231  			informerListerForGvr := target.ForResource(ts.gvr)
   232  			informerListerForGvr.Informer().AddEventHandler(ts.handler(informerReciveObjectCh))
   233  			target.Start(ctx.Done())
   234  			if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] {
   235  				t.Errorf("informer for %s hasn't synced", ts.gvr)
   236  			}
   237  
   238  			testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj)
   239  			select {
   240  			case objFromInformer := <-informerReciveObjectCh:
   241  				if !equality.Semantic.DeepEqual(testObject, objFromInformer) {
   242  					t.Fatalf("%v", cmp.Diff(testObject, objFromInformer))
   243  				}
   244  			case <-ctx.Done():
   245  				t.Errorf("tested informer haven't received an object, waited %v", timeout)
   246  			}
   247  		})
   248  	}
   249  }
   250  
   251  func newUnstructured(apiVersion, kind, namespace, name string) *unstructured.Unstructured {
   252  	return &unstructured.Unstructured{
   253  		Object: map[string]interface{}{
   254  			"apiVersion": apiVersion,
   255  			"kind":       kind,
   256  			"metadata": map[string]interface{}{
   257  				"namespace": namespace,
   258  				"name":      name,
   259  			},
   260  			"spec": name,
   261  		},
   262  	}
   263  }
   264
View as plain text