@@ -4,8 +4,8 @@
 package activemq_test
 
 import (
+	"context"
 	"encoding/base64"
-	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -475,19 +475,35 @@ func setupActiveMQ(t *testing.T, kc *kubernetes.Clientset) {
 	require.NoErrorf(t, err, "%s", err)
 }
 
+// The StatefulSet reports the broker ready as soon as its container is up, which is before the
+// broker has finished starting, so readiness is established by asking the broker's health MBean.
+// Ten minutes is the budget the previous 60 x 10s loop had; the broker usually answers Good well
+// inside the first minute, and the wait ends as soon as it does.
+const (
+	activeMQReadyTimeout  = 10 * time.Minute
+	activeMQReadyInterval = 10 * time.Second
+)
+
 func checkIfActiveMQStatusIsReady(t *testing.T, name string) error {
 	t.Log("--- checking activemq status ---")
-	time.Sleep(time.Second * 10)
-	for i := 0; i < 60; i++ {
-		out, errOut, _ := ExecCommandOnSpecificPod(t, name, testNamespace, fmt.Sprintf("%s query –objname type=Broker,brokerName=localhost,Service=Health", activemqPath))
+
+	ctx, cancel := context.WithTimeout(context.Background(), activeMQReadyTimeout)
+	defer cancel()
+
+	// A failed exec is handed back rather than swallowed, so that a broker which never answers
+	// is reported with the reason instead of as a bare timeout.
+	err := KedaEventually(ctx, func(_ context.Context) (bool, error) {
+		out, errOut, err := ExecCommandOnSpecificPod(t, name, testNamespace, fmt.Sprintf("%s query –objname type=Broker,brokerName=localhost,Service=Health", activemqPath))
 		t.Logf("Output: %s, Error: %s", out, errOut)
-		if !strings.Contains(out, "CurrentStatus = Good") {
-			time.Sleep(time.Second * 10)
-			continue
+		if err != nil {
+			return false, fmt.Errorf("cannot query activemq health: %w", err)
 		}
-		return nil
+		return strings.Contains(out, "CurrentStatus = Good"), nil
+	}, activeMQReadyInterval)
+	if err != nil {
+		return fmt.Errorf("activemq is not ready: %w", err)
 	}
-	return errors.New("activemq is not ready")
+	return nil
 }
 
 func testActivation(t *testing.T, kc *kubernetes.Clientset) {
 
@@ -9,7 +9,6 @@ import (
 	"fmt"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/joho/godotenv"
@@ -165,8 +164,10 @@ func TestCouchDBScaler(t *testing.T) {
 	data, templates := getTemplateData(t, kc)
 	KubectlApplyMultipleWithTemplate(t, data, templates)
 
-	// wait until client is ready
-	time.Sleep(10 * time.Second)
+	// The database is created through the client pod, which has no readiness probe, so Ready means its
+	// container is running and exec will reach it. A minute covers pulling the curl image on a cold node.
+	require.True(t, WaitForPodReady(t, kc, clientName, testNamespace, 60, 1),
+		"client pod %s should be ready", clientName)
 	// create database
 	_, _, err := ExecCommandOnSpecificPod(t, clientName, testNamespace, fmt.Sprintf("curl -X PUT http://admin:%s@test-release-svc-couchdb.%s.svc.cluster.local:5984/animals", getPassword(t, kc), testNamespace))
 	require.NoErrorf(t, err, "cannot execute command - %s", err)
 
@@ -4,6 +4,7 @@
 package elastic_forecast_test
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -402,23 +403,29 @@ func activateTrialAndVerify(t *testing.T) {
 // requireDatafeedStopped polls the datafeed status until it reports "stopped", which means the bounded historical run has completed.
 func requireDatafeedStopped(t *testing.T, timeout time.Duration) {
 	t.Helper()
-	deadline := time.Now().Add(timeout)
-	for time.Now().Before(deadline) {
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	// A failed exec is returned rather than skipped, so a datafeed that never reports is failed
+	// with the reason instead of a bare timeout.
+	err := KedaEventually(ctx, func(_ context.Context) (bool, error) {
 		out, err := ExecuteCommand(fmt.Sprintf(
 			"%s -XGET http://127.0.0.1:9200/_ml/datafeeds/datafeed-%s/_stats",
 			forecastKubectlExecCmd, forecastJobID,
 		))
-		if err == nil {
-			body := string(out)
-			if strings.Contains(body, `"state":"stopped"`) {
-				t.Log("[datafeed] state is stopped — historical processing complete")
-				return
-			}
-			t.Logf("[datafeed] not yet stopped, waiting… (body: %s)", body)
+		if err != nil {
+			return false, fmt.Errorf("cannot read datafeed stats: %w", err)
 		}
-		time.Sleep(3 * time.Second)
-	}
-	require.Fail(t, fmt.Sprintf("datafeed-%s did not reach state 'stopped' within %s", forecastJobID, timeout))
+		body := string(out)
+		if strings.Contains(body, `"state":"stopped"`) {
+			t.Log("[datafeed] state is stopped — historical processing complete")
+			return true, nil
+		}
+		t.Logf("[datafeed] not yet stopped, waiting… (body: %s)", body)
+		return false, nil
+	}, 3*time.Second)
+	require.NoErrorf(t, err, "datafeed-%s did not reach state 'stopped' within %s", forecastJobID, timeout)
 }
 
 // requireJobHasBuckets asserts that the ML job has processed enough result buckets for the model to produce a reliable forecast.
@@ -467,26 +474,35 @@ func ingestForecastTrainingData(t *testing.T, count int, window time.Duration) {
 func waitForForecastDocs(t *testing.T) {
 	t.Helper()
 
-	deadline := time.Now().Add(3 * time.Minute)
-	for time.Now().Before(deadline) {
+	const timeout = 3 * time.Minute
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	// A failed exec or an unparseable body is returned rather than skipped, so a count that never
+	// comes back is failed with the reason instead of a bare timeout.
+	err := KedaEventually(ctx, func(_ context.Context) (bool, error) {
 		out, err := ExecuteCommand(fmt.Sprintf(
 			`%s -XGET 'http://127.0.0.1:9200/.ml-anomalies-shared/_count' -d '{"query":{"bool":{"filter":[{"term":{"job_id":"%s"}},{"term":{"result_type":"model_forecast"}}]}}}'`,
 			forecastKubectlExecCmd, forecastJobID,
 		))
-		if err == nil {
-			body := string(out)
-			var countResp struct {
-				Count int64 `json:"count"`
-			}
-			if jsonErr := json.Unmarshal([]byte(body), &countResp); jsonErr == nil && countResp.Count > 0 {
-				t.Logf("[forecast] %d forecast document(s) indexed — proceeding", countResp.Count)
-				return
-			}
-			t.Logf("[forecast] 0 documents yet, waiting... (response: %s)", body)
+		if err != nil {
+			return false, fmt.Errorf("cannot count forecast documents: %w", err)
 		}
-		time.Sleep(5 * time.Second)
-	}
-	require.Fail(t, "forecast documents did not appear in .ml-anomalies-shared within 3 minutes")
+		body := string(out)
+		var countResp struct {
+			Count int64 `json:"count"`
+		}
+		if jsonErr := json.Unmarshal([]byte(body), &countResp); jsonErr != nil {
+			return false, fmt.Errorf("cannot parse count response: %w (body: %s)", jsonErr, body)
+		}
+		if countResp.Count > 0 {
+			t.Logf("[forecast] %d forecast document(s) indexed — proceeding", countResp.Count)
+			return true, nil
+		}
+		t.Logf("[forecast] 0 documents yet, waiting... (response: %s)", body)
+		return false, nil
+	}, 5*time.Second)
+	require.NoErrorf(t, err, "forecast documents did not appear in .ml-anomalies-shared within %s", timeout)
 }
 
 func testElasticForecastScaler(t *testing.T, kc *kubernetes.Clientset) {
 
@@ -266,50 +266,55 @@ func getSplitArrayFromAverage(t *testing.T, average, splitInNValues int) []int {
 	return result
 }
 
-func getUpdateUrlsForAllMetricAllMetricsServerReplicas(t *testing.T, kc *kubernetes.Clientset, expectedAverageMetric int, nbReplicasForMetricsServer int, nbRetry int) map[string]string {
-	nbRetriesMax := 5
+// Nothing else waits for the metrics server's Deployment, so this is where the test first depends on
+// its pods being scheduled and running: the budget has to absorb image pull and start-up for up to ten
+// replicas on a cold node. The previous retry allowed about 25 seconds.
+const metricsServerPodsTimeout = 2 * time.Minute
+
+func getUpdateUrlsForAllMetricAllMetricsServerReplicas(t *testing.T, kc *kubernetes.Clientset, expectedAverageMetric int, nbReplicasForMetricsServer int) map[string]string {
 	// get an array for which all elements' average would give expectedAverageMetric without any of its elements being exactly expectedAverageMetric
 	individualMetrics := getSplitArrayFromAverage(t, expectedAverageMetric, nbReplicasForMetricsServer)
-	// use kc to curl all metrics-server replicas
-	// Get pods with the specified label selector
-	pods, err := kc.CoreV1().Pods(testNamespace).List(context.Background(), metav1.ListOptions{
-		LabelSelector: "app=" + metricsServerDeploymentName,
-	})
-	if err != nil {
-		t.Fatalf("Error listing pods: %v", err)
-		return nil
-	}
 
-	retryFunc := func(message string) map[string]string {
-		if nbRetry >= nbRetriesMax {
-			t.Fatal(message)
-			return nil
+	ctx, cancel := context.WithTimeout(context.Background(), metricsServerPodsTimeout)
+	defer cancel()
+
+	// Every metrics-server replica has to be running with an IP before any of them is posted to,
+	// because the values are split so that only their average equals the expected metric: posting
+	// to a subset would give the scaler a different average. The reason a check fell short is
+	// returned as the error, so a timeout says which pod was not ready rather than just that one was not.
+	var postUrls map[string]string
+	err := KedaEventually(ctx, func(ctx context.Context) (bool, error) {
+		pods, err := kc.CoreV1().Pods(testNamespace).List(ctx, metav1.ListOptions{
+			LabelSelector: "app=" + metricsServerDeploymentName,
+		})
+		if err != nil {
+			return false, fmt.Errorf("cannot list metrics server pods: %w", err)
 		}
-		t.Logf("%s. Retry calling getUpdateUrlsForAllMetricAllMetricsServerReplicas() after 1 second", message)
-		time.Sleep(5 * time.Second)
-		return getUpdateUrlsForAllMetricAllMetricsServerReplicas(t, kc, expectedAverageMetric, nbReplicasForMetricsServer, nbRetry+1)
-	}
-	if len(pods.Items) == 0 {
-		return retryFunc("No pods found with the given selector.")
-	}
 
-	if len(pods.Items) != nbReplicasForMetricsServer {
-		return retryFunc(fmt.Sprintf("Number of replicas of metrics server (%d) does not match expected value (%d).", len(pods.Items), nbReplicasForMetricsServer))
-	}
-	postUrls := make(map[string]string, nbReplicasForMetricsServer)
-	// Iterate through the pods and send HTTP requests
-	for nbPod, pod := range pods.Items {
-		if pod.Status.Phase != v1.PodRunning || pod.Status.PodIP == "" {
-			return retryFunc(fmt.Sprintf("Pod %s was expected to be running and to have an IP.", pod.Name))
+		if len(pods.Items) != nbReplicasForMetricsServer {
+			return false, fmt.Errorf("metrics server has %d pods, expected %d", len(pods.Items), nbReplicasForMetricsServer)
+		}
+
+		urls := make(map[string]string, nbReplicasForMetricsServer)
+		for nbPod, pod := range pods.Items {
+			if pod.Status.Phase != v1.PodRunning || pod.Status.PodIP == "" {
+				return false, fmt.Errorf("pod %s was expected to be running and to have an IP (phase %s, ip %q)", pod.Name, pod.Status.Phase, pod.Status.PodIP)
+			}
+			urls[pod.Name] = fmt.Sprintf("http://%s:8080/api/value/%d", pod.Status.PodIP, individualMetrics[nbPod])
 		}
-		url := fmt.Sprintf("http://%s:8080/api/value/%d", pod.Status.PodIP, individualMetrics[nbPod])
-		postUrls[pod.Name] = url
+		postUrls = urls
+		return true, nil
+	}, IntervalShort)
+	if err != nil {
+		t.Fatalf("metrics server replicas were not all running within %s: %v", metricsServerPodsTimeout, err)
+		return nil
 	}
+
 	return postUrls
 }
 
 func updateAllMetricsServerReplicas(t *testing.T, kc *kubernetes.Clientset, data templateData, metricValue int, nbReplicasForMetricsServer int) {
-	for targetPodName, urlToPost := range getUpdateUrlsForAllMetricAllMetricsServerReplicas(t, kc, metricValue, nbReplicasForMetricsServer, 0) {
+	for targetPodName, urlToPost := range getUpdateUrlsForAllMetricAllMetricsServerReplicas(t, kc, metricValue, nbReplicasForMetricsServer) {
 		if urlToPost == "" {
 			t.Fatalf("target pod %s should have non emoty url but got one", targetPodName)
 			return
 
@@ -4,8 +4,8 @@
 package solr_test
 
 import (
+	"context"
 	"encoding/base64"
-	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -226,19 +226,34 @@ func setupSolr(t *testing.T, kc *kubernetes.Clientset) {
 	t.Log("--- solr is ready ---")
 }
 
+// Solr is asked directly whether it is serving, because the StatefulSet reports ready before the
+// server does, and again after the in-place restart that enables BasicAuth, which the StatefulSet
+// does not see at all. One minute is the budget the previous 12 x 5s loop had.
+const (
+	solrReadyTimeout  = time.Minute
+	solrReadyInterval = 5 * time.Second
+)
+
 func checkIfSolrStatusIsReady(t *testing.T, name string) error {
 	t.Log("--- checking solr status ---")
 
-	for i := 0; i < 12; i++ {
-		out, errOut, _ := ExecCommandOnSpecificPod(t, name, testNamespace, fmt.Sprintf("%s status", solrPath))
+	ctx, cancel := context.WithTimeout(context.Background(), solrReadyTimeout)
+	defer cancel()
+
+	// A failed exec is handed back rather than swallowed, so that a server which never answers
+	// is reported with the reason instead of as a bare timeout.
+	err := KedaEventually(ctx, func(_ context.Context) (bool, error) {
+		out, errOut, err := ExecCommandOnSpecificPod(t, name, testNamespace, fmt.Sprintf("%s status", solrPath))
 		t.Logf("Output: %s, Error: %s", out, errOut)
-		if !strings.Contains(out, "running on port") {
-			time.Sleep(time.Second * 5)
-			continue
+		if err != nil {
+			return false, fmt.Errorf("cannot query solr status: %w", err)
 		}
-		return nil
+		return strings.Contains(out, "running on port"), nil
+	}, solrReadyInterval)
+	if err != nil {
+		return fmt.Errorf("solr is not ready: %w", err)
 	}
-	return errors.New("solr is not ready")
+	return nil
 }
 
 func getTemplateData() (templateData, []Template) {