func (rst *watcherSyncerTester) expectStop(r watchersyncer.ResourceType) {
	name := model.ListOptionsToDefaultPathRoot(r.ListInterface)
	log.WithField("Name", name).Infof("Expecting Stop")
	Eventually(func() bool {
		return len(rst.lws[name].stopEvents) > 0
	}).Should(BeTrue())
	<-rst.lws[name].stopEvents
}
Call to verify that stop has been invoked on the watcher.
package watchersyncer_test
import (
	"context"
	"errors"
	"sync"
	"time"
	"github.com/google/uuid"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	log "github.com/sirupsen/logrus"
	apiv3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
	"github.com/projectcalico/libcalico-go/lib/backend/api"
	"github.com/projectcalico/libcalico-go/lib/backend/model"
	"github.com/projectcalico/libcalico-go/lib/backend/watchersyncer"
	cerrors "github.com/projectcalico/libcalico-go/lib/errors"
	cnet "github.com/projectcalico/libcalico-go/lib/net"
	"github.com/projectcalico/libcalico-go/lib/testutils"
)
var (
	dsError = cerrors.ErrorDatastoreError{Err: errors.New("Generic datastore error")}
	l1Key1  = model.ResourceKey{
		Kind:      apiv3.KindNetworkPolicy,
		Namespace: "namespace1",
		Name:      "policy-1",
	}
	l1Key2 = model.ResourceKey{
		Kind:      apiv3.KindNetworkPolicy,
		Namespace: "namespace1",
		Name:      "policy-2",
	}
	l1Key3 = model.ResourceKey{
		Kind:      apiv3.KindNetworkPolicy,
		Namespace: "namespace2",
		Name:      "policy-1",
	}
	l1Key4 = model.ResourceKey{
		Kind:      apiv3.KindNetworkPolicy,
		Namespace: "namespace2999",
		Name:      "policy-1000",
	}
	l2Key1 = model.ResourceKey{
		Kind: apiv3.KindIPPool,
		Name: "ippool-1",
	}
	l2Key2 = model.ResourceKey{
		Kind: apiv3.KindIPPool,
		Name: "ippool-2",
	}
	l3Key1 = model.BlockAffinityKey{
		CIDR: cnet.MustParseCIDR("1.2.3.0/24"),
		Host: "mynode",
	}
	emptyList = &model.KVPairList{
		Revision: "abcdef12345",
	}
	notSupported = cerrors.ErrorOperationNotSupported{}
	notExists    = cerrors.ErrorResourceDoesNotExist{}
	genError     = errors.New("Generic error")
)
var _ = Describe("Test the backend datastore multi-watch syncer", func() {
	r1 := watchersyncer.ResourceType{
		ListInterface: model.ResourceListOptions{Kind: apiv3.KindNetworkPolicy},
	}
	r2 := watchersyncer.ResourceType{
		ListInterface: model.ResourceListOptions{Kind: apiv3.KindIPPool},
	}
	r3 := watchersyncer.ResourceType{
		ListInterface: model.BlockAffinityListOptions{},
	}
	It("should receive a sync event when the watchers have listed current settings", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2, r3})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.clientListResponse(r2, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUpdate(api.InSync)
	})
	It("should not change status if watch returns multiple ErrorOperationNotSupported errors", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r1, notSupported)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notSupported)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notSupported)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notSupported)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notSupported)
		rs.ExpectStatusUnchanged()
	})
	It("should not change status if watch returns multiple ErrorResourceDoesNotExist errors", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r1, notExists)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notExists)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notExists)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notExists)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, notExists)
		rs.ExpectStatusUnchanged()
	})
	It("should handle reconnection if watchers fail to be created", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2, r3})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		defer setWatchIntervals(watchersyncer.ListRetryInterval, watchersyncer.WatchPollInterval)
		setWatchIntervals(500*time.Millisecond, 2000*time.Millisecond)
		By("Driving a bunch of List complete, Watch fail events for 2/3 resource types")
		expectedDuration := watchersyncer.WatchPollInterval * 2
		minDuration := 70 * expectedDuration / 100
		maxDuration := 150 * expectedDuration / 100
		before := time.Now()
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.clientWatchResponse(r1, genError)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, nil)
		rs.clientListResponse(r2, emptyList)
		rs.clientWatchResponse(r2, notSupported)
		rs.clientListResponse(r2, emptyList)
		rs.clientWatchResponse(r2, notSupported)
		rs.clientListResponse(r2, emptyList)
		rs.clientWatchResponse(r2, nil)
		By("Expecting the time for all events to be handled is within a sensible window")
		for i := time.Duration(0); i < maxDuration/(10*time.Millisecond); i++ {
			if rs.allEventsHandled() {
				break
			}
			time.Sleep(minDuration / 50)
		}
		duration := time.Now().Sub(before)
		rs.expectAllEventsHandled()
		Expect(duration).To(BeNumerically(">", minDuration))
		Expect(duration).To(BeNumerically("<", maxDuration))
		rs.ExpectStatusUnchanged()
		By("Driving a bunch of List complete, Watch fail events for the 3rd resource type")
		expectedDuration = watchersyncer.WatchPollInterval * watchersyncer.ListRetryInterval
		minDuration = 70 * expectedDuration / 100
		maxDuration = 130 * expectedDuration / 100
		before = time.Now()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r3, notSupported)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, genError)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r3, nil)
		rs.ExpectStatusUnchanged()
		By("Expecting the time for the events of the final resource sync to be handled is within sensible window")
		for i := time.Duration(0); i < maxDuration/(10*time.Millisecond); i++ {
			if rs.allEventsHandled() {
				break
			}
			time.Sleep(minDuration / 50)
		}
		duration = time.Now().Sub(before)
		rs.expectAllEventsHandled()
		Expect(duration).To(BeNumerically(">", minDuration))
		Expect(duration).To(BeNumerically("<", maxDuration))
	})
	It("Should handle reconnection and syncing when the watcher sends a watch terminated error", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2, r3})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.clientWatchResponse(r1, nil)
		rs.clientListResponse(r2, emptyList)
		rs.clientWatchResponse(r2, nil)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r3, nil)
		rs.sendEvent(r3, api.WatchEvent{
			Type:  api.WatchError,
			Error: dsError,
		})
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r3, nil)
		rs.ExpectStatusUnchanged()
		rs.expectAllEventsHandled()
	})
	It("Should handle receiving events while one watcher fails and fails to recreate", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2, r3})
		eventL1Added1 := addEvent(l1Key1)
		eventL2Added1 := addEvent(l2Key1)
		eventL2Added2 := addEvent(l2Key2)
		eventL3Added1 := addEvent(l3Key1)
		defer setWatchIntervals(watchersyncer.ListRetryInterval, watchersyncer.WatchPollInterval)
		setWatchIntervals(100*time.Millisecond, 500*time.Millisecond)
		By("Syncing a single result for resource 1 and creating the watch")
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, &model.KVPairList{
			Revision: "aabababababa",
			KVPairs: []*model.KVPair{
				{
					Revision: eventL1Added1.New.Revision,
					Key:      eventL1Added1.New.Key,
					Value:    eventL1Added1.New.Value,
				},
			},
		})
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.clientWatchResponse(r1, nil)
		By("Syncing no results for resource 2, failing to create a watch, retrying successfully.")
		rs.clientListResponse(r2, emptyList)
		rs.clientWatchResponse(r2, genError)
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r2, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r2, nil)
		time.Sleep(130 * watchersyncer.WatchPollInterval / 100)
		rs.expectAllEventsHandled()
		By("Sending two watch events for resource 2.")
		rs.sendEvent(r2, eventL2Added1)
		rs.sendEvent(r2, eventL2Added2)
		By("Syncing no results for resource 3, creating a watcher and then terminating the watcher.")
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r3, nil)
		rs.sendEvent(r3, api.WatchEvent{
			Type:  api.WatchError,
			Error: dsError,
		})
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r3, emptyList)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r3, nil)
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r3, nil)
		rs.expectAllEventsHandled()
		By("Validating we are still resyncing and we have received all the current events thus far")
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     *eventL1Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Added2.New,
				UpdateType: api.UpdateTypeKVNew,
			},
		}, true)
		By("Checking that resource 3 can reconnect and then receive events")
		rs.clientListResponse(r3, emptyList)
		rs.clientWatchResponse(r3, nil)
		rs.ExpectStatusUnchanged()
		rs.sendEvent(r3, eventL3Added1)
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     *eventL3Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
		}, true)
	})
	It("Should not resend add events during a resync and should delete stale entries", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1})
		eventL1Added1 := addEvent(l1Key1)
		eventL1Deleted1 := deleteEvent(l1Key1)
		eventL1Added2 := addEvent(l1Key2)
		eventL1Added3 := addEvent(l1Key3)
		eventL1Added4 := addEvent(l1Key4)
		eventL1Modified4 := modifiedEvent(l1Key4)
		eventL1Modified4_2 := modifiedEvent(l1Key4)
		defer setWatchIntervals(watchersyncer.ListRetryInterval, watchersyncer.WatchPollInterval)
		setWatchIntervals(100*time.Millisecond, 500*time.Millisecond)
		By("returning a sync list with three entries and then failing the watch")
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, &model.KVPairList{
			Revision: "12345",
			KVPairs: []*model.KVPair{
				eventL1Added1.New,
				eventL1Added2.New,
				eventL1Added3.New,
			},
		})
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r1, genError)
		time.Sleep(watchersyncer.WatchPollInterval)
		rs.ExpectStatusUnchanged()
		By("returning a sync list with one entry removed and a new one added")
		rs.clientListResponse(r1, &model.KVPairList{
			Revision: "12346",
			KVPairs: []*model.KVPair{
				eventL1Added1.New,
				eventL1Added3.New,
				eventL1Added4.New,
			},
		})
		rs.ExpectStatusUnchanged()
		rs.clientWatchResponse(r1, nil)
		By("Expecting new events for the first three entries followed by an add and then the delete")
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     *eventL1Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL1Added2.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL1Added3.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL1Added4.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair: model.KVPair{
					Key: eventL1Added2.New.Key,
				},
				UpdateType: api.UpdateTypeKVDeleted,
			},
		}, true)
		By("Sending a watch event updating one of the entries and deleting another")
		rs.sendEvent(r1, eventL1Modified4)
		rs.sendEvent(r1, eventL1Deleted1)
		By("Sending the same events (bug) but same revision, so no updates expected")
		rs.sendEvent(r1, eventL1Modified4)
		rs.sendEvent(r1, eventL1Deleted1)
		By("Failing the watch, and resyncing with another modified entry")
		rs.sendEvent(r1, api.WatchEvent{
			Type:  api.WatchError,
			Error: dsError,
		})
		rs.ExpectStatusUnchanged()
		rs.clientListResponse(r1, &model.KVPairList{
			Revision: "12347",
			KVPairs: []*model.KVPair{
				eventL1Added3.New,
				eventL1Modified4_2.New,
			},
		})
		rs.ExpectStatusUnchanged()
		By("Expecting mod, delete, mod updates")
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     *eventL1Modified4.New,
				UpdateType: api.UpdateTypeKVUpdated,
			},
			{
				KVPair: model.KVPair{
					Key: eventL1Deleted1.Old.Key,
				},
				UpdateType: api.UpdateTypeKVDeleted,
			},
			{
				KVPair:     *eventL1Modified4_2.New,
				UpdateType: api.UpdateTypeKVUpdated,
			},
		}, true)
	})
	It("Should accumulate updates into a single update when the handler thread is blocked", func() {
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2})
		eventL1Added1 := addEvent(l1Key1)
		eventL2Added1 := addEvent(l2Key1)
		eventL2Added2 := addEvent(l2Key2)
		eventL2Modified1 := modifiedEvent(l2Key1)
		eventL1Delete1 := deleteEvent(l1Key1)
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.clientWatchResponse(r1, nil)
		rs.clientListResponse(r2, emptyList)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r2, nil)
		rs.BlockUpdateHandling()
		rs.sendEvent(r1, eventL1Added1)
		rs.ExpectOnUpdates([][]api.Update{{
			{
				KVPair:     *eventL1Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
		}})
		rs.sendEvent(r2, eventL2Added1)
		rs.sendEvent(r2, eventL2Added2)
		rs.sendEvent(r2, eventL2Modified1)
		time.Sleep(100 * time.Millisecond)
		rs.UnblockUpdateHandling()
		rs.ExpectOnUpdates([][]api.Update{{
			{
				KVPair:     *eventL2Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Added2.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Modified1.New,
				UpdateType: api.UpdateTypeKVUpdated,
			},
		}})
		rs.BlockUpdateHandling()
		rs.sendEvent(r1, eventL1Delete1)
		rs.ExpectOnUpdates([][]api.Update{{
			{
				KVPair: model.KVPair{
					Key: eventL1Delete1.Old.Key,
				},
				UpdateType: api.UpdateTypeKVDeleted,
			},
		}})
	})
	It("should emit all events when stop is called", func() {
		eventL1Added1 := addEvent(l1Key1)
		eventL2Added1 := addEvent(l2Key1)
		eventL2Added2 := addEvent(l2Key2)
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{r1, r2})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.clientListResponse(r2, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r1, nil)
		rs.clientWatchResponse(r2, nil)
		rs.sendEvent(r1, eventL1Added1)
		rs.sendEvent(r2, eventL2Added1)
		rs.sendEvent(r2, eventL2Added2)
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     *eventL1Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Added1.New,
				UpdateType: api.UpdateTypeKVNew,
			},
			{
				KVPair:     *eventL2Added2.New,
				UpdateType: api.UpdateTypeKVNew,
			},
		}, false)
		rs.watcherSyncer.Stop()
		rs.ExpectUpdates([]api.Update{
			{
				KVPair:     model.KVPair{Key: eventL1Added1.New.Key},
				UpdateType: api.UpdateTypeKVDeleted,
			},
			{
				KVPair:     model.KVPair{Key: eventL2Added1.New.Key},
				UpdateType: api.UpdateTypeKVDeleted,
			},
			{
				KVPair:     model.KVPair{Key: eventL2Added2.New.Key},
				UpdateType: api.UpdateTypeKVDeleted,
			},
		}, false)
	})
	It("Should invoke the supplied converter to alter the update", func() {
		rc1 := watchersyncer.ResourceType{
			UpdateProcessor: &fakeConverter{},
			ListInterface:   model.ResourceListOptions{Kind: apiv3.KindNetworkPolicy},
		}
		rs := newWatcherSyncerTester([]watchersyncer.ResourceType{rc1})
		rs.ExpectStatusUpdate(api.WaitForDatastore)
		rs.clientListResponse(r1, emptyList)
		rs.ExpectStatusUpdate(api.ResyncInProgress)
		rs.ExpectStatusUpdate(api.InSync)
		rs.clientWatchResponse(r1, nil)
		rs.BlockUpdateHandling()
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.ExpectOnUpdates([][]api.Update{{{
			KVPair:     *fakeConverterKVP1,
			UpdateType: api.UpdateTypeKVNew,  
		}}})
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.sendEvent(r1, addEvent(l1Key1))
		rs.sendEvent(r1, addEvent(l1Key1))
		time.Sleep(100 * time.Millisecond)
		rs.UnblockUpdateHandling()
		rs.ExpectOnUpdates([][]api.Update{
			{
				{
					KVPair:     *fakeConverterKVP2,
					UpdateType: api.UpdateTypeKVUpdated,  
				},
				{
					KVPair:     *fakeConverterKVP3,
					UpdateType: api.UpdateTypeKVNew,  
				},
			},
			{
				{
					KVPair: model.KVPair{
						Key: fakeConverterKVP4.Key,
					},
					UpdateType: api.UpdateTypeKVDeleted,  
				},
			},
			{
				{
					KVPair:     *fakeConverterKVP5,
					UpdateType: api.UpdateTypeKVUpdated,  
				},
				{
					KVPair:     *fakeConverterKVP6,
					UpdateType: api.UpdateTypeKVUpdated,  
				},
			},
		})
		rs.ExpectParseError("abcdef", "aabbccdd")
		rs.sendEvent(r1, deleteEvent(l1Key1))
		time.Sleep(100 * time.Millisecond)
		rs.ExpectOnUpdates([][]api.Update{
			{
				{
					KVPair: model.KVPair{
						Key: l1Key1,
					},
					UpdateType: api.UpdateTypeKVDeleted,
				},
			},
		})
		rs.ExpectParseError("zzzzz", "xxxxx")
	})
})
var (
	fakeConverterKVP1 = &model.KVPair{
		Key:      l1Key1,
		Value:    "abcdef",
		Revision: "abcdefg",
	}
	fakeConverterKVP2 = &model.KVPair{
		Key:      l1Key1,
		Value:    "abcdefgh",
		Revision: "abcdfg",
	}
	fakeConverterKVP3 = &model.KVPair{
		Key:      l1Key2,
		Value:    "abcdef",
		Revision: "abcdefg",
	}
	fakeConverterKVP4 = &model.KVPair{
		Key:      l1Key2,
		Revision: "abfdgscdfg",
	}
	fakeConverterKVP5 = &model.KVPair{
		Key:      l1Key1,
		Value:    "abcdeddfgh",
		Revision: "abfdgffffscdfg",
	}
	fakeConverterKVP6 = &model.KVPair{
		Key:      l1Key1,
		Value:    "abcdeddgjdfgjdfgdfgh",
		Revision: "abfdgscdfg",
	}
)
func setWatchIntervals(listRetryInterval, watchPollInterval time.Duration) {
	watchersyncer.ListRetryInterval = listRetryInterval
	watchersyncer.WatchPollInterval = watchPollInterval
}
type fakeConverter struct {
	i int
}
func (fc *fakeConverter) Process(kvp *model.KVPair) ([]*model.KVPair, error) {
	if kvp.Value == nil {
		return []*model.KVPair{
				{
					Key: l1Key1,
				},
				{
					Key: l1Key2,
				},
			}, cerrors.ErrorParsingDatastoreEntry{
				RawKey:   "zzzzz",
				RawValue: "xxxxx",
			}
	}
	fc.i++
	switch fc.i {
	case 1:  
		return []*model.KVPair{
			fakeConverterKVP1,
		}, nil
	case 2:  
		return []*model.KVPair{
			fakeConverterKVP2,
			fakeConverterKVP3,
		}, nil
	case 3:  
		return nil, errors.New("Fake error that we should handle gracefully")
	case 4:  
		return []*model.KVPair{
				fakeConverterKVP4,
			}, cerrors.ErrorParsingDatastoreEntry{
				RawKey:   "abcdef",
				RawValue: "aabbccdd",
			}
	case 5:  
		return []*model.KVPair{
			fakeConverterKVP5,
		}, nil
	case 6:  
		return nil, nil
	case 7:  
		return []*model.KVPair{
			fakeConverterKVP6,
		}, nil
	}
	return nil, nil
}
func (fc *fakeConverter) OnSyncerStarting() {
}
func deleteEvent(key model.Key) api.WatchEvent {
	return api.WatchEvent{
		Type: api.WatchDeleted,
		Old: &model.KVPair{
			Key:      key,
			Value:    uuid.NewString(),
			Revision: uuid.NewString(),
		},
	}
}
func addEvent(key model.Key) api.WatchEvent {
	return api.WatchEvent{
		Type: api.WatchAdded,
		New: &model.KVPair{
			Key:      key,
			Value:    uuid.NewString(),
			Revision: uuid.NewString(),
		},
	}
}
func modifiedEvent(key model.Key) api.WatchEvent {
	return api.WatchEvent{
		Type: api.WatchModified,
		New: &model.KVPair{
			Key:      key,
			Value:    uuid.NewString(),
			Revision: uuid.NewString(),
		},
	}
}
func newWatcherSyncerTester(l []watchersyncer.ResourceType) *watcherSyncerTester {
	lws := map[string]*listWatchSource{}
	for _, r := range l {
		name := model.ListOptionsToDefaultPathRoot(r.ListInterface)
		lws[name] = &listWatchSource{
			name:            name,
			watchCallError:  make(chan error, 50),
			listCallResults: make(chan interface{}, 200),
			stopEvents:      make(chan struct{}, 200),
			results:         make(chan api.WatchEvent, 200),
		}
	}
	fc := &fakeClient{
		lws: lws,
	}
	st := testutils.NewSyncerTester()
	rst := &watcherSyncerTester{
		SyncerTester:  st,
		fc:            fc,
		watcherSyncer: watchersyncer.New(fc, l, st),
		lws:           lws,
	}
	rst.watcherSyncer.Start()
	return rst
}
type watcherSyncerTester struct {
	*testutils.SyncerTester
	fc            *fakeClient
	lws           map[string]*listWatchSource
	watcherSyncer api.Syncer
}
func (rst *watcherSyncerTester) expectAllEventsHandled() {
	log.Infof("Expecting all events to have been handled")
	for _, l := range rst.lws {
		Expect(l.listCallResults).To(HaveLen(0), "pending list results to be processed")
		Expect(l.stopEvents).To(HaveLen(0), "pending stop events to be processed")
		Expect(l.results).To(HaveLen(0), "pending watch results to be processed")
	}
}
func (rst *watcherSyncerTester) allEventsHandled() bool {
	eventsHandled := true
	for _, l := range rst.lws {
		eventsHandled = eventsHandled && (len(l.listCallResults) == 0)
		eventsHandled = eventsHandled && (len(l.stopEvents) == 0)
		eventsHandled = eventsHandled && (len(l.results) == 0)
	}
	return eventsHandled
}
func (rst *watcherSyncerTester) sendEvent(r watchersyncer.ResourceType, event api.WatchEvent) {
	name := model.ListOptionsToDefaultPathRoot(r.ListInterface)
	log.WithField("Name", name).Infof("Sending event")
	log.Info("Waiting for previous watcher to terminate (if any)")
	rst.lws[name].termWg.Wait()
	log.Info("Previous watcher terminated (if any)")
	if event.Type == api.WatchError {
		log.WithField("Name", name).Info("Watcher error will trigger restart - increment termination count")
		rst.lws[name].termWg.Add(1)
	}
	log.WithField("Name", name).Info("Sending event")
	rst.lws[name].results <- event
	if event.Type == api.WatchError {
		log.WithField("Name", name).Info("Expecting a stop invocation")
		rst.expectStop(r)
		log.WithField("Name", name).Info("Stop invoked")
	}
}