return nil
		}
		success.Inc()
		return nil
	})
	go queue.Run(test.NewStop(t))
	retry.UntilOrFail(t, queue.HasSynced)
	queue.Add(types.NamespacedName{Name: "success"})
	retry.UntilOrFail(t, func() bool { return success.Load() == 1 })
	queue.Add(types.NamespacedName{Name: "conflict-success"})
	retry.UntilOrFail(t, func() bool { return success.Load() == 2 && retries.Load() == 5 })

			t.Fatal(err)
		}
		retry.UntilOrFail(t, func() bool { return w.Mesh().GetIngressClass() == "core" }, retry.Delay(time.Millisecond), retry.Timeout(time.Second))
	})
	t.Run("only user", func(t *testing.T) {
		cms, w := setup(t)
		if _, err := cms.Create(context.Background(), cmUser, metav1.CreateOptions{}); err != nil {
			t.Fatal(err)
		}

			return getErrorMessage(options)
		}
		return nil
	}, options...)
}

// UntilOrFail calls Until, and fails t with Fatalf if it ends up returning an error
func UntilOrFail(t test.Failer, fn func() bool, options ...Option) {
	t.Helper()
	err := Until(fn, options...)
	if err != nil {
		t.Fatalf("retry.UntilOrFail: %v", err)
	}
}

func getErrorMessage(options []Option) error {
	cfg := defaultConfig

		for _, override := range configOverrides {
			override(fakeRestConfig)
		}
		return kube.NewFakeClient(), nil
	}
	client.RunAndWait(stopCh)
	assert.NoError(t, c.Run(stopCh))
	retry.UntilOrFail(t, c.HasSynced, retry.Timeout(2*time.Second))
	secret0 := makeSecret(secretNamespace, "s0", clusterCredential{"c0", []byte("kubeconfig0-0")})
	secrets := clienttest.NewWriter[*v1.Secret](t, client)

		meshWatcher,
		stop,
	)
	kube.SetObjectFilter(client, discoveryNamespacesFilter)
	nc := NewNamespaceController(client, watcher)
	client.RunAndWait(stop)
	go nc.Run(stop)
	retry.UntilOrFail(t, nc.queue.HasSynced)

	expectedData := map[string]string{
		constants.CACertNamespaceConfigMapDataName: string(caBundle),
	}
	createNamespace(t, client.Kube(), "foo", nil)

	ec := c.exports.(*serviceExportCacheImpl)
	// Wait for the resources to be processed by the controller.
	retry.UntilOrFail(t, func() bool {
		if svc := ec.GetService(ec.serviceHostname()); svc == nil {
			return false
		}
		inst := ec.getEndpoint(c.Endpoints)
		return inst != nil
	}, serviceExportTimeout)
	return ec, c.Endpoints

			t.Fatalf("multi network is not enabled")
		}
	})

	workingDNSServer.setFailure(false)
	t.Run("resolution recovered", func(t *testing.T) {
		// addresses should be updated
		retry.UntilOrFail(t, func() bool {
			return !reflect.DeepEqual(env.NetworkManager.AllGateways(), gateways)
		}, retry.Timeout(10*model.MinGatewayTTL), retry.Delay(time.Millisecond*10))
		xdsUpdater.WaitOrFail(t, "xds full")
	})

// Ensure that the client can run without CRDs present
func TestClientNoCRDs(t *testing.T) {
	schema := collection.NewSchemasBuilder().MustAdd(collections.Sidecar).Build()
	store, _ := makeClient(t, schema, nil)
	retry.UntilOrFail(t, store.HasSynced, retry.Timeout(time.Second))
	r := collections.VirtualService
	configMeta := config.Meta{
		Name:             "name",
		Namespace:        "ns",
		GroupVersionKind: r.GroupVersionKind(),
	}

	handles := atomic.NewInt32(0)
	task := func() error {
		handles.Inc()
		return nil
	}
	q := NewQueue(0)
	q.Push(task)
	stop := make(chan struct{})
	go q.Run(stop)
	retry.UntilOrFail(t, q.HasSynced, retry.Delay(time.Microsecond))
	// Must always be 1 since we are synced
	assert.Equal(t, handles.Load(), 1)
	close(stop)
	assert.NoError(t, WaitForClose(q, time.Second))

		enabled:        true,
	}
	l.AddRunFunction(func(stop <-chan struct{}) {
		<-stop
	})
	for _, fn := range fns {
		l.AddRunFunction(fn)
	}
	stop := make(chan struct{})
	go l.Run(stop)

	retry.UntilOrFail(t, func() bool {
		return l.isLeader() == expectLeader
	}, retry.Converge(5), retry.Delay(time.Millisecond*100), retry.Timeout(time.Second*10))
	return l, stop
}

type fakeDefaultWatcher struct {