mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-12 19:14:23 +00:00
aed7010508
* fix: improve scheduling * build pre-release * fix: locker * fix: user handler and print stack in case of panic in reducer * chore: remove sentry * fix: improve handler projection and implement tests * more tests * fix: race condition in tests * Update internal/eventstore/repository/sql/query.go Co-authored-by: Silvan <silvan.reusser@gmail.com> * fix: implemented suggested changes * fix: lock statement Co-authored-by: Silvan <silvan.reusser@gmail.com>
1005 lines
22 KiB
Go
1005 lines
22 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
|
|
"github.com/zitadel/zitadel/internal/api/authz"
|
|
"github.com/zitadel/zitadel/internal/api/service"
|
|
"github.com/zitadel/zitadel/internal/eventstore"
|
|
"github.com/zitadel/zitadel/internal/eventstore/repository"
|
|
es_repo_mock "github.com/zitadel/zitadel/internal/eventstore/repository/mock"
|
|
)
|
|
|
|
var (
|
|
ErrQuery = errors.New("query err")
|
|
ErrFilter = errors.New("filter err")
|
|
ErrReduce = errors.New("reduce err")
|
|
ErrLock = errors.New("lock failed")
|
|
ErrUnlock = errors.New("unlock failed")
|
|
ErrExec = errors.New("exec error")
|
|
ErrBulk = errors.New("bulk err")
|
|
ErrUpdate = errors.New("update err")
|
|
)
|
|
|
|
func TestProjectionHandler_Trigger(t *testing.T) {
|
|
type fields struct {
|
|
reduce Reduce
|
|
update Update
|
|
query SearchQuery
|
|
eventstore func(t *testing.T) *eventstore.Eventstore
|
|
}
|
|
type args struct {
|
|
ctx context.Context
|
|
instances []string
|
|
}
|
|
type want struct {
|
|
isErr func(err error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
fields fields
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
"query error",
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return nil
|
|
},
|
|
query: testQuery(nil, 0, ErrQuery),
|
|
},
|
|
args{
|
|
context.Background(),
|
|
nil,
|
|
},
|
|
want{isErr: func(err error) bool {
|
|
return errors.Is(err, ErrQuery)
|
|
}},
|
|
},
|
|
{
|
|
"no events",
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(),
|
|
)
|
|
},
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil),
|
|
},
|
|
args{
|
|
context.Background(),
|
|
nil,
|
|
},
|
|
want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"process error",
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 1,
|
|
PreviousAggregateSequence: 0,
|
|
CreationDate: time.Now(),
|
|
Type: "test.added",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
),
|
|
)
|
|
},
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil),
|
|
reduce: testReduceErr(ErrReduce),
|
|
},
|
|
args{
|
|
context.Background(),
|
|
nil,
|
|
},
|
|
want{
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, ErrReduce)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"process ok",
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 1,
|
|
PreviousAggregateSequence: 0,
|
|
CreationDate: time.Now(),
|
|
Type: "test.added",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
),
|
|
)
|
|
},
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil),
|
|
reduce: testReduce(newTestStatement("testAgg", 1, 0)),
|
|
update: testUpdate(t, 1, 0, nil),
|
|
},
|
|
args{
|
|
context.Background(),
|
|
nil,
|
|
},
|
|
want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"process limit exceeded ok",
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).
|
|
ExpectFilterEvents(
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 1,
|
|
PreviousAggregateSequence: 0,
|
|
CreationDate: time.Now(),
|
|
Type: "test.added",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
).ExpectFilterEvents(),
|
|
)
|
|
},
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
1,
|
|
nil),
|
|
reduce: testReduce(newTestStatement("testAgg", 1, 0)),
|
|
update: testUpdate(t, 1, 0, nil),
|
|
},
|
|
args{
|
|
context.Background(),
|
|
nil,
|
|
},
|
|
want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := &ProjectionHandler{
|
|
Handler: Handler{
|
|
Eventstore: tt.fields.eventstore(t),
|
|
},
|
|
ProjectionName: "test",
|
|
reduce: tt.fields.reduce,
|
|
update: tt.fields.update,
|
|
searchQuery: tt.fields.query,
|
|
}
|
|
|
|
err := h.Trigger(tt.args.ctx, tt.args.instances...)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error %v", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProjectionHandler_Process(t *testing.T) {
|
|
type fields struct {
|
|
reduce Reduce
|
|
update Update
|
|
}
|
|
type args struct {
|
|
ctx context.Context
|
|
events []eventstore.Event
|
|
}
|
|
type want struct {
|
|
isErr func(err error) bool
|
|
index int
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
fields fields
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no events",
|
|
fields: fields{},
|
|
args: args{},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
index: 0,
|
|
},
|
|
},
|
|
{
|
|
name: "reduce fails",
|
|
fields: fields{
|
|
reduce: testReduceErr(ErrReduce),
|
|
},
|
|
args: args{
|
|
events: []eventstore.Event{newTestEvent("id", "description", nil)},
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, ErrReduce)
|
|
},
|
|
index: -1,
|
|
},
|
|
},
|
|
{
|
|
name: "stmt failed",
|
|
fields: fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 1, -1, ErrSomeStmtsFailed),
|
|
},
|
|
args: args{
|
|
events: []eventstore.Event{newTestEvent("id", "description", nil)},
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, ErrSomeStmtsFailed)
|
|
},
|
|
index: -1,
|
|
},
|
|
},
|
|
{
|
|
name: "stmt error",
|
|
fields: fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 1, -1, errors.New("some error")),
|
|
},
|
|
args: args{
|
|
events: []eventstore.Event{newTestEvent("id", "description", nil)},
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return err.Error() == "some error"
|
|
},
|
|
index: -1,
|
|
},
|
|
},
|
|
{
|
|
name: "stmt succeeded",
|
|
fields: fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 1, 0, nil),
|
|
},
|
|
args: args{
|
|
events: []eventstore.Event{newTestEvent("id", "description", nil)},
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
index: 0,
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := NewProjectionHandler(
|
|
context.Background(),
|
|
ProjectionHandlerConfig{
|
|
HandlerConfig: HandlerConfig{
|
|
Eventstore: nil,
|
|
},
|
|
ProjectionName: "test",
|
|
RequeueEvery: -1,
|
|
},
|
|
tt.fields.reduce,
|
|
tt.fields.update,
|
|
nil,
|
|
nil,
|
|
nil,
|
|
)
|
|
|
|
index, err := h.Process(tt.args.ctx, tt.args.events...)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error %v", err)
|
|
}
|
|
assert.Equal(t, tt.want.index, index)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProjectionHandler_FetchEvents(t *testing.T) {
|
|
type args struct {
|
|
ctx context.Context
|
|
instanceIDs []string
|
|
}
|
|
type want struct {
|
|
limitExceeded bool
|
|
isErr func(error) bool
|
|
}
|
|
type fields struct {
|
|
eventstore *eventstore.Eventstore
|
|
query SearchQuery
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
fields fields
|
|
want want
|
|
}{
|
|
{
|
|
name: "query returns err",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields: fields{
|
|
query: testQuery(nil, 0, ErrQuery),
|
|
},
|
|
want: want{
|
|
limitExceeded: false,
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, ErrQuery)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "eventstore returns err",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields: fields{
|
|
eventstore: eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEventsError(ErrFilter),
|
|
),
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil,
|
|
),
|
|
},
|
|
want: want{
|
|
limitExceeded: false,
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, ErrFilter)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no events found",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields: fields{
|
|
eventstore: eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(),
|
|
),
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil,
|
|
),
|
|
},
|
|
want: want{
|
|
limitExceeded: false,
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "found events smaller than limit",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields: fields{
|
|
eventstore: eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 1,
|
|
PreviousAggregateSequence: 0,
|
|
CreationDate: time.Now(),
|
|
Type: "test.added",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 2,
|
|
PreviousAggregateSequence: 1,
|
|
CreationDate: time.Now(),
|
|
Type: "test.changed",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
),
|
|
),
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
5,
|
|
nil,
|
|
),
|
|
},
|
|
want: want{
|
|
limitExceeded: false,
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "found events exceeds limit",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields: fields{
|
|
eventstore: eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectFilterEvents(
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 1,
|
|
PreviousAggregateSequence: 0,
|
|
CreationDate: time.Now(),
|
|
Type: "test.added",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
&repository.Event{
|
|
ID: "id",
|
|
Sequence: 2,
|
|
PreviousAggregateSequence: 1,
|
|
CreationDate: time.Now(),
|
|
Type: "test.changed",
|
|
Version: "v1",
|
|
AggregateID: "testid",
|
|
AggregateType: "testAgg",
|
|
},
|
|
),
|
|
),
|
|
query: testQuery(
|
|
eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
|
|
AddQuery().
|
|
AggregateTypes("test").
|
|
Builder(),
|
|
2,
|
|
nil,
|
|
),
|
|
},
|
|
want: want{
|
|
limitExceeded: true,
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := &ProjectionHandler{
|
|
Handler: Handler{
|
|
Eventstore: tt.fields.eventstore,
|
|
},
|
|
searchQuery: tt.fields.query,
|
|
}
|
|
_, limitExceeded, err := h.FetchEvents(tt.args.ctx, tt.args.instanceIDs...)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("ProjectionHandler.prepareBulkStmts() error = %v", err)
|
|
return
|
|
}
|
|
if limitExceeded != tt.want.limitExceeded {
|
|
t.Errorf("ProjectionHandler.prepareBulkStmts() = %v, want %v", limitExceeded, tt.want.limitExceeded)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProjection_subscribe(t *testing.T) {
|
|
type args struct {
|
|
ctx context.Context
|
|
}
|
|
type fields struct {
|
|
reduce Reduce
|
|
update Update
|
|
events []eventstore.Event
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
fields fields
|
|
}{
|
|
{
|
|
"panic",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
reduce: nil,
|
|
update: nil,
|
|
events: []eventstore.Event{
|
|
newTestEvent("id", "", nil),
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"error",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
reduce: testReduceErr(ErrReduce),
|
|
update: nil,
|
|
events: []eventstore.Event{
|
|
newTestEvent("id", "", nil),
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"not all statement",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 1, 0, ErrSomeStmtsFailed),
|
|
events: []eventstore.Event{
|
|
newTestEvent("id", "", nil),
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"single event ok",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 1, 1, nil),
|
|
events: []eventstore.Event{
|
|
newTestEvent("id", "", nil),
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"multiple events ok",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
reduce: testReduce(newTestStatement("aggregate1", 1, 0)),
|
|
update: testUpdate(t, 2, 2, nil),
|
|
events: []eventstore.Event{
|
|
newTestEvent("id", "", nil),
|
|
newTestEvent("id2", "", nil),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := &ProjectionHandler{
|
|
Handler: Handler{
|
|
EventQueue: make(chan eventstore.Event, 10),
|
|
},
|
|
reduce: tt.fields.reduce,
|
|
update: tt.fields.update,
|
|
}
|
|
ctx, cancel := context.WithCancel(tt.args.ctx)
|
|
go func() {
|
|
//changed go h.subscribe(ctx) to this to be able to ignore logs easily
|
|
t.Helper()
|
|
h.subscribe(ctx)
|
|
}()
|
|
for _, event := range tt.fields.events {
|
|
h.EventQueue <- event
|
|
}
|
|
time.Sleep(1 * time.Second)
|
|
cancel()
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProjection_schedule(t *testing.T) {
|
|
type args struct {
|
|
ctx context.Context
|
|
}
|
|
type fields struct {
|
|
reduce Reduce
|
|
update Update
|
|
eventstore func(t *testing.T) *eventstore.Eventstore
|
|
triggerProjection *time.Timer
|
|
lock *lockMock
|
|
unlock *unlockMock
|
|
query SearchQuery
|
|
}
|
|
type want struct {
|
|
locksCount int
|
|
lockCanceled bool
|
|
unlockCount int
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
fields fields
|
|
want want
|
|
}{
|
|
{
|
|
"panic",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return nil
|
|
},
|
|
},
|
|
want{},
|
|
},
|
|
{
|
|
"filter instance ids error",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectInstanceIDsError(ErrFilter),
|
|
)
|
|
},
|
|
triggerProjection: time.NewTimer(0),
|
|
},
|
|
want{
|
|
locksCount: 0,
|
|
lockCanceled: false,
|
|
unlockCount: 0,
|
|
},
|
|
},
|
|
{
|
|
"lock error",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectInstanceIDs("instanceID1"),
|
|
)
|
|
},
|
|
triggerProjection: time.NewTimer(0),
|
|
lock: &lockMock{
|
|
errWait: 100 * time.Millisecond,
|
|
firstErr: ErrLock,
|
|
canceled: make(chan bool, 1),
|
|
},
|
|
},
|
|
want{
|
|
locksCount: 1,
|
|
lockCanceled: true,
|
|
unlockCount: 0,
|
|
},
|
|
},
|
|
{
|
|
"trigger error",
|
|
args{
|
|
ctx: context.Background(),
|
|
},
|
|
fields{
|
|
eventstore: func(t *testing.T) *eventstore.Eventstore {
|
|
return eventstore.NewEventstore(
|
|
es_repo_mock.NewRepo(t).ExpectInstanceIDs("instanceID1"),
|
|
)
|
|
},
|
|
triggerProjection: time.NewTimer(0),
|
|
lock: &lockMock{
|
|
canceled: make(chan bool, 1),
|
|
firstErr: nil,
|
|
errWait: 100 * time.Millisecond,
|
|
},
|
|
unlock: &unlockMock{},
|
|
query: testQuery(nil, 0, ErrQuery),
|
|
},
|
|
want{
|
|
locksCount: 1,
|
|
lockCanceled: true,
|
|
unlockCount: 1,
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := &ProjectionHandler{
|
|
Handler: Handler{
|
|
EventQueue: make(chan eventstore.Event, 10),
|
|
Eventstore: tt.fields.eventstore(t),
|
|
},
|
|
reduce: tt.fields.reduce,
|
|
update: tt.fields.update,
|
|
searchQuery: tt.fields.query,
|
|
lock: tt.fields.lock.lock(),
|
|
unlock: tt.fields.unlock.unlock(),
|
|
triggerProjection: tt.fields.triggerProjection,
|
|
requeueAfter: 10 * time.Second,
|
|
concurrentInstances: 1,
|
|
}
|
|
ctx, cancel := context.WithCancel(tt.args.ctx)
|
|
go func() {
|
|
//changed go h.schedule(ctx) to this to be able to ignore logs easily
|
|
t.Helper()
|
|
h.schedule(ctx)
|
|
}()
|
|
|
|
time.Sleep(1 * time.Second)
|
|
if tt.fields.lock != nil {
|
|
tt.fields.lock.check(t, tt.want.locksCount, tt.want.lockCanceled)
|
|
}
|
|
if tt.fields.unlock != nil {
|
|
tt.fields.unlock.check(t, tt.want.unlockCount)
|
|
}
|
|
cancel()
|
|
})
|
|
}
|
|
}
|
|
|
|
func Test_cancelOnErr(t *testing.T) {
|
|
type args struct {
|
|
ctx context.Context
|
|
errs chan error
|
|
err error
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
cancelMocker *cancelMocker
|
|
}{
|
|
{
|
|
name: "error occured",
|
|
args: args{
|
|
ctx: context.Background(),
|
|
errs: make(chan error),
|
|
err: ErrNoCondition,
|
|
},
|
|
cancelMocker: &cancelMocker{
|
|
shouldBeCalled: true,
|
|
wasCalled: make(chan bool, 1),
|
|
},
|
|
},
|
|
{
|
|
name: "ctx done",
|
|
args: args{
|
|
ctx: canceledCtx(),
|
|
errs: make(chan error),
|
|
},
|
|
cancelMocker: &cancelMocker{
|
|
shouldBeCalled: false,
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
h := &ProjectionHandler{}
|
|
go h.cancelOnErr(tt.args.ctx, tt.args.errs, tt.cancelMocker.mockCancel)
|
|
if tt.args.err != nil {
|
|
tt.args.errs <- tt.args.err
|
|
}
|
|
tt.cancelMocker.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func newTestStatement(aggType eventstore.AggregateType, seq, previousSeq uint64) *Statement {
|
|
return &Statement{
|
|
AggregateType: aggType,
|
|
Sequence: seq,
|
|
PreviousSequence: previousSeq,
|
|
}
|
|
}
|
|
|
|
// testEvent implements the Event interface
|
|
type testEvent struct {
|
|
eventstore.BaseEvent
|
|
|
|
description string
|
|
data func() interface{}
|
|
}
|
|
|
|
func newTestEvent(id, description string, data func() interface{}) *testEvent {
|
|
return &testEvent{
|
|
description: description,
|
|
data: data,
|
|
BaseEvent: *eventstore.NewBaseEventForPush(
|
|
service.WithService(authz.NewMockContext("instanceID", "resourceOwner", "editorUser"), "editorService"),
|
|
eventstore.NewAggregate(authz.NewMockContext("zitadel", "caos", "adlerhurst"), id, "test.aggregate", "v1"),
|
|
"test.event",
|
|
),
|
|
}
|
|
}
|
|
|
|
func (e *testEvent) Data() interface{} {
|
|
if e.data == nil {
|
|
return nil
|
|
}
|
|
return e.data()
|
|
}
|
|
|
|
func (e *testEvent) UniqueConstraints() []*eventstore.EventUniqueConstraint {
|
|
return nil
|
|
}
|
|
|
|
func testUpdate(t *testing.T, expectedStmtCount, returnedIndex int, returnedErr error) Update {
|
|
return func(ctx context.Context, stmts []*Statement, reduce Reduce) (int, error) {
|
|
if expectedStmtCount != len(stmts) {
|
|
t.Errorf("expected %d stmts got %d", expectedStmtCount, len(stmts))
|
|
}
|
|
return returnedIndex, returnedErr
|
|
}
|
|
}
|
|
|
|
func testReduce(stmts *Statement) Reduce {
|
|
return func(event eventstore.Event) (*Statement, error) {
|
|
return stmts, nil
|
|
}
|
|
}
|
|
|
|
func testReduceErr(err error) Reduce {
|
|
return func(event eventstore.Event) (*Statement, error) {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
func testQuery(query *eventstore.SearchQueryBuilder, limit uint64, err error) SearchQuery {
|
|
return func(ctx context.Context, instanceIDs []string) (*eventstore.SearchQueryBuilder, uint64, error) {
|
|
return query, limit, err
|
|
}
|
|
}
|
|
|
|
type lockMock struct {
|
|
callCount int
|
|
canceled chan bool
|
|
mu sync.Mutex
|
|
|
|
firstErr error
|
|
err error
|
|
errWait time.Duration
|
|
}
|
|
|
|
func (m *lockMock) lock() Lock {
|
|
return func(ctx context.Context, _ time.Duration, _ ...string) <-chan error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.callCount++
|
|
errs := make(chan error)
|
|
go func() {
|
|
for i := 0; ; i++ {
|
|
select {
|
|
case <-ctx.Done():
|
|
m.canceled <- true
|
|
close(errs)
|
|
return
|
|
case <-time.After(m.errWait):
|
|
err := m.err
|
|
if i == 0 {
|
|
err = m.firstErr
|
|
}
|
|
errs <- err
|
|
}
|
|
}
|
|
}()
|
|
return errs
|
|
}
|
|
}
|
|
|
|
func (m *lockMock) check(t *testing.T, callCount int, shouldBeCanceled bool) {
|
|
t.Helper()
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if callCount != m.callCount {
|
|
t.Errorf("wrong call count: expected %v got: %v", callCount, m.callCount)
|
|
}
|
|
if shouldBeCanceled {
|
|
select {
|
|
case <-m.canceled:
|
|
case <-time.After(5 * time.Second):
|
|
t.Error("lock should be canceled but wasn't")
|
|
}
|
|
}
|
|
}
|
|
|
|
type unlockMock struct {
|
|
callCount int
|
|
err error
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func (m *unlockMock) unlock() Unlock {
|
|
return func(...string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.callCount++
|
|
return m.err
|
|
}
|
|
}
|
|
|
|
func (m *unlockMock) check(t *testing.T, callCount int) {
|
|
t.Helper()
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if callCount != m.callCount {
|
|
t.Errorf("wrong call count: expected %v got: %v", callCount, m.callCount)
|
|
}
|
|
}
|
|
|
|
func canceledCtx() context.Context {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
return ctx
|
|
}
|
|
|
|
type cancelMocker struct {
|
|
shouldBeCalled bool
|
|
wasCalled chan bool
|
|
}
|
|
|
|
func (m *cancelMocker) mockCancel() {
|
|
m.wasCalled <- true
|
|
}
|
|
|
|
func (m *cancelMocker) check(t *testing.T) {
|
|
t.Helper()
|
|
if m.shouldBeCalled {
|
|
if wasCalled := <-m.wasCalled; !wasCalled {
|
|
t.Errorf("cancel: should: %t got: %t", m.shouldBeCalled, wasCalled)
|
|
}
|
|
}
|
|
}
|