mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-12 11:04:25 +00:00
bb756482c7
* document analytics config
* rework configuration and docs
* describe HandleActiveInstances better
* describe active instances on quotas better
* only projected events are considered
* cleanup
* describe changes at runtime
* push milestones
* stop tracking events
* calculate and push 4 in 6 milestones
* reduce milestone pushed
* remove docs
* fix scheduled pseudo event projection
* push 5 in 6 milestones
* push 6 in 6 milestones
* ignore client ids
* fix text array contains
* push human readable milestone type
* statement unit tests
* improve dev and db performance
* organize imports
* cleanup
* organize imports
* test projection
* check rows.Err()
* test search query
* pass linting
* review
* test 4 milestones
* simplify milestone by instance ids query
* use type NamespacedCondition
* cleanup
* lint
* lint
* dont overwrite original error
* no opt-in in examples
* cleanup
* prerelease
* enable request headers
* make limit configurable
* review fixes
* only requeue special handlers secondly
* include integration tests
* Revert "include integration tests"
This reverts commit 96db9504ec.
* pass reducers
* test handlers
* fix unit test
* feat: increment version
* lint
* remove prerelease
* fix integration tests
1526 lines
31 KiB
Go
1526 lines
31 KiB
Go
package crdb
|
|
|
|
import (
|
|
"database/sql"
|
|
"errors"
|
|
"reflect"
|
|
"testing"
|
|
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
"github.com/zitadel/zitadel/internal/eventstore"
|
|
"github.com/zitadel/zitadel/internal/eventstore/handler"
|
|
)
|
|
|
|
type wantExecuter struct {
|
|
params []params
|
|
i int
|
|
t *testing.T
|
|
wasExecuted bool
|
|
shouldExecute bool
|
|
}
|
|
|
|
type params struct {
|
|
query string
|
|
args []interface{}
|
|
}
|
|
|
|
var errTestErr = errors.New("some error")
|
|
|
|
func (ex *wantExecuter) check(t *testing.T) {
|
|
t.Helper()
|
|
if ex.wasExecuted && !ex.shouldExecute {
|
|
t.Error("executer should not be executed")
|
|
} else if !ex.wasExecuted && ex.shouldExecute {
|
|
t.Error("executer should be executed")
|
|
} else if ex.wasExecuted != ex.shouldExecute {
|
|
t.Errorf("executed missmatched should be %t, but was %t", ex.shouldExecute, ex.wasExecuted)
|
|
}
|
|
}
|
|
|
|
func (ex *wantExecuter) Exec(query string, args ...interface{}) (sql.Result, error) {
|
|
ex.t.Helper()
|
|
ex.wasExecuted = true
|
|
if ex.i >= len(ex.params) {
|
|
ex.t.Errorf("did not expect more exec, but got:\n %q with %q", query, args)
|
|
return nil, nil
|
|
}
|
|
p := ex.params[ex.i]
|
|
if query != p.query {
|
|
ex.t.Errorf("wrong query:\n expected:\n %q\n got:\n %q", p.query, query)
|
|
}
|
|
if !reflect.DeepEqual(p.args, args) {
|
|
ex.t.Errorf("wrong args:\n expected:\n %v\n got:\n %v", p.args, args)
|
|
}
|
|
ex.i++
|
|
return nil, nil
|
|
}
|
|
|
|
func TestNewCreateStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
values []handler.Column
|
|
}
|
|
type want struct {
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
table string
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no table",
|
|
args: args{
|
|
table: "",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoProjection)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no values",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoValues)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "INSERT INTO my_table (col1) VALUES ($1)",
|
|
args: []interface{}{"val"},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.want.executer.t = t
|
|
stmt := NewCreateStatement(tt.args.event, tt.args.values)
|
|
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewUpsertStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
conflictCols []handler.Column
|
|
values []handler.Column
|
|
}
|
|
type want struct {
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
table string
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no table",
|
|
args: args{
|
|
table: "",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoProjection)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no values",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoValues)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no update cols",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conflictCols: []handler.Column{
|
|
handler.NewCol("col1", nil),
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoValues)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct UPDATE multi col",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conflictCols: []handler.Column{
|
|
handler.NewCol("col1", nil),
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
{
|
|
Name: "col2",
|
|
Value: "val",
|
|
},
|
|
{
|
|
Name: "col3",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "INSERT INTO my_table (col1, col2, col3) VALUES ($1, $2, $3) ON CONFLICT (col1) DO UPDATE SET (col2, col3) = (EXCLUDED.col2, EXCLUDED.col3)",
|
|
args: []interface{}{"val", "val", "val"},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct UPDATE single col",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conflictCols: []handler.Column{
|
|
handler.NewCol("col1", nil),
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
{
|
|
Name: "col2",
|
|
Value: "val",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "INSERT INTO my_table (col1, col2) VALUES ($1, $2) ON CONFLICT (col1) DO UPDATE SET col2 = EXCLUDED.col2",
|
|
args: []interface{}{"val", "val"},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.want.executer.t = t
|
|
stmt := NewUpsertStatement(tt.args.event, tt.args.conflictCols, tt.args.values)
|
|
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewUpdateStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
values []handler.Column
|
|
conditions []handler.Condition
|
|
}
|
|
type want struct {
|
|
table string
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no table",
|
|
args: args{
|
|
table: "",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoProjection)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no values",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoValues)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no conditions",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
conditions: []handler.Condition{},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoCondition)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct single column",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "UPDATE my_table SET col1 = $1 WHERE (col2 = $2)",
|
|
args: []interface{}{"val", 1},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct multi column",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
values: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: "val",
|
|
},
|
|
{
|
|
Name: "col3",
|
|
Value: "val5",
|
|
},
|
|
},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "UPDATE my_table SET (col1, col3) = ($1, $2) WHERE (col2 = $3)",
|
|
args: []interface{}{"val", "val5", 1},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.want.executer.t = t
|
|
stmt := NewUpdateStatement(tt.args.event, tt.args.values, tt.args.conditions)
|
|
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewDeleteStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
conditions []handler.Condition
|
|
}
|
|
|
|
type want struct {
|
|
table string
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no table",
|
|
args: args{
|
|
table: "",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoProjection)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no conditions",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conditions: []handler.Condition{},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoCondition)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
aggregateType: "agg",
|
|
},
|
|
conditions: []handler.Condition{
|
|
handler.NewCond("col1", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "DELETE FROM my_table WHERE (col1 = $1)",
|
|
args: []interface{}{1},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.want.executer.t = t
|
|
stmt := NewDeleteStatement(tt.args.event, tt.args.conditions)
|
|
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewNoOpStatement(t *testing.T) {
|
|
type args struct {
|
|
event *testEvent
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want *handler.Statement
|
|
}{
|
|
{
|
|
name: "generate correctly",
|
|
args: args{
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 5,
|
|
previousSequence: 3,
|
|
instanceID: "instanceID",
|
|
},
|
|
},
|
|
want: &handler.Statement{
|
|
AggregateType: "agg",
|
|
Execute: nil,
|
|
Sequence: 5,
|
|
PreviousSequence: 3,
|
|
InstanceID: "instanceID",
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
if got := NewNoOpStatement(tt.args.event); !reflect.DeepEqual(got, tt.want) {
|
|
t.Errorf("NewNoOpStatement() = %v, want %v", got, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewMultiStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
execs []func(eventstore.Event) Exec
|
|
}
|
|
|
|
type want struct {
|
|
table string
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no op",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
execs: nil,
|
|
},
|
|
want: want{
|
|
executer: nil,
|
|
},
|
|
},
|
|
{
|
|
name: "no condition",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
execs: []func(eventstore.Event) Exec{
|
|
AddDeleteStatement(
|
|
[]handler.Condition{},
|
|
),
|
|
AddCreateStatement(
|
|
[]handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
}),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoCondition)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
aggregateType: "agg",
|
|
},
|
|
execs: []func(eventstore.Event) Exec{
|
|
AddDeleteStatement(
|
|
[]handler.Condition{
|
|
handler.NewCond("col1", 1),
|
|
},
|
|
),
|
|
AddCreateStatement(
|
|
[]handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
}),
|
|
AddUpsertStatement(
|
|
[]handler.Column{
|
|
handler.NewCol("col1", nil),
|
|
},
|
|
[]handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
{
|
|
Name: "col2",
|
|
Value: 2,
|
|
},
|
|
}),
|
|
AddUpdateStatement(
|
|
[]handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
},
|
|
[]handler.Condition{
|
|
handler.NewCond("col1", 1),
|
|
},
|
|
),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "DELETE FROM my_table WHERE (col1 = $1)",
|
|
args: []interface{}{1},
|
|
},
|
|
{
|
|
query: "INSERT INTO my_table (col1) VALUES ($1)",
|
|
args: []interface{}{1},
|
|
},
|
|
{
|
|
query: "INSERT INTO my_table (col1, col2) VALUES ($1, $2) ON CONFLICT (col1) DO UPDATE SET col2 = EXCLUDED.col2",
|
|
args: []interface{}{1, 2},
|
|
},
|
|
{
|
|
query: "UPDATE my_table SET col1 = $1 WHERE (col1 = $2)",
|
|
args: []interface{}{1, 1},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
stmt := NewMultiStatement(tt.args.event, tt.args.execs...)
|
|
|
|
if tt.want.executer != nil && stmt.Execute == nil {
|
|
t.Error("expected executer, but was nil")
|
|
}
|
|
if stmt.Execute == nil {
|
|
return
|
|
}
|
|
tt.want.executer.t = t
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNewCopyStatement(t *testing.T) {
|
|
type args struct {
|
|
table string
|
|
event *testEvent
|
|
conflictingCols []handler.Column
|
|
from []handler.Column
|
|
to []handler.Column
|
|
conds []handler.NamespacedCondition
|
|
}
|
|
type want struct {
|
|
aggregateType eventstore.AggregateType
|
|
sequence uint64
|
|
previousSequence uint64
|
|
table string
|
|
executer *wantExecuter
|
|
isErr func(error) bool
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no table",
|
|
args: args{
|
|
table: "",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conds: []handler.NamespacedCondition{
|
|
handler.NewNamespacedCondition("col2", 1),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoProjection)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no conditions",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conds: []handler.NamespacedCondition{},
|
|
from: []handler.Column{
|
|
{
|
|
Name: "col",
|
|
},
|
|
},
|
|
to: []handler.Column{
|
|
{
|
|
Name: "col",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoCondition)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "more to than from cols",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conds: []handler.NamespacedCondition{},
|
|
from: []handler.Column{
|
|
{
|
|
Name: "col",
|
|
},
|
|
},
|
|
to: []handler.Column{
|
|
{
|
|
Name: "col",
|
|
},
|
|
{
|
|
Name: "col2",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoCondition)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "no columns",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
conds: []handler.NamespacedCondition{
|
|
handler.NewNamespacedCondition("col2", nil),
|
|
},
|
|
from: []handler.Column{},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
shouldExecute: false,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, handler.ErrNoValues)
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct same column names",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
from: []handler.Column{
|
|
{
|
|
Name: "state",
|
|
Value: 1,
|
|
},
|
|
{
|
|
Name: "id",
|
|
},
|
|
{
|
|
Name: "col_a",
|
|
},
|
|
{
|
|
Name: "col_b",
|
|
},
|
|
},
|
|
to: []handler.Column{
|
|
{
|
|
Name: "state",
|
|
},
|
|
{
|
|
Name: "id",
|
|
},
|
|
{
|
|
Name: "col_a",
|
|
},
|
|
{
|
|
Name: "col_b",
|
|
},
|
|
},
|
|
conds: []handler.NamespacedCondition{
|
|
handler.NewNamespacedCondition("id", 2),
|
|
handler.NewNamespacedCondition("state", 3),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "INSERT INTO my_table (state, id, col_a, col_b) SELECT $1, id, col_a, col_b FROM my_table AS copy_table WHERE (copy_table.id = $2) AND (copy_table.state = $3) ON CONFLICT () DO UPDATE SET (state, id, col_a, col_b) = ($1, EXCLUDED.id, EXCLUDED.col_a, EXCLUDED.col_b)",
|
|
args: []interface{}{1, 2, 3},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "correct different column names",
|
|
args: args{
|
|
table: "my_table",
|
|
event: &testEvent{
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 0,
|
|
},
|
|
from: []handler.Column{
|
|
{
|
|
Value: 1,
|
|
},
|
|
{
|
|
Name: "id",
|
|
},
|
|
{
|
|
Name: "col_a",
|
|
},
|
|
{
|
|
Name: "col_b",
|
|
},
|
|
},
|
|
to: []handler.Column{
|
|
{
|
|
Name: "state",
|
|
},
|
|
{
|
|
Name: "id",
|
|
},
|
|
{
|
|
Name: "col_c",
|
|
},
|
|
{
|
|
Name: "col_d",
|
|
},
|
|
},
|
|
conds: []handler.NamespacedCondition{
|
|
handler.NewNamespacedCondition("id", 2),
|
|
handler.NewNamespacedCondition("state", 3),
|
|
},
|
|
},
|
|
want: want{
|
|
table: "my_table",
|
|
aggregateType: "agg",
|
|
sequence: 1,
|
|
previousSequence: 1,
|
|
executer: &wantExecuter{
|
|
params: []params{
|
|
{
|
|
query: "INSERT INTO my_table (state, id, col_c, col_d) SELECT $1, id, col_a, col_b FROM my_table AS copy_table WHERE (copy_table.id = $2) AND (copy_table.state = $3) ON CONFLICT () DO UPDATE SET (state, id, col_c, col_d) = ($1, EXCLUDED.id, EXCLUDED.col_a, EXCLUDED.col_b)",
|
|
args: []interface{}{1, 2, 3},
|
|
},
|
|
},
|
|
shouldExecute: true,
|
|
},
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
tt.want.executer.t = t
|
|
stmt := NewCopyStatement(tt.args.event, tt.args.conflictingCols, tt.args.from, tt.args.to, tt.args.conds)
|
|
|
|
err := stmt.Execute(tt.want.executer, tt.args.table)
|
|
if !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
tt.want.executer.check(t)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStatement_Execute(t *testing.T) {
|
|
type fields struct {
|
|
execute func(ex handler.Executer, projectionName string) error
|
|
}
|
|
type want struct {
|
|
isErr func(error) bool
|
|
}
|
|
type args struct {
|
|
projectionName string
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
fields fields
|
|
want want
|
|
}{
|
|
{
|
|
name: "execute returns no error",
|
|
fields: fields{
|
|
execute: func(ex handler.Executer, projectionName string) error { return nil },
|
|
},
|
|
args: args{
|
|
projectionName: "my_projection",
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return err == nil
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "execute returns error",
|
|
args: args{
|
|
projectionName: "my_projection",
|
|
},
|
|
fields: fields{
|
|
execute: func(ex handler.Executer, projectionName string) error { return errTestErr },
|
|
},
|
|
want: want{
|
|
isErr: func(err error) bool {
|
|
return errors.Is(err, errTestErr)
|
|
},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
stmt := &handler.Statement{
|
|
Execute: tt.fields.execute,
|
|
}
|
|
if err := stmt.Execute(nil, tt.args.projectionName); !tt.want.isErr(err) {
|
|
t.Errorf("unexpected error: %v", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func Test_columnsToQuery(t *testing.T) {
|
|
type args struct {
|
|
cols []handler.Column
|
|
}
|
|
type want struct {
|
|
names []string
|
|
params []string
|
|
values []interface{}
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no columns",
|
|
args: args{},
|
|
want: want{
|
|
names: []string{},
|
|
params: []string{},
|
|
values: []interface{}{},
|
|
},
|
|
},
|
|
{
|
|
name: "one column",
|
|
args: args{
|
|
cols: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
names: []string{"col1"},
|
|
params: []string{"$1"},
|
|
values: []interface{}{1},
|
|
},
|
|
},
|
|
{
|
|
name: "multiple columns",
|
|
args: args{
|
|
cols: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
{
|
|
Name: "col2",
|
|
Value: 3.14,
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
names: []string{"col1", "col2"},
|
|
params: []string{"$1", "$2"},
|
|
values: []interface{}{1, 3.14},
|
|
},
|
|
},
|
|
{
|
|
name: "with copy column",
|
|
args: args{
|
|
cols: []handler.Column{
|
|
{
|
|
Name: "col1",
|
|
Value: 1,
|
|
},
|
|
{
|
|
Name: "col2",
|
|
Value: handler.Column{
|
|
Name: "col1",
|
|
},
|
|
},
|
|
{
|
|
Name: "col3",
|
|
Value: "something",
|
|
},
|
|
},
|
|
},
|
|
want: want{
|
|
names: []string{"col1", "col2", "col3"},
|
|
params: []string{"$1", "col1", "$2"},
|
|
values: []interface{}{1, "something"},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
gotNames, gotParameters, gotValues := columnsToQuery(tt.args.cols)
|
|
if !reflect.DeepEqual(gotNames, tt.want.names) {
|
|
t.Errorf("columnsToQuery() gotNames = %v, want %v", gotNames, tt.want.names)
|
|
}
|
|
if !reflect.DeepEqual(gotParameters, tt.want.params) {
|
|
t.Errorf("columnsToQuery() gotParameters = %v, want %v", gotParameters, tt.want.params)
|
|
}
|
|
if !reflect.DeepEqual(gotValues, tt.want.values) {
|
|
t.Errorf("columnsToQuery() gotValues = %v, want %v", gotValues, tt.want.values)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func Test_conditionsToWhere(t *testing.T) {
|
|
type args struct {
|
|
conds []handler.Condition
|
|
paramOffset int
|
|
}
|
|
type want struct {
|
|
wheres []string
|
|
values []interface{}
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
want want
|
|
}{
|
|
{
|
|
name: "no wheres",
|
|
args: args{},
|
|
want: want{
|
|
wheres: []string{},
|
|
values: []interface{}{},
|
|
},
|
|
},
|
|
{
|
|
name: "no offset",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
handler.NewCond("col1", "val1"),
|
|
},
|
|
paramOffset: 0,
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 = $1)"},
|
|
values: []interface{}{"val1"},
|
|
},
|
|
},
|
|
{
|
|
name: "multiple cols",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
handler.NewCond("col1", "val1"),
|
|
handler.NewCond("col2", "val2"),
|
|
},
|
|
paramOffset: 0,
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 = $1)", "(col2 = $2)"},
|
|
values: []interface{}{"val1", "val2"},
|
|
},
|
|
},
|
|
{
|
|
name: "2 offset",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
handler.NewCond("col1", "val1"),
|
|
},
|
|
paramOffset: 2,
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 = $3)"},
|
|
values: []interface{}{"val1"},
|
|
},
|
|
},
|
|
{
|
|
name: "less than",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
NewLessThanCond("col1", "val1"),
|
|
},
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 < $1)"},
|
|
values: []interface{}{"val1"},
|
|
},
|
|
},
|
|
{
|
|
name: "is null",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
NewIsNullCond("col1"),
|
|
},
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 IS NULL)"},
|
|
values: []interface{}{},
|
|
},
|
|
},
|
|
{
|
|
name: "text array contains",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
NewTextArrayContainsCond("col1", "val1"),
|
|
},
|
|
},
|
|
want: want{
|
|
wheres: []string{"(col1 @> $1)"},
|
|
values: []interface{}{database.StringArray{"val1"}},
|
|
},
|
|
},
|
|
{
|
|
name: "not",
|
|
args: args{
|
|
conds: []handler.Condition{
|
|
Not(handler.NewCond("col1", "val1")),
|
|
},
|
|
},
|
|
want: want{
|
|
wheres: []string{"(NOT (col1 = $1))"},
|
|
values: []interface{}{"val1"},
|
|
},
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
gotWheres, gotValues := conditionsToWhere(tt.args.conds, tt.args.paramOffset)
|
|
if !reflect.DeepEqual(gotWheres, tt.want.wheres) {
|
|
t.Errorf("columnsToWhere() gotWheres = %v, want %v", gotWheres, tt.want.wheres)
|
|
}
|
|
if !reflect.DeepEqual(gotValues, tt.want.values) {
|
|
t.Errorf("columnsToWhere() gotValues = %v, want %v", gotValues, tt.want.values)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestParameterOpts(t *testing.T) {
|
|
type args struct {
|
|
column string
|
|
value interface{}
|
|
placeholder string
|
|
}
|
|
tests := []struct {
|
|
name string
|
|
args args
|
|
constructor func(column string, value interface{}) handler.Column
|
|
want string
|
|
}{
|
|
{
|
|
name: "NewArrayAppendCol",
|
|
args: args{
|
|
column: "testCol",
|
|
value: "val",
|
|
placeholder: "$1",
|
|
},
|
|
constructor: NewArrayAppendCol,
|
|
want: "array_append(testCol, $1)",
|
|
},
|
|
{
|
|
name: "NewArrayRemoveCol",
|
|
args: args{
|
|
column: "testCol",
|
|
value: "val",
|
|
placeholder: "$1",
|
|
},
|
|
constructor: NewArrayRemoveCol,
|
|
want: "array_remove(testCol, $1)",
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
col := tt.constructor(tt.args.column, tt.args.value)
|
|
if param := col.ParameterOpt(tt.args.placeholder); param != tt.want {
|
|
t.Errorf("constructor() = %v, want %v", param, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|