stopCh := ctx.Done()
			controllerCompletionCh := make(chan error)

			informerFactory.Start(stopCh)

			status := informerFactory.WaitForCacheSync(ctx.Done())
			if names := unsynced(status); len(names) > 0 {
				t.Fatalf("WaitForCacheSync did not successfully complete, resources=%#v", names)
			}

			go func() {
				controllerCompletionCh <- controller.Run(stopCh)
			}()

						NodeName:           p.node,
					},
				})
			}
			c := kube.NewFakeClient(pods...)
			na := NewClusterNodeAuthorizer(c, tt.trustedAccounts)
			c.RunAndWait(test.NewStop(t))
			kube.WaitForCacheSync("test", test.NewStop(t), na.pods.HasSynced)

			err := na.authenticateImpersonation(tt.caller, tt.requestedIdentityString)
			if tt.wantErr == "" && err != nil {
				t.Fatalf("wanted no error, got %v", err)
			}

		{deleted: cm1, expectCalled: true},
		{deleted: cm2},
	}

	stop := make(chan struct{})
	c := NewController(client, configMapNamespace, configMapName, callback)
	go c.Run(stop)
	kube.WaitForCacheSync("test", stop, c.HasSynced)

	cms := client.Kube().CoreV1().ConfigMaps(configMapNamespace)
	for i, step := range steps {
		resetCalled()

		t.Run(fmt.Sprintf("[%v]", i), func(t *testing.T) {

	p.webhooks.AddEventHandler(controllers.FilteredObjectHandler(p.queue.AddObject, isDefaultTagWebhook))

	return p
}

func (p *defaultWatcher) Run(stopCh <-chan struct{}) {
	kube.WaitForCacheSync("default revision", stopCh, p.webhooks.HasSynced)
	p.queue.Run(stopCh)
}

// GetDefault returns the current default revision.
func (p *defaultWatcher) GetDefault() string {
	p.mu.RLock()
	defer p.mu.RUnlock()

			if rev == "" {
				return nil
			}
			return []string{rev}
		}, controllers.ObjectHandler(p.queue.AddObject))
	return p
}

func (p *tagWatcher) Run(stopCh <-chan struct{}) {
	if !kube.WaitForCacheSync("tag watcher", stopCh, p.webhooks.HasSynced) {
		return
	}
	// Notify handlers of initial state
	p.notifyHandlers()
	p.queue.Run(stopCh)
}

// AddHandler registers a new handler for updates to tag changes.

			assert.NoError(t, err)
			t.Cleanup(func() {
				assert.NoError(t, c.queue.WaitForClose(time.Second))
			})
			stop := test.NewStop(t)
			tt.client.RunAndWait(stop)
			go c.Run(stop)
			kube.WaitForCacheSync("test", stop, c.queue.HasSynced)

			assert.EventuallyEqual(t, func() map[string]string {
				havePods := c.pods.List(metav1.NamespaceAll, klabels.Everything())
				slices.SortBy(havePods, func(a *corev1.Pod) string {

	// First, wait for pods to sync. Once this is complete, we know the event handler for Pods will have
	// ran for each item and enqueued everything.
	kube.WaitForCacheSync("pod controller", stop, c.pods.HasSynced)

	// Now we can run the queue. This will block until `stop` is closed.
	c.queue.Run(stop)

			}
		}
		clienttest.MakeCRDWithAnnotations(t, fake, s.GroupVersionResource(), annotations)
	}
	stop := test.NewStop(t)
	config := New(fake, Option{})
	go config.Run(stop)
	fake.RunAndWait(stop)
	kube.WaitForCacheSync("test", stop, config.HasSynced)
	return config, fake
}

func createResource(t *testing.T, store model.ConfigStoreController, r resource.Schema, configMeta config.Meta) config.Spec {
	pb, err := r.NewInstance()

		mu.Lock()
		defer mu.Unlock()
		newConfig = config
		newValues = values
		return nil
	})
	stop := test.NewStop(t)
	go w.Run(stop)
	controller := w.(*configMapWatcher).c
	client.RunAndWait(stop)
	kube.WaitForCacheSync("test", stop, controller.HasSynced)

	cms := client.Kube().CoreV1().ConfigMaps(namespace)
	steps := []struct {
		added   *v1.ConfigMap
		updated *v1.ConfigMap
		deleted *v1.ConfigMap
	}{
		{added: cm},

	}
	opts.Client.RunAndWait(c.stop)
	var fx *xdsfake.Updater
	if x, ok := xdsUpdater.(*xdsfake.Updater); ok {
		fx = x
	}

	if !opts.SkipRun {
		go c.Run(c.stop)
		kubelib.WaitForCacheSync("test", c.stop, c.HasSynced)
	}

	return &FakeController{Controller: c, Endpoints: endpoints}, fx