mirror of
https://github.com/pomerium/pomerium.git
synced 2025-05-01 03:16:31 +02:00
5d60cff21e
* refactor backend, implement encrypted store * refactor in-memory store * wip * wip * wip * add syncer test * fix redis expiry * fix linting issues * fix test by skipping non-config records * fix backoff import * fix init issues * fix query * wait for initial sync before starting directory sync * add type to SyncLatest * add more log messages, fix deadlock in in-memory store, always return server version from SyncLatest * update sync types and tests * add redis tests * skip macos in github actions * add comments to proto * split getBackend into separate methods * handle errors in initVersion * return different error for not found vs other errors in get * use exponential backoff for redis transaction retry * rename raw to result * use context instead of close channel * store type urls as constants in databroker * use timestampb instead of ptypes * fix group merging not waiting * change locked names * update GetAll to return latest record version * add method to grpcutil to get the type url for a protobuf type
116 lines
2.6 KiB
Go
116 lines
2.6 KiB
Go
package databroker
|
|
|
|
import (
|
|
"context"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
func TestApplyOffsetAndLimit(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
records []*Record
|
|
offset, limit int
|
|
expect []*Record
|
|
}{
|
|
{
|
|
name: "empty",
|
|
records: nil,
|
|
offset: 10,
|
|
limit: 5,
|
|
expect: nil,
|
|
},
|
|
{
|
|
name: "less than limit",
|
|
records: []*Record{{Id: "A"}, {Id: "B"}, {Id: "C"}, {Id: "D"}},
|
|
offset: 1,
|
|
limit: 10,
|
|
expect: []*Record{{Id: "B"}, {Id: "C"}, {Id: "D"}},
|
|
},
|
|
{
|
|
name: "more than limit",
|
|
records: []*Record{{Id: "A"}, {Id: "B"}, {Id: "C"}, {Id: "D"}, {Id: "E"}, {Id: "F"}, {Id: "G"}, {Id: "H"}},
|
|
offset: 3,
|
|
limit: 2,
|
|
expect: []*Record{{Id: "D"}, {Id: "E"}},
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
actual, cnt := ApplyOffsetAndLimit(tc.records, tc.offset, tc.limit)
|
|
assert.Equal(t, len(tc.records), cnt)
|
|
assert.Equal(t, tc.expect, actual)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestInitialSync(t *testing.T) {
|
|
ctx, clearTimeout := context.WithTimeout(context.Background(), time.Second*10)
|
|
defer clearTimeout()
|
|
|
|
li, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if !assert.NoError(t, err) {
|
|
return
|
|
}
|
|
defer li.Close()
|
|
|
|
r1 := new(Record)
|
|
r2 := new(Record)
|
|
|
|
m := &mockServer{
|
|
syncLatest: func(req *SyncLatestRequest, stream DataBrokerService_SyncLatestServer) error {
|
|
stream.Send(&SyncLatestResponse{
|
|
Response: &SyncLatestResponse_Record{
|
|
Record: r1,
|
|
},
|
|
})
|
|
stream.Send(&SyncLatestResponse{
|
|
Response: &SyncLatestResponse_Record{
|
|
Record: r2,
|
|
},
|
|
})
|
|
stream.Send(&SyncLatestResponse{
|
|
Response: &SyncLatestResponse_Versions{
|
|
Versions: &Versions{
|
|
LatestRecordVersion: 2,
|
|
ServerVersion: 1,
|
|
},
|
|
},
|
|
})
|
|
return nil
|
|
},
|
|
}
|
|
|
|
srv := grpc.NewServer()
|
|
RegisterDataBrokerServiceServer(srv, m)
|
|
go srv.Serve(li)
|
|
|
|
cc, err := grpc.Dial(li.Addr().String(), grpc.WithInsecure())
|
|
if !assert.NoError(t, err) {
|
|
return
|
|
}
|
|
defer cc.Close()
|
|
|
|
c := NewDataBrokerServiceClient(cc)
|
|
|
|
records, recordVersion, serverVersion, err := InitialSync(ctx, c, new(SyncLatestRequest))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(2), recordVersion)
|
|
assert.Equal(t, uint64(1), serverVersion)
|
|
assert.Equal(t, []*Record{r1, r2}, records)
|
|
}
|
|
|
|
type mockServer struct {
|
|
DataBrokerServiceServer
|
|
|
|
syncLatest func(empty *SyncLatestRequest, server DataBrokerService_SyncLatestServer) error
|
|
}
|
|
|
|
func (m *mockServer) SyncLatest(req *SyncLatestRequest, stream DataBrokerService_SyncLatestServer) error {
|
|
return m.syncLatest(req, stream)
|
|
}
|