Adding server tests

agent/grpc-external/services/configentry/server_test.go

@@ -0,0 +1,306 @@
+package configentry
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/armon/go-metrics"
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/acl/resolver"
+	"github.com/hashicorp/consul/agent/grpc-external/testutils"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/proto/private/pbconfigentry"
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+type MockBackend struct {
+	mock.Mock
+	authorizer acl.Authorizer
+}
+
+func (m *MockBackend) ResolveTokenAndDefaultMeta(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) (resolver.Result, error) {
+	return resolver.Result{Authorizer: m.authorizer}, nil
+}
+
+func (m *MockBackend) EnterpriseCheckPartitions(partition string) error {
+	called := m.Called(partition)
+	ret := called.Get(0)
+
+	if ret == nil {
+		return nil
+	} else {
+		return ret.(error)
+	}
+}
+
+func TestGetResolvedExportedServices_ACL_Deny(t *testing.T) {
+	authorizer := acl.MockAuthorizer{}
+	authorizer.On("MeshRead", mock.Anything).Return(acl.Deny)
+
+	backend := &MockBackend{authorizer: &authorizer}
+	backend.On("EnterpriseCheckPartitions", mock.Anything).Return(nil)
+
+	fakeFSM := testutils.NewFakeBlockingFSM(t)
+
+	c := Config{
+		Backend:    backend,
+		Logger:     hclog.New(nil),
+		ForwardRPC: doForwardRPC,
+		FSMServer:  fakeFSM,
+	}
+
+	server := NewServer(c)
+
+	_, err := server.GetResolvedExportedServices(context.Background(), &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.Error(t, err)
+}
+
+func TestGetResolvedExportedServices_ACL_Allow(t *testing.T) {
+	authorizer := acl.MockAuthorizer{}
+	authorizer.On("MeshRead", mock.Anything).Return(acl.Allow)
+
+	backend := &MockBackend{authorizer: &authorizer}
+	backend.On("EnterpriseCheckPartitions", mock.Anything).Return(nil)
+
+	fakeFSM := testutils.NewFakeBlockingFSM(t)
+
+	c := Config{
+		Backend:    backend,
+		Logger:     hclog.New(nil),
+		ForwardRPC: doForwardRPC,
+		FSMServer:  fakeFSM,
+	}
+	server := NewServer(c)
+
+	// Add config entry
+	entry := &structs.ExportedServicesConfigEntry{
+		Name: "default",
+		Services: []structs.ExportedService{
+			{
+				Name: "db",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+					{
+						Peer: "west",
+					},
+				},
+			},
+			{
+				Name: "cache",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+				},
+			},
+		},
+	}
+	fakeFSM.GetState().EnsureConfigEntry(1, entry)
+
+	expected := []*pbconfigentry.ResolvedExportedService{
+		{
+			Service: "db",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east", "west"},
+			},
+		},
+		{
+			Service: "cache",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east"},
+			},
+		},
+	}
+
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), &testutils.MockServerTransportStream{})
+	resp, err := server.GetResolvedExportedServices(ctx, &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.NoError(t, err)
+	require.ElementsMatch(t, expected, resp.Services)
+}
+
+func TestGetResolvedExportedServices_PartitionCheck(t *testing.T) {
+	authorizer := acl.MockAuthorizer{}
+	authorizer.On("MeshRead", mock.Anything).Return(acl.Allow)
+
+	backend := &MockBackend{authorizer: &authorizer}
+	backend.On("EnterpriseCheckPartitions", mock.Anything).Return(fmt.Errorf("partition not supported"))
+
+	fakeFSM := testutils.NewFakeBlockingFSM(t)
+
+	c := Config{
+		Backend:    backend,
+		Logger:     hclog.New(nil),
+		ForwardRPC: doForwardRPC,
+		FSMServer:  fakeFSM,
+	}
+
+	server := NewServer(c)
+
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), &testutils.MockServerTransportStream{})
+
+	resp, err := server.GetResolvedExportedServices(ctx, &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.EqualError(t, err, "rpc error: code = InvalidArgument desc = partition not supported")
+	require.Nil(t, resp)
+}
+
+func TestGetResolvedExportedServices_Index(t *testing.T) {
+	authorizer := acl.MockAuthorizer{}
+	authorizer.On("MeshRead", mock.Anything).Return(acl.Allow)
+
+	backend := &MockBackend{authorizer: &authorizer}
+	backend.On("EnterpriseCheckPartitions", mock.Anything).Return(nil)
+
+	fakeFSM := testutils.NewFakeBlockingFSM(t)
+
+	c := Config{
+		Backend:    backend,
+		Logger:     hclog.New(nil),
+		ForwardRPC: doForwardRPC,
+		FSMServer:  fakeFSM,
+	}
+	server := NewServer(c)
+
+	// Add config entry
+	entry := &structs.ExportedServicesConfigEntry{
+		Name: "default",
+		Services: []structs.ExportedService{
+			{
+				Name: "db",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+					{
+						Peer: "west",
+					},
+				},
+			},
+			{
+				Name: "cache",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+				},
+			},
+		},
+	}
+	fakeFSM.GetState().EnsureConfigEntry(1, entry)
+
+	expected := []*pbconfigentry.ResolvedExportedService{
+		{
+			Service: "db",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east", "west"},
+			},
+		},
+		{
+			Service: "cache",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east"},
+			},
+		},
+	}
+
+	headerStream := &testutils.MockServerTransportStream{}
+
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), headerStream)
+	resp, err := server.GetResolvedExportedServices(ctx, &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.NoError(t, err)
+	require.ElementsMatch(t, expected, resp.Services)
+	require.Equal(t, []string{"1"}, headerStream.MD.Get("index"))
+
+	// Updating the index
+	fakeFSM.GetState().EnsureConfigEntry(2, entry)
+
+	headerStream = &testutils.MockServerTransportStream{}
+
+	ctx = grpc.NewContextWithServerTransportStream(context.Background(), headerStream)
+	resp, err = server.GetResolvedExportedServices(ctx, &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.NoError(t, err)
+	require.ElementsMatch(t, expected, resp.Services)
+	require.Equal(t, []string{"2"}, headerStream.MD.Get("index"))
+}
+
+func TestGetResolvedExportedServices_Metrics(t *testing.T) {
+	sink := metrics.NewInmemSink(5*time.Second, time.Minute)
+	cfg := metrics.DefaultConfig("consul")
+	metrics.NewGlobal(cfg, sink)
+
+	authorizer := acl.MockAuthorizer{}
+	authorizer.On("MeshRead", mock.Anything).Return(acl.Allow)
+
+	backend := &MockBackend{authorizer: &authorizer}
+	backend.On("EnterpriseCheckPartitions", mock.Anything).Return(nil)
+
+	fakeFSM := testutils.NewFakeBlockingFSM(t)
+
+	c := Config{
+		Backend:    backend,
+		Logger:     hclog.New(nil),
+		ForwardRPC: doForwardRPC,
+		FSMServer:  fakeFSM,
+	}
+	server := NewServer(c)
+
+	// Add config entry
+	entry := &structs.ExportedServicesConfigEntry{
+		Name: "default",
+		Services: []structs.ExportedService{
+			{
+				Name: "db",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+					{
+						Peer: "west",
+					},
+				},
+			},
+			{
+				Name: "cache",
+				Consumers: []structs.ServiceConsumer{
+					{
+						Peer: "east",
+					},
+				},
+			},
+		},
+	}
+	fakeFSM.GetState().EnsureConfigEntry(1, entry)
+
+	expected := []*pbconfigentry.ResolvedExportedService{
+		{
+			Service: "db",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east", "west"},
+			},
+		},
+		{
+			Service: "cache",
+			Consumers: &pbconfigentry.Consumers{
+				Peers: []string{"east"},
+			},
+		},
+	}
+
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), &testutils.MockServerTransportStream{})
+	resp, err := server.GetResolvedExportedServices(ctx, &pbconfigentry.GetResolvedExportedServicesRequest{})
+	require.NoError(t, err)
+	require.ElementsMatch(t, expected, resp.Services)
+
+	// Checking if metrics were added
+	require.NotNil(t, sink.Data()[0].Samples[`consul.configentry.get_resolved_exported_services`])
+}
+
+func doForwardRPC(structs.RPCInfo, func(*grpc.ClientConn) error) (bool, error) {
+	return false, nil
+}

