refact(plugin): Move host loopback plugin into internal/plugin/loopback

internal/plugin/loopback/client.go

@@ -0,0 +1,96 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"errors"
+
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"google.golang.org/grpc"
+)
+
+var _ plgpb.HostPluginServiceClient = (*WrappingPluginClient)(nil)
+
+// WrappingPluginClient provides a wrapper around a Server implementation that
+// can be used when loading a plugin in-memory. Supports HostServiceServer
+type WrappingPluginClient struct {
+	Server any
+}
+
+func NewWrappingPluginClient(s any) *WrappingPluginClient {
+	return &WrappingPluginClient{Server: s}
+}
+
+func (tpc *WrappingPluginClient) NormalizeCatalogData(ctx context.Context, req *plgpb.NormalizeCatalogDataRequest, opts ...grpc.CallOption) (*plgpb.NormalizeCatalogDataResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.NormalizeCatalogData(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnCreateCatalog(ctx context.Context, req *plgpb.OnCreateCatalogRequest, opts ...grpc.CallOption) (*plgpb.OnCreateCatalogResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnCreateCatalog(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnUpdateCatalog(ctx context.Context, req *plgpb.OnUpdateCatalogRequest, opts ...grpc.CallOption) (*plgpb.OnUpdateCatalogResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnUpdateCatalog(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnDeleteCatalog(ctx context.Context, req *plgpb.OnDeleteCatalogRequest, opts ...grpc.CallOption) (*plgpb.OnDeleteCatalogResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnDeleteCatalog(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) NormalizeSetData(ctx context.Context, req *plgpb.NormalizeSetDataRequest, opts ...grpc.CallOption) (*plgpb.NormalizeSetDataResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.NormalizeSetData(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnCreateSet(ctx context.Context, req *plgpb.OnCreateSetRequest, opts ...grpc.CallOption) (*plgpb.OnCreateSetResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnCreateSet(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnUpdateSet(ctx context.Context, req *plgpb.OnUpdateSetRequest, opts ...grpc.CallOption) (*plgpb.OnUpdateSetResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnUpdateSet(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) OnDeleteSet(ctx context.Context, req *plgpb.OnDeleteSetRequest, opts ...grpc.CallOption) (*plgpb.OnDeleteSetResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.OnDeleteSet(ctx, req)
+}
+
+func (tpc *WrappingPluginClient) ListHosts(ctx context.Context, req *plgpb.ListHostsRequest, opts ...grpc.CallOption) (*plgpb.ListHostsResponse, error) {
+	svc, ok := tpc.Server.(plgpb.HostPluginServiceServer)
+	if !ok {
+		return nil, errors.New("wrapping client not initialized with a HostServiceServer")
+	}
+	return svc.ListHosts(ctx, req)
+}

internal/plugin/loopback/host.go

@@ -0,0 +1,253 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/mitchellh/mapstructure"
+)
+
+var _ plgpb.HostPluginServiceServer = (*TestPluginHostServer)(nil)
+
+// TestPluginHostServer provides a host plugin service server where each method can be overwritten for tests.
+type TestPluginHostServer struct {
+	NormalizeCatalogDataFn func(context.Context, *plgpb.NormalizeCatalogDataRequest) (*plgpb.NormalizeCatalogDataResponse, error)
+	OnCreateCatalogFn      func(context.Context, *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error)
+	OnUpdateCatalogFn      func(context.Context, *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error)
+	OnDeleteCatalogFn      func(context.Context, *plgpb.OnDeleteCatalogRequest) (*plgpb.OnDeleteCatalogResponse, error)
+	NormalizeSetDataFn     func(context.Context, *plgpb.NormalizeSetDataRequest) (*plgpb.NormalizeSetDataResponse, error)
+	OnCreateSetFn          func(context.Context, *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error)
+	OnUpdateSetFn          func(context.Context, *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error)
+	OnDeleteSetFn          func(context.Context, *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error)
+	ListHostsFn            func(context.Context, *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error)
+	plgpb.UnimplementedHostPluginServiceServer
+}
+
+func (t TestPluginHostServer) NormalizeCatalogData(ctx context.Context, req *plgpb.NormalizeCatalogDataRequest) (*plgpb.NormalizeCatalogDataResponse, error) {
+	if t.NormalizeCatalogDataFn == nil {
+		return t.UnimplementedHostPluginServiceServer.NormalizeCatalogData(ctx, req)
+	}
+	return t.NormalizeCatalogDataFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnCreateCatalog(ctx context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+	if t.OnCreateCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnCreateCatalog(ctx, req)
+	}
+	return t.OnCreateCatalogFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnUpdateCatalog(ctx context.Context, req *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error) {
+	if t.OnUpdateCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnUpdateCatalog(ctx, req)
+	}
+	return t.OnUpdateCatalogFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnDeleteCatalog(ctx context.Context, req *plgpb.OnDeleteCatalogRequest) (*plgpb.OnDeleteCatalogResponse, error) {
+	if t.OnDeleteCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnDeleteCatalog(ctx, req)
+	}
+	return t.OnDeleteCatalogFn(ctx, req)
+}
+
+func (t TestPluginHostServer) NormalizeSetData(ctx context.Context, req *plgpb.NormalizeSetDataRequest) (*plgpb.NormalizeSetDataResponse, error) {
+	if t.NormalizeSetDataFn == nil {
+		return t.UnimplementedHostPluginServiceServer.NormalizeSetData(ctx, req)
+	}
+	return t.NormalizeSetDataFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnCreateSet(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+	if t.OnCreateSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnCreateSet(ctx, req)
+	}
+	return t.OnCreateSetFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnUpdateSet(ctx context.Context, req *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error) {
+	if t.OnUpdateSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnUpdateSet(ctx, req)
+	}
+	return t.OnUpdateSetFn(ctx, req)
+}
+
+func (t TestPluginHostServer) OnDeleteSet(ctx context.Context, req *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error) {
+	if t.OnDeleteSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnDeleteSet(ctx, req)
+	}
+	return t.OnDeleteSetFn(ctx, req)
+}
+
+func (t TestPluginHostServer) ListHosts(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) {
+	if t.ListHostsFn == nil {
+		return t.UnimplementedHostPluginServiceServer.ListHosts(ctx, req)
+	}
+	return t.ListHostsFn(ctx, req)
+}
+
+type loopbackPluginHostInfo struct {
+	ExternalId  string   `mapstructure:"external_id"`
+	IpAddresses []string `mapstructure:"ip_addresses"`
+	DnsNames    []string `mapstructure:"dns_names"`
+}
+
+// LoopbackHost provides a host plugin with functionality useful for certain
+// kinds of testing.
+//
+// It is not (currently) thread-safe.
+//
+// Over time, if useful, it can be enhanced to do things like handle multiple
+// hosts per set.
+type LoopbackHost struct {
+	hostMap map[string][]*loopbackPluginHostInfo
+}
+
+func (l *LoopbackHost) onCreateCatalog(ctx context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateCatalog"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	if cat := req.GetCatalog(); cat != nil {
+		if secrets := cat.GetSecrets(); secrets != nil {
+			return &plgpb.OnCreateCatalogResponse{
+				Persisted: &plgpb.HostCatalogPersisted{
+					Secrets: secrets,
+				},
+			}, nil
+		}
+	}
+	return &plgpb.OnCreateCatalogResponse{}, nil
+}
+
+func (l *LoopbackHost) onUpdateCatalog(ctx context.Context, req *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error) {
+	const op = "loopback.(loopbackPlugin).onUpdateCatalog"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	if cat := req.GetNewCatalog(); cat != nil {
+		if secrets := cat.GetSecrets(); secrets != nil {
+			return &plgpb.OnUpdateCatalogResponse{
+				Persisted: &plgpb.HostCatalogPersisted{
+					Secrets: secrets,
+				},
+			}, nil
+		}
+	}
+	return &plgpb.OnUpdateCatalogResponse{}, nil
+}
+
+func (l *LoopbackHost) onCreateSet(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	if attrs := set.GetAttributes(); attrs != nil {
+		attrsMap := attrs.AsMap()
+		if field := attrsMap[loopbackPluginHostInfoAttrField]; field != nil {
+			switch t := field.(type) {
+			case []any:
+				for _, h := range t {
+					hostInfo := new(loopbackPluginHostInfo)
+					if err := mapstructure.Decode(h, hostInfo); err != nil {
+						return nil, errors.Wrap(ctx, err, op)
+					}
+					l.hostMap[set.GetId()] = append(l.hostMap[set.GetId()], hostInfo)
+				}
+			case map[string]any:
+				hostInfo := new(loopbackPluginHostInfo)
+				if err := mapstructure.Decode(t, hostInfo); err != nil {
+					return nil, errors.Wrap(ctx, err, op)
+				}
+				l.hostMap[set.GetId()] = append(l.hostMap[set.GetId()], hostInfo)
+			default:
+				return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown host info type %T", t))
+			}
+		}
+	}
+	return &plgpb.OnCreateSetResponse{}, nil
+}
+
+func (l *LoopbackHost) onUpdateSet(ctx context.Context, req *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetNewSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	var hosts []*loopbackPluginHostInfo
+	if attrs := set.GetAttributes(); attrs != nil {
+		attrsMap := attrs.AsMap()
+		if field := attrsMap[loopbackPluginHostInfoAttrField]; field != nil {
+			switch t := field.(type) {
+			case []any:
+				for _, h := range t {
+					hostInfo := new(loopbackPluginHostInfo)
+					if err := mapstructure.Decode(h, hostInfo); err != nil {
+						return nil, errors.Wrap(ctx, err, op)
+					}
+					hosts = append(hosts, hostInfo)
+				}
+			case map[string]any:
+				hostInfo := new(loopbackPluginHostInfo)
+				if err := mapstructure.Decode(t, hostInfo); err != nil {
+					return nil, errors.Wrap(ctx, err, op)
+				}
+				hosts = append(hosts, hostInfo)
+			default:
+				return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown host info type %T", t))
+			}
+		}
+	}
+	if hosts != nil {
+		l.hostMap[set.GetId()] = hosts
+	}
+	return &plgpb.OnUpdateSetResponse{}, nil
+}
+
+func (l *LoopbackHost) onDeleteSet(ctx context.Context, req *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onDeleteSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	delete(l.hostMap, set.GetId())
+	return nil, nil
+}
+
+func (l *LoopbackHost) listHosts(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) {
+	const op = "loopback.(loopbackPlugin).listHosts"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	resp := new(plgpb.ListHostsResponse)
+	for _, set := range req.GetSets() {
+		hostInfos := l.hostMap[set.GetId()]
+		if len(hostInfos) == 0 {
+			continue
+		}
+		for _, host := range hostInfos {
+			resp.Hosts = append(resp.Hosts, &plgpb.ListHostsResponseHost{
+				SetIds:      []string{set.GetId()},
+				ExternalId:  host.ExternalId,
+				IpAddresses: host.IpAddresses,
+				DnsNames:    host.DnsNames,
+			})
+		}
+	}
+	return resp, nil
+}

internal/plugin/loopback/host_test.go

@@ -0,0 +1,375 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostcatalogs"
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostsets"
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	ta "github.com/stretchr/testify/assert"
+	tr "github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+// TestLoopbackHostPlugin is a quick test of basic host service functionality.
+func TestLoopbackHostPlugin(t *testing.T) {
+	require, assert := tr.New(t), ta.New(t)
+	ctx := context.Background()
+
+	plg, err := NewLoopbackPlugin()
+	require.NoError(err)
+	secretsMap := map[string]any{
+		"key1": "key2",
+		"baz":  true,
+	}
+	secrets, err := structpb.NewStruct(secretsMap)
+	require.NoError(err)
+
+	// First, test that if we give it secrets, those secrets come back as
+	// persisted data
+	catResp, err := plg.OnCreateCatalog(ctx, &plgpb.OnCreateCatalogRequest{
+		Catalog: &hostcatalogs.HostCatalog{
+			Secrets: secrets,
+		},
+	})
+	require.NoError(err)
+	require.NotNil(catResp)
+	require.NotNil(catResp.GetPersisted())
+	require.NotNil(catResp.GetPersisted().GetSecrets())
+	assert.EqualValues(secretsMap, catResp.GetPersisted().GetSecrets().AsMap())
+
+	newSecretsMap := map[string]any{
+		"key1": "key2",
+		"baz":  true,
+	}
+	newSecrets, err := structpb.NewStruct(newSecretsMap)
+	require.NoError(err)
+
+	// First, test that if we give it secrets, those secrets come back as
+	// persisted data
+	upResp, err := plg.OnUpdateCatalog(ctx, &plgpb.OnUpdateCatalogRequest{
+		CurrentCatalog: &hostcatalogs.HostCatalog{},
+		NewCatalog: &hostcatalogs.HostCatalog{
+			Secrets: newSecrets,
+		},
+		Persisted: &plgpb.HostCatalogPersisted{
+			Secrets: secrets,
+		},
+	})
+	require.NoError(err)
+	require.NotNil(upResp)
+	require.NotNil(upResp.GetPersisted())
+	require.NotNil(upResp.GetPersisted().GetSecrets())
+	assert.EqualValues(newSecretsMap, upResp.GetPersisted().GetSecrets().AsMap())
+
+	// Add data to some sets
+	hostInfo1 := map[string]any{
+		loopbackPluginHostInfoAttrField: map[string]any{
+			"set_ids":      []any{"set1"},
+			"external_id":  "host1",
+			"ip_addresses": []any{"1.2.3.4", "2.3.4.5"},
+			"dns_names":    []any{"foo.com"},
+		},
+	}
+	attrs, err := structpb.NewStruct(hostInfo1)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+	hostInfo2 := map[string]any{
+		loopbackPluginHostInfoAttrField: map[string]any{
+			"set_ids":      []any{"set2"},
+			"external_id":  "host2",
+			"ip_addresses": []any{"5.6.7.8", "6.7.8.9"},
+			"dns_names":    []any{"bar.com"},
+		},
+	}
+	attrs, err = structpb.NewStruct(hostInfo2)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set2",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+
+	// Define test struct and validation function
+	type testInfo struct {
+		name  string
+		sets  []string
+		found []map[string]any
+	}
+	validateSets := func(t *testing.T, tt testInfo) {
+		require, assert := tr.New(t), ta.New(t)
+		var hostSets []*hostsets.HostSet
+		for _, set := range tt.sets {
+			hostSets = append(hostSets, &hostsets.HostSet{Id: set})
+		}
+		resp, err := plg.ListHosts(ctx, &plgpb.ListHostsRequest{
+			Sets: hostSets,
+		})
+		require.NoError(err)
+		if len(tt.found) == 0 {
+			assert.Len(resp.GetHosts(), 0)
+			return
+		}
+
+		require.Greater(len(resp.GetHosts()), 0)
+
+		var found []map[string]any
+		for _, host := range resp.GetHosts() {
+			hostMap := map[string]any{
+				"external_id": host.GetExternalId(),
+			}
+			var sets []any
+			for _, set := range host.SetIds {
+				sets = append(sets, set)
+			}
+			var ips []any
+			for _, ip := range host.GetIpAddresses() {
+				ips = append(ips, ip)
+			}
+			var names []any
+			for _, name := range host.GetDnsNames() {
+				names = append(names, name)
+			}
+			if len(sets) > 0 {
+				hostMap["set_ids"] = sets
+			}
+			if len(ips) > 0 {
+				hostMap["ip_addresses"] = ips
+			}
+			if len(names) > 0 {
+				hostMap["dns_names"] = names
+			}
+			found = append(found, hostMap)
+		}
+		assert.ElementsMatch(tt.found, found)
+	}
+
+	// First set of tests: check that we can look up sets individually and
+	// together
+	setTests := []testInfo{
+		{
+			name: "set 1",
+			sets: []string{"set1"},
+			found: []map[string]any{
+				hostInfo1[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "set 2",
+			sets: []string{"set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "sets 1 and 2",
+			sets: []string{"set1", "set2"},
+			found: []map[string]any{
+				hostInfo1[loopbackPluginHostInfoAttrField].(map[string]any),
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+
+	// Remove a set
+	_, err = plg.OnDeleteSet(ctx, &plgpb.OnDeleteSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+		},
+	})
+	require.NoError(err)
+
+	// Run tests again, making sure we no longer find that host either
+	// individually or together
+	setTests = []testInfo{
+		{
+			name:  "set 1 deleted",
+			sets:  []string{"set1"},
+			found: []map[string]any{},
+		},
+		{
+			name: "set 2 not deleted",
+			sets: []string{"set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "sets 1 and 2 set 1 deleted",
+			sets: []string{"set1", "set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+}
+
+func TestLoopbackHostPluginArrays(t *testing.T) {
+	require := tr.New(t)
+	ctx := context.Background()
+
+	plg, err := NewLoopbackPlugin()
+	require.NoError(err)
+
+	// Add data to some sets
+	hostInfo1 := map[string]any{
+		loopbackPluginHostInfoAttrField: []any{
+			map[string]any{
+				"set_ids":      []any{"set1"},
+				"external_id":  "host1a",
+				"ip_addresses": []any{"1.2.3.4", "2.3.4.5"},
+				"dns_names":    []any{"foo.com"},
+			},
+			map[string]any{
+				"set_ids":      []any{"set1"},
+				"external_id":  "host1b",
+				"ip_addresses": []any{"3.4.5.6", "4.5.6.7"},
+				"dns_names":    []any{"bar.com"},
+			},
+		},
+	}
+	attrs, err := structpb.NewStruct(hostInfo1)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+	hostInfo2 := map[string]any{
+		loopbackPluginHostInfoAttrField: []any{
+			map[string]any{
+				"set_ids":      []any{"set2"},
+				"external_id":  "host2a",
+				"ip_addresses": []any{"10.20.30.40", "20.30.40.50"},
+				"dns_names":    []any{"foz.com"},
+			},
+			map[string]any{
+				"set_ids":      []any{"set2"},
+				"external_id":  "host2b",
+				"ip_addresses": []any{"30.40.50.60", "40.50.60.70"},
+				"dns_names":    []any{"baz.com"},
+			},
+		},
+	}
+	attrs, err = structpb.NewStruct(hostInfo2)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set2",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+
+	// Define test struct and validation function
+	type testInfo struct {
+		name  string
+		sets  []string
+		found []any
+	}
+	validateSets := func(t *testing.T, tt testInfo) {
+		require, assert := tr.New(t), ta.New(t)
+		var hostSets []*hostsets.HostSet
+		for _, set := range tt.sets {
+			hostSets = append(hostSets, &hostsets.HostSet{Id: set})
+		}
+		resp, err := plg.ListHosts(ctx, &plgpb.ListHostsRequest{
+			Sets: hostSets,
+		})
+		require.NoError(err)
+		if len(tt.found) == 0 {
+			assert.Len(resp.GetHosts(), 0)
+			return
+		}
+
+		require.Greater(len(resp.GetHosts()), 0)
+
+		var found []any
+		for _, host := range resp.GetHosts() {
+			hostMap := map[string]any{
+				"external_id": host.GetExternalId(),
+			}
+			var sets []any
+			for _, set := range host.SetIds {
+				sets = append(sets, set)
+			}
+			var ips []any
+			for _, ip := range host.GetIpAddresses() {
+				ips = append(ips, ip)
+			}
+			var names []any
+			for _, name := range host.GetDnsNames() {
+				names = append(names, name)
+			}
+			if len(sets) > 0 {
+				hostMap["set_ids"] = sets
+			}
+			if len(ips) > 0 {
+				hostMap["ip_addresses"] = ips
+			}
+			if len(names) > 0 {
+				hostMap["dns_names"] = names
+			}
+			found = append(found, hostMap)
+		}
+		assert.ElementsMatch(tt.found, found)
+	}
+
+	// First set of tests: check that we can look up sets individually and
+	// together
+	setTests := []testInfo{
+		{
+			name:  "set 1",
+			sets:  []string{"set1"},
+			found: hostInfo1[loopbackPluginHostInfoAttrField].([]any),
+		},
+		{
+			name:  "set 2",
+			sets:  []string{"set2"},
+			found: hostInfo2[loopbackPluginHostInfoAttrField].([]any),
+		},
+		{
+			name: "sets 1 and 2",
+			sets: []string{"set1", "set2"},
+			found: append(hostInfo1[loopbackPluginHostInfoAttrField].([]any),
+				hostInfo2[loopbackPluginHostInfoAttrField].([]any)...),
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+}

internal/plugin/loopback/loopback.go

@@ -0,0 +1,274 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/mitchellh/mapstructure"
+)
+
+const loopbackPluginHostInfoAttrField = "host_info"
+
+var (
+	_ plgpb.HostPluginServiceServer = (*LoopbackPlugin)(nil)
+	_ plgpb.HostPluginServiceServer = (*TestPluginServer)(nil)
+)
+
+// TestPluginServer provides a host plugin service server where each method can be overwritten for tests.
+type TestPluginServer struct {
+	NormalizeCatalogDataFn func(context.Context, *plgpb.NormalizeCatalogDataRequest) (*plgpb.NormalizeCatalogDataResponse, error)
+	OnCreateCatalogFn      func(context.Context, *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error)
+	OnUpdateCatalogFn      func(context.Context, *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error)
+	OnDeleteCatalogFn      func(context.Context, *plgpb.OnDeleteCatalogRequest) (*plgpb.OnDeleteCatalogResponse, error)
+	NormalizeSetDataFn     func(context.Context, *plgpb.NormalizeSetDataRequest) (*plgpb.NormalizeSetDataResponse, error)
+	OnCreateSetFn          func(context.Context, *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error)
+	OnUpdateSetFn          func(context.Context, *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error)
+	OnDeleteSetFn          func(context.Context, *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error)
+	ListHostsFn            func(context.Context, *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error)
+	plgpb.UnimplementedHostPluginServiceServer
+}
+
+func (t TestPluginServer) NormalizeCatalogData(ctx context.Context, req *plgpb.NormalizeCatalogDataRequest) (*plgpb.NormalizeCatalogDataResponse, error) {
+	if t.NormalizeCatalogDataFn == nil {
+		return t.UnimplementedHostPluginServiceServer.NormalizeCatalogData(ctx, req)
+	}
+	return t.NormalizeCatalogDataFn(ctx, req)
+}
+
+func (t TestPluginServer) OnCreateCatalog(ctx context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+	if t.OnCreateCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnCreateCatalog(ctx, req)
+	}
+	return t.OnCreateCatalogFn(ctx, req)
+}
+
+func (t TestPluginServer) OnUpdateCatalog(ctx context.Context, req *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error) {
+	if t.OnUpdateCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnUpdateCatalog(ctx, req)
+	}
+	return t.OnUpdateCatalogFn(ctx, req)
+}
+
+func (t TestPluginServer) OnDeleteCatalog(ctx context.Context, req *plgpb.OnDeleteCatalogRequest) (*plgpb.OnDeleteCatalogResponse, error) {
+	if t.OnDeleteCatalogFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnDeleteCatalog(ctx, req)
+	}
+	return t.OnDeleteCatalogFn(ctx, req)
+}
+
+func (t TestPluginServer) NormalizeSetData(ctx context.Context, req *plgpb.NormalizeSetDataRequest) (*plgpb.NormalizeSetDataResponse, error) {
+	if t.NormalizeSetDataFn == nil {
+		return t.UnimplementedHostPluginServiceServer.NormalizeSetData(ctx, req)
+	}
+	return t.NormalizeSetDataFn(ctx, req)
+}
+
+func (t TestPluginServer) OnCreateSet(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+	if t.OnCreateSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnCreateSet(ctx, req)
+	}
+	return t.OnCreateSetFn(ctx, req)
+}
+
+func (t TestPluginServer) OnUpdateSet(ctx context.Context, req *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error) {
+	if t.OnUpdateSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnUpdateSet(ctx, req)
+	}
+	return t.OnUpdateSetFn(ctx, req)
+}
+
+func (t TestPluginServer) OnDeleteSet(ctx context.Context, req *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error) {
+	if t.OnDeleteSetFn == nil {
+		return t.UnimplementedHostPluginServiceServer.OnDeleteSet(ctx, req)
+	}
+	return t.OnDeleteSetFn(ctx, req)
+}
+
+func (t TestPluginServer) ListHosts(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) {
+	if t.ListHostsFn == nil {
+		return t.UnimplementedHostPluginServiceServer.ListHosts(ctx, req)
+	}
+	return t.ListHostsFn(ctx, req)
+}
+
+// LoopbackPlugin provides a host plugin with functionality useful for certain
+// kinds of testing. It is not (currently) thread-safe.
+//
+// Over time, if useful, it can be enhanced to do things like handle multiple
+// hosts per set.
+type LoopbackPlugin struct {
+	*TestPluginServer
+
+	hostMap map[string][]*loopbackPluginHostInfo
+}
+
+type loopbackPluginHostInfo struct {
+	ExternalId  string   `mapstructure:"external_id"`
+	IpAddresses []string `mapstructure:"ip_addresses"`
+	DnsNames    []string `mapstructure:"dns_names"`
+}
+
+// NewLoopbackPlugin returns a new loopback plugin
+func NewLoopbackPlugin() (*LoopbackPlugin, error) {
+	ret := &LoopbackPlugin{
+		TestPluginServer: new(TestPluginServer),
+		hostMap:          make(map[string][]*loopbackPluginHostInfo),
+	}
+	ret.OnCreateCatalogFn = ret.onCreateCatalog
+	ret.OnUpdateCatalogFn = ret.onUpdateCatalog
+	ret.OnCreateSetFn = ret.onCreateSet
+	ret.OnUpdateSetFn = ret.onUpdateSet
+	ret.OnDeleteSetFn = ret.onDeleteSet
+	ret.ListHostsFn = ret.listHosts
+
+	return ret, nil
+}
+
+func (l *LoopbackPlugin) onCreateCatalog(ctx context.Context, req *plgpb.OnCreateCatalogRequest) (*plgpb.OnCreateCatalogResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateCatalog"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	if cat := req.GetCatalog(); cat != nil {
+		if secrets := cat.GetSecrets(); secrets != nil {
+			return &plgpb.OnCreateCatalogResponse{
+				Persisted: &plgpb.HostCatalogPersisted{
+					Secrets: secrets,
+				},
+			}, nil
+		}
+	}
+	return &plgpb.OnCreateCatalogResponse{}, nil
+}
+
+func (l *LoopbackPlugin) onUpdateCatalog(ctx context.Context, req *plgpb.OnUpdateCatalogRequest) (*plgpb.OnUpdateCatalogResponse, error) {
+	const op = "loopback.(loopbackPlugin).onUpdateCatalog"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	if cat := req.GetNewCatalog(); cat != nil {
+		if secrets := cat.GetSecrets(); secrets != nil {
+			return &plgpb.OnUpdateCatalogResponse{
+				Persisted: &plgpb.HostCatalogPersisted{
+					Secrets: secrets,
+				},
+			}, nil
+		}
+	}
+	return &plgpb.OnUpdateCatalogResponse{}, nil
+}
+
+func (l *LoopbackPlugin) onCreateSet(ctx context.Context, req *plgpb.OnCreateSetRequest) (*plgpb.OnCreateSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	if attrs := set.GetAttributes(); attrs != nil {
+		attrsMap := attrs.AsMap()
+		if field := attrsMap[loopbackPluginHostInfoAttrField]; field != nil {
+			switch t := field.(type) {
+			case []any:
+				for _, h := range t {
+					hostInfo := new(loopbackPluginHostInfo)
+					if err := mapstructure.Decode(h, hostInfo); err != nil {
+						return nil, errors.Wrap(ctx, err, op)
+					}
+					l.hostMap[set.GetId()] = append(l.hostMap[set.GetId()], hostInfo)
+				}
+			case map[string]any:
+				hostInfo := new(loopbackPluginHostInfo)
+				if err := mapstructure.Decode(t, hostInfo); err != nil {
+					return nil, errors.Wrap(ctx, err, op)
+				}
+				l.hostMap[set.GetId()] = append(l.hostMap[set.GetId()], hostInfo)
+			default:
+				return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown host info type %T", t))
+			}
+		}
+	}
+	return &plgpb.OnCreateSetResponse{}, nil
+}
+
+func (l *LoopbackPlugin) onUpdateSet(ctx context.Context, req *plgpb.OnUpdateSetRequest) (*plgpb.OnUpdateSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onCreateSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetNewSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	var hosts []*loopbackPluginHostInfo
+	if attrs := set.GetAttributes(); attrs != nil {
+		attrsMap := attrs.AsMap()
+		if field := attrsMap[loopbackPluginHostInfoAttrField]; field != nil {
+			switch t := field.(type) {
+			case []any:
+				for _, h := range t {
+					hostInfo := new(loopbackPluginHostInfo)
+					if err := mapstructure.Decode(h, hostInfo); err != nil {
+						return nil, errors.Wrap(ctx, err, op)
+					}
+					hosts = append(hosts, hostInfo)
+				}
+			case map[string]any:
+				hostInfo := new(loopbackPluginHostInfo)
+				if err := mapstructure.Decode(t, hostInfo); err != nil {
+					return nil, errors.Wrap(ctx, err, op)
+				}
+				hosts = append(hosts, hostInfo)
+			default:
+				return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown host info type %T", t))
+			}
+		}
+	}
+	if hosts != nil {
+		l.hostMap[set.GetId()] = hosts
+	}
+	return &plgpb.OnUpdateSetResponse{}, nil
+}
+
+func (l *LoopbackPlugin) onDeleteSet(ctx context.Context, req *plgpb.OnDeleteSetRequest) (*plgpb.OnDeleteSetResponse, error) {
+	const op = "loopback.(loopbackPlugin).onDeleteSet"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	set := req.GetSet()
+	if set == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "set is nil")
+	}
+	delete(l.hostMap, set.GetId())
+	return nil, nil
+}
+
+func (l *LoopbackPlugin) listHosts(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) {
+	const op = "loopback.(loopbackPlugin).listHosts"
+	if req == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "req is nil")
+	}
+	resp := new(plgpb.ListHostsResponse)
+	for _, set := range req.GetSets() {
+		hostInfos := l.hostMap[set.GetId()]
+		if len(hostInfos) == 0 {
+			continue
+		}
+		for _, host := range hostInfos {
+			resp.Hosts = append(resp.Hosts, &plgpb.ListHostsResponseHost{
+				SetIds:      []string{set.GetId()},
+				ExternalId:  host.ExternalId,
+				IpAddresses: host.IpAddresses,
+				DnsNames:    host.DnsNames,
+			})
+		}
+	}
+	return resp, nil
+}

internal/plugin/loopback/loopback_test.go

@@ -0,0 +1,375 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostcatalogs"
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostsets"
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	ta "github.com/stretchr/testify/assert"
+	tr "github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+// TestLoopbackPlugin is a quick test of basic functionality.
+func TestLoopbackPlugin(t *testing.T) {
+	require, assert := tr.New(t), ta.New(t)
+	ctx := context.Background()
+
+	plg, err := NewLoopbackPlugin()
+	require.NoError(err)
+	secretsMap := map[string]any{
+		"key1": "key2",
+		"baz":  true,
+	}
+	secrets, err := structpb.NewStruct(secretsMap)
+	require.NoError(err)
+
+	// First, test that if we give it secrets, those secrets come back as
+	// persisted data
+	catResp, err := plg.OnCreateCatalog(ctx, &plgpb.OnCreateCatalogRequest{
+		Catalog: &hostcatalogs.HostCatalog{
+			Secrets: secrets,
+		},
+	})
+	require.NoError(err)
+	require.NotNil(catResp)
+	require.NotNil(catResp.GetPersisted())
+	require.NotNil(catResp.GetPersisted().GetSecrets())
+	assert.EqualValues(secretsMap, catResp.GetPersisted().GetSecrets().AsMap())
+
+	newSecretsMap := map[string]any{
+		"key1": "key2",
+		"baz":  true,
+	}
+	newSecrets, err := structpb.NewStruct(newSecretsMap)
+	require.NoError(err)
+
+	// First, test that if we give it secrets, those secrets come back as
+	// persisted data
+	upResp, err := plg.OnUpdateCatalog(ctx, &plgpb.OnUpdateCatalogRequest{
+		CurrentCatalog: &hostcatalogs.HostCatalog{},
+		NewCatalog: &hostcatalogs.HostCatalog{
+			Secrets: newSecrets,
+		},
+		Persisted: &plgpb.HostCatalogPersisted{
+			Secrets: secrets,
+		},
+	})
+	require.NoError(err)
+	require.NotNil(upResp)
+	require.NotNil(upResp.GetPersisted())
+	require.NotNil(upResp.GetPersisted().GetSecrets())
+	assert.EqualValues(newSecretsMap, upResp.GetPersisted().GetSecrets().AsMap())
+
+	// Add data to some sets
+	hostInfo1 := map[string]any{
+		loopbackPluginHostInfoAttrField: map[string]any{
+			"set_ids":      []any{"set1"},
+			"external_id":  "host1",
+			"ip_addresses": []any{"1.2.3.4", "2.3.4.5"},
+			"dns_names":    []any{"foo.com"},
+		},
+	}
+	attrs, err := structpb.NewStruct(hostInfo1)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+	hostInfo2 := map[string]any{
+		loopbackPluginHostInfoAttrField: map[string]any{
+			"set_ids":      []any{"set2"},
+			"external_id":  "host2",
+			"ip_addresses": []any{"5.6.7.8", "6.7.8.9"},
+			"dns_names":    []any{"bar.com"},
+		},
+	}
+	attrs, err = structpb.NewStruct(hostInfo2)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set2",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+
+	// Define test struct and validation function
+	type testInfo struct {
+		name  string
+		sets  []string
+		found []map[string]any
+	}
+	validateSets := func(t *testing.T, tt testInfo) {
+		require, assert := tr.New(t), ta.New(t)
+		var hostSets []*hostsets.HostSet
+		for _, set := range tt.sets {
+			hostSets = append(hostSets, &hostsets.HostSet{Id: set})
+		}
+		resp, err := plg.ListHosts(ctx, &plgpb.ListHostsRequest{
+			Sets: hostSets,
+		})
+		require.NoError(err)
+		if len(tt.found) == 0 {
+			assert.Len(resp.GetHosts(), 0)
+			return
+		}
+
+		require.Greater(len(resp.GetHosts()), 0)
+
+		var found []map[string]any
+		for _, host := range resp.GetHosts() {
+			hostMap := map[string]any{
+				"external_id": host.GetExternalId(),
+			}
+			var sets []any
+			for _, set := range host.SetIds {
+				sets = append(sets, set)
+			}
+			var ips []any
+			for _, ip := range host.GetIpAddresses() {
+				ips = append(ips, ip)
+			}
+			var names []any
+			for _, name := range host.GetDnsNames() {
+				names = append(names, name)
+			}
+			if len(sets) > 0 {
+				hostMap["set_ids"] = sets
+			}
+			if len(ips) > 0 {
+				hostMap["ip_addresses"] = ips
+			}
+			if len(names) > 0 {
+				hostMap["dns_names"] = names
+			}
+			found = append(found, hostMap)
+		}
+		assert.ElementsMatch(tt.found, found)
+	}
+
+	// First set of tests: check that we can look up sets individually and
+	// together
+	setTests := []testInfo{
+		{
+			name: "set 1",
+			sets: []string{"set1"},
+			found: []map[string]any{
+				hostInfo1[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "set 2",
+			sets: []string{"set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "sets 1 and 2",
+			sets: []string{"set1", "set2"},
+			found: []map[string]any{
+				hostInfo1[loopbackPluginHostInfoAttrField].(map[string]any),
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+
+	// Remove a set
+	_, err = plg.OnDeleteSet(ctx, &plgpb.OnDeleteSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+		},
+	})
+	require.NoError(err)
+
+	// Run tests again, making sure we no longer find that host either
+	// individually or together
+	setTests = []testInfo{
+		{
+			name:  "set 1 deleted",
+			sets:  []string{"set1"},
+			found: []map[string]any{},
+		},
+		{
+			name: "set 2 not deleted",
+			sets: []string{"set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+		{
+			name: "sets 1 and 2 set 1 deleted",
+			sets: []string{"set1", "set2"},
+			found: []map[string]any{
+				hostInfo2[loopbackPluginHostInfoAttrField].(map[string]any),
+			},
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+}
+
+func TestLoopbackPluginArrays(t *testing.T) {
+	require := tr.New(t)
+	ctx := context.Background()
+
+	plg, err := NewLoopbackPlugin()
+	require.NoError(err)
+
+	// Add data to some sets
+	hostInfo1 := map[string]any{
+		loopbackPluginHostInfoAttrField: []any{
+			map[string]any{
+				"set_ids":      []any{"set1"},
+				"external_id":  "host1a",
+				"ip_addresses": []any{"1.2.3.4", "2.3.4.5"},
+				"dns_names":    []any{"foo.com"},
+			},
+			map[string]any{
+				"set_ids":      []any{"set1"},
+				"external_id":  "host1b",
+				"ip_addresses": []any{"3.4.5.6", "4.5.6.7"},
+				"dns_names":    []any{"bar.com"},
+			},
+		},
+	}
+	attrs, err := structpb.NewStruct(hostInfo1)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set1",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+	hostInfo2 := map[string]any{
+		loopbackPluginHostInfoAttrField: []any{
+			map[string]any{
+				"set_ids":      []any{"set2"},
+				"external_id":  "host2a",
+				"ip_addresses": []any{"10.20.30.40", "20.30.40.50"},
+				"dns_names":    []any{"foz.com"},
+			},
+			map[string]any{
+				"set_ids":      []any{"set2"},
+				"external_id":  "host2b",
+				"ip_addresses": []any{"30.40.50.60", "40.50.60.70"},
+				"dns_names":    []any{"baz.com"},
+			},
+		},
+	}
+	attrs, err = structpb.NewStruct(hostInfo2)
+	require.NoError(err)
+	_, err = plg.OnCreateSet(ctx, &plgpb.OnCreateSetRequest{
+		Set: &hostsets.HostSet{
+			Id: "set2",
+			Attrs: &hostsets.HostSet_Attributes{
+				Attributes: attrs,
+			},
+		},
+	})
+	require.NoError(err)
+
+	// Define test struct and validation function
+	type testInfo struct {
+		name  string
+		sets  []string
+		found []any
+	}
+	validateSets := func(t *testing.T, tt testInfo) {
+		require, assert := tr.New(t), ta.New(t)
+		var hostSets []*hostsets.HostSet
+		for _, set := range tt.sets {
+			hostSets = append(hostSets, &hostsets.HostSet{Id: set})
+		}
+		resp, err := plg.ListHosts(ctx, &plgpb.ListHostsRequest{
+			Sets: hostSets,
+		})
+		require.NoError(err)
+		if len(tt.found) == 0 {
+			assert.Len(resp.GetHosts(), 0)
+			return
+		}
+
+		require.Greater(len(resp.GetHosts()), 0)
+
+		var found []any
+		for _, host := range resp.GetHosts() {
+			hostMap := map[string]any{
+				"external_id": host.GetExternalId(),
+			}
+			var sets []any
+			for _, set := range host.SetIds {
+				sets = append(sets, set)
+			}
+			var ips []any
+			for _, ip := range host.GetIpAddresses() {
+				ips = append(ips, ip)
+			}
+			var names []any
+			for _, name := range host.GetDnsNames() {
+				names = append(names, name)
+			}
+			if len(sets) > 0 {
+				hostMap["set_ids"] = sets
+			}
+			if len(ips) > 0 {
+				hostMap["ip_addresses"] = ips
+			}
+			if len(names) > 0 {
+				hostMap["dns_names"] = names
+			}
+			found = append(found, hostMap)
+		}
+		assert.ElementsMatch(tt.found, found)
+	}
+
+	// First set of tests: check that we can look up sets individually and
+	// together
+	setTests := []testInfo{
+		{
+			name:  "set 1",
+			sets:  []string{"set1"},
+			found: hostInfo1[loopbackPluginHostInfoAttrField].([]any),
+		},
+		{
+			name:  "set 2",
+			sets:  []string{"set2"},
+			found: hostInfo2[loopbackPluginHostInfoAttrField].([]any),
+		},
+		{
+			name: "sets 1 and 2",
+			sets: []string{"set1", "set2"},
+			found: append(hostInfo1[loopbackPluginHostInfoAttrField].([]any),
+				hostInfo2[loopbackPluginHostInfoAttrField].([]any)...),
+		},
+	}
+	for _, tt := range setTests {
+		t.Run(tt.name, func(t *testing.T) {
+			validateSets(t, tt)
+		})
+	}
+}

internal/plugin/loopback/options.go

@@ -0,0 +1,69 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
+	"google.golang.org/grpc/codes"
+)
+
+// PluginMockError is used to mock an error when interacting with an external object store.
+type PluginMockError struct {
+	bucketName   string
+	bucketPrefix string
+	objectKey    string
+	errMsg       string
+	errCode      codes.Code
+}
+
+func (e PluginMockError) match(bucket *storagebuckets.StorageBucket, key string) bool {
+	if key != "" && e.objectKey != key {
+		return false
+	}
+	if e.bucketName != bucket.BucketName {
+		return false
+	}
+	if bucket.BucketPrefix != "" && e.bucketPrefix != bucket.BucketPrefix {
+		return false
+	}
+	return true
+}
+
+type TestOption func(*TestOptions) error
+
+type TestOptions struct {
+	withMockBuckets map[BucketName]Bucket
+	withMockError   []PluginMockError
+}
+
+// getTestOpts - iterate the inbound Options and return a struct
+func getTestOpts(opt ...TestOption) (TestOptions, error) {
+	opts := getDefaultTestOptions()
+	for _, o := range opt {
+		if err := o(&opts); err != nil {
+			return opts, err
+		}
+	}
+	return opts, nil
+}
+
+func getDefaultTestOptions() TestOptions {
+	return TestOptions{}
+}
+
+// WithMockBuckets provides an option to create mocked external object store buckets.
+func WithMockBuckets(buckets map[BucketName]Bucket) TestOption {
+	return func(o *TestOptions) error {
+		o.withMockBuckets = buckets
+		return nil
+	}
+}
+
+// WithMockError provides an option to mock an error when interacting with an external object store.
+func WithMockError(err PluginMockError) TestOption {
+	return func(o *TestOptions) error {
+		o.withMockError = append(o.withMockError, err)
+		return nil
+	}
+}

internal/plugin/loopback/storage.go

@@ -0,0 +1,395 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"io"
+	"sync"
+	"time"
+
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+var _ plgpb.StoragePluginServiceServer = (*TestPluginStorageServer)(nil)
+
+// TestPluginStorageServer provides a storage plugin service server where each method can be overwritten for tests.
+type TestPluginStorageServer struct {
+	OnCreateStorageBucketFn func(context.Context, *plgpb.OnCreateStorageBucketRequest) (*plgpb.OnCreateStorageBucketResponse, error)
+	OnUpdateStorageBucketFn func(context.Context, *plgpb.OnUpdateStorageBucketRequest) (*plgpb.OnUpdateStorageBucketResponse, error)
+	OnDeleteStorageBucketFn func(context.Context, *plgpb.OnDeleteStorageBucketRequest) (*plgpb.OnDeleteStorageBucketResponse, error)
+	ValidatePermissionsFn   func(context.Context, *plgpb.ValidatePermissionsRequest) (*plgpb.ValidatePermissionsResponse, error)
+	HeadObjectFn            func(context.Context, *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error)
+	GetObjectFn             func(*plgpb.GetObjectRequest, plgpb.StoragePluginService_GetObjectServer) error
+	PutObjectFn             func(plgpb.StoragePluginService_PutObjectServer) error
+	plgpb.UnimplementedStoragePluginServiceServer
+}
+
+func (t TestPluginStorageServer) OnCreateStorageBucket(ctx context.Context, req *plgpb.OnCreateStorageBucketRequest) (*plgpb.OnCreateStorageBucketResponse, error) {
+	if t.OnCreateStorageBucketFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.OnCreateStorageBucket(ctx, req)
+	}
+	return t.OnCreateStorageBucketFn(ctx, req)
+}
+
+func (t TestPluginStorageServer) OnUpdateStorageBucket(ctx context.Context, req *plgpb.OnUpdateStorageBucketRequest) (*plgpb.OnUpdateStorageBucketResponse, error) {
+	if t.OnUpdateStorageBucketFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.OnUpdateStorageBucket(ctx, req)
+	}
+	return t.OnUpdateStorageBucketFn(ctx, req)
+}
+
+func (t TestPluginStorageServer) OnDeleteStorageBucket(ctx context.Context, req *plgpb.OnDeleteStorageBucketRequest) (*plgpb.OnDeleteStorageBucketResponse, error) {
+	if t.OnDeleteStorageBucketFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.OnDeleteStorageBucket(ctx, req)
+	}
+	return t.OnDeleteStorageBucketFn(ctx, req)
+}
+
+func (t TestPluginStorageServer) ValidatePermissions(ctx context.Context, req *plgpb.ValidatePermissionsRequest) (*plgpb.ValidatePermissionsResponse, error) {
+	if t.ValidatePermissionsFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.ValidatePermissions(ctx, req)
+	}
+	return t.ValidatePermissionsFn(ctx, req)
+}
+
+func (t TestPluginStorageServer) HeadObject(ctx context.Context, req *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error) {
+	if t.HeadObjectFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.HeadObject(ctx, req)
+	}
+	return t.HeadObjectFn(ctx, req)
+}
+
+func (t TestPluginStorageServer) GetObject(req *plgpb.GetObjectRequest, stream plgpb.StoragePluginService_GetObjectServer) error {
+	if t.GetObjectFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.GetObject(req, stream)
+	}
+	return t.GetObjectFn(req, stream)
+}
+
+func (t TestPluginStorageServer) PutObject(stream plgpb.StoragePluginService_PutObjectServer) error {
+	if t.PutObjectFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.PutObject(stream)
+	}
+	return t.PutObjectFn(stream)
+}
+
+type (
+	BucketName string
+	ObjectName string
+	Chunk      []byte
+)
+
+type Bucket map[ObjectName]*storagePluginStorageInfo
+
+// storagePluginStorageInfo is an in-memory represenation of an object
+// that would be normally stored in an external object store.
+type storagePluginStorageInfo struct {
+	DataChunks []Chunk `mapstructure:"dataChunks"`
+
+	lastModified  *time.Time `mapstructure:"lastModified"`
+	contentLength *int64     `mapstructure:"contentLength"`
+}
+
+// LoopbackStorage provides a storage plugin with functionality useful for certain
+// kinds of testing.
+//
+// It is thread-safe.
+type LoopbackStorage struct {
+	m sync.Mutex
+
+	buckets map[BucketName]Bucket
+	errs    []PluginMockError
+}
+
+func (l *LoopbackStorage) onCreateStorageBucket(ctx context.Context, req *plgpb.OnCreateStorageBucketRequest) (*plgpb.OnCreateStorageBucketResponse, error) {
+	const op = "loopback.(LoopbackPlugin).onCreateStorageBucket"
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.Bucket == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetBucket().Secrets == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	if _, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]; !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+	for _, err := range l.errs {
+		if err.match(req.GetBucket(), "") {
+			return nil, status.Errorf(err.errCode, err.errMsg)
+		}
+	}
+	return &plgpb.OnCreateStorageBucketResponse{
+		Persisted: &plgpb.StorageBucketPersisted{
+			Data: req.GetBucket().GetSecrets(),
+		},
+	}, nil
+}
+
+func (l *LoopbackStorage) onUpdateStorageBucket(ctx context.Context, req *plgpb.OnUpdateStorageBucketRequest) (*plgpb.OnUpdateStorageBucketResponse, error) {
+	const op = "loopback.(LoopbackPlugin).onUpdateStorageBucket"
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.NewBucket == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetNewBucket().Secrets == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	if _, ok := l.buckets[BucketName(req.GetNewBucket().GetBucketName())]; !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+	for _, err := range l.errs {
+		if err.match(req.GetNewBucket(), "") {
+			return nil, status.Errorf(err.errCode, "%s: %s", op, err.errMsg)
+		}
+	}
+	return &plgpb.OnUpdateStorageBucketResponse{
+		Persisted: &plgpb.StorageBucketPersisted{
+			Data: req.GetNewBucket().GetSecrets(),
+		},
+	}, nil
+}
+
+func (l *LoopbackStorage) onDeleteStorageBucket(ctx context.Context, req *plgpb.OnDeleteStorageBucketRequest) (*plgpb.OnDeleteStorageBucketResponse, error) {
+	const op = "loopback.(LoopbackPlugin).onDeleteStorageBucket"
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.Bucket == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetBucket().Secrets == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	return &plgpb.OnDeleteStorageBucketResponse{}, nil
+}
+
+func (l *LoopbackStorage) validatePermissions(ctx context.Context, req *plgpb.ValidatePermissionsRequest) (*plgpb.ValidatePermissionsResponse, error) {
+	const op = "loopback.(LoopbackPlugin).validatePermissions"
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.Bucket == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetBucket().Secrets == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	if _, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]; !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+	for _, err := range l.errs {
+		if err.match(req.GetBucket(), "") {
+			return nil, status.Errorf(err.errCode, "%s: %s", op, err.errMsg)
+		}
+	}
+	return &plgpb.ValidatePermissionsResponse{}, nil
+}
+
+func (l *LoopbackStorage) headObject(ctx context.Context, req *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error) {
+	const op = "loopback.(LoopbackPlugin).headObject"
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.Bucket == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "$%s: missing storage bucket", op)
+	}
+	if req.GetBucket().Secrets == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	if req.Key == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "%s; missing object key", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	bucket, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]
+	if !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+	objectPath := ObjectName(req.GetBucket().GetBucketPrefix() + req.GetKey())
+	object, ok := bucket[objectPath]
+	if !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: object %s not found", op, objectPath)
+	}
+	for _, err := range l.errs {
+		if err.match(req.GetBucket(), req.GetKey()) {
+			return nil, status.Errorf(err.errCode, "%s: %s", op, err.errMsg)
+		}
+	}
+	return &plgpb.HeadObjectResponse{
+		ContentLength: *object.contentLength,
+		LastModified:  timestamppb.New(*object.lastModified),
+	}, nil
+}
+
+func (l *LoopbackStorage) getObject(req *plgpb.GetObjectRequest, stream plgpb.StoragePluginService_GetObjectServer) error {
+	const op = "loopback.(LoopbackPlugin).getObject"
+	if req == nil {
+		return status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.Bucket == nil {
+		return status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetBucket().Secrets == nil {
+		return status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)
+	}
+	if req.Key == "" {
+		return status.Errorf(codes.InvalidArgument, "%s; missing object key", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	bucket, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]
+	if !ok {
+		return status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+	objectPath := ObjectName(req.GetBucket().GetBucketPrefix() + req.GetKey())
+	object, ok := bucket[objectPath]
+	if !ok {
+		return status.Errorf(codes.NotFound, "%s: object %s not found", op, objectPath)
+	}
+	for _, err := range l.errs {
+		if err.match(req.GetBucket(), req.GetKey()) {
+			return status.Errorf(err.errCode, "%s: %s", op, err.errMsg)
+		}
+	}
+	go func() {
+		for _, chunk := range object.DataChunks {
+			if err := stream.Send(&plgpb.GetObjectResponse{
+				FileChunk: chunk,
+			}); err != nil {
+				stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to send object data: %v", op, err))
+				return
+			}
+		}
+		stream.SendMsg(io.EOF)
+	}()
+	return nil
+}
+
+func (l *LoopbackStorage) putObject(stream plgpb.StoragePluginService_PutObjectServer) error {
+	const op = "loopback.(LoopbackPlugin).putObject"
+	if stream == nil {
+		return status.Errorf(codes.Internal, "%s: missing stream", op)
+	}
+	go func() {
+		var (
+			request      *plgpb.PutObjectRequest_Request
+			objectChunks []Chunk
+			objectData   []byte
+		)
+		for {
+			req, err := stream.Recv()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				_ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to recv message from stream: %v", op, err))
+				return
+			}
+			switch data := req.Data.(type) {
+			case *plgpb.PutObjectRequest_Request:
+				if data.Request == nil {
+					_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing request metadata", op))
+					return
+				}
+				if data.Request.GetBucket() == nil {
+					_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op))
+					return
+				}
+				if data.Request.GetBucket().GetSecrets() == nil {
+					_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing secrets", op))
+					return
+				}
+				if data.Request.Key == "" {
+					_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s; missing object key", op))
+					return
+				}
+				request = data
+			case *plgpb.PutObjectRequest_FileChunk:
+				objectChunks = append(objectChunks, data.FileChunk)
+				objectData = append(objectData, data.FileChunk...)
+			case nil:
+				continue
+			default:
+				_ = stream.SendMsg(status.Errorf(codes.Internal, "%s: unknown message type: %v", op, data))
+				return
+			}
+		}
+		if request == nil {
+			_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: request is nil", op))
+			return
+		}
+		if len(objectData) <= 0 {
+			_ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing object data", op))
+			return
+		}
+		l.m.Lock()
+		defer l.m.Unlock()
+		bucket, ok := l.buckets[BucketName(request.Request.GetBucket().GetBucketName())]
+		if !ok {
+			_ = stream.SendMsg(status.Errorf(codes.NotFound, "%s: bucket not found", op))
+			return
+		}
+		for _, err := range l.errs {
+			if err.match(request.Request.GetBucket(), request.Request.GetKey()) {
+				_ = stream.SendMsg(status.Errorf(err.errCode, "%s: %s", op, err.errMsg))
+				return
+			}
+		}
+		hash := sha256.New()
+		if _, err := io.Copy(hash, bytes.NewReader(objectData)); err != nil {
+			_ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to hash object: %v", op, err))
+			return
+		}
+		contentLength := int64(len(objectData))
+		lastModified := time.Now()
+		objectPath := ObjectName(request.Request.GetBucket().GetBucketPrefix() + request.Request.GetKey())
+		bucket[objectPath] = &storagePluginStorageInfo{
+			DataChunks:    objectChunks,
+			contentLength: &contentLength,
+			lastModified:  &lastModified,
+		}
+		if err := stream.SendAndClose(&plgpb.PutObjectResponse{
+			ChecksumSha_256: hash.Sum(nil),
+		}); err != nil {
+			_ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to close stream: %v", op, err))
+			return
+		}
+	}()
+	return nil
+}
+
+func MockObject(data []Chunk) *storagePluginStorageInfo {
+	lastModified := time.Now()
+	dataChunks := make([]Chunk, len(data))
+	var contentLength int64
+	for i, chunk := range data {
+		dataChunks[i] = copyBytes(chunk)
+		contentLength += int64(len(chunk))
+	}
+	return &storagePluginStorageInfo{
+		DataChunks:    dataChunks,
+		contentLength: &contentLength,
+		lastModified:  &lastModified,
+	}
+}
+
+func copyBytes(in []byte) []byte {
+	return append([]byte{}, in...)
+}

internal/plugin/loopback/testing_grpc_stream.go

@@ -0,0 +1,502 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"sync"
+
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"google.golang.org/grpc/metadata"
+)
+
+// getObjectStreamResponse is used to mock a message sent from the server to the client.
+type getObjectStreamResponse struct {
+	msg *plgpb.GetObjectResponse
+	err error
+}
+
+// getObjectStream is used to mock the interactions between
+// the client and server for the GetObject method.
+type getObjectStream struct {
+	client plgpb.StoragePluginService_GetObjectClient
+	server plgpb.StoragePluginService_GetObjectServer
+
+	// messages is used to mock the server sending messages to the client.
+	messages chan *getObjectStreamResponse
+
+	m            *sync.Mutex
+	streamClosed bool
+}
+
+// IsStreamClosed returns true if the stream is closed.
+func (s *getObjectStream) IsStreamClosed() bool {
+	s.m.Lock()
+	defer s.m.Unlock()
+	return s.streamClosed
+}
+
+// Close closes the channels of the stream and sets the streamClosed flag to true.
+// A closeStream is used to prevent the channels from being closed multiple times.
+func (s *getObjectStream) Close() {
+	s.m.Lock()
+	defer s.m.Unlock()
+	if s.streamClosed {
+		return
+	}
+	close(s.messages)
+	s.streamClosed = true
+}
+
+// getObjectClient is used to mock the client stream
+// interactions for the GetObject method.
+type getObjectClient struct {
+	// sentFromServer is used to mock the server sending messages to the client.
+	sentFromServer chan *getObjectStreamResponse
+
+	// closeStream is used to close the channels of the stream.
+	// This is used to prevent the channels from being closed multiple times.
+	// This is needed because the channel can be closed by the client or the server.
+	closeStream func()
+
+	// isStreamClosed is used to check if the stream is closed.
+	// This is needed because the channel can be closed by the client or the server.
+	isStreamClosed func() bool
+}
+
+// Recv will block until a message is received from the server.
+// Recv will return io.EOF if the server closes the stream.
+// Recv will return an error if the server sends an error.
+func (c *getObjectClient) Recv() (*plgpb.GetObjectResponse, error) {
+	resp, ok := <-c.sentFromServer
+	if !ok {
+		return nil, io.EOF
+	}
+	return resp.msg, resp.err
+}
+
+// Header should not be used.
+// Header is implemeted to satisfy the grpc.ClientStream interface.
+// Header will always return an empty metadata and an nil error.
+func (c *getObjectClient) Header() (metadata.MD, error) {
+	return make(metadata.MD), nil
+}
+
+// Trailer should not be used.
+// Trailer is implemeted to satisfy the grpc.ClientStream interface.
+// Trailer will always return an empty metadata.
+func (c *getObjectClient) Trailer() metadata.MD {
+	return make(metadata.MD)
+}
+
+// CloseSend will close the channel used to retrieve messages from
+// the server. This will cause the Recv method to return io.EOF.
+// CloseSend will return an error if the channel is already closed.
+func (c *getObjectClient) CloseSend() error {
+	if c.isStreamClosed() {
+		return fmt.Errorf("stream is closed")
+	}
+	c.closeStream()
+	return nil
+}
+
+// Context will always return a Background context.
+func (c *getObjectClient) Context() context.Context {
+	return context.Background()
+}
+
+// SendMsg should not be used.
+// SendMsg is implemeted to satisfy the grpc.ClientStream interface.
+// SendMsg will always return an nil error.
+func (c *getObjectClient) SendMsg(m interface{}) error {
+	return nil
+}
+
+// RecvMsg should not be used.
+// RecvMsg is implemeted to satisfy the grpc.ClientStream interface.
+// RecvMsg will always return an nil error.
+func (c *getObjectClient) RecvMsg(m interface{}) error {
+	return nil
+}
+
+// getObjectServer is used to mock the server stream
+// interactions for the GetObject method.
+type getObjectServer struct {
+	// sendToClient is used to mock the server sending messages to the client.
+	sendToClient chan *getObjectStreamResponse
+
+	// closeStream is used to close the channels of the stream.
+	// This is used to prevent the channels from being closed multiple times.
+	// This is needed because the channel can be closed by the client or the server.
+	closeStream func()
+
+	// isStreamClosed is used to check if the stream is closed.
+	// This is needed because the channel can be closed by the client or the server.
+	isStreamClosed func() bool
+}
+
+// Send will send a message to the client.
+// Send will return an error if the client closes the stream.
+// Send will return an error if the response is nil.
+func (s *getObjectServer) Send(resp *plgpb.GetObjectResponse) error {
+	if resp == nil {
+		return fmt.Errorf(`parameter arg "resp GetObjectResponse" cannot be nil`)
+	}
+	if s.isStreamClosed() {
+		return fmt.Errorf("stream is closed")
+	}
+	s.sendToClient <- &getObjectStreamResponse{msg: resp}
+	return nil
+}
+
+// SetHeader should not be used.
+// SetHeader is implemeted to satisfy the grpc.ServerStream interface.
+// SetHeader will always return an nil error.
+func (s *getObjectServer) SetHeader(metadata.MD) error {
+	return nil
+}
+
+// SendHeader should not be used.
+// SendHeader is implemeted to satisfy the grpc.ServerStream interface.
+// SendHeader will always return an nil error.
+func (s *getObjectServer) SendHeader(metadata.MD) error {
+	return nil
+}
+
+// SetTrailer should not be used.
+// SetTrailer is implemeted to satisfy the grpc.ServerStream interface.
+// SetTrailer will always return an nil error.
+func (s *getObjectServer) SetTrailer(metadata.MD) {
+}
+
+// Context will always return a Background context.
+func (s *getObjectServer) Context() context.Context {
+	return context.Background()
+}
+
+// SendMsg allows sending GetObjectResponse messages to the client.
+// SendMsg allows sending errors other than io.EOF to the client.
+// Sending an error message will close the stream.
+// SendMsg returns an invalid argument error if the message is not
+// an error or GetObjectResponse.
+// SendMsg will return an error if the stream is closed.
+func (s *getObjectServer) SendMsg(m interface{}) error {
+	switch msg := m.(type) {
+	case *plgpb.GetObjectResponse:
+		if s.isStreamClosed() {
+			return fmt.Errorf("stream is closed")
+		}
+		s.sendToClient <- &getObjectStreamResponse{msg: msg}
+	case error:
+		if s.isStreamClosed() {
+			return fmt.Errorf("stream is closed")
+		}
+		defer s.closeStream()
+		s.sendToClient <- &getObjectStreamResponse{err: msg}
+	default:
+		return fmt.Errorf("invalid argument %v", m)
+	}
+	return nil
+}
+
+// RecvMsg should not be used.
+// RecvMsg is implemeted to satisfy the grpc.ServerStream interface.
+// RecvMsg will always return an nil error.
+func (s *getObjectServer) RecvMsg(m interface{}) error {
+	return nil
+}
+
+// newGetObjectStream will create a mock stream for the GetObject method.
+// The client and server stream is mocked by creating a GetObjectResponse
+// channel and an error channel that is shared between the client and server.
+func newGetObjectStream() *getObjectStream {
+	stream := &getObjectStream{
+		m:        new(sync.Mutex),
+		messages: make(chan *getObjectStreamResponse),
+	}
+	stream.client = &getObjectClient{
+		sentFromServer: stream.messages,
+		closeStream:    stream.Close,
+		isStreamClosed: stream.IsStreamClosed,
+	}
+	stream.server = &getObjectServer{
+		sendToClient:   stream.messages,
+		closeStream:    stream.Close,
+		isStreamClosed: stream.IsStreamClosed,
+	}
+	return stream
+}
+
+// putObjectStreamRequest is used to mock a message sent from the client to the server.
+type putObjectStreamRequest struct {
+	msg *plgpb.PutObjectRequest
+	err error
+}
+
+// puObjectStreamResponse is used to mock a message sent from the server to the client.
+type putObjectStreamResponse struct {
+	msg *plgpb.PutObjectResponse
+	err error
+}
+
+// putObjectStream is used to mock the interactions between
+// the client and server for the PutObject method.
+type putObjectStream struct {
+	client plgpb.StoragePluginService_PutObjectClient
+	server plgpb.StoragePluginService_PutObjectServer
+
+	// requests is used to mock the client sending messages to the server.
+	requests chan *putObjectStreamRequest
+
+	// responses is used to mock the server sending messages to the client.
+	responses chan *putObjectStreamResponse
+
+	m            *sync.Mutex
+	clientClosed bool
+	serverClosed bool
+}
+
+func (s *putObjectStream) CloseClient() {
+	s.m.Lock()
+	defer s.m.Unlock()
+	if s.clientClosed {
+		return
+	}
+	close(s.requests)
+	s.clientClosed = true
+}
+
+func (s *putObjectStream) IsClientClosed() bool {
+	s.m.Lock()
+	defer s.m.Unlock()
+	return s.clientClosed
+}
+
+func (s *putObjectStream) CloseServer() {
+	s.m.Lock()
+	defer s.m.Unlock()
+	if s.serverClosed {
+		return
+	}
+	close(s.responses)
+	s.serverClosed = true
+}
+
+func (s *putObjectStream) IsServerClosed() bool {
+	s.m.Lock()
+	defer s.m.Unlock()
+	return s.serverClosed
+}
+
+// putObjectClient is used to mock the client stream
+// interactions for the PutObject method.
+type putObjectClient struct {
+	sendToServer   chan *putObjectStreamRequest
+	sentFromServer chan *putObjectStreamResponse
+
+	// closeStream is used to close the channels of the stream.
+	// This is used to prevent the channels from being closed multiple times.
+	// This is needed because the channel can be closed by the client or the server.
+	closeStream func()
+
+	// isStreamClosed is used to check if the stream is closed.
+	// This is needed because the channel can be closed by the client or the server.
+	isStreamClosed func() bool
+}
+
+// Send will send a message to the server.
+// Send will return an error if the request is nil.
+// Send will return an error if the stream is closed.
+func (c *putObjectClient) Send(req *plgpb.PutObjectRequest) error {
+	if req == nil {
+		return fmt.Errorf(`parameter arg "req PutObjectRequest" cannot be nil`)
+	}
+	if c.isStreamClosed() {
+		return fmt.Errorf("stream is closed")
+	}
+	c.sendToServer <- &putObjectStreamRequest{msg: req}
+	return nil
+}
+
+// CloseAndRecv will return an PutObjectResponse if the server returns one.
+// CloseAndRecv will return an error if the server returns an error.
+// CloseAndRecv will return an io.EOF error if the server closes the stream.
+// CloseAndRecv will close the stream used to send messages to the server.
+func (c *putObjectClient) CloseAndRecv() (*plgpb.PutObjectResponse, error) {
+	c.closeStream()
+	resp, ok := <-c.sentFromServer
+	if !ok {
+		return nil, io.EOF
+	}
+	return resp.msg, resp.err
+}
+
+// Header should not be used.
+// Header is implemeted to satisfy the grpc.ClientStream interface.
+// Header will always return an empty metadata and an nil error.
+func (c *putObjectClient) Header() (metadata.MD, error) {
+	return make(metadata.MD), nil
+}
+
+// Trailer should not be used.
+// Trailer is implemeted to satisfy the grpc.ClientStream interface.
+// Trailer will always return an empty metadata.
+func (c *putObjectClient) Trailer() metadata.MD {
+	return make(metadata.MD)
+}
+
+// CloseSend will close the channel used to retrieve messages from
+// the server. This will cause the CloseAndRecv method to return io.EOF.
+// CloseSend will always return a nill error.
+func (c *putObjectClient) CloseSend() error {
+	if c.isStreamClosed() {
+		return fmt.Errorf("stream is closed")
+	}
+	c.closeStream()
+	return nil
+}
+
+// Context will always return a Background context.
+func (c *putObjectClient) Context() context.Context {
+	return context.Background()
+}
+
+// SendMsg should not be used.
+// SendMsg is implemeted to satisfy the grpc.ClientStream interface.
+// SendMsg will always return an nil error.
+func (c *putObjectClient) SendMsg(m interface{}) error {
+	return nil
+}
+
+// RecvMsg should not be used.
+// RecvMsg is implemeted to satisfy the grpc.ClientStream interface.
+// RecvMsg will always return an nil error.
+func (c *putObjectClient) RecvMsg(m interface{}) error {
+	return nil
+}
+
+// putObjectServer is used to mock the server stream
+// interactions for the PutObject method.
+type putObjectServer struct {
+	sentFromClient chan *putObjectStreamRequest
+	sentToClient   chan *putObjectStreamResponse
+
+	// closeStream is used to close the channels of the stream.
+	// This is used to prevent the channels from being closed multiple times.
+	// This is needed because the channel can be closed by the client or the server.
+	closeStream func()
+
+	// isStreamClosed is used to check if the stream is closed.
+	// This is needed because the channel can be closed by the client or the server.
+	isStreamClosed func() bool
+}
+
+// SendAndClose will send a PutObjectResponse to the client.
+// SendAndClose will return an error if the response is nil.
+// SendAndClose will close the stream used to send messages from the client.
+// SendAndClose will return an error if the stream is closed.
+func (s *putObjectServer) SendAndClose(resp *plgpb.PutObjectResponse) error {
+	if resp == nil {
+		return fmt.Errorf(`parameter arg "resp PutObjectResponse" cannot be nil`)
+	}
+	if s.isStreamClosed() {
+		return fmt.Errorf("stream is closed")
+	}
+	s.sentToClient <- &putObjectStreamResponse{msg: resp}
+	s.closeStream()
+	return nil
+}
+
+// Recv will read a PutObjectRequest from the stream.
+// Recv will return an io.EOF error if the stream is closed.
+func (s *putObjectServer) Recv() (*plgpb.PutObjectRequest, error) {
+	req, ok := <-s.sentFromClient
+	if !ok {
+		return nil, io.EOF
+	}
+	return req.msg, req.err
+}
+
+// SetHeader should not be used.
+// SetHeader is implemeted to satisfy the grpc.ServerStream interface.
+// SetHeader will always return an nil error.
+func (s *putObjectServer) SetHeader(metadata.MD) error {
+	return nil
+}
+
+// SendHeader should not be used.
+// SendHeader is implemeted to satisfy the grpc.ServerStream interface.
+// SendHeader will always return an nil error.
+func (s *putObjectServer) SendHeader(metadata.MD) error {
+	return nil
+}
+
+// SetTrailer should not be used.
+// SetTrailer is implemeted to satisfy the grpc.ServerStream interface.
+func (s *putObjectServer) SetTrailer(metadata.MD) {
+}
+
+// Context will always return a Background context.
+func (s *putObjectServer) Context() context.Context {
+	return context.Background()
+}
+
+// SendMsg allows sending errors other than io.EOF to the client.
+// Sending an error message will close the stream.
+// SendMsg allows sending PutObjectResponse messages to the client.
+// SendMsg returns an invalid argument error if the message is not
+// an error or PutObjectResponse.
+// SendMsg will return an error if the stream is closed.
+func (s *putObjectServer) SendMsg(m interface{}) error {
+	switch msg := m.(type) {
+	case *plgpb.PutObjectResponse:
+		if s.isStreamClosed() {
+			return fmt.Errorf("stream is closed")
+		}
+		s.sentToClient <- &putObjectStreamResponse{msg: msg}
+	case error:
+		if s.isStreamClosed() {
+			return fmt.Errorf("stream is closed")
+		}
+		s.sentToClient <- &putObjectStreamResponse{err: msg}
+		defer s.closeStream()
+	default:
+		return fmt.Errorf("invalid argument %v", m)
+	}
+	return nil
+}
+
+// RecvMsg should not be used.
+// RecvMsg is implemeted to satisfy the grpc.ServerStream interface.
+// RecvMsg will always return an nil error.
+func (s *putObjectServer) RecvMsg(m interface{}) error {
+	return nil
+}
+
+// newPutObjectStream will create a mock stream for the PutObject method.
+// The client and server stream is mocked by creating a PutObjectRequest,
+// PutObjectResponse and an error channel that is shared between the client
+// and server.
+func newPutObjectStream() *putObjectStream {
+	stream := &putObjectStream{
+		m:         new(sync.Mutex),
+		requests:  make(chan *putObjectStreamRequest),
+		responses: make(chan *putObjectStreamResponse),
+	}
+	stream.client = &putObjectClient{
+		sendToServer:   stream.requests,
+		sentFromServer: stream.responses,
+		closeStream:    stream.CloseClient,
+		isStreamClosed: stream.IsClientClosed,
+	}
+	stream.server = &putObjectServer{
+		sentFromClient: stream.requests,
+		sentToClient:   stream.responses,
+		closeStream:    stream.CloseServer,
+		isStreamClosed: stream.IsServerClosed,
+	}
+	return stream
+}

