working on replacing 'wait.Poll()' with 'Watch()' in e2e tests

go.sum

@@ -22,7 +22,7 @@ github.com/coreos/bbolt v1.3.2 h1:wZwiHHUieZCquLkDL0B8UhzreNWsPHooDAG3q34zk0s=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.9+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
-github.com/coreos/etcd v3.3.11+incompatible h1:U0wJghY374q+UrjOM2mfROHSwEspsQVkCACB1PGka1g=
+github.com/coreos/etcd v3.3.11+incompatible h1:0gCnqKsq7XxMi69JsnbmMc1o+RJH3XH64sV9aiTTYko=
 github.com/coreos/etcd v3.3.11+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-semver v0.2.0 h1:3Jm3tLmsgAYcjC+4Up7hJrFBPr+n7rAqYeSw/SZazuY=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -112,7 +112,7 @@ github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoA
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f h1:ShTPMJQes6tubcjzGMODIVG5hlrCeImaBnZzKF2N8SM=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
-github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:BWIsLfhgKhV5g/oF34aRjniBHLTZe5DNekSjbAjIS6c=
+github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:Iju5GlWwrvL6UBg4zJJt3btmonfrMlCDdsejg4CZE7c=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -318,5 +318,5 @@ k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd h1:ggv/Vfza0i5xuhUZyYyxcc
 k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd/go.mod h1:BXM9ceUBTj2QnfH2MK1odQs778ajze1RxcmP6S8RVVc=
 k8s.io/kubernetes v1.11.7-beta.0.0.20181219023948-b875d52ea96d/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
 k8s.io/kubernetes v1.11.8-beta.0.0.20190124204751-3a10094374f2/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
-k8s.io/kubernetes v1.11.8-beta.0.0.20190214232326-4e0b35876724 h1:THwYErr8LUBf1Je2gh9lzarMPzLxN6SsdtuiLLZMtoQ=
+k8s.io/kubernetes v1.11.8-beta.0.0.20190214232326-4e0b35876724 h1:nXsoZagyrOPRWL760O5Vvjg5ejOqnVwvTwHLz274hAE=
 k8s.io/kubernetes v1.11.8-beta.0.0.20190214232326-4e0b35876724/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=

test/e2e/catalog_e2e_test.go

