SimpleEndpoints := SimpleEndpointsCollection(SimplePods, SimpleServices)
	assert.Equal(t, SimpleEndpoints.Synced().WaitUntilSynced(stop), true)
	// Assert Equal -- not EventuallyEqual -- to ensure our WaitForCacheSync is proper
	assert.Equal(t, fetcherSorted(SimpleEndpoints)(), []SimpleEndpoint{{"pod", "svc", "namespace", "1.2.3.4"}})
}

func TestCollectionMerged(t *testing.T) {
	c := kube.NewFakeClient()

	if cl.started.Swap(true) {
		// was already started by other thread
		return
	}

	t0 := time.Now()
	cl.logger.Infof("Starting Pilot K8S CRD controller")

	if !kube.WaitForCacheSync("crdclient", stop, cl.informerSynced) {
		cl.logger.Errorf("Failed to sync Pilot K8S CRD controller cache")
		return
	}
	cl.logger.Infof("Pilot K8S CRD controller synced in %v", time.Since(t0))
	cl.queue.Run(stop)

	return f.csiDriverLister
}

func (f *fakeKubeletVolumeHost) CSIDriversSynced() cache.InformerSynced {
	// not needed for testing
	return nil
}

func (f *fakeKubeletVolumeHost) WaitForCacheSync() error {
	return nil
}

func (f *fakeKubeletVolumeHost) GetHostUtil() hostutil.HostUtils {
	return f.hostUtil
}

				frameworkruntime.WithLogger(logger),
			)
			if err != nil {
				t.Fatal(err)
			}

			informerFactory.Start(ctx.Done())
			informerFactory.WaitForCacheSync(ctx.Done())
			snapshot := internalcache.NewSnapshot(tt.initPods, tt.nodes)
			nodeInfos, err := snapshot.NodeInfos().List()
			if err != nil {
				t.Fatal(err)
			}

	// Now mark the config cluster as synced
	component.All()[0].Synced.Store(true)
	t.Run("sync timeout", func(t *testing.T) {
		retry.UntilOrFail(t, c.HasSynced, retry.Timeout(2*time.Second))
	})
	kube.WaitForCacheSync("test", stopCh, c.HasSynced)

	for _, step := range steps {
		t.Run(step.name, func(t *testing.T) {
			switch {
			case step.add != nil:
				secrets.Create(step.add)
			case step.update != nil:

		_ = listener.Close()
	})
	// Start the discovery server
	s.Start(stop)
	cg.ServiceEntryRegistry.XdsUpdater = s
	// Now that handlers are added, get everything started
	cg.Run()
	kubelib.WaitForCacheSync("fake", stop,
		cg.Registry.HasSynced,
		cg.Store().HasSynced)
	cg.ServiceEntryRegistry.ResyncEDS()

	// Send an update. This ensures that even if there are no configs provided, the push context is

			volumeAttachmentInformer := factory.Storage().V1().VolumeAttachments()
			if driverInfo != nil {
				csiDriverInformer.Informer().GetStore().Add(driverInfo)
			}

			factory.Start(wait.NeverStop)
			factory.WaitForCacheSync(wait.NeverStop)

			attachDetachVolumeHost := volumetest.NewFakeAttachDetachVolumeHostWithCSINodeName(t,
				tmpDir,
				client,
				ProbeVolumePlugins(),
				"fakeNode",
				csiDriverInformer.Lister(),

			// Ensure informers are up-to-date.
			informerFactory.Start(ctx.Done())
			stopInformers := func() {
				cancel()
				informerFactory.Shutdown()
			}
			defer stopInformers()
			informerFactory.WaitForCacheSync(ctx.Done())

			// Add claims that only exist in the mutation cache.
			for _, claim := range tc.claimsInCache {
				ec.claimCache.Mutation(claim)
			}

		for _, rs := range test.oldRSs {
			informers.Apps().V1().ReplicaSets().Informer().GetIndexer().Add(rs)
		}

		stopCh := make(chan struct{})
		defer close(stopCh)
		informers.Start(stopCh)
		informers.WaitForCacheSync(stopCh)

		t.Logf(" &test.revisionHistoryLimit: %d", test.revisionHistoryLimit)
		d := newDeployment("foo", 1, &test.revisionHistoryLimit, nil, nil, map[string]string{"foo": "bar"})

	result.SkippedAnalyzers = a.RemoveSkipped(schemas)
	result.MappedMessages = make(map[string]diag.Messages, len(result.ExecutedAnalyzers))

	for _, store := range sa.multiClusterStores {
		kubelib.WaitForCacheSync("istiod analyzer", cancel, store.HasSynced)
	}

	stores := map[cluster.ID]model.ConfigStore{}
	for k, v := range sa.multiClusterStores {
		stores[k] = v
	}