internal/plugin/loopback/testing_grpc_stream_test.go

@@ -0,0 +1,602 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package loopback
+
+import (
+	"fmt"
+	"io"
+	"sync"
+	"testing"
+
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	tr "github.com/stretchr/testify/require"
+)
+
+func Test_GetObjectStream_Client(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newGetObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.client)
+
+	// Validate Header is not nil and does not return an error
+	header, err := stream.client.Header()
+	require.NotNil(header)
+	require.NoError(err)
+
+	// Validate Trailer is not nil
+	trailer := stream.client.Trailer()
+	require.NotNil(trailer)
+
+	// Validate Context is not nil
+	ctx := stream.client.Context()
+	require.NotNil(ctx)
+
+	// validate SendMsg does not write to the channel messages
+	err = stream.client.SendMsg(nil)
+	require.NoError(err)
+	require.Empty(stream.messages)
+
+	// Validate RecvMsg does not read from the channel messages
+	streamSize := len(stream.messages)
+	var msg *getObjectStreamResponse
+	err = stream.client.RecvMsg(msg)
+	require.NoError(err)
+	require.Nil(msg)
+	require.Equal(streamSize, len(stream.messages))
+
+	// Validate Recv reads a message from the channel messages
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		stream.messages <- &getObjectStreamResponse{
+			msg: &plgpb.GetObjectResponse{},
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		resp, err := stream.client.Recv()
+		require.NoError(err)
+		require.NotNil(resp)
+	}()
+	wg.Wait()
+
+	// Validates Recv reads a error from the channel messages
+	wg.Add(1)
+	go func() {
+		stream.messages <- &getObjectStreamResponse{
+			err: fmt.Errorf("mock error"),
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		resp, err := stream.client.Recv()
+		require.Error(err)
+		require.Nil(resp)
+	}()
+	wg.Wait()
+
+	// Validate CloseSend does not return an error
+	// Validate streamClosed is set to true
+	// Validate the channel messages is closed
+	var ok bool
+	err = stream.client.CloseSend()
+	require.NoError(err)
+	require.True(stream.streamClosed)
+	_, ok = <-stream.messages
+	require.False(ok)
+
+	// Validate CloseSend does return an error when the channel is closed
+	// Validate streamClosed is set to true
+	// Validate the channel messages is closed
+	err = stream.client.CloseSend()
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+	require.True(stream.streamClosed)
+	_, ok = <-stream.messages
+	require.False(ok)
+
+	// Validate Recv returns EOF error when the channel is closed
+	resp, err := stream.client.Recv()
+	require.Equal(io.EOF, err)
+	require.Nil(resp)
+}
+
+func Test_GetObjectStream_Server(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newGetObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.server)
+
+	// Validate SetHeader does not return an error
+	err := stream.server.SetHeader(nil)
+	require.NoError(err)
+
+	// Validate SendHeader does not return an error
+	err = stream.server.SendHeader(nil)
+	require.NoError(err)
+
+	// Validate Context returns a context
+	ctx := stream.server.Context()
+	require.NotNil(ctx)
+
+	// Validate RecvMsg does not read from the channel messages
+	streamSize := len(stream.messages)
+	var msg *getObjectStreamResponse
+	err = stream.server.RecvMsg(msg)
+	require.NoError(err)
+	require.Nil(msg)
+	require.Equal(streamSize, len(stream.messages))
+
+	// Validate Send writes to the channel messages
+	var wg sync.WaitGroup
+	go func() {
+		err = stream.server.Send(&plgpb.GetObjectResponse{})
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.messages
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.msg)
+		require.Nil(resp.err)
+	}()
+	wg.Wait()
+
+	// Validate Send returns an error when the channel is closed
+	stream.Close()
+	require.True(stream.streamClosed)
+	err = stream.server.Send(&plgpb.GetObjectResponse{})
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+
+	// Validate SendMsg returns an error when the channel is closed
+	err = stream.server.SendMsg(&plgpb.GetObjectResponse{})
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+	err = stream.server.SendMsg(fmt.Errorf("mock error"))
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+
+	// Reset stream
+	stream = newGetObjectStream()
+
+	// Validate SendMsg returns an error for unknown message type
+	err = stream.server.SendMsg(map[string]string{})
+	require.False(stream.streamClosed)
+	require.Error(err)
+	require.Contains(err.Error(), "invalid argument")
+
+	// Validate SendMsg writes to the channel messages for GetObjectResponse type
+	go func() {
+		err = stream.server.SendMsg(&plgpb.GetObjectResponse{})
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.messages
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.msg)
+		require.Nil(resp.err)
+	}()
+	wg.Wait()
+
+	// Validate SendMsg closes the channel messages for error type
+	go func() {
+		err = stream.server.SendMsg(fmt.Errorf("mock error"))
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.messages
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.err)
+		require.Nil(resp.msg)
+	}()
+	wg.Wait()
+	require.True(stream.streamClosed)
+	_, ok := <-stream.messages
+	require.False(ok)
+}
+
+func TestNewGetObjectStream(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newGetObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.m)
+	require.NotNil(stream.client)
+	require.NotNil(stream.server)
+	require.NotNil(stream.messages)
+	require.False(stream.streamClosed)
+	require.False(stream.IsStreamClosed())
+
+	// Validate Close() closes the channel messages
+	stream.Close()
+	require.True(stream.streamClosed)
+	require.True(stream.IsStreamClosed())
+	_, ok := <-stream.messages
+	require.False(ok)
+}
+
+func Test_PutObjectStream_Client(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newPutObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.client)
+
+	// Validate Header is not nil and does not return an error
+	header, err := stream.client.Header()
+	require.NotNil(header)
+	require.NoError(err)
+
+	// Validate Trailer is not nil
+	trailer := stream.client.Trailer()
+	require.NotNil(trailer)
+
+	// Validate Context is not nil
+	ctx := stream.client.Context()
+	require.NotNil(ctx)
+
+	// Validate SendMsg does not write to the channel requests
+	err = stream.client.SendMsg(nil)
+	require.NoError(err)
+	require.Empty(stream.requests)
+
+	// Validate RecvMsg does not read from the channel requests
+	streamSize := len(stream.requests)
+	var msg *putObjectStreamResponse
+	err = stream.client.RecvMsg(msg)
+	require.NoError(err)
+	require.Nil(msg)
+	require.Equal(streamSize, len(stream.requests))
+
+	// Validate Send returns an error when the request is nil
+	err = stream.client.Send(nil)
+	require.Error(err)
+
+	// Validate Send writes to the channel requests
+	var wg sync.WaitGroup
+	go func() {
+		err = stream.client.Send(&plgpb.PutObjectRequest{})
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		req, ok := <-stream.requests
+		require.True(ok)
+		require.NotNil(req)
+		require.NotNil(req.msg)
+		require.Nil(req.err)
+	}()
+	wg.Wait()
+
+	// Validate CloseSend does not return an error
+	// Validate clientClosed is set to true
+	// Validate the channel requests is closed
+	var ok bool
+	err = stream.client.CloseSend()
+	require.NoError(err)
+	require.True(stream.clientClosed)
+	_, ok = <-stream.requests
+	require.False(ok)
+
+	// Validate CloseSend does return an error when the channel is closed
+	// Validate clientClosed is set to true
+	// Validate the channel requests is closed
+	err = stream.client.CloseSend()
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+	require.True(stream.clientClosed)
+	_, ok = <-stream.requests
+	require.False(ok)
+
+	// Validate Send returns an error when the channel is closed
+	err = stream.client.Send(&plgpb.PutObjectRequest{})
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+	require.Empty(stream.requests)
+
+	// Reset stream
+	stream = newPutObjectStream()
+
+	// Validate CloseAndRecv returns an EOF error when the channel responses is closed
+	// Validate CloseAndRecv closes the channel requests
+	stream.CloseServer()
+	resp, err := stream.client.CloseAndRecv()
+	require.Equal(io.EOF, err)
+	require.Nil(resp)
+	require.True(stream.clientClosed)
+	require.True(stream.IsClientClosed())
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+}
+
+func Test_PutObjectStream_Server(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newPutObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.server)
+
+	// Validate SetHeader does not return an error
+	err := stream.server.SetHeader(nil)
+	require.NoError(err)
+
+	// Validate SendHeader does not return an error
+	err = stream.server.SendHeader(nil)
+	require.NoError(err)
+
+	// Validate Context returns a context
+	ctx := stream.server.Context()
+	require.NotNil(ctx)
+
+	// Validate RecvMsg does not read from the channel requests
+	streamSize := len(stream.requests)
+	var msg *putObjectStreamRequest
+	err = stream.server.RecvMsg(msg)
+	require.NoError(err)
+	require.Nil(msg)
+	require.Equal(streamSize, len(stream.requests))
+
+	// Validate SendMsg returns an error for unknown message types
+	err = stream.server.SendMsg(map[string]string{})
+	require.Error(err)
+	require.Contains(err.Error(), "invalid argument")
+
+	// Validate SendMsg does write a message to the channel responses
+	var wg sync.WaitGroup
+	go func() {
+		err = stream.server.SendMsg(&plgpb.PutObjectResponse{})
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.responses
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.msg)
+		require.Nil(resp.err)
+	}()
+	wg.Wait()
+
+	// Validate SendMsg does write a error to the channel responses
+	// Validate SendMsg closes the channel responses
+	go func() {
+		err = stream.server.SendMsg(fmt.Errorf("mock error"))
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.responses
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.err)
+		require.Nil(resp.msg)
+	}()
+	wg.Wait()
+	_, ok := <-stream.responses
+	require.False(ok)
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+
+	// Validate SendMsg does return an error when the channel responses is closed
+	err = stream.server.SendMsg(&plgpb.PutObjectResponse{})
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+	err = stream.server.SendMsg(fmt.Errorf("mock error"))
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+
+	// Validate SendAndClose returns an error when response is nil
+	err = stream.server.SendAndClose(nil)
+	require.Error(err)
+
+	// Validate SendAndClose returns an error when the channel responses is closed
+	err = stream.server.SendAndClose(&plgpb.PutObjectResponse{})
+	require.Error(err)
+	require.Contains(err.Error(), "stream is closed")
+
+	// Reset stream
+	stream = newPutObjectStream()
+
+	// Validate SendAndClose closes the responses channel
+	go func() {
+		err = stream.server.SendAndClose(&plgpb.PutObjectResponse{})
+		require.NoError(err)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		resp, ok := <-stream.responses
+		require.True(ok)
+		require.NotNil(resp)
+		require.NotNil(resp.msg)
+		require.Nil(resp.err)
+	}()
+	wg.Wait()
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+	_, ok = <-stream.responses
+	require.False(ok)
+}
+
+func TestNewPutObjectStream(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newPutObjectStream()
+	require.NotNil(stream)
+	require.NotNil(stream.m)
+	require.NotNil(stream.client)
+	require.NotNil(stream.server)
+	require.NotNil(stream.requests)
+	require.NotNil(stream.responses)
+	require.False(stream.clientClosed)
+	require.False(stream.IsClientClosed())
+	require.False(stream.serverClosed)
+	require.False(stream.IsServerClosed())
+
+	// Validate CloseClient() closes the requests channel
+	stream.CloseClient()
+	require.True(stream.clientClosed)
+	require.True(stream.IsClientClosed())
+	_, ok := <-stream.requests
+	require.False(ok)
+
+	// Validate CloseServer() closes the responses channel
+	stream.CloseServer()
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+	_, ok = <-stream.responses
+	require.False(ok)
+}
+
+func Test_GetObjectStream(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newGetObjectStream()
+
+	// Validate the server receives an error when the client closes the stream
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			err := stream.server.Send(&plgpb.GetObjectResponse{})
+			if err == nil {
+				continue
+			}
+			require.Error(err)
+			require.Contains(err.Error(), "stream is closed")
+			break
+		}
+	}()
+	stream.client.CloseSend()
+	wg.Wait()
+	require.True(stream.streamClosed)
+	require.True(stream.IsStreamClosed())
+	_, ok := <-stream.messages
+	require.False(ok)
+
+	// Reset Stream
+	stream = newGetObjectStream()
+
+	// Validate the client recieves an error when the server sends an error
+	// Validate the channel messages is closed
+	go func() {
+		err := stream.server.SendMsg(fmt.Errorf("mock error"))
+		require.NoError(err)
+	}()
+	resp, err := stream.client.Recv()
+	require.Nil(resp)
+	require.Error(err)
+	require.Contains(err.Error(), "mock error")
+	require.True(stream.streamClosed)
+	require.True(stream.IsStreamClosed())
+	_, ok = <-stream.messages
+	require.False(ok)
+}
+
+func Test_PutObjectStream(t *testing.T) {
+	require := tr.New(t)
+
+	stream := newPutObjectStream()
+
+	// Validate the client recieves the response when the server closes the stream
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for {
+			resp, err := stream.client.CloseAndRecv()
+			require.NotNil(resp)
+			require.NoError(err)
+			break
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		err := stream.server.SendAndClose(&plgpb.PutObjectResponse{})
+		require.NoError(err)
+	}()
+	wg.Wait()
+	require.True(stream.clientClosed)
+	require.True(stream.IsClientClosed())
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+	_, ok := <-stream.requests
+	require.False(ok)
+	_, ok = <-stream.responses
+	require.False(ok)
+
+	// Reset Stream
+	stream = newPutObjectStream()
+
+	// Validate the client recieves an error when the server sends an error
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for {
+			resp, err := stream.client.CloseAndRecv()
+			require.Nil(resp)
+			require.Error(err)
+			require.Contains(err.Error(), "mock error")
+			break
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		err := stream.server.SendMsg(fmt.Errorf("mock error"))
+		require.NoError(err)
+	}()
+	wg.Wait()
+	require.True(stream.clientClosed)
+	require.True(stream.IsClientClosed())
+	require.True(stream.serverClosed)
+	require.True(stream.IsServerClosed())
+	_, ok = <-stream.requests
+	require.False(ok)
+	_, ok = <-stream.responses
+	require.False(ok)
+
+	// Reset Stream
+	stream = newPutObjectStream()
+
+	// Validate the server recieves an EOF error when the client closes the stream
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for {
+			req, err := stream.server.Recv()
+			if err != nil {
+				require.Nil(req)
+				require.Equal(io.EOF, err)
+				break
+			}
+		}
+	}()
+	go func() {
+		wg.Done()
+		err := stream.client.CloseSend()
+		require.NoError(err)
+	}()
+	wg.Wait()
+	require.True(stream.clientClosed)
+	require.True(stream.IsClientClosed())
+	require.False(stream.serverClosed)
+	require.False(stream.IsServerClosed())
+	_, ok = <-stream.requests
+	require.False(ok)
+}