agent/grpc-external/testutils/fsm.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/consul/agent/blockingquery"
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/consul/stream"
 	"github.com/stretchr/testify/require"
@@ -70,6 +71,47 @@ func (f *FakeFSM) ReplaceStore(store *state.Store) {
 	}
 }
 
+type FakeBlockingFSM struct {
+	store *state.Store
+}
+
+func NewFakeBlockingFSM(t *testing.T) *FakeBlockingFSM {
+	t.Helper()
+
+	store := TestStateStore(t, nil)
+
+	fsm := &FakeBlockingFSM{store: store}
+
+	return fsm
+}
+
+func (f *FakeBlockingFSM) GetState() *state.Store {
+	return f.store
+}
+
+func (f *FakeBlockingFSM) ConsistentRead() error {
+	return nil
+}
+
+func (f *FakeBlockingFSM) DecrementBlockingQueries() uint64 {
+	return 0
+}
+
+func (f *FakeBlockingFSM) IncrementBlockingQueries() uint64 {
+	return 0
+}
+
+func (f *FakeBlockingFSM) GetShutdownChannel() chan struct{} {
+	return nil
+}
+
+func (f *FakeBlockingFSM) RPCQueryTimeout(queryTimeout time.Duration) time.Duration {
+	return queryTimeout
+}
+
+func (f *FakeBlockingFSM) SetQueryMeta(blockingquery.ResponseMeta, string) {
+}
+
 func SetupFSMAndPublisher(t *testing.T, config FakeFSMConfig) (*FakeFSM, state.EventPublisher) {
 	t.Helper()
 	config.publisher = stream.NewEventPublisher(10 * time.Second)

agent/grpc-external/testutils/mock_server_transport_stream.go

@@ -0,0 +1,27 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package testutils
+
+import "google.golang.org/grpc/metadata"
+
+type MockServerTransportStream struct {
+	MD metadata.MD
+}
+
+func (m *MockServerTransportStream) Method() string {
+	return ""
+}
+
+func (m *MockServerTransportStream) SetHeader(md metadata.MD) error {
+	return nil
+}
+
+func (m *MockServerTransportStream) SendHeader(md metadata.MD) error {
+	m.MD = metadata.Join(m.MD, md)
+	return nil
+}
+
+func (m *MockServerTransportStream) SetTrailer(md metadata.MD) error {
+	return nil
+}