@@ -160,7 +160,7 @@ func TestConfigMapUpdateTriggersRegistryPodRollout(t *testing.T) {
 	// Check pod created
 	initialPods, err := c.KubernetesInterface().CoreV1().Pods(testNamespace).List(metav1.ListOptions{LabelSelector: "olm.configMapResourceVersion=" + configMap.ResourceVersion})
 	require.NoError(t, err)
-	require.Equal(t, 1, len(initialPods.Items))
+	require.Equal(t, 1, len(initialPods.Items), "Could not find pod with olm.configMapResourceVersion=" + configMap.ResourceVersion)
 
 	// Update raw manifests
 	manifestsRaw, err := yaml.Marshal(append(mainManifests, dependentManifests...))
@@ -187,17 +187,18 @@ func TestConfigMapUpdateTriggersRegistryPodRollout(t *testing.T) {
 	require.NoError(t, err)
 
 	fetchedUpdatedCatalog, err := fetchCatalogSource(t, crc, mainCatalogName, testNamespace, func(catalog *v1alpha1.CatalogSource) bool {
-		if catalog.Status.LastSync != fetchedInitialCatalog.Status.LastSync {
+		if catalog.Status.LastSync != fetchedInitialCatalog.Status.LastSync && 
+			catalog.Status.ConfigMapResource.ResourceVersion != fetchedInitialCatalog.Status.ConfigMapResource.ResourceVersion {
 			fmt.Println("catalog updated")
 			return true
 		}
-		fmt.Println("waiting for catalog pod to be available")
+		fmt.Println("waiting for catalog to be updated")
 		return false
 	})
 	require.NoError(t, err)
 
+	// FIXME(alecmerdler): Takes several MODIFY events to CatalogSource before `status` is correct, needed to move assertions to be a part of the checker
 	require.NotEqual(t, updatedConfigMap.ResourceVersion, configMap.ResourceVersion)
-	require.NotEqual(t, fetchedUpdatedCatalog.Status.ConfigMapResource.ResourceVersion, fetchedInitialCatalog.Status.ConfigMapResource.ResourceVersion)
 	require.Equal(t, updatedConfigMap.GetResourceVersion(), fetchedUpdatedCatalog.Status.ConfigMapResource.ResourceVersion)
 
 	// Await 1 CatalogSource registry pod matching the updated labels
@@ -342,6 +343,7 @@ func rescaleDeployment(c operatorclient.ClientInterface, deployment *appsv1.Depl
 	}
 
 	// wait for deployment to scale down
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, waitForScaleup)
 	if err != nil {
 		return err
@@ -356,6 +358,7 @@ func rescaleDeployment(c operatorclient.ClientInterface, deployment *appsv1.Depl
 	}
 
 	// wait for deployment to scale up
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, waitForScaleup)
 
 	return err

test/e2e/csv_e2e_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
@@ -17,6 +18,7 @@ import (
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
@@ -214,22 +216,36 @@ var csvAnyChecker = buildCSVConditionChecker(v1alpha1.CSVPhasePending, v1alpha1.
 var csvCopiedChecker = buildCSVReasonChecker(v1alpha1.CSVReasonCopied)
 
 func fetchCSV(t *testing.T, c versioned.Interface, name, namespace string, checker csvConditionChecker) (*v1alpha1.ClusterServiceVersion, error) {
-	var fetched *v1alpha1.ClusterServiceVersion
-	var err error
+	fetchedList, err := c.OperatorsV1alpha1().ClusterServiceVersions(namespace).List(metav1.ListOptions{FieldSelector: "metadata.name="+name})
+	require.NoError(t, err)
 
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetched, err = c.OperatorsV1alpha1().ClusterServiceVersions(namespace).Get(name, metav1.GetOptions{})
-		if err != nil {
-			return false, err
-		}
-		t.Logf("%s (%s): %s", fetched.Status.Phase, fetched.Status.Reason, fetched.Status.Message)
-		return checker(fetched), nil
+	watcher, err := c.OperatorsV1alpha1().ClusterServiceVersions(namespace).Watch(metav1.ListOptions{
+		FieldSelector: "metadata.name="+name,
+		ResourceVersion: fetchedList.GetResourceVersion(),
 	})
+	require.NoError(t, err)
 
-	if err != nil {
-		t.Logf("never got correct status: %#v", fetched.Status)
+	events := watcher.ResultChan()
+	for {
+		if len(fetchedList.Items) == 1 {
+			break
+		}
+
+		select {
+		case evt := <-events:
+			if evt.Type == watch.Added {
+				csv := evt.Object.(*v1alpha1.ClusterServiceVersion)
+				t.Logf("%s (%s): %s", csv.Status.Phase, csv.Status.Reason, csv.Status.Message)
+				if checker(csv) {
+					fetchedList.Items = append(fetchedList.Items, *csv)
+				}
+			}
+		case <-time.After(pollDuration):
+			break
+		}
 	}
-	return fetched, err
+
+	return &fetchedList.Items[0], err
 }
 
 func awaitCSV(t *testing.T, c versioned.Interface, namespace, name string, checker csvConditionChecker) (*v1alpha1.ClusterServiceVersion, error) {
@@ -255,6 +271,7 @@ func awaitCSV(t *testing.T, c versioned.Interface, namespace, name string, check
 }
 
 func waitForDeploymentToDelete(t *testing.T, c operatorclient.ClientInterface, name string) error {
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	return wait.Poll(pollInterval, pollDuration, func() (bool, error) {
 		t.Logf("waiting for deployment %s to delete", name)
 		_, err := c.GetDeployment(testNamespace, name)
@@ -270,22 +287,33 @@ func waitForDeploymentToDelete(t *testing.T, c operatorclient.ClientInterface, n
 	})
 }
 
+// TODO(alecmerdler): Create generic "waitForResourceToDelete" function which uses `.Watch()`
 func waitForCSVToDelete(t *testing.T, c versioned.Interface, name string) error {
-	var err error
+	fetchedList, err := c.OperatorsV1alpha1().ClusterServiceVersions(testNamespace).List(metav1.ListOptions{FieldSelector: "metadata.name=" + name})
+	require.NoError(t, err)
 
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetched, err := c.OperatorsV1alpha1().ClusterServiceVersions(testNamespace).Get(name, metav1.GetOptions{})
-		if errors.IsNotFound(err) {
-			return true, nil
-		}
-		t.Logf("%s (%s): %s", fetched.Status.Phase, fetched.Status.Reason, fetched.Status.Message)
-		if err != nil {
-			return false, err
-		}
-		return false, nil
+	if len(fetchedList.Items) == 0 {
+		return nil
+	}
+
+	watcher, err := c.OperatorsV1alpha1().ClusterServiceVersions(testNamespace).Watch(metav1.ListOptions{
+		FieldSelector: "metadata.name=" + name,
+		ResourceVersion: fetchedList.GetResourceVersion(),
 	})
+	require.NoError(t, err)
 
-	return err
+	events := watcher.ResultChan()
+	for {
+		t.Logf("waiting for CSV %s to be deleted", name)
+		select {
+		case evt := <-events:
+			if evt.Type == watch.Deleted {
+				return nil
+			}
+		case <-time.After(pollDuration):
+			return fmt.Errorf("CSV %s was not deleted before deadline", name)
+		}
+	}
 }
 
 func TestCreateCSVWithUnmetRequirementsMinKubeVersion(t *testing.T) {
@@ -951,6 +979,7 @@ func TestCreateCSVRequirementsMetCRD(t *testing.T) {
 
 	fmt.Println("checking for deployment")
 	// Poll for deployment to be ready
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
 		dep, err := c.GetDeployment(testNamespace, depName)
 		if k8serrors.IsNotFound(err) {

test/e2e/installplan_e2e_test.go

@@ -13,7 +13,6 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/apis/rbac"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
@@ -60,21 +59,6 @@ func buildInstallPlanCleanupFunc(crc versioned.Interface, namespace string, inst
 	}
 }
 
-func fetchInstallPlan(t *testing.T, c versioned.Interface, name string, checkPhase checkInstallPlanFunc) (*v1alpha1.InstallPlan, error) {
-	var fetchedInstallPlan *v1alpha1.InstallPlan
-	var err error
-
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetchedInstallPlan, err = c.OperatorsV1alpha1().InstallPlans(testNamespace).Get(name, metav1.GetOptions{})
-		if err != nil || fetchedInstallPlan == nil {
-			return false, err
-		}
-
-		return checkPhase(fetchedInstallPlan), nil
-	})
-	return fetchedInstallPlan, err
-}
-
 func newNginxInstallStrategy(name string, permissions []install.StrategyDeploymentPermissions, clusterPermissions []install.StrategyDeploymentPermissions) v1alpha1.NamedInstallStrategy {
 	// Create an nginx details deployment
 	details := install.StrategyDetailsDeployment{

test/e2e/operator_groups_e2e_test.go

@@ -54,6 +54,7 @@ func patchOlmDeployment(t *testing.T, c operatorclient.ClientInterface, newNames
 	}
 	require.NoError(t, err)
 
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
 		t.Log("Polling for OLM deployment update...")
 		fetchedDeployment, err := c.GetDeployment(olmDeployment.Namespace, olmDeployment.Name)
@@ -196,18 +197,11 @@ func TestOperatorGroup(t *testing.T) {
 	}
 
 	t.Log("Waiting on operator group to have correct status")
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetched, fetchErr := crc.OperatorsV1alpha2().OperatorGroups(opGroupNamespace).Get(operatorGroup.Name, metav1.GetOptions{})
-		if fetchErr != nil {
-			return false, fetchErr
-		}
-		if len(fetched.Status.Namespaces) > 0 {
-			require.ElementsMatch(t, expectedOperatorGroupStatus.Namespaces, fetched.Status.Namespaces)
-			return true, nil
-		}
-		return false, nil
+	fetchedOperatorGroup, err := fetchOperatorGroup(t, crc, operatorGroup.Name, func(og *v1alpha2.OperatorGroup) bool {
+		return len(og.Status.Namespaces) > 0
 	})
 	require.NoError(t, err)
+	require.ElementsMatch(t, expectedOperatorGroupStatus.Namespaces, fetchedOperatorGroup.Status.Namespaces)
 
 	t.Log("Creating CSV")
 	// Generate permissions
@@ -267,72 +261,32 @@ func TestOperatorGroup(t *testing.T) {
 	namedStrategy := newNginxInstallStrategy(deploymentName, permissions, nil)
 
 	aCSV := newCSV(csvName, opGroupNamespace, "", *semver.New("0.0.0"), []apiextensions.CustomResourceDefinition{mainCRD}, nil, namedStrategy)
-	createdCSV, err := crc.OperatorsV1alpha1().ClusterServiceVersions(opGroupNamespace).Create(&aCSV)
+	_, err = crc.OperatorsV1alpha1().ClusterServiceVersions(opGroupNamespace).Create(&aCSV)
 	require.NoError(t, err)
 
 	t.Log("wait for CSV to succeed")
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetched, err := crc.OperatorsV1alpha1().ClusterServiceVersions(opGroupNamespace).Get(createdCSV.GetName(), metav1.GetOptions{})
-		if err != nil {
-			return false, err
-		}
-		t.Logf("%s (%s): %s", fetched.Status.Phase, fetched.Status.Reason, fetched.Status.Message)
-		return csvSucceededChecker(fetched), nil
-	})
+	_, err = fetchCSV(t, crc, csvName, opGroupNamespace, csvSucceededChecker)
 	require.NoError(t, err)
 
 	t.Log("Waiting for operator namespace csv to have annotations")
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetchedCSV, fetchErr := crc.OperatorsV1alpha1().ClusterServiceVersions(opGroupNamespace).Get(csvName, metav1.GetOptions{})
-		if fetchErr != nil {
-			if errors.IsNotFound(fetchErr) {
-				return false, nil
-			}
-			t.Logf("Error (in %v): %v", testNamespace, fetchErr.Error())
-			return false, fetchErr
-		}
-		if checkOperatorGroupAnnotations(fetchedCSV, &operatorGroup, true, bothNamespaceNames) == nil {
-			return true, nil
-		}
-		return false, nil
+	_, err = fetchCSV(t, crc, csvName, opGroupNamespace, func(fetchedCSV *v1alpha1.ClusterServiceVersion) bool {
+		return checkOperatorGroupAnnotations(fetchedCSV, &operatorGroup, true, bothNamespaceNames) == nil
 	})
 	require.NoError(t, err)
 
 	t.Log("Waiting for target namespace csv to have annotations (but not target namespaces)")
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetchedCSV, fetchErr := crc.OperatorsV1alpha1().ClusterServiceVersions(otherNamespaceName).Get(csvName, metav1.GetOptions{})
-		if fetchErr != nil {
-			if errors.IsNotFound(fetchErr) {
-				return false, nil
-			}
-			t.Logf("Error (in %v): %v", otherNamespaceName, fetchErr.Error())
-			return false, fetchErr
-		}
-		if checkOperatorGroupAnnotations(fetchedCSV, &operatorGroup, false, "") == nil {
-			return true, nil
-		}
-
-		return false, nil
+	_, err = fetchCSV(t, crc, csvName, otherNamespaceName, func(fetchedCSV *v1alpha1.ClusterServiceVersion) bool {
+		return checkOperatorGroupAnnotations(fetchedCSV, &operatorGroup, true, bothNamespaceNames) == nil
 	})
 
 	t.Log("Checking status on csv in target namespace")
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetchedCSV, fetchErr := crc.OperatorsV1alpha1().ClusterServiceVersions(otherNamespaceName).Get(csvName, metav1.GetOptions{})
-		if fetchErr != nil {
-			if errors.IsNotFound(fetchErr) {
-				return false, nil
-			}
-			t.Logf("Error (in %v): %v", otherNamespaceName, fetchErr.Error())
-			return false, fetchErr
-		}
-		if fetchedCSV.Status.Reason == v1alpha1.CSVReasonCopied {
-			return true, nil
-		}
-		return false, nil
+	_, err = fetchCSV(t, crc, csvName, otherNamespaceName, func(fetchedCSV *v1alpha1.ClusterServiceVersion) bool {
+		return fetchedCSV.Status.Reason == v1alpha1.CSVReasonCopied
 	})
 	require.NoError(t, err)
 
 	t.Log("Waiting on deployment to have correct annotations")
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
 		createdDeployment, err := c.GetDeployment(opGroupNamespace, deploymentName)
 		if err != nil {
@@ -377,6 +331,7 @@ func TestOperatorGroup(t *testing.T) {
 	ruleChecker := install.NewCSVRuleChecker(roleInformer.Lister(), roleBindingInformer.Lister(), clusterRoleInformer.Lister(), clusterRoleBindingInformer.Lister(), &aCSV)
 
 	t.Log("Waiting for operator to have rbac in target namespace")
+	// TODO(alecmerdler): Replace `wait.Poll()` with `.Watch()`
 	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
 		for _, perm := range permissions {
 			sa, err := c.GetServiceAccount(opGroupNamespace, perm.ServiceAccountName)

test/e2e/subscription_e2e_test.go

@@ -314,25 +314,6 @@ func subscriptionStateAny(subscription *v1alpha1.Subscription) bool {
 		subscriptionStateUpgradeAvailableChecker(subscription)
 }
 
-func fetchSubscription(t *testing.T, crc versioned.Interface, namespace, name string, checker subscriptionStateChecker) (*v1alpha1.Subscription, error) {
-	var fetchedSubscription *v1alpha1.Subscription
-	var err error
-
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
-		fetchedSubscription, err = crc.OperatorsV1alpha1().Subscriptions(namespace).Get(name, metav1.GetOptions{})
-		if err != nil || fetchedSubscription == nil {
-			return false, err
-		}
-		t.Logf("%s (%s): %s", fetchedSubscription.Status.State, fetchedSubscription.Status.CurrentCSV, fetchedSubscription.Status.Install)
-		return checker(fetchedSubscription), nil
-	})
-	if err != nil {
-		t.Logf("never got correct status: %#v", fetchedSubscription.Status)
-		t.Logf("subscription spec: %#v", fetchedSubscription.Spec)
-	}
-	return fetchedSubscription, err
-}
-
 func buildSubscriptionCleanupFunc(t *testing.T, crc versioned.Interface, subscription *v1alpha1.Subscription) cleanupFunc {
 	return func() {
 		// Check for an installplan