/* Copyright 2016 The Kubernetes Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package watch import ( "context" "errors" "reflect" "strings" "testing" "time" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" fakeclient "k8s.io/client-go/kubernetes/fake" "k8s.io/client-go/tools/cache" ) type fakePod struct { } func (obj *fakePod) GetObjectKind() schema.ObjectKind { return schema.EmptyObjectKind } func (obj *fakePod) DeepCopyObject() runtime.Object { panic("DeepCopyObject not supported by fakePod") } func TestUntil(t *testing.T) { fw := watch.NewFake() go func() { var obj *fakePod fw.Add(obj) fw.Modify(obj) }() conditions := []ConditionFunc{ func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return event.Type == watch.Modified, nil }, } ctx, cancel := context.WithTimeout(context.Background(), time.Minute) defer cancel() lastEvent, err := UntilWithoutRetry(ctx, fw, conditions...) if err != nil { t.Fatalf("expected nil error, got %#v", err) } if lastEvent == nil { t.Fatal("expected an event") } if lastEvent.Type != watch.Modified { t.Fatalf("expected MODIFIED event type, got %v", lastEvent.Type) } if got, isPod := lastEvent.Object.(*fakePod); !isPod { t.Fatalf("expected a pod event, got %#v", got) } } func TestUntilMultipleConditions(t *testing.T) { fw := watch.NewFake() go func() { var obj *fakePod fw.Add(obj) }() conditions := []ConditionFunc{ func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, } ctx, cancel := context.WithTimeout(context.Background(), time.Minute) defer cancel() lastEvent, err := UntilWithoutRetry(ctx, fw, conditions...) if err != nil { t.Fatalf("expected nil error, got %#v", err) } if lastEvent == nil { t.Fatal("expected an event") } if lastEvent.Type != watch.Added { t.Fatalf("expected MODIFIED event type, got %v", lastEvent.Type) } if got, isPod := lastEvent.Object.(*fakePod); !isPod { t.Fatalf("expected a pod event, got %#v", got) } } func TestUntilMultipleConditionsFail(t *testing.T) { fw := watch.NewFake() go func() { var obj *fakePod fw.Add(obj) }() conditions := []ConditionFunc{ func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return event.Type == watch.Deleted, nil }, } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() lastEvent, err := UntilWithoutRetry(ctx, fw, conditions...) if err != wait.ErrWaitTimeout { t.Fatalf("expected ErrWaitTimeout error, got %#v", err) } if lastEvent == nil { t.Fatal("expected an event") } if lastEvent.Type != watch.Added { t.Fatalf("expected ADDED event type, got %v", lastEvent.Type) } if got, isPod := lastEvent.Object.(*fakePod); !isPod { t.Fatalf("expected a pod event, got %#v", got) } } func TestUntilTimeout(t *testing.T) { fw := watch.NewFake() go func() { var obj *fakePod fw.Add(obj) fw.Modify(obj) }() conditions := []ConditionFunc{ func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return event.Type == watch.Modified, nil }, } lastEvent, err := UntilWithoutRetry(context.Background(), fw, conditions...) if err != nil { t.Fatalf("expected nil error, got %#v", err) } if lastEvent == nil { t.Fatal("expected an event") } if lastEvent.Type != watch.Modified { t.Fatalf("expected MODIFIED event type, got %v", lastEvent.Type) } if got, isPod := lastEvent.Object.(*fakePod); !isPod { t.Fatalf("expected a pod event, got %#v", got) } } func TestUntilErrorCondition(t *testing.T) { fw := watch.NewFake() go func() { var obj *fakePod fw.Add(obj) }() expected := "something bad" conditions := []ConditionFunc{ func(event watch.Event) (bool, error) { return event.Type == watch.Added, nil }, func(event watch.Event) (bool, error) { return false, errors.New(expected) }, } ctx, cancel := context.WithTimeout(context.Background(), time.Minute) defer cancel() _, err := UntilWithoutRetry(ctx, fw, conditions...) if err == nil { t.Fatal("expected an error") } if !strings.Contains(err.Error(), expected) { t.Fatalf("expected %q in error string, got %q", expected, err.Error()) } } func TestUntilWithSync(t *testing.T) { // FIXME: test preconditions tt := []struct { name string lw *cache.ListWatch preconditionFunc PreconditionFunc conditionFunc ConditionFunc expectedErr error expectedEvent *watch.Event }{ { name: "doesn't wait for sync with no precondition", lw: &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { select {} }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { select {} }, }, preconditionFunc: nil, conditionFunc: func(e watch.Event) (bool, error) { return true, nil }, expectedErr: errors.New("timed out waiting for the condition"), expectedEvent: nil, }, { name: "waits indefinitely with precondition if it can't sync", lw: &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { select {} }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { select {} }, }, preconditionFunc: func(store cache.Store) (bool, error) { return true, nil }, conditionFunc: func(e watch.Event) (bool, error) { return true, nil }, expectedErr: errors.New("UntilWithSync: unable to sync caches: context deadline exceeded"), expectedEvent: nil, }, { name: "precondition can stop the loop", lw: func() *cache.ListWatch { fakeclient := fakeclient.NewSimpleClientset(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "first"}}) return &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return fakeclient.CoreV1().Secrets("").List(context.TODO(), options) }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return fakeclient.CoreV1().Secrets("").Watch(context.TODO(), options) }, } }(), preconditionFunc: func(store cache.Store) (bool, error) { _, exists, err := store.Get(&metav1.ObjectMeta{Namespace: "", Name: "first"}) if err != nil { return true, err } if exists { return true, nil } return false, nil }, conditionFunc: func(e watch.Event) (bool, error) { return true, errors.New("should never reach this") }, expectedErr: nil, expectedEvent: nil, }, { name: "precondition lets it proceed to regular condition", lw: func() *cache.ListWatch { fakeclient := fakeclient.NewSimpleClientset(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "first"}}) return &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { return fakeclient.CoreV1().Secrets("").List(context.TODO(), options) }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { return fakeclient.CoreV1().Secrets("").Watch(context.TODO(), options) }, } }(), preconditionFunc: func(store cache.Store) (bool, error) { return false, nil }, conditionFunc: func(e watch.Event) (bool, error) { if e.Type == watch.Added { return true, nil } panic("no other events are expected") }, expectedErr: nil, expectedEvent: &watch.Event{Type: watch.Added, Object: &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "first"}}}, }, } for _, tc := range tt { t.Run(tc.name, func(t *testing.T) { // Informer waits for caches to sync by polling in 100ms intervals, // timeout needs to be reasonably higher ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) defer cancel() event, err := UntilWithSync(ctx, tc.lw, &corev1.Secret{}, tc.preconditionFunc, tc.conditionFunc) if !reflect.DeepEqual(err, tc.expectedErr) { t.Errorf("expected error %#v, got %#v", tc.expectedErr, err) } if !reflect.DeepEqual(event, tc.expectedEvent) { t.Errorf("expected event %#v, got %#v", tc.expectedEvent, event) } }) } }