add sync latest test

internal/testutil/grpc.go

@@ -0,0 +1,44 @@
+package testutil
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+// NewGRPCServer starts a gRPC server and returns a client connection to it.
+func NewGRPCServer(t testing.TB, register func(s *grpc.Server)) *grpc.ClientConn {
+	t.Helper()
+
+	li := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	register(s)
+	go func() {
+		err := s.Serve(li)
+		if errors.Is(err, grpc.ErrServerStopped) {
+			err = nil
+		}
+		require.NoError(t, err)
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+	})
+
+	cc, err := grpc.NewClient("passthrough://bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return li.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		cc.Close()
+	})
+
+	return cc
+}

internal/zero/controller/usage_reporter_test.go

@@ -0,0 +1,54 @@
+package controller
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/pomerium/pomerium/internal/databroker"
+	"github.com/pomerium/pomerium/internal/testutil"
+	databrokerpb "github.com/pomerium/pomerium/pkg/grpc/databroker"
+	"github.com/pomerium/pomerium/pkg/grpc/user"
+)
+
+func Test_SyncLatestRecords(t *testing.T) {
+	t.Parallel()
+
+	ctx, clearTimeout := context.WithTimeout(context.Background(), time.Minute)
+	defer clearTimeout()
+
+	cc := testutil.NewGRPCServer(t, func(s *grpc.Server) {
+		databrokerpb.RegisterDataBrokerServiceServer(s, databroker.New())
+	})
+
+	c := databrokerpb.NewDataBrokerServiceClient(cc)
+
+	expected := []*user.User{
+		{Id: "u1"},
+		{Id: "u2"},
+		{Id: "u3"},
+	}
+
+	for _, u := range expected {
+		_, err := c.Put(ctx, &databrokerpb.PutRequest{
+			Records: []*databrokerpb.Record{
+				databrokerpb.NewRecord(u),
+			},
+		})
+		require.NoError(t, err)
+	}
+
+	var actual []*user.User
+	serverVersion, latestRecordVersion, err := syncLatestRecords(context.Background(), c, func(u *user.User) {
+		actual = append(actual, u)
+	})
+	assert.NoError(t, err)
+	assert.NotZero(t, serverVersion)
+	assert.Equal(t, uint64(3), latestRecordVersion)
+	testutil.AssertProtoEqual(t, expected, actual)
+}