diff --git a/cmd/queryRunner/main.go b/cmd/queryRunner/main.go index 804a646d..dc56aa34 100644 --- a/cmd/queryRunner/main.go +++ b/cmd/queryRunner/main.go @@ -3,7 +3,6 @@ package main import ( "context" "log" - "os" "queryorchestration/api/queue" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" @@ -20,10 +19,10 @@ import ( func main() { ctx := context.Background() - closeTracer := otel.Init(ctx) + closeTracer := otel.New(ctx) defer closeTracer() - database.RunMigrations(os.Getenv("PWD")) + database.RunMigrations(&database.MigrationConfig{}) cfg, err := config.LoadDefaultConfig(ctx) if err != nil { diff --git a/cmd/queryService/main.go b/cmd/queryService/main.go index be2a9b5d..7705a2ff 100644 --- a/cmd/queryService/main.go +++ b/cmd/queryService/main.go @@ -4,7 +4,6 @@ import ( "context" "log" "net" - "os" "queryorchestration/api/controllers" serviceinterfaces "queryorchestration/api/serviceInterfaces" "queryorchestration/internal/collector" @@ -23,10 +22,10 @@ import ( func main() { ctx := context.Background() - closeTracer := otel.Init(ctx) + closeTracer := otel.New(ctx) defer closeTracer() - database.RunMigrations(os.Getenv("PWD")) + database.RunMigrations(&database.MigrationConfig{}) host := "0.0.0.0" port := 8080 diff --git a/database/migrations/20250108174833_create_fullactivequery_view.down.sql b/database/migrations/20250108174833_create_fullactivequery_view.down.sql new file mode 100644 index 00000000..04c4b9a6 --- /dev/null +++ b/database/migrations/20250108174833_create_fullactivequery_view.down.sql @@ -0,0 +1 @@ +DROP VIEW fullActiveQueries; \ No newline at end of file diff --git a/database/migrations/20250108174833_create_fullactivequery_view.up.sql b/database/migrations/20250108174833_create_fullactivequery_view.up.sql new file mode 100644 index 00000000..00f7f761 --- /dev/null +++ b/database/migrations/20250108174833_create_fullactivequery_view.up.sql @@ -0,0 +1,11 @@ + +CREATE VIEW fullActiveQueries AS + SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + WHERE (c.id is null or c.addedVersion >= q.activeVersion + and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion) + and (r.id is null or r.addedVersion >= q.activeVersion + and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion) + GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config; \ No newline at end of file diff --git a/database/queries/query.sql b/database/queries/query.sql index 08488057..efbab606 100644 --- a/database/queries/query.sql +++ b/database/queries/query.sql @@ -6,31 +6,14 @@ INSERT INTO queryDeprecations (queryId) VALUES ($1); -- name: IsQueryDeprecated :one SELECT EXISTS ( - SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null + SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is null LIMIT 1 ); -- name: GetQuery :one -SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds - FROM queries AS q - LEFT JOIN queryConfigs AS c ON q.id = c.queryId - LEFT JOIN requiredQueries AS r ON q.id = r.queryId - WHERE q.id = $1 - and c.addedVersion >= q.activeVersion - and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion - and r.addedVersion >= q.activeVersion - and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion - GROUP BY q.id, c.config; +SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1; -- name: ListQueries :many -SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds - FROM queries AS q - LEFT JOIN queryConfigs AS c ON q.id = c.queryId - LEFT JOIN requiredQueries AS r ON q.id = r.queryId - WHERE c.addedVersion >= q.activeVersion - and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion - and r.addedVersion >= q.activeVersion - and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion - GROUP BY q.id, c.config; +SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries; -- name: CreateQuery :one INSERT INTO queries (type) VALUES ($1) RETURNING id; diff --git a/internal/database/config_test.go b/internal/database/config_test.go new file mode 100644 index 00000000..512e0b9d --- /dev/null +++ b/internal/database/config_test.go @@ -0,0 +1,84 @@ +package database + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestMustGetConnectionString(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "port") + t.Setenv("DB_NAME", "name") + + connStr := mustGetConnectionString() + assert.Equal(t, "postgres://user:pass@host:port/name?", connStr) +} + +func TestMustGetConnectionStringNoSSL(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "port") + t.Setenv("DB_NAME", "name") + t.Setenv("DB_NOSSL", "1") + + connStr := mustGetConnectionString() + assert.Equal(t, "postgres://user:pass@host:port/name?sslmode=disable", connStr) +} + +func TestMustGetConnectionStringNoEnv(t *testing.T) { + t.Setenv("DB_USER", "") + t.Setenv("DB_PASS", "") + t.Setenv("DB_HOST", "") + t.Setenv("DB_PORT", "") + t.Setenv("DB_NAME", "") + + assert.Panics(t, func() { mustGetConnectionString() }) +} + +func TestGetPoolConfig(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "5432") + t.Setenv("DB_NAME", "name") + + config, err := getPoolConfig() + assert.Nil(t, err) + assert.Equal(t, config.ConnString(), mustGetConnectionString()) +} + +func TestGetPoolConfigInvalidPort(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "invalid_port") + t.Setenv("DB_NAME", "name") + + _, err := getPoolConfig() + assert.EqualError(t, err, "cannot parse `postgres://user:xxxxxx@host:invalid_port/name?`: failed to parse as URL (invalid port \":invalid_port\" after host)") +} + +func TestMustGetPoolConfig(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "5432") + t.Setenv("DB_NAME", "name") + + config := mustGetPoolConfig() + assert.Equal(t, config.ConnString(), mustGetConnectionString()) +} + +func TestMustGetPoolConfigInvalidPort(t *testing.T) { + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "invalid_port") + t.Setenv("DB_NAME", "name") + + assert.Panics(t, func() { mustGetPoolConfig() }) +} diff --git a/internal/database/connection.go b/internal/database/connection.go index 1219ac04..1448d85d 100644 --- a/internal/database/connection.go +++ b/internal/database/connection.go @@ -10,23 +10,70 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -func createConnectionString() string { - driver := "postgres" - dbUser := env.GetPanic("DB_USER") - dbPass := env.GetPanic("DB_PASS") - dbHost := env.GetPanic("DB_HOST") - dbPort := env.GetPanic("DB_PORT") - dbName := env.GetPanic("DB_NAME") - opts := "sslmode=disable" - connStr := fmt.Sprintf("%s://%s:%s@%s:%s/%s?%s", driver, dbUser, dbPass, dbHost, dbPort, dbName, opts) +type config struct { + user string + password string + host string + port string + name string + disableSSL string +} - return connStr +func mustGetConfig() *config { + user := env.GetPanic("DB_USER") + pass := env.GetPanic("DB_PASS") + host := env.GetPanic("DB_HOST") + port := env.GetPanic("DB_PORT") + name := env.GetPanic("DB_NAME") + + disableSSL, _ := env.Get("DB_NOSSL") + + return &config{ + user: user, + password: pass, + host: host, + port: port, + name: name, + disableSSL: disableSSL, + } +} + +func mustGetConnectionString() string { + driver := "postgres" + conf := mustGetConfig() + opts := "" + + if conf.disableSSL == "1" { + opts += "sslmode=disable" + } + + return fmt.Sprintf("%s://%s:%s@%s:%s/%s?%s", driver, conf.user, conf.password, conf.host, conf.port, conf.name, opts) +} + +func getPoolConfig() (*pgxpool.Config, error) { + connStr := mustGetConnectionString() + + config, err := pgxpool.ParseConfig(connStr) + if err != nil { + return nil, err + } + + return config, nil +} + +func mustGetPoolConfig() *pgxpool.Config { + config, err := getPoolConfig() + if err != nil { + log.Panicf("Unable to create database config: %v\n", err) + } + + return config } func GetDBPool(ctx context.Context) *pgxpool.Pool { - connStr := createConnectionString() + config := mustGetPoolConfig() - pool, err := pgxpool.New(ctx, connStr) + pool, err := pgxpool.NewWithConfig(ctx, config) if err != nil { log.Panicf("Unable to create database pool: %v\n", err) } @@ -35,9 +82,9 @@ func GetDBPool(ctx context.Context) *pgxpool.Pool { } func GetDBConn(ctx context.Context) *pgx.Conn { - connStr := createConnectionString() + config := mustGetConnectionString() - conn, err := pgx.Connect(ctx, connStr) + conn, err := pgx.Connect(ctx, config) if err != nil { log.Panicf("Unable to connect to database: %v\n", err) } diff --git a/internal/database/connection_test.go b/internal/database/connection_test.go index 52ddd885..4043ad5a 100644 --- a/internal/database/connection_test.go +++ b/internal/database/connection_test.go @@ -3,7 +3,6 @@ package database_test import ( "context" "fmt" - "os" "queryorchestration/internal/database" "testing" @@ -20,8 +19,16 @@ func TestDBConn(t *testing.T) { conn := database.GetDBConn(ctx) assert.NotNil(t, conn) +} + +func TestDBConnNoDB(t *testing.T) { + ctx := context.Background() + t.Setenv("DB_USER", "user") + t.Setenv("DB_PASS", "pass") + t.Setenv("DB_HOST", "host") + t.Setenv("DB_PORT", "5432") + t.Setenv("DB_NAME", "name") - os.Setenv("DB_HOST", "") assert.Panics(t, func() { database.GetDBConn(ctx) }) } @@ -32,9 +39,6 @@ func TestDBPool(t *testing.T) { pool := database.GetDBPool(ctx) assert.NotNil(t, pool) - - os.Setenv("DB_HOST", "") - assert.Panics(t, func() { database.GetDBPool(ctx) }) } type dbConfig struct { @@ -92,11 +96,11 @@ func createDB(t *testing.T, ctx context.Context) (*dbConfig, testcontainers.Cont config.Host = mappedHost config.Port = mappedPort.Int() - os.Setenv("DB_USER", config.User) - os.Setenv("DB_PASS", config.Password) - os.Setenv("DB_HOST", config.Host) - os.Setenv("DB_PORT", fmt.Sprint(config.Port)) - os.Setenv("DB_NAME", config.Name) + t.Setenv("DB_USER", config.User) + t.Setenv("DB_PASS", config.Password) + t.Setenv("DB_HOST", config.Host) + t.Setenv("DB_PORT", fmt.Sprint(config.Port)) + t.Setenv("DB_NAME", config.Name) return &config, container } diff --git a/internal/database/migrations.go b/internal/database/migrations.go index 87f3e4ac..d1e67182 100644 --- a/internal/database/migrations.go +++ b/internal/database/migrations.go @@ -4,8 +4,8 @@ import ( "database/sql" "fmt" "log" + "os" "path" - "queryorchestration/internal/env" "github.com/golang-migrate/migrate/v4" "github.com/golang-migrate/migrate/v4/database/postgres" @@ -13,13 +13,13 @@ import ( ) func createDB() { - dbUser := env.GetPanic("DB_USER") - dbPass := env.GetPanic("DB_PASS") - dbHost := env.GetPanic("DB_HOST") - dbPort := env.GetPanic("DB_PORT") - dbName := env.GetPanic("DB_NAME") driver := "postgres" - connStr := fmt.Sprintf("%s://%s:%s@%s:%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort) + opts := "" + conf := mustGetConfig() + if conf.disableSSL == "1" { + opts += "sslmode=disable" + } + connStr := fmt.Sprintf("%s://%s:%s@%s:%s?%s", driver, conf.user, conf.password, conf.host, conf.port, opts) db, err := sql.Open(driver, connStr) if err != nil { @@ -27,7 +27,7 @@ func createDB() { } defer db.Close() - result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", dbName, dbName)) + result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", conf.name, conf.name)) if err != nil { log.Panicf("Error creating database: %v", err) } @@ -36,16 +36,30 @@ func createDB() { if err != nil { log.Panic(err) } else if rows == 0 { - log.Printf("Database already exists: %s", dbName) + log.Printf("Database already exists: %s", conf.name) } else { - log.Printf("Database created: %s", dbName) + log.Printf("Database created: %s", conf.name) } } -func RunMigrations(basePath string) { +type MigrationConfig struct { + BasePath string + ConnectionString string +} + +func RunMigrations(config *MigrationConfig) { createDB() - connStr := createConnectionString() + connStr := config.ConnectionString + if connStr == "" { + connStr = mustGetConnectionString() + } + + basePath := config.BasePath + if basePath == "" { + basePath = os.Getenv("PWD") + } + driver := "postgres" db, err := sql.Open(driver, connStr) @@ -55,12 +69,12 @@ func RunMigrations(basePath string) { defer db.Close() if err = db.Ping(); err != nil { - log.Panic("Failed to connect to database:", err) + log.Panic("failed to connect to database:", err) } conn, err := postgres.WithInstance(db, &postgres.Config{}) if err != nil { - log.Panic(err) + log.Panic("failed to create migrations: ", err) } migPath := "file://" + path.Join(basePath, "database/migrations") diff --git a/internal/database/migrations_test.go b/internal/database/migrations_test.go index cfce2d29..47cabf57 100644 --- a/internal/database/migrations_test.go +++ b/internal/database/migrations_test.go @@ -6,6 +6,8 @@ import ( "path" "queryorchestration/internal/database" "testing" + + "github.com/stretchr/testify/assert" ) func TestRunMigrations(t *testing.T) { @@ -13,5 +15,17 @@ func TestRunMigrations(t *testing.T) { _, container := createDB(t, ctx) defer container.Terminate(ctx) - database.RunMigrations(path.Join(os.Getenv("PWD"), "../..")) + t.Setenv("DB_NOSSL", "1") + + database.RunMigrations(&database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../.."), + }) +} + +func TestRunMigrationsNoDB(t *testing.T) { + assert.Panics(t, func() { + database.RunMigrations(&database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../.."), + }) + }) } diff --git a/internal/database/repository/collector_test.go b/internal/database/repository/collector_test.go new file mode 100644 index 00000000..7348914a --- /dev/null +++ b/internal/database/repository/collector_test.go @@ -0,0 +1,30 @@ +package repository_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" +) + +func TestCollector(t *testing.T) { + ctx := context.Background() + pool, container := createDB(t, ctx) + defer container.Terminate(ctx) + queries := repository.New(pool) + + collectorID := database.MustToDBUUID(uuid.New()) + + collectorQueries, err := queries.GetCollectorQueries(ctx, collectorID) + assert.Nil(t, err) + assert.Equal(t, 0, len(collectorQueries)) + assert.ElementsMatch(t, []repository.GetCollectorQueriesRow{}, collectorQueries) + + jobID := database.MustToDBUUID(uuid.New()) + + _, err = queries.GetCollectorFromJobID(ctx, jobID) + assert.NotNil(t, err) +} diff --git a/internal/database/repository/db_test.go b/internal/database/repository/db_test.go new file mode 100644 index 00000000..21e8e834 --- /dev/null +++ b/internal/database/repository/db_test.go @@ -0,0 +1,41 @@ +package repository_test + +import ( + "context" + "queryorchestration/internal/database/repository" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestQueriesNew(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + queries := repository.New(pool) + assert.NotNil(t, queries) +} + +func TestQueriesWithTx(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + queries := repository.New(pool) + assert.NotNil(t, queries) + + pool.ExpectBeginTx(pgx.TxOptions{}) + + tx, err := pool.Begin(ctx) + assert.Nil(t, err) + + txQueries := queries.WithTx(tx) + assert.NotNil(t, txQueries) +} diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index fda9edcf..c7b8cdeb 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -92,6 +92,15 @@ type Collectorquerydependencytree struct { Queryversion int32 } +type Fullactivequery struct { + ID pgtype.UUID + Type Querytype + Activeversion int32 + Latestversion int32 + Config []byte + Requiredids []pgtype.UUID +} + type Query struct { ID pgtype.UUID Latestversion int32 diff --git a/internal/database/repository/query.sql.go b/internal/database/repository/query.sql.go index c3ab0ad7..e9445c64 100644 --- a/internal/database/repository/query.sql.go +++ b/internal/database/repository/query.sql.go @@ -62,30 +62,12 @@ func (q *Queries) DeprecateQuery(ctx context.Context, queryid pgtype.UUID) error } const getQuery = `-- name: GetQuery :one -SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds - FROM queries AS q - LEFT JOIN queryConfigs AS c ON q.id = c.queryId - LEFT JOIN requiredQueries AS r ON q.id = r.queryId - WHERE q.id = $1 - and c.addedVersion >= q.activeVersion - and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion - and r.addedVersion >= q.activeVersion - and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion - GROUP BY q.id, c.config +SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1 ` -type GetQueryRow struct { - ID pgtype.UUID - Type Querytype - Activeversion int32 - Latestversion int32 - Config []byte - Requiredids []pgtype.UUID -} - -func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (GetQueryRow, error) { +func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (Fullactivequery, error) { row := q.db.QueryRow(ctx, getQuery, id) - var i GetQueryRow + var i Fullactivequery err := row.Scan( &i.ID, &i.Type, @@ -120,7 +102,7 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg GetQueryConfigParams) const isQueryDeprecated = `-- name: IsQueryDeprecated :one SELECT EXISTS ( - SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null + SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is null LIMIT 1 ) ` @@ -132,35 +114,18 @@ func (q *Queries) IsQueryDeprecated(ctx context.Context, queryid pgtype.UUID) (b } const listQueries = `-- name: ListQueries :many -SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds - FROM queries AS q - LEFT JOIN queryConfigs AS c ON q.id = c.queryId - LEFT JOIN requiredQueries AS r ON q.id = r.queryId - WHERE c.addedVersion >= q.activeVersion - and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion - and r.addedVersion >= q.activeVersion - and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion - GROUP BY q.id, c.config +SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries ` -type ListQueriesRow struct { - ID pgtype.UUID - Type Querytype - Activeversion int32 - Latestversion int32 - Config []byte - Requiredids []pgtype.UUID -} - -func (q *Queries) ListQueries(ctx context.Context) ([]ListQueriesRow, error) { +func (q *Queries) ListQueries(ctx context.Context) ([]Fullactivequery, error) { rows, err := q.db.Query(ctx, listQueries) if err != nil { return nil, err } defer rows.Close() - var items []ListQueriesRow + var items []Fullactivequery for rows.Next() { - var i ListQueriesRow + var i Fullactivequery if err := rows.Scan( &i.ID, &i.Type, diff --git a/internal/database/repository/query_test.go b/internal/database/repository/query_test.go new file mode 100644 index 00000000..ea41fc9d --- /dev/null +++ b/internal/database/repository/query_test.go @@ -0,0 +1,90 @@ +package repository_test + +import ( + "context" + "log" + "queryorchestration/internal/database/repository" + "testing" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/assert" +) + +func TestQueries(t *testing.T) { + ctx := context.Background() + pool, container := createDB(t, ctx) + defer container.Terminate(ctx) + queries := repository.New(pool) + + contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull)) + assert.Nil(t, err) + assert.True(t, contextQueryID.Valid) + + jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor)) + assert.Nil(t, err) + assert.True(t, jsonQueryID.Valid) + + jsonConfig := []byte("{\"path\": \"example_path\"}") + + err = queries.CreateRequiredQuery(ctx, repository.CreateRequiredQueryParams{ + Queryid: jsonQueryID, + Requiredqueryid: contextQueryID, + Addedversion: 1, + }) + assert.Nil(t, err) + + err = queries.CreateQueryConfig(ctx, repository.CreateQueryConfigParams{ + Queryid: jsonQueryID, + Config: jsonConfig, + Addedversion: 1, + }) + assert.Nil(t, err) + + jsonQuery, err := queries.GetQuery(ctx, jsonQueryID) + assert.Nil(t, err) + assert.EqualExportedValues(t, repository.Fullactivequery{ + ID: jsonQueryID, + Type: repository.QuerytypeJsonExtractor, + Activeversion: 1, + Latestversion: 1, + Config: jsonConfig, + Requiredids: []pgtype.UUID{contextQueryID}, + }, jsonQuery) + + jsonQueryConfig, err := queries.GetQueryConfig(ctx, repository.GetQueryConfigParams{ + Queryid: jsonQueryID, + Addedversion: jsonQuery.Activeversion, + }) + assert.Nil(t, err) + assert.Equal(t, jsonConfig, jsonQueryConfig.Config) + + qs, err := queries.ListQueries(ctx) + assert.Nil(t, err) + assert.Equal(t, 2, len(qs)) + log.Print(qs) + assert.ElementsMatch(t, []repository.Fullactivequery{ + {ID: jsonQueryID, + Type: repository.QuerytypeJsonExtractor, + Activeversion: 1, + Latestversion: 1, + Config: jsonConfig, + Requiredids: []pgtype.UUID{contextQueryID}}, + {ID: contextQueryID, + Type: repository.QuerytypeContextFull, + Activeversion: 1, + Latestversion: 1, + Config: nil, + Requiredids: []pgtype.UUID{{}}}, + }, qs) + + isDeprecated, err := queries.IsQueryDeprecated(ctx, jsonQueryID) + assert.Nil(t, err) + assert.False(t, isDeprecated) + + err = queries.DeprecateQuery(ctx, jsonQueryID) + assert.Nil(t, err) + + isDeprecated, err = queries.IsQueryDeprecated(ctx, jsonQueryID) + assert.Nil(t, err) + assert.True(t, isDeprecated) +} diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go new file mode 100644 index 00000000..d960c5a4 --- /dev/null +++ b/internal/database/repository/result_test.go @@ -0,0 +1,68 @@ +package repository_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/assert" +) + +func TestResults(t *testing.T) { + ctx := context.Background() + pool, container := createDB(t, ctx) + defer container.Terminate(ctx) + queries := repository.New(pool) + + jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor)) + assert.Nil(t, err) + + documentID := database.MustToDBUUID(uuid.New()) + + jsonQuery, err := queries.GetQuery(ctx, jsonQueryID) + assert.Nil(t, err) + + cleanVersion := int32(1) + textVersion := int32(1) + jsonResultValue := "example_value" + + jsonResultID, err := queries.SetResult(ctx, repository.SetResultParams{ + Queryid: jsonQueryID, + Documentid: documentID, + Value: jsonResultValue, + Cleanversion: cleanVersion, + Textversion: textVersion, + Queryversion: jsonQuery.Activeversion, + }) + assert.Nil(t, err) + assert.True(t, jsonResultID.Valid) + + resultsByDoc, err := queries.ListResultsByDocumentID(ctx, repository.ListResultsByDocumentIDParams{ + Documentid: documentID, + Cleanversion: cleanVersion, + Textversion: textVersion, + }) + assert.Nil(t, err) + assert.Equal(t, 1, len(resultsByDoc)) + assert.ElementsMatch(t, []repository.ListResultsByDocumentIDRow{ + { + ID: jsonResultID, + Queryid: jsonQueryID, + Queryversion: jsonQuery.Activeversion, + }, + }, resultsByDoc) + + results, err := queries.ListResultValuesByID(ctx, []pgtype.UUID{jsonResultID}) + assert.Nil(t, err) + assert.Equal(t, 1, len(results)) + assert.ElementsMatch(t, []repository.ListResultValuesByIDRow{ + { + ID: jsonResultID, + Queryid: jsonQueryID, + Value: jsonResultValue, + }, + }, results) +} diff --git a/internal/database/repository/sqlquery_test.go b/internal/database/repository/sqlquery_test.go new file mode 100644 index 00000000..8c6b84c1 --- /dev/null +++ b/internal/database/repository/sqlquery_test.go @@ -0,0 +1,69 @@ +package repository_test + +import ( + "context" + "fmt" + "os" + "path" + "queryorchestration/internal/database" + "testing" + + "github.com/docker/go-connections/nat" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" +) + +func createDB(t *testing.T, ctx context.Context) (*pgxpool.Pool, testcontainers.Container) { + port, err := nat.NewPort("tcp", "5432") + if err != nil { + t.Fatalf("Failed to create port: %v", err) + } + + name := "queryorchestration" + pass := "pass" + user := "postgres" + + req := testcontainers.ContainerRequest{ + Image: "postgres:latest", + Env: map[string]string{ + "POSTGRES_DB": name, + "POSTGRES_USER": user, + "POSTGRES_PASSWORD": pass, + }, + ExposedPorts: []string{port.Port()}, + WaitingFor: wait.ForListeningPort(port), + } + + container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Fatalf("Failed to start container: %v", err) + } + + host, err := container.Host(ctx) + if err != nil { + t.Fatalf("Failed to extract host: %v", err) + } + mappedPort, err := container.MappedPort(ctx, port) + if err != nil { + t.Fatalf("Failed to extract port: %v", err) + } + + t.Setenv("DB_USER", user) + t.Setenv("DB_PASS", pass) + t.Setenv("DB_HOST", host) + t.Setenv("DB_PORT", fmt.Sprint(mappedPort.Int())) + t.Setenv("DB_NAME", name) + t.Setenv("DB_NOSSL", "1") + + database.RunMigrations(&database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../../.."), + }) + + pool := database.GetDBPool(ctx) + + return pool, container +} diff --git a/internal/env/get_test.go b/internal/env/get_test.go index 6aa44eb1..daaef96e 100644 --- a/internal/env/get_test.go +++ b/internal/env/get_test.go @@ -1,7 +1,6 @@ package env_test import ( - "os" "queryorchestration/internal/env" "testing" @@ -12,7 +11,7 @@ func TestGet(t *testing.T) { name := "test_get_env_var_does_not_exist" value := "value" - os.Setenv(name, value) + t.Setenv(name, value) returnValue, err := env.Get(name) @@ -20,7 +19,7 @@ func TestGet(t *testing.T) { assert.Nil(t, err) value = "" - os.Setenv(name, value) + t.Setenv(name, value) returnValue, err = env.Get(name) diff --git a/internal/env/getfatal_test.go b/internal/env/getfatal_test.go index f6bd5054..79978c9c 100644 --- a/internal/env/getfatal_test.go +++ b/internal/env/getfatal_test.go @@ -1,7 +1,6 @@ package env_test import ( - "os" "queryorchestration/internal/env" "testing" @@ -12,14 +11,14 @@ func TestGetPanic(t *testing.T) { name := "test_get_env_var_does_not_exist" value := "value" - os.Setenv(name, value) + t.Setenv(name, value) returnValue := env.GetPanic(name) assert.Equal(t, value, returnValue) value = "" - os.Setenv(name, value) + t.Setenv(name, value) assert.Panics(t, func() { env.GetPanic(name) }) } diff --git a/internal/export/service_test.go b/internal/export/service_test.go new file mode 100644 index 00000000..9ffcb523 --- /dev/null +++ b/internal/export/service_test.go @@ -0,0 +1,26 @@ +package export_test + +import ( + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/export" + "testing" + + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestNew(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := export.New(db) + assert.NotNil(t, svc) +} diff --git a/internal/otel/service.go b/internal/otel/service.go index 609f5483..1023ac71 100644 --- a/internal/otel/service.go +++ b/internal/otel/service.go @@ -10,7 +10,7 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" ) -func Init(ctx context.Context) func() { +func New(ctx context.Context) func() { if os.Getenv("ENABLE_OTEL") != "1" { log.Println("OpenTelemetry is disabled. Set ENABLE_OTEL to enable.") return func() {} // No-op shutdown function diff --git a/internal/otel/service_test.go b/internal/otel/service_test.go new file mode 100644 index 00000000..90d9a343 --- /dev/null +++ b/internal/otel/service_test.go @@ -0,0 +1,16 @@ +package otel_test + +import ( + "context" + "queryorchestration/internal/otel" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNew(t *testing.T) { + ctx := context.Background() + + svc := otel.New(ctx) + assert.NotNil(t, svc) +} diff --git a/internal/query/create_test.go b/internal/query/create_test.go index 72921fe6..36ffa0d9 100644 --- a/internal/query/create_test.go +++ b/internal/query/create_test.go @@ -30,10 +30,8 @@ func TestCreate(t *testing.T) { config := "{\"path\":\"example_path\"}" q := query.Query{ - ID: uuid.New(), - Type: queryprocessor.TypeJsonExtractor, - ActiveVersion: int32(1), - LatestVersion: int32(1), + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{ uuid.New(), }, diff --git a/internal/query/createprivate_test.go b/internal/query/createprivate_test.go new file mode 100644 index 00000000..b0fce7bf --- /dev/null +++ b/internal/query/createprivate_test.go @@ -0,0 +1,122 @@ +package query + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestGetCreator(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + svc := New(db) + + queryType := queryprocessor.Type(queryprocessor.TypeContextFull) + creator, err := svc.getCreator(queryType) + assert.Nil(t, err) + assert.NotNil(t, creator) + + queryType = queryprocessor.Type(queryprocessor.TypeJsonExtractor) + creator, err = svc.getCreator(queryType) + assert.Nil(t, err) + assert.NotNil(t, creator) + + queryType = queryprocessor.Type(-1) + _, err = svc.getCreator(queryType) + assert.NotNil(t, err) +} + +func TestParseCreateQuery(t *testing.T) { + cQuery := &queryprocessor.Create{ + Type: queryprocessor.TypeContextFull, + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: "{\"key\":\"value\"}", + } + + resultQuery, err := parseCreateQuery(cQuery) + assert.Nil(t, err) + assert.EqualExportedValues(t, createQuery{ + Type: repository.QuerytypeContextFull, + RequiredQueryIDs: database.MustToDBUUIDArray(cQuery.RequiredQueryIDs), + Config: []byte(cQuery.Config), + }, *resultQuery) +} + +func TestParseCreateQueryInvalidType(t *testing.T) { + cQuery := &queryprocessor.Create{ + Type: queryprocessor.Type(-1), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: "{\"key\":\"value\"}", + } + + _, err := parseCreateQuery(cQuery) + assert.EqualError(t, err, "invalid database query type") +} + +func TestSubmitCreate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + svc := New(db) + + config := "{\"path\":\"example_path\"}" + q := Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + create := &queryprocessor.Create{ + Type: q.Type, + RequiredQueryIDs: q.RequiredQueryIDs, + Config: q.Config, + } + + dbType, err := queryprocessor.ToDBQueryType(create.Type) + assert.Nil(t, err) + + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectQuery("name: CreateQuery :one").WithArgs(dbType).WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(q.ID)), + ) + for _, req := range create.RequiredQueryIDs { + pool.ExpectExec("name: CreateRequiredQuery :exec").WithArgs(database.MustToDBUUID(q.ID), database.MustToDBUUID(req), int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + } + pool.ExpectExec("name: CreateQueryConfig :exec").WithArgs(database.MustToDBUUID(q.ID), []byte(create.Config), int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + id, err := svc.submitCreate(ctx, create) + assert.Nil(t, err) + assert.Equal(t, q.ID, id) +} diff --git a/internal/query/get.go b/internal/query/get.go index eb38dcdb..6549c233 100644 --- a/internal/query/get.go +++ b/internal/query/get.go @@ -24,10 +24,10 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) { return nil, err } - return ParseDBGetQuery(&query) + return ParseFullActiveQuery(&query) } -func ParseDBGetQuery(q *repository.GetQueryRow) (*Query, error) { +func ParseFullActiveQuery(q *repository.Fullactivequery) (*Query, error) { reqQueryIDs := database.MustToUUIDArray(q.Requiredids) qType, err := queryprocessor.ParseDBType(q.Type) diff --git a/internal/query/list.go b/internal/query/list.go index 7bffbd30..3a29f72e 100644 --- a/internal/query/list.go +++ b/internal/query/list.go @@ -2,8 +2,6 @@ package query import ( "context" - "queryorchestration/internal/database" - "queryorchestration/internal/database/repository" queryprocessor "queryorchestration/internal/queryProcessor" ) @@ -21,7 +19,7 @@ func (s *Service) List(ctx context.Context, filters ListFilters) ([]*Query, erro queries := make([]*Query, len(dbQueries)) for index, query := range dbQueries { - q, err := ParseDBListQuery(&query) + q, err := ParseFullActiveQuery(&query) if err != nil { return nil, err } @@ -31,21 +29,3 @@ func (s *Service) List(ctx context.Context, filters ListFilters) ([]*Query, erro return queries, nil } - -func ParseDBListQuery(q *repository.ListQueriesRow) (*Query, error) { - reqQueryIDs := database.MustToUUIDArray(q.Requiredids) - - qType, err := queryprocessor.ParseDBType(q.Type) - if err != nil { - return nil, err - } - - return &Query{ - ID: database.MustToUUID(q.ID), - ActiveVersion: q.Activeversion, - LatestVersion: q.Latestversion, - Type: qType, - RequiredQueryIDs: reqQueryIDs, - Config: string(q.Config), - }, nil -} diff --git a/internal/query/parse_test.go b/internal/query/parse_test.go index 76ae7a2a..d5b4d62a 100644 --- a/internal/query/parse_test.go +++ b/internal/query/parse_test.go @@ -34,8 +34,8 @@ func TestParseQuery(t *testing.T) { }, *out) } -func TestParseDBListQuery(t *testing.T) { - q := &repository.ListQueriesRow{ +func TestParseFullActiveQuery(t *testing.T) { + q := &repository.Fullactivequery{ ID: database.MustToDBUUID(uuid.New()), Type: repository.QuerytypeContextFull, Activeversion: int32(1), @@ -46,33 +46,7 @@ func TestParseDBListQuery(t *testing.T) { Config: []byte("example"), } - out, err := query.ParseDBListQuery(q) - assert.Nil(t, err) - assert.EqualExportedValues(t, query.Query{ - ID: database.MustToUUID(q.ID), - Type: queryprocessor.TypeContextFull, - ActiveVersion: q.Activeversion, - LatestVersion: q.Latestversion, - RequiredQueryIDs: []uuid.UUID{ - database.MustToUUID(q.Requiredids[0]), - }, - Config: string(q.Config), - }, *out) -} - -func TestParseDBGetQuery(t *testing.T) { - q := &repository.GetQueryRow{ - ID: database.MustToDBUUID(uuid.New()), - Type: repository.QuerytypeContextFull, - Activeversion: int32(1), - Latestversion: int32(2), - Requiredids: []pgtype.UUID{ - database.MustToDBUUID(uuid.New()), - }, - Config: []byte("example"), - } - - out, err := query.ParseDBGetQuery(q) + out, err := query.ParseFullActiveQuery(q) assert.Nil(t, err) assert.EqualExportedValues(t, query.Query{ ID: database.MustToUUID(q.ID), diff --git a/internal/query/updateprivate_test.go b/internal/query/updateprivate_test.go new file mode 100644 index 00000000..d91196ee --- /dev/null +++ b/internal/query/updateprivate_test.go @@ -0,0 +1,72 @@ +package query + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestGetUpdator(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + svc := New(db) + + queryType := queryprocessor.Type(queryprocessor.TypeContextFull) + updator, err := svc.getUpdator(queryType) + assert.Nil(t, err) + assert.NotNil(t, updator) + + queryType = queryprocessor.Type(queryprocessor.TypeJsonExtractor) + updator, err = svc.getUpdator(queryType) + assert.Nil(t, err) + assert.NotNil(t, updator) + + queryType = queryprocessor.Type(-1) + _, err = svc.getUpdator(queryType) + assert.NotNil(t, err) +} + +func TestSubmitUpdate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + svc := New(db) + + config := "{\"path\":\"example_path\"}" + q := Query{ + ID: uuid.New(), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + update := &queryprocessor.Update{ + ID: q.ID, + RequiredQueryIDs: q.RequiredQueryIDs, + Config: q.Config, + } + + err = svc.submitUpdate(ctx, update) + assert.Nil(t, err) +} diff --git a/internal/queryQueue/create.go b/internal/queryQueue/create.go index 74f220f6..b4930a67 100644 --- a/internal/queryQueue/create.go +++ b/internal/queryQueue/create.go @@ -8,15 +8,7 @@ import ( func (q *Queue) getUnsyncedQueries() { for _, query := range q.collectorQueries { - isSynced := false - for _, result := range q.results { - if result.QueryID != query.ID || result.QueryVersion != query.Version { - continue - } - isSynced = true - break - } - if isSynced { + if q.isQuerySynced(query) { continue } @@ -24,6 +16,16 @@ func (q *Queue) getUnsyncedQueries() { } } +func (q *Queue) isQuerySynced(query *queryprocessor.Query) bool { + for _, result := range q.results { + if result.QueryID == query.ID && result.QueryVersion == query.Version { + return true + } + } + + return false +} + func (c *Queue) getCollectorQueries(ctx context.Context) error { if c.collectorQueries != nil { return nil diff --git a/internal/queryQueue/create_test.go b/internal/queryQueue/create_test.go new file mode 100644 index 00000000..dff8e42e --- /dev/null +++ b/internal/queryQueue/create_test.go @@ -0,0 +1,139 @@ +package queryqueue + +import ( + "context" + "queryorchestration/internal/collector" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "queryorchestration/internal/result" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestGetUnsyncedQueries(t *testing.T) { + queryOne := &queryprocessor.Query{ + ID: uuid.New(), + Version: int32(1), + } + svc := Queue{ + collectorQueries: []*queryprocessor.Query{ + queryOne, + }, + results: []*result.Result{ + {ID: uuid.New(), QueryID: queryOne.ID, QueryVersion: queryOne.Version}, + }, + } + + svc.getUnsyncedQueries() + assert.EqualExportedValues(t, []*queryprocessor.Query(nil), svc.unsyncedQueue) + + svc.results = []*result.Result{} + svc.unsyncedQueue = []*queryprocessor.Query{} + + svc.getUnsyncedQueries() + assert.EqualExportedValues(t, []*queryprocessor.Query{queryOne}, svc.unsyncedQueue) +} + +func TestGetCollectorQueries(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Queue{ + db: db, + collector: &collector.Collector{ + ID: uuid.New(), + }, + } + dbCollectorID := database.MustToDBUUID(svc.collector.ID) + + collectorQueries := []*queryprocessor.Query{ + {ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)}, + } + + rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}) + for _, q := range collectorQueries { + dbID := database.MustToDBUUID(q.ID) + dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs) + ty, err := queryprocessor.ToDBNullQueryType(q.Type) + assert.Nil(t, err) + rows = rows. + AddRow(dbCollectorID, dbID, ty, pgtype.Int4{Int32: int32(q.Version), Valid: true}, dbReqIDs) + } + + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows) + + svc.getCollectorQueries(ctx) + assert.EqualExportedValues(t, collectorQueries, svc.collectorQueries) +} + +func TestIsQuerySynced(t *testing.T) { + query := &queryprocessor.Query{ + ID: uuid.New(), + Version: int32(1), + } + svc := Queue{ + results: []*result.Result{ + {QueryID: query.ID, QueryVersion: query.Version}, + }, + } + + isSynced := svc.isQuerySynced(query) + assert.True(t, isSynced) +} + +func TestIsQuerySyncedNoResult(t *testing.T) { + query := &queryprocessor.Query{ + ID: uuid.New(), + Version: int32(1), + } + svc := Queue{ + results: []*result.Result{}, + } + + isSynced := svc.isQuerySynced(query) + assert.False(t, isSynced) +} + +func TestIsQuerySyncedOldResult(t *testing.T) { + query := &queryprocessor.Query{ + ID: uuid.New(), + Version: int32(1), + } + svc := Queue{ + results: []*result.Result{ + {QueryID: query.ID, QueryVersion: query.Version - 1}, + }, + } + + isSynced := svc.isQuerySynced(query) + assert.False(t, isSynced) +} + +func TestIsQuerySyncedNewResult(t *testing.T) { + query := &queryprocessor.Query{ + ID: uuid.New(), + Version: int32(1), + } + svc := Queue{ + results: []*result.Result{ + {QueryID: query.ID, QueryVersion: query.Version + 1}, + }, + } + + isSynced := svc.isQuerySynced(query) + assert.False(t, isSynced) +} diff --git a/internal/queryQueue/execute.go b/internal/queryQueue/execute.go index a7ae461e..f9fa393a 100644 --- a/internal/queryQueue/execute.go +++ b/internal/queryQueue/execute.go @@ -2,11 +2,7 @@ package queryqueue import ( "context" - "fmt" - contextfull "queryorchestration/internal/contextFull" "queryorchestration/internal/database" - "queryorchestration/internal/database/repository" - jsonextractor "queryorchestration/internal/jsonExtractor" queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" @@ -71,21 +67,3 @@ func (q *Queue) executeQuery(ctx context.Context, qu *queryprocessor.Query) erro return nil } - -func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) { - var queryType queryprocessor.Type - for _, qu := range q.collectorQueries { - if qu.ID == database.MustToUUID(res.Queryid) { - queryType = qu.Type - } - } - - switch queryType { - case queryprocessor.TypeJsonExtractor: - return jsonextractor.NewResult(res.Value), nil - case queryprocessor.TypeContextFull: - return contextfull.NewResult(res.Value), nil - default: - return nil, fmt.Errorf("attempting to process invalid query type") - } -} diff --git a/internal/queryQueue/queue_test.go b/internal/queryQueue/execute_test.go similarity index 86% rename from internal/queryQueue/queue_test.go rename to internal/queryQueue/execute_test.go index 5113500a..0f0425a7 100644 --- a/internal/queryQueue/queue_test.go +++ b/internal/queryQueue/execute_test.go @@ -2,7 +2,6 @@ package queryqueue_test import ( "context" - "errors" "fmt" "queryorchestration/internal/collector" "queryorchestration/internal/database" @@ -18,7 +17,7 @@ import ( "github.com/stretchr/testify/assert" ) -func TestQueue(t *testing.T) { +func TestExecute(t *testing.T) { ctx := context.Background() pool, err := pgxmock.NewPool() @@ -71,10 +70,7 @@ func TestQueue(t *testing.T) { rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}) for _, q := range collectorQueries { dbID := database.MustToDBUUID(q.ID) - dbReqIDs := make([]pgtype.UUID, len(q.RequiredQueryIDs)) - for index, id := range q.RequiredQueryIDs { - dbReqIDs[index] = database.MustToDBUUID(id) - } + dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs) ty, err := queryprocessor.ToDBNullQueryType(q.Type) assert.Nil(t, err) rows = rows. @@ -181,43 +177,3 @@ func TestQueue(t *testing.T) { err = q.Execute(ctx) assert.Nil(t, err) } - -func TestQueueFail(t *testing.T) { - ctx := context.Background() - - pool, err := pgxmock.NewPool() - if err != nil { - t.Fatalf("failed to open pgxmock database: %v", err) - } - queries := repository.New(pool) - db := &database.Connection{ - Queries: queries, - Pool: pool, - } - jobID := uuid.New() - dbJobID := database.MustToDBUUID(jobID) - collectorID := uuid.New() - dbCollectorID := database.MustToDBUUID(collectorID) - - pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). - WillReturnRows( - pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). - AddRow(dbCollectorID, dbJobID, int32(1), int32(1)), - ) - - coll, err := collector.NewByJobId(ctx, db, jobID) - assert.Nil(t, err) - - errr := "database failure" - pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID). - WillReturnError(errors.New(errr)) - - results := []*result.Result{} - - docID := uuid.New() - cleanVersion := int32(1) - textVersion := int32(1) - - _, err = queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion) - assert.EqualError(t, err, errr) -} diff --git a/internal/queryQueue/executeprivate_test.go b/internal/queryQueue/executeprivate_test.go new file mode 100644 index 00000000..6f326d7a --- /dev/null +++ b/internal/queryQueue/executeprivate_test.go @@ -0,0 +1,94 @@ +package queryqueue + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestExecute(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + expectedQueries := []*queryprocessor.Query{ + {ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)}, + {ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(2)}, + } + + q := &Queue{ + unsyncedQueue: expectedQueries, + db: db, + documentId: uuid.New(), + cleanVersion: int32(1), + textVersion: int32(2), + } + + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows( + pgxmock.NewRows([]string{"id", "queryId", "value"}), + ) + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(expectedQueries[0].ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, expectedQueries[0].Version). + WillReturnRows( + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), + ) + + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows( + pgxmock.NewRows([]string{"id", "queryId", "value"}), + ) + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(expectedQueries[1].ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, expectedQueries[1].Version). + WillReturnRows( + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), + ) + + err = q.Execute(ctx) + assert.Nil(t, err) +} + +func TestExecuteQuery(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + qu := &queryprocessor.Query{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)} + + q := &Queue{ + db: db, + documentId: uuid.New(), + cleanVersion: int32(1), + textVersion: int32(2), + } + + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows( + pgxmock.NewRows([]string{"id", "queryId", "value"}), + ) + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(qu.ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, qu.Version). + WillReturnRows( + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), + ) + + err = q.executeQuery(ctx, qu) + assert.Nil(t, err) +} diff --git a/internal/queryQueue/result.go b/internal/queryQueue/result.go index 90c1fd52..d18e0fd7 100644 --- a/internal/queryQueue/result.go +++ b/internal/queryQueue/result.go @@ -4,6 +4,8 @@ import ( "context" "fmt" contextfull "queryorchestration/internal/contextFull" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" jsonextractor "queryorchestration/internal/jsonExtractor" queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" @@ -51,3 +53,21 @@ func (q *Queue) getProcessor(queryType queryprocessor.Type) (queryprocessor.Proc return nil, fmt.Errorf("attempting to process invalid query type") } } + +func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) { + var queryType queryprocessor.Type + for _, qu := range q.collectorQueries { + if qu.ID == database.MustToUUID(res.Queryid) { + queryType = qu.Type + } + } + + switch queryType { + case queryprocessor.TypeJsonExtractor: + return jsonextractor.NewResult(res.Value), nil + case queryprocessor.TypeContextFull: + return contextfull.NewResult(res.Value), nil + default: + return nil, fmt.Errorf("attempting to process invalid query type") + } +} diff --git a/internal/queryQueue/resultprivate_test.go b/internal/queryQueue/resultprivate_test.go new file mode 100644 index 00000000..aece9c2f --- /dev/null +++ b/internal/queryQueue/resultprivate_test.go @@ -0,0 +1,74 @@ +package queryqueue + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "queryorchestration/internal/result" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestSetResult(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + q := &Queue{ + db: db, + documentId: uuid.New(), + cleanVersion: int32(1), + textVersion: int32(2), + } + + qu := &queryprocessor.Query{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)} + resultValues := []result.Value{} + + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(qu.ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, qu.Version). + WillReturnRows( + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), + ) + + err = q.setResult(ctx, qu, resultValues) + assert.Nil(t, err) +} + +func TestGetProcessor(t *testing.T) { + q := Queue{} + + qType := queryprocessor.Type(queryprocessor.TypeContextFull) + + processor, err := q.getProcessor(qType) + assert.Nil(t, err) + assert.NotNil(t, processor) +} + +func TestGetResultValue(t *testing.T) { + result := &repository.ListResultValuesByIDRow{ + Queryid: database.MustToDBUUID(uuid.New()), + Value: "EXAMPLE_VALUE", + } + + q := &Queue{ + collectorQueries: []*queryprocessor.Query{ + {ID: database.MustToUUID(result.Queryid), Type: queryprocessor.TypeContextFull}, + }, + } + + value, err := q.getResultValue(result) + assert.Nil(t, err) + assert.NotNil(t, value) +} diff --git a/internal/queryQueue/service_test.go b/internal/queryQueue/service_test.go new file mode 100644 index 00000000..e62aa49a --- /dev/null +++ b/internal/queryQueue/service_test.go @@ -0,0 +1,145 @@ +package queryqueue_test + +import ( + "context" + "errors" + "queryorchestration/internal/collector" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + queryqueue "queryorchestration/internal/queryQueue" + "queryorchestration/internal/result" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestService(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + jobID := uuid.New() + dbJobID := database.MustToDBUUID(jobID) + collectorID := uuid.New() + dbCollectorID := database.MustToDBUUID(collectorID) + + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). + AddRow(dbCollectorID, dbJobID, int32(1), int32(1)), + ) + + coll, err := collector.NewByJobId(ctx, db, jobID) + assert.Nil(t, err) + + queryOneID := uuid.New() + queryOneVersion := int32(1) + queryTwoID := uuid.New() + queryTwoVersion := int32(2) + queryThreeID := uuid.New() + queryThreeVersion := int32(3) + queryFourID := uuid.New() + queryFourVersion := int32(4) + queryFiveID := uuid.New() + queryFiveVersion := int32(5) + querySixID := uuid.New() + querySixVersion := int32(6) + contextID := uuid.New() + contextVersion := int32(1) + collectorQueries := []queryprocessor.Query{ + {ID: contextID, Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: contextVersion}, + {ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion}, + {ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion}, + {ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion}, + {ID: queryFourID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryFourVersion}, + {ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion}, + {ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion}, + } + + rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}) + for _, q := range collectorQueries { + dbID := database.MustToDBUUID(q.ID) + dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs) + ty, err := queryprocessor.ToDBNullQueryType(q.Type) + assert.Nil(t, err) + rows = rows. + AddRow(dbCollectorID, dbID, ty, pgtype.Int4{Int32: int32(q.Version), Valid: true}, dbReqIDs) + } + + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows) + + contextResultID := uuid.New() + results := []*result.Result{ + {ID: contextResultID, QueryID: contextID, QueryVersion: contextVersion}, + {ID: uuid.New(), QueryID: queryFourID, QueryVersion: queryFourVersion}, + {ID: uuid.New(), QueryID: querySixID, QueryVersion: querySixVersion - 1}, + {ID: uuid.New(), QueryID: queryOneID, QueryVersion: queryOneVersion - 1}, + } + + expectedQueries := []*queryprocessor.Query{ + {ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion}, + {ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion}, + {ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion}, + {ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion}, + {ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion}, + } + + docID := uuid.New() + cleanVersion := int32(1) + textVersion := int32(1) + + q, err := queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion) + assert.Nil(t, err) + assert.Equal(t, expectedQueries, q.GetQueue()) +} + +func TestQueueFail(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + jobID := uuid.New() + dbJobID := database.MustToDBUUID(jobID) + collectorID := uuid.New() + dbCollectorID := database.MustToDBUUID(collectorID) + + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). + AddRow(dbCollectorID, dbJobID, int32(1), int32(1)), + ) + + coll, err := collector.NewByJobId(ctx, db, jobID) + assert.Nil(t, err) + + errr := "database failure" + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID). + WillReturnError(errors.New(errr)) + + results := []*result.Result{} + + docID := uuid.New() + cleanVersion := int32(1) + textVersion := int32(1) + + _, err = queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion) + assert.EqualError(t, err, errr) +} diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go new file mode 100644 index 00000000..a914377d --- /dev/null +++ b/internal/queue/queue_test.go @@ -0,0 +1,92 @@ +package queue_test + +import ( + "context" + "fmt" + "queryorchestration/internal/queue" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/docker/go-connections/nat" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" +) + +type queueConfig struct { + Container *testcontainers.Container + Config *queue.QueueConfig +} + +func createQueue(t *testing.T, ctx context.Context) (*queueConfig, func()) { + queueName := "test-queue" + region := "us-east-1" + alias := "localstack" + provider := credentials.NewStaticCredentialsProvider("test", "test", "") + + port, err := nat.NewPort("tcp", "4566") + if err != nil { + t.Fatalf("Failed to create port: %v", err) + } + + req := testcontainers.ContainerRequest{ + Image: "localstack/localstack:latest", + Env: map[string]string{ + "AWS_ACCESS_KEY_ID": provider.Value.AccessKeyID, + "AWS_SECRET_ACCESS_KEY": provider.Value.SecretAccessKey, + "AWS_SESSION_TOKEN": provider.Value.SessionToken, + "AWS_DEFAULT_REGION": region, + "AWS_REGION": region, + "SERVICES": "sqs", + "SKIP_SSL_CERT_DOWNLOAD": "1", + "LOCALSTACK_HOST": alias, + "SQS_ENDPOINT_STRATEGY": "path", + }, + ExposedPorts: []string{port.Port()}, + WaitingFor: wait.ForListeningPort(port), + } + + container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Fatalf("Failed to start container: %v", err) + } + + host, err := container.Host(ctx) + if err != nil { + t.Fatalf("Failed to extract host: %v", err) + } + mappedPort, err := container.MappedPort(ctx, port) + if err != nil { + t.Fatalf("Failed to extract port: %v", err) + } + + endpoint := fmt.Sprintf("http://%s:%s", host, mappedPort.Port()) + cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region), config.WithCredentialsProvider(provider), config.WithBaseEndpoint(endpoint)) + if err != nil { + t.Fatal(err) + } + + client := sqs.NewFromConfig(cfg) + + queueM, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{ + QueueName: aws.String(queueName), + }) + if err != nil { + t.Fatal(err) + } + + return &queueConfig{ + Container: &container, + Config: &queue.QueueConfig{ + Client: client, + URL: *queueM.QueueUrl, + }, + }, func() { + container.Terminate(ctx) + } +}