crdb tests

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,
+				&sql.NullTime{
-				event.Data,
+					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()

internal/eventstore/v2/repository/sql/local_crdb_test.go

@@ -1,164 +1,221 @@
 package sql
 
-// import (
+import (
-// 	"database/sql"
+	"context"
-// 	"fmt"
+	"database/sql"
-// 	"io/ioutil"
+	"fmt"
-// 	"os"
+	"io/ioutil"
-// 	"path/filepath"
+	"os"
-// 	"sort"
+	"path/filepath"
-// 	"strconv"
+	"sort"
-// 	"strings"
+	"strconv"
-// 	"testing"
+	"strings"
-// 	"time"
+	"testing"
+	"time"
 
-// 	"github.com/caos/logging"
+	"github.com/caos/logging"
-// 	"github.com/caos/zitadel/internal/eventstore/v2/repository"
+	"github.com/caos/zitadel/internal/eventstore/v2/repository"
-// 	"github.com/cockroachdb/cockroach-go/v2/testserver"
+	"github.com/cockroachdb/cockroach-go/v2/testserver"
-// )
+)
 
-// var (
+var (
-// 	migrationsPath = os.ExpandEnv("${GOPATH}/src/github.com/caos/zitadel/migrations/cockroach")
+	migrationsPath = os.ExpandEnv("${GOPATH}/src/github.com/caos/zitadel/migrations/cockroach")
-// 	db             *sql.DB
+	db             *sql.DB
-// )
+)
 
-// func TestMain(m *testing.M) {
+func TestMain(m *testing.M) {
-// 	ts, err := testserver.NewTestServer()
+	ts, err := testserver.NewTestServer()
-// 	if err != nil {
+	if err != nil {
-// 		logging.LogWithFields("REPOS-RvjLG", "error", err).Fatal("unable to start db")
+		logging.LogWithFields("REPOS-RvjLG", "error", err).Fatal("unable to start db")
-// 	}
+	}
 
-// 	db, err = sql.Open("postgres", ts.PGURL().String())
+	db, err = sql.Open("postgres", ts.PGURL().String())
-// 	if err != nil {
+	if err != nil {
-// 		logging.LogWithFields("REPOS-CF6dQ", "error", err).Fatal("unable to connect to db")
+		logging.LogWithFields("REPOS-CF6dQ", "error", err).Fatal("unable to connect to db")
-// 	}
+	}
 
-// 	defer func() {
+	defer func() {
-// 		db.Close()
+		db.Close()
-// 		ts.Stop()
+		ts.Stop()
-// 	}()
+	}()
 
-// 	if err = executeMigrations(); err != nil {
+	if err = executeMigrations(); err != nil {
-// 		logging.LogWithFields("REPOS-jehDD", "error", err).Fatal("migrations failed")
+		logging.LogWithFields("REPOS-jehDD", "error", err).Fatal("migrations failed")
-// 	}
+	}
 
-// 	os.Exit(m.Run())
+	os.Exit(m.Run())
-// }
+}
 
-// func TestInsert(t *testing.T) {
+func TestInsert(t *testing.T) {
-// 	tx, _ := db.Begin()
+	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
+	fmt.Println("inserted ts:")
-// 	var d time.Time
+	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)
+	fmt.Println("==============")
-// 	err := row.Scan(&seq, &d)
+	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)
+	// tx, _ := db.Begin()
-// 	err = row.Scan(&seq, &d)
 
-// 	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)
+	// var seq Sequence
-// 	err = row.Scan(&seq, &d)
+	// 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")
+	// 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)
-// 	defer rows.Close()
+	// err = row.Scan(&seq, &d)
-// 	fmt.Println(err)
 
-// 	fmt.Println(rows.Columns())
+	// 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)
-// 	for rows.Next() {
+	// err = row.Scan(&seq, &d)
-// 		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
 
-// 		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 {
+	fmt.Println(rows.Columns())
-// 	files, err := migrationFilePaths()
+	for rows.Next() {
-// 	if err != nil {
+		i := make([]interface{}, 12)
-// 		return err
+		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])
-// 	sort.Sort(files)
+		i[0] = id
-// 	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
-// }
 
-// type migrationPaths []string
+		fmt.Println(i)
+	}
 
-// type version struct {
+	t.Fail()
-// 	major int
+}
-// 	minor int
-// }
 
-// func versionFromPath(s string) version {
+func executeMigrations() error {
-// 	v := s[strings.Index(s, "/V")+2 : strings.Index(s, "__")]
+	files, err := migrationFilePaths()
-// 	splitted := strings.Split(v, ".")
+	if err != nil {
-// 	res := version{}
+		return err
-// 	var err error
+	}
-// 	if len(splitted) >= 1 {
+	sort.Sort(files)
-// 		res.major, err = strconv.Atoi(splitted[0])
+	for _, file := range files {
-// 		if err != nil {
+		migration, err := ioutil.ReadFile(string(file))
-// 			panic(err)
+		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 {
+type migrationPaths []string
-// 		res.minor, err = strconv.Atoi(splitted[1])
-// 		if err != nil {
-// 			panic(err)
-// 		}
-// 	}
 
-// 	return res
+type version struct {
-// }
+	major int
+	minor int
+}
 
-// func (a migrationPaths) Len() int      { return len(a) }
+func versionFromPath(s string) version {
-// func (a migrationPaths) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
+	v := s[strings.Index(s, "/V")+2 : strings.Index(s, "__")]
-// func (a migrationPaths) Less(i, j int) bool {
+	splitted := strings.Split(v, ".")
-// 	versionI := versionFromPath(a[i])
+	res := version{}
-// 	versionJ := versionFromPath(a[j])
+	var err error
+	if len(splitted) >= 1 {
+		res.major, err = strconv.Atoi(splitted[0])
+		if err != nil {
+			panic(err)
+		}
+	}
 
-// 	return versionI.major < versionJ.major ||
+	if len(splitted) >= 2 {
-// 		(versionI.major == versionJ.major && versionI.minor < versionJ.minor)
+		res.minor, err = strconv.Atoi(splitted[1])
-// }
+		if err != nil {
+			panic(err)
+		}
+	}
 
-// func migrationFilePaths() (migrationPaths, error) {
+	return res
-// 	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") {
+func (a migrationPaths) Len() int      { return len(a) }
-// 			return err
+func (a migrationPaths) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
-// 		}
+func (a migrationPaths) Less(i, j int) bool {
-// 		files = append(files, path)
+	versionI := versionFromPath(a[i])
-// 		return nil
+	versionJ := versionFromPath(a[j])
-// 	})
+
-// 	return files, err
+	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
+}

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) {