From f2559c20272e5f94ca1cfda7d68024e893a3a617 Mon Sep 17 00:00:00 2001 From: adlerhurst Date: Mon, 5 Oct 2020 22:03:21 +0200 Subject: [PATCH] crdb tests --- internal/eventstore/v2/repository/sql/crdb.go | 13 +- .../v2/repository/sql/local_crdb_test.go | 329 ++++++++++-------- .../v2/repository/sql/query_test.go | 42 ++- 3 files changed, 230 insertions(+), 154 deletions(-) diff --git a/internal/eventstore/v2/repository/sql/crdb.go b/internal/eventstore/v2/repository/sql/crdb.go index 7e8bfc75c..ea67e7651 100644 --- a/internal/eventstore/v2/repository/sql/crdb.go +++ b/internal/eventstore/v2/repository/sql/crdb.go @@ -98,7 +98,7 @@ const ( " END " + " ) " + " ) " + - "RETURNING event_sequence, creation_date " + "RETURNING id, event_sequence, creation_date " ) type CRDB struct { @@ -127,14 +127,17 @@ func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error { event.AggregateType, event.AggregateID, event.Version, - event.CreationDate, - event.Data, + &sql.NullTime{ + Time: event.CreationDate, + Valid: !event.CreationDate.IsZero(), + }, + Data(event.Data), event.EditorUser, event.EditorService, event.ResourceOwner, - previousSequence, + Sequence(previousSequence), event.CheckPreviousSequence, - ).Scan(&event.Sequence, &event.CreationDate) + ).Scan(&event.ID, &event.Sequence, &event.CreationDate) if err != nil { tx.Rollback() diff --git a/internal/eventstore/v2/repository/sql/local_crdb_test.go b/internal/eventstore/v2/repository/sql/local_crdb_test.go index 1ee47f7b5..49d666e18 100644 --- a/internal/eventstore/v2/repository/sql/local_crdb_test.go +++ b/internal/eventstore/v2/repository/sql/local_crdb_test.go @@ -1,164 +1,221 @@ package sql -// import ( -// "database/sql" -// "fmt" -// "io/ioutil" -// "os" -// "path/filepath" -// "sort" -// "strconv" -// "strings" -// "testing" -// "time" +import ( + "context" + "database/sql" + "fmt" + "io/ioutil" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "testing" + "time" -// "github.com/caos/logging" -// "github.com/caos/zitadel/internal/eventstore/v2/repository" -// "github.com/cockroachdb/cockroach-go/v2/testserver" -// ) + "github.com/caos/logging" + "github.com/caos/zitadel/internal/eventstore/v2/repository" + "github.com/cockroachdb/cockroach-go/v2/testserver" +) -// var ( -// migrationsPath = os.ExpandEnv("${GOPATH}/src/github.com/caos/zitadel/migrations/cockroach") -// db *sql.DB -// ) +var ( + migrationsPath = os.ExpandEnv("${GOPATH}/src/github.com/caos/zitadel/migrations/cockroach") + db *sql.DB +) -// func TestMain(m *testing.M) { -// ts, err := testserver.NewTestServer() -// if err != nil { -// logging.LogWithFields("REPOS-RvjLG", "error", err).Fatal("unable to start db") -// } +func TestMain(m *testing.M) { + ts, err := testserver.NewTestServer() + if err != nil { + logging.LogWithFields("REPOS-RvjLG", "error", err).Fatal("unable to start db") + } -// db, err = sql.Open("postgres", ts.PGURL().String()) -// if err != nil { -// logging.LogWithFields("REPOS-CF6dQ", "error", err).Fatal("unable to connect to db") -// } + db, err = sql.Open("postgres", ts.PGURL().String()) + if err != nil { + logging.LogWithFields("REPOS-CF6dQ", "error", err).Fatal("unable to connect to db") + } -// defer func() { -// db.Close() -// ts.Stop() -// }() + defer func() { + db.Close() + ts.Stop() + }() -// if err = executeMigrations(); err != nil { -// logging.LogWithFields("REPOS-jehDD", "error", err).Fatal("migrations failed") -// } + if err = executeMigrations(); err != nil { + logging.LogWithFields("REPOS-jehDD", "error", err).Fatal("migrations failed") + } -// os.Exit(m.Run()) -// } + os.Exit(m.Run()) +} -// func TestInsert(t *testing.T) { -// tx, _ := db.Begin() +func TestInsert(t *testing.T) { + crdb := &CRDB{client: db} + events := []*repository.Event{ + { + AggregateID: "agg.id", + AggregateType: "agg.type", + CheckPreviousSequence: false, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + }, + { + AggregateID: "agg.id", + AggregateType: "agg.type", + CheckPreviousSequence: false, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + CreationDate: time.Now().Add(-2 * time.Second), + }, + { + AggregateID: "agg.id", + AggregateType: "agg.type", + CheckPreviousSequence: false, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + CreationDate: time.Now().Add(-500 * time.Millisecond), + }, + } + fmt.Println("==============") + fmt.Println("will insert ts:") + for _, event := range events { + fmt.Printf("%v | %v\n", event.CreationDate, event.ID) + } + err := crdb.Push(context.Background(), events...) + if err != nil { + t.Error(err) + return + } -// var seq Sequence -// var d time.Time + fmt.Println("inserted ts:") + for _, event := range events { + fmt.Printf("%v | %v\n", event.CreationDate, event.ID) + } -// row := tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) -// err := row.Scan(&seq, &d) + fmt.Println("==============") + for _, event := range events { + fmt.Printf("%+v", event) + } -// row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(1), true) -// err = row.Scan(&seq, &d) + // tx, _ := db.Begin() -// row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) -// err = row.Scan(&seq, &d) + // var seq Sequence + // var d time.Time -// tx.Commit() + // row := tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) + // err := row.Scan(&seq, &d) -// rows, err := db.Query("select * from eventstore.events order by event_sequence") -// defer rows.Close() -// fmt.Println(err) + // row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(1), true) + // err = row.Scan(&seq, &d) -// fmt.Println(rows.Columns()) -// for rows.Next() { -// i := make([]interface{}, 12) -// var id string -// rows.Scan(&id, &i[1], &i[2], &i[3], &i[4], &i[5], &i[6], &i[7], &i[8], &i[9], &i[10], &i[11]) -// i[0] = id + // row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) + // err = row.Scan(&seq, &d) -// fmt.Println(i) -// } + // tx.Commit() -// t.Fail() -// } + rows, err := db.Query("select * from eventstore.events order by event_sequence") + defer rows.Close() + fmt.Println(err) -// func executeMigrations() error { -// files, err := migrationFilePaths() -// if err != nil { -// return err -// } -// sort.Sort(files) -// for _, file := range files { -// migration, err := ioutil.ReadFile(string(file)) -// if err != nil { -// return err -// } -// transactionInMigration := strings.Contains(string(migration), "BEGIN;") -// exec := db.Exec -// var tx *sql.Tx -// if !transactionInMigration { -// tx, err = db.Begin() -// if err != nil { -// return fmt.Errorf("begin file: %v || err: %w", file, err) -// } -// exec = tx.Exec -// } -// if _, err = exec(string(migration)); err != nil { -// return fmt.Errorf("exec file: %v || err: %w", file, err) -// } -// if !transactionInMigration { -// if err = tx.Commit(); err != nil { -// return fmt.Errorf("commit file: %v || err: %w", file, err) -// } -// } -// } -// return nil -// } + fmt.Println(rows.Columns()) + for rows.Next() { + i := make([]interface{}, 12) + var id string + rows.Scan(&id, &i[1], &i[2], &i[3], &i[4], &i[5], &i[6], &i[7], &i[8], &i[9], &i[10], &i[11]) + i[0] = id -// type migrationPaths []string + fmt.Println(i) + } -// type version struct { -// major int -// minor int -// } + t.Fail() +} -// func versionFromPath(s string) version { -// v := s[strings.Index(s, "/V")+2 : strings.Index(s, "__")] -// splitted := strings.Split(v, ".") -// res := version{} -// var err error -// if len(splitted) >= 1 { -// res.major, err = strconv.Atoi(splitted[0]) -// if err != nil { -// panic(err) -// } -// } +func executeMigrations() error { + files, err := migrationFilePaths() + if err != nil { + return err + } + sort.Sort(files) + for _, file := range files { + migration, err := ioutil.ReadFile(string(file)) + if err != nil { + return err + } + transactionInMigration := strings.Contains(string(migration), "BEGIN;") + exec := db.Exec + var tx *sql.Tx + if !transactionInMigration { + tx, err = db.Begin() + if err != nil { + return fmt.Errorf("begin file: %v || err: %w", file, err) + } + exec = tx.Exec + } + if _, err = exec(string(migration)); err != nil { + return fmt.Errorf("exec file: %v || err: %w", file, err) + } + if !transactionInMigration { + if err = tx.Commit(); err != nil { + return fmt.Errorf("commit file: %v || err: %w", file, err) + } + } + } + return nil +} -// if len(splitted) >= 2 { -// res.minor, err = strconv.Atoi(splitted[1]) -// if err != nil { -// panic(err) -// } -// } +type migrationPaths []string -// return res -// } +type version struct { + major int + minor int +} -// func (a migrationPaths) Len() int { return len(a) } -// func (a migrationPaths) Swap(i, j int) { a[i], a[j] = a[j], a[i] } -// func (a migrationPaths) Less(i, j int) bool { -// versionI := versionFromPath(a[i]) -// versionJ := versionFromPath(a[j]) +func versionFromPath(s string) version { + v := s[strings.Index(s, "/V")+2 : strings.Index(s, "__")] + splitted := strings.Split(v, ".") + res := version{} + var err error + if len(splitted) >= 1 { + res.major, err = strconv.Atoi(splitted[0]) + if err != nil { + panic(err) + } + } -// return versionI.major < versionJ.major || -// (versionI.major == versionJ.major && versionI.minor < versionJ.minor) -// } + if len(splitted) >= 2 { + res.minor, err = strconv.Atoi(splitted[1]) + if err != nil { + panic(err) + } + } -// func migrationFilePaths() (migrationPaths, error) { -// files := make(migrationPaths, 0) -// err := filepath.Walk(migrationsPath, func(path string, info os.FileInfo, err error) error { -// if err != nil || info.IsDir() || !strings.HasSuffix(info.Name(), ".sql") { -// return err -// } -// files = append(files, path) -// return nil -// }) -// return files, err -// } + return res +} + +func (a migrationPaths) Len() int { return len(a) } +func (a migrationPaths) Swap(i, j int) { a[i], a[j] = a[j], a[i] } +func (a migrationPaths) Less(i, j int) bool { + versionI := versionFromPath(a[i]) + versionJ := versionFromPath(a[j]) + + return versionI.major < versionJ.major || + (versionI.major == versionJ.major && versionI.minor < versionJ.minor) +} + +func migrationFilePaths() (migrationPaths, error) { + files := make(migrationPaths, 0) + err := filepath.Walk(migrationsPath, func(path string, info os.FileInfo, err error) error { + if err != nil || info.IsDir() || !strings.HasSuffix(info.Name(), ".sql") { + return err + } + files = append(files, path) + return nil + }) + return files, err +} diff --git a/internal/eventstore/v2/repository/sql/query_test.go b/internal/eventstore/v2/repository/sql/query_test.go index 25ac03010..b96f00c2b 100644 --- a/internal/eventstore/v2/repository/sql/query_test.go +++ b/internal/eventstore/v2/repository/sql/query_test.go @@ -309,19 +309,6 @@ func Test_buildQuery(t *testing.T) { args args res res }{ - - // { - //removed because it's no valid test case - // name: "no query", - // args: args{ - // query: nil, - // }, - // res: res{ - // query: "", - // rowScanner: false, - // values: nil, - // }, - // }, { name: "with order by desc", args: args{ @@ -388,6 +375,35 @@ func Test_buildQuery(t *testing.T) { values: []interface{}{repository.AggregateType("user"), uint64(5)}, }, }, + { + name: "error no columns", + args: args{ + query: &repository.SearchQuery{ + Columns: repository.Columns(-1), + }, + }, + res: res{ + query: "", + rowScanner: false, + values: []interface{}(nil), + }, + }, + { + name: "invalid condition", + args: args{ + query: &repository.SearchQuery{ + Columns: repository.Columns_Event, + Filters: []*repository.Filter{ + {}, + }, + }, + }, + res: res{ + query: "", + rowScanner: false, + values: []interface{}(nil), + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) {