fix(registry): Fix dependency load/query and add ListBundles to registry client

configmap.example.yaml

@@ -6341,6 +6341,23 @@ data:
                           fieldRef:
                             fieldPath: metadata.name
         customresourcedefinitions:
+          required:
+            - name: etcdclusters.etcd.database.coreos.com
+              version: v1beta2
+              kind: EtcdCluster
+              displayName: etcd Cluster
+              description: Represents a cluster of etcd nodes.
+              resources:
+                - kind: Service
+                  version: v1
+                - kind: Pod
+                  version: v1
+              specDescriptors:
+                - description: The desired number of member Pods for the etcd cluster.
+                  displayName: Size
+                  path: size
+                  x-descriptors:
+                    - 'urn:alm:descriptor:com.tectonic.ui:podCount'
           owned:
           - name: etcdclusters.etcd.database.coreos.com
             version: v1beta2
@@ -7264,5 +7281,3 @@ data:
       channels:
       - name: alpha
         currentCSV: prometheusoperator.0.22.2
-
-

pkg/client/client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"io"
 	"time"
 
 	"google.golang.org/grpc"
@@ -16,6 +17,7 @@ type Interface interface {
 	GetBundleInPackageChannel(ctx context.Context, packageName, channelName string) (*api.Bundle, error)
 	GetReplacementBundleInPackageChannel(ctx context.Context, currentName, packageName, channelName string) (*api.Bundle, error)
 	GetBundleThatProvides(ctx context.Context, group, version, kind string) (*api.Bundle, error)
+	ListBundles(ctx context.Context) (*BundleIterator, error)
 	HealthCheck(ctx context.Context, reconnectTimeout time.Duration) (bool, error)
 	Close() error
 }
@@ -28,6 +30,37 @@ type Client struct {
 
 var _ Interface = &Client{}
 
+type BundleStream interface {
+	Recv() (*api.Bundle, error)
+}
+
+type BundleIterator struct {
+	stream BundleStream
+	error  error
+}
+
+func NewBundleIterator(stream BundleStream) *BundleIterator {
+	return &BundleIterator{stream: stream}
+}
+
+func (it *BundleIterator) Next() *api.Bundle {
+	if it.error != nil {
+		return nil
+	}
+	next, err := it.stream.Recv()
+	if err == io.EOF {
+		return nil
+	}
+	if err != nil {
+		it.error = err
+	}
+	return next
+}
+
+func (it *BundleIterator) Error() error {
+	return it.error
+}
+
 func (c *Client) GetBundle(ctx context.Context, packageName, channelName, csvName string) (*api.Bundle, error) {
 	return c.Registry.GetBundle(ctx, &api.GetBundleRequest{PkgName: packageName, ChannelName: channelName, CsvName: csvName})
 }
@@ -44,6 +77,14 @@ func (c *Client) GetBundleThatProvides(ctx context.Context, group, version, kind
 	return c.Registry.GetDefaultBundleThatProvides(ctx, &api.GetDefaultProviderRequest{Group: group, Version: version, Kind: kind})
 }
 
+func (c *Client) ListBundles(ctx context.Context) (*BundleIterator, error) {
+	stream, err := c.Registry.ListBundles(ctx, &api.ListBundlesRequest{})
+	if err != nil {
+		return nil, err
+	}
+	return NewBundleIterator(stream), nil
+}
+
 func (c *Client) Close() error {
 	if c.Conn == nil {
 		return nil

pkg/client/client_test.go

@@ -0,0 +1,127 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/operator-framework/operator-registry/pkg/api"
+	"github.com/operator-framework/operator-registry/pkg/api/grpc_health_v1"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+type RegistryClientStub struct {
+	ListBundlesClient api.Registry_ListBundlesClient
+	error             error
+}
+
+func (s *RegistryClientStub) ListPackages(ctx context.Context, in *api.ListPackageRequest, opts ...grpc.CallOption) (api.Registry_ListPackagesClient, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetPackage(ctx context.Context, in *api.GetPackageRequest, opts ...grpc.CallOption) (*api.Package, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetBundle(ctx context.Context, in *api.GetBundleRequest, opts ...grpc.CallOption) (*api.Bundle, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetBundleForChannel(ctx context.Context, in *api.GetBundleInChannelRequest, opts ...grpc.CallOption) (*api.Bundle, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetChannelEntriesThatReplace(ctx context.Context, in *api.GetAllReplacementsRequest, opts ...grpc.CallOption) (api.Registry_GetChannelEntriesThatReplaceClient, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetBundleThatReplaces(ctx context.Context, in *api.GetReplacementRequest, opts ...grpc.CallOption) (*api.Bundle, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetChannelEntriesThatProvide(ctx context.Context, in *api.GetAllProvidersRequest, opts ...grpc.CallOption) (api.Registry_GetChannelEntriesThatProvideClient, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetLatestChannelEntriesThatProvide(ctx context.Context, in *api.GetLatestProvidersRequest, opts ...grpc.CallOption) (api.Registry_GetLatestChannelEntriesThatProvideClient, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) GetDefaultBundleThatProvides(ctx context.Context, in *api.GetDefaultProviderRequest, opts ...grpc.CallOption) (*api.Bundle, error) {
+	return nil, nil
+}
+
+func (s *RegistryClientStub) ListBundles(ctx context.Context, in *api.ListBundlesRequest, opts ...grpc.CallOption) (api.Registry_ListBundlesClient, error) {
+	return s.ListBundlesClient, s.error
+}
+
+func (s *RegistryClientStub) Check(ctx context.Context, in *grpc_health_v1.HealthCheckRequest, opts ...grpc.CallOption) (*grpc_health_v1.HealthCheckResponse, error) {
+	return nil, nil
+}
+
+type BundleReceiverStub struct {
+	Bundle *api.Bundle
+	error  error
+	grpc.ClientStream
+}
+
+func (s *BundleReceiverStub) Recv() (*api.Bundle, error) {
+	return s.Bundle, s.error
+}
+
+func TestListBundlesError(t *testing.T) {
+	expected := errors.New("test error")
+	stub := &RegistryClientStub{
+		error: expected,
+	}
+	c := Client{
+		Registry: stub,
+		Health:   stub,
+	}
+
+	_, actual := c.ListBundles(context.TODO())
+	require.Equal(t, expected, actual)
+}
+
+func TestListBundlesRecvError(t *testing.T) {
+	expected := errors.New("test error")
+	rstub := &BundleReceiverStub{
+		error: expected,
+	}
+	cstub := &RegistryClientStub{
+		ListBundlesClient: rstub,
+	}
+	c := Client{
+		Registry: cstub,
+		Health:   cstub,
+	}
+
+	it, err := c.ListBundles(context.TODO())
+	require.NoError(t, err)
+
+	require.Nil(t, it.Next())
+	require.Equal(t, expected, it.Error())
+}
+
+func TestListBundlesNext(t *testing.T) {
+	expected := &api.Bundle{CsvName: "test"}
+	rstub := &BundleReceiverStub{
+		Bundle: expected,
+	}
+	cstub := &RegistryClientStub{
+		ListBundlesClient: rstub,
+	}
+	c := Client{
+		Registry: cstub,
+		Health:   cstub,
+	}
+
+	it, err := c.ListBundles(context.TODO())
+	require.NoError(t, err)
+
+	actual := it.Next()
+	require.NoError(t, it.Error())
+	require.Equal(t, expected, actual)
+}

pkg/registry/populator.go

@@ -351,7 +351,6 @@ func parseDependenciesFile(path string, depFile *DependenciesFile) error {
 	for _, v := range deps.RawMessage {
 		// convert map to json
 		jsonStr, _ := json.Marshal(v)
-		fmt.Println(string(jsonStr))
 
 		// Check dependency type
 		dep := Dependency{}

pkg/registry/populator_test.go

@@ -121,6 +121,10 @@ func TestQuerierForImage(t *testing.T) {
 				Type:  "olm.gvk",
 				Value: `{"group":"testapi.coreos.com","kind":"testapi","type":"olm.gvk","version":"v1"}`,
 			},
+			{
+				Type:  "olm.gvk",
+				Value: `{"group":"etcd.database.coreos.com","kind":"EtcdCluster","type":"olm.gvk","version":"v1beta2"}`,
+			},
 		},
 		ProvidedApis: []*api.GroupVersionKind{
 			{Group: "etcd.database.coreos.com", Version: "v1beta2", Kind: "EtcdCluster", Plural: "etcdclusters"},
@@ -403,6 +407,10 @@ func TestQuerierForDependencies(t *testing.T) {
 			Type:  "olm.gvk",
 			Value: `{"group":"testapi.coreos.com","kind":"testapi","type":"olm.gvk","version":"v1"}`,
 		},
+		{
+			Type:  "olm.gvk",
+			Value: `{"group":"etcd.database.coreos.com","kind":"EtcdCluster","type":"olm.gvk","version":"v1beta2"}`,
+		},
 	}
 
 	type operatorbundle struct {
@@ -441,6 +449,11 @@ func TestListBundles(t *testing.T) {
 	store, err := createAndPopulateDB(db)
 	require.NoError(t, err)
 
+	var count int
+	row := db.QueryRow("SELECT COUNT(*) FROM operatorbundle")
+	err = row.Scan(&count)
+	require.NoError(t, err)
+
 	expectedDependencies := []*api.Dependency{
 		{
 			Type:  "olm.package",
@@ -450,15 +463,23 @@ func TestListBundles(t *testing.T) {
 			Type:  "olm.gvk",
 			Value: `{"group":"testapi.coreos.com","kind":"testapi","type":"olm.gvk","version":"v1"}`,
 		},
+		{
+			Type:  "olm.gvk",
+			Value: `{"group":"etcd.database.coreos.com","kind":"EtcdCluster","type":"olm.gvk","version":"v1beta2"}`,
+		},
 	}
 
 	dependencies := []*api.Dependency{}
 	bundles, err := store.ListBundles(context.TODO())
 	require.NoError(t, err)
 	for _, b := range bundles {
-		dep := b.Dependencies
-		dependencies = append(dependencies, dep...)
+		for _, d := range b.Dependencies {
+			if d.GetType() != "" {
+				dependencies = append(dependencies, d)
+			}
+		}
 	}
+	require.Equal(t, count, len(bundles))
 	require.ElementsMatch(t, expectedDependencies, dependencies)
 }
 
@@ -477,7 +498,7 @@ func CheckInvariants(t *testing.T, db *sql.DB) {
 func CheckChannelHeadsHaveDescriptions(t *testing.T, db *sql.DB) {
 	// check channel heads have csv / bundle
 	rows, err := db.Query(`
-		select operatorbundle.name,length(operatorbundle.csv),length(operatorbundle.bundle) from operatorbundle 
+		select operatorbundle.name,length(operatorbundle.csv),length(operatorbundle.bundle) from operatorbundle
 		join channel on channel.head_operatorbundle_name = operatorbundle.name`)
 	require.NoError(t, err)
 
@@ -496,7 +517,7 @@ func CheckChannelHeadsHaveDescriptions(t *testing.T, db *sql.DB) {
 func CheckBundlesHaveContentsIfNoPath(t *testing.T, db *sql.DB) {
 	// check that any bundle entry has csv/bundle content unpacked if there is no bundlepath
 	rows, err := db.Query(`
-		select name,length(csv),length(bundle) from operatorbundle 
+		select name,length(csv),length(bundle) from operatorbundle
 		where bundlepath="" or bundlepath=null`)
 	require.NoError(t, err)
 
@@ -510,4 +531,4 @@ func CheckBundlesHaveContentsIfNoPath(t *testing.T, db *sql.DB) {
 		require.NotZero(t, csvlen.Int64, "length of csv for %s should not be zero, it has no bundle path", name.String)
 		require.NotZero(t, bundlelen.Int64, "length of bundle for %s should not be zero, it has no bundle path", name.String)
 	}
-}
+}

pkg/registry/types.go

@@ -11,6 +11,11 @@ var (
 	ErrPackageNotInDatabase = errors.New("Package not in database")
 )
 
+const (
+	GVKType     = "olm.gvk"
+	PackageType = "olm.package"
+)
+
 // APIKey stores GroupVersionKind for use as map keys
 type APIKey struct {
 	Group   string