// expect the decorated channel to be closed
	select {
	case e, ok := <-dw.ResultChan():
		if ok {
			t.Errorf("expected result chan closed, got %#v", e)
		}
	case <-time.After(wait.ForeverTestTimeout):
		t.Errorf("timeout after %v", wait.ForeverTestTimeout)
	}

	// expect the underlying watcher to have been stopped as a result of the context cancellation
	if !w.IsStopped() {
		t.Errorf("expected underlying watcher to be stopped")
	}
}

	select {
	case res := <-w.ResultChan():
		if res.Type != expectEventType {
			t.Errorf("event type want=%v, get=%v", expectEventType, res.Type)
		}
	case <-time.After(wait.ForeverTestTimeout):
		t.Errorf("time out after waiting %v on ResultChan", wait.ForeverTestTimeout)
	}
}

func testCheckResult(t *testing.T, w watch.Interface, expectEvent watch.Event) {
	testCheckResultFunc(t, w, func(actualEvent watch.Event) {

		events[i] = &auditinternal.Event{}
	}

	return events
}

func testBatchConfig() BatchConfig {
	return BatchConfig{
		BufferSize:     100,
		MaxBatchSize:   10,
		MaxBatchWait:   wait.ForeverTestTimeout,
		ThrottleEnable: false,
		AsyncDelegate:  true,
	}
}

func TestBatchedBackendCollectEvents(t *testing.T) {
	config := testBatchConfig()
	batchSize := config.MaxBatchSize

			require.NoError(t, err)
			req, err := NewProbeRequest(target, nil)
			require.NoError(t, err)
			result, _, _ := prober.Probe(req, wait.ForeverTestTimeout)
			assert.Equal(t, test.expectLocalResult, result)
		})
		t.Run(desc+"-nonlocal", func(t *testing.T) {
			followNonLocalRedirects := true
			prober := New(followNonLocalRedirects)

	updateCRD()

	drained := make(chan struct{})
	go func() {
		defer close(drained)
		wg.Wait()
	}()

	select {
	case <-drained:
	case <-time.After(wait.ForeverTestTimeout):
		t.Fatal("timed out waiting for watchers to be terminated")
	}

	stopChan := make(chan struct{})

	// Set up loop to modify CRD in the background
	wg.Add(1)
	go func() {
		defer wg.Done()

				err := wait.Poll(10*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {
					return len(ctrl.claims.ListKeys()) == 0, nil
				})
				if err != nil {
					return err
				}
				// wait for volume delete operation to appear once volumeWorker() runs
				return wait.PollImmediate(10*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {

func waitForBase64PluginToBeUp(plugin *Base64Plugin) error {
	var gRPCErr error
	var resp *kmsapi.StatusResponse
	pollErr := wait.PollImmediate(1*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
		resp, gRPCErr = plugin.Status(context.Background(), &kmsapi.StatusRequest{})
		return gRPCErr == nil && resp.Healthz == "ok", nil
	})

	if pollErr != nil {

	addDefault("v1beta2", "c", "C")

	t.Logf("wait until GET sees 'c' in both status and spec")
	if err := wait.PollImmediate(100*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {
		obj, err := fooClient.Get(context.TODO(), foo.GetName(), metav1.GetOptions{})
		if err != nil {
			return false, err
		}

	go func() {
		// Start the streaming on the WebSocket "exec" client.
		errorChan <- exec.StreamWithContext(context.Background(), *options)
	}()

	select {
	case <-time.After(wait.ForeverTestTimeout):
		t.Fatalf("expect stream to be closed after connection is closed.")
	case err := <-errorChan:
		if err != nil {
			t.Errorf("unexpected error: %v", err)
		}
	}

	epsConfig.RegisterEventHandler(epsHandler)

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

	err := wait.PollImmediate(time.Millisecond*10, wait.ForeverTestTimeout, func() (bool, error) {
		svcHandler.lock.Lock()
		defer svcHandler.lock.Unlock()
		if reflect.DeepEqual(svcHandler.state, expectedSvcState) {
			return true, nil
		}
		return false, nil
	})