package collectorset import ( "fmt" "testing" "queryorchestration/internal/collector" "queryorchestration/internal/database/repository" "queryorchestration/internal/serviceconfig" "queryorchestration/internal/serviceconfig/queue/clientsync" queuemock "queryorchestration/mocks/queue" "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) type CollectorSetConfig struct { serviceconfig.BaseConfig clientsync.ClientSyncConfig } func TestGetSetParams(t *testing.T) { t.Run("all params", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{ Collector: collector.New(cfg), }, } minCleanV := int64(1) minTextV := int64(1) aV := int32(3) current := collector.Collector{ ActiveVersion: 1, LatestVersion: 10, } params := SetParams{ ActiveVersion: &aV, MinCleanVersion: &minCleanV, MinTextVersion: &minTextV, Fields: &map[string]uuid.UUID{ "example_key": uuid.New(), }, } pool.ExpectQuery("name: AllQueriesExist :one").WithArgs([]uuid.UUID{(*params.Fields)["example_key"]}).WillReturnRows( pgxmock.NewRows([]string{"all_exist"}). AddRow(true), ) clientId := "string" dbparams, err := svc.getSetParams(ctx, clientId, ¤t, ¶ms) require.NoError(t, err) assert.EqualExportedValues(t, &dbSetParams{ ClientID: clientId, ActiveVersion: &aV, MinCleanVersion: &minCleanV, MinTextVersion: &minTextV, Fields: &map[string]uuid.UUID{ "example_key": (*params.Fields)["example_key"], }, }, dbparams) (*params.Fields)["second_key"] = (*params.Fields)["example_key"] assert.Len(t, *params.Fields, 2) _, err = svc.getSetParams(ctx, clientId, ¤t, ¶ms) assert.Error(t, err) }) t.Run("no params", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } current := collector.Collector{} params := SetParams{} _, err = svc.getSetParams(ctx, current.ClientID, ¤t, ¶ms) assert.EqualError(t, err, "no changes") }) t.Run("ONLY active version to next latest", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } current := collector.Collector{ ActiveVersion: 2, LatestVersion: 4, } version := int32(5) params := SetParams{ ActiveVersion: &version, } _, err = svc.getSetParams(ctx, current.ClientID, ¤t, ¶ms) assert.EqualError(t, err, "no changes") }) t.Run("no fields change", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } current := collector.Collector{ Fields: map[string]uuid.UUID{ "example": uuid.New(), }, } params := SetParams{ Fields: &map[string]uuid.UUID{ "example": current.Fields["example"], }, } _, err = svc.getSetParams(ctx, current.ClientID, ¤t, ¶ms) assert.EqualError(t, err, "no changes") }) } func TestSubmitSet(t *testing.T) { t.Run("all fields", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } minCleanV := int64(2) minTextV := int64(4) current := collector.Collector{ ClientID: "hello", ActiveVersion: 1, LatestVersion: 1, MinCleanVersion: 1, MinTextVersion: 2, Fields: map[string]uuid.UUID{ "original_key": uuid.New(), "og_key": uuid.New(), }, } aV := int32(2) params := dbSetParams{ ClientID: current.ClientID, ActiveVersion: &aV, MinCleanVersion: &minCleanV, MinTextVersion: &minTextV, Fields: &map[string]uuid.UUID{ "example_key": uuid.New(), "second_key": uuid.New(), "changed_key": current.Fields["original_key"], "og_key": current.Fields["og_key"], }, } pool.ExpectBeginTx(pgx.TxOptions{}) rv := int32(2) pool.ExpectQuery("name: AddLatestCollectorVersion :one").WithArgs(current.ClientID).WillReturnRows( pgxmock.NewRows([]string{"version"}). AddRow(rv), ) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(current.ClientID, int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorCleanVersion :exec").WithArgs(current.ClientID, rv, *params.MinCleanVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorTextVersion :exec").WithArgs(current.ClientID, rv, *params.MinTextVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: RemoveCollectorQuery :exec").WithArgs(&rv, current.Fields["original_key"], current.ClientID). WillReturnResult(pgxmock.NewResult("", 1)) pool.MatchExpectationsInOrder(false) pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(current.ClientID, "example_key", (*params.Fields)["example_key"], int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(current.ClientID, "second_key", (*params.Fields)["second_key"], int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(current.ClientID, "changed_key", (*params.Fields)["changed_key"], int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() err = svc.submitSet(ctx, ¤t, ¶ms) require.NoError(t, err) }) t.Run("only active version", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } current := collector.Collector{ ClientID: "hell", ActiveVersion: 1, LatestVersion: 1, MinCleanVersion: 1, MinTextVersion: 2, Fields: map[string]uuid.UUID{}, } av := int32(2) params := dbSetParams{ ClientID: current.ClientID, ActiveVersion: &av, } pool.ExpectBeginTx(pgx.TxOptions{}) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(current.ClientID, int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() err = svc.submitSet(ctx, ¤t, ¶ms) require.NoError(t, err) }) } func TestNormalizeActiveVersion(t *testing.T) { svc := Service{} t.Run("all nil", func(t *testing.T) { err := svc.normalizeActiveVersion(nil, nil) assert.Error(t, err) }) t.Run("nil update", func(t *testing.T) { current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } err := svc.normalizeActiveVersion(¤t, nil) require.NoError(t, err) }) t.Run("no update", func(t *testing.T) { current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } update := SetParams{} err := svc.normalizeActiveVersion(¤t, &update) require.NoError(t, err) assert.Nil(t, update.ActiveVersion) }) t.Run("same version", func(t *testing.T) { version := int32(2) current := collector.Collector{ ActiveVersion: version, LatestVersion: 4, } update := SetParams{ ActiveVersion: &version, } err := svc.normalizeActiveVersion(¤t, &update) require.NoError(t, err) assert.Nil(t, update.ActiveVersion) }) t.Run("latest version plus 1", func(t *testing.T) { current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } version := int32(5) update := SetParams{ ActiveVersion: &version, } err := svc.normalizeActiveVersion(¤t, &update) require.NoError(t, err) assert.Equal(t, version, *update.ActiveVersion) }) t.Run("latest version plus 2", func(t *testing.T) { current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } version := int32(6) update := SetParams{ ActiveVersion: &version, } err := svc.normalizeActiveVersion(¤t, &update) assert.Error(t, err) }) } func TestNormalizeCodeVersions(t *testing.T) { svc := Service{} t.Run("all nil", func(t *testing.T) { err := svc.normalizeCodeVersions(nil, nil) assert.Error(t, err) }) t.Run("nil update", func(t *testing.T) { current := collector.Collector{} err := svc.normalizeCodeVersions(¤t, nil) require.NoError(t, err) }) t.Run("empty", func(t *testing.T) { current := collector.Collector{} update := SetParams{} err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) t.Run("valid clean", func(t *testing.T) { current := collector.Collector{} update := SetParams{} cv := int64(1) update.MinCleanVersion = &cv err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Equal(t, cv, *update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) t.Run("valid text", func(t *testing.T) { current := collector.Collector{} update := SetParams{} tv := int64(1) update.MinTextVersion = &tv err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Equal(t, tv, *update.MinTextVersion) }) t.Run("valid clean and text", func(t *testing.T) { current := collector.Collector{} update := SetParams{} current.MinCleanVersion = 1 current.MinTextVersion = 1 err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) t.Run("current clean", func(t *testing.T) { current := collector.Collector{ MinCleanVersion: 1, } update := SetParams{} update.MinCleanVersion = ¤t.MinCleanVersion err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) t.Run("current text", func(t *testing.T) { current := collector.Collector{ MinTextVersion: 1, } update := SetParams{} update.MinTextVersion = ¤t.MinTextVersion err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) t.Run("current clean and text", func(t *testing.T) { current := collector.Collector{ MinCleanVersion: 1, MinTextVersion: 1, } update := SetParams{} update.MinCleanVersion = ¤t.MinCleanVersion update.MinTextVersion = ¤t.MinTextVersion err := svc.normalizeCodeVersions(¤t, &update) require.NoError(t, err) assert.Nil(t, update.MinCleanVersion) assert.Nil(t, update.MinTextVersion) }) } func TestGetRemoveFields(t *testing.T) { current := map[string]uuid.UUID{} var update *map[string]uuid.UUID remove := getRemoveFields(current, update) assert.Len(t, remove, 0) current["a"] = uuid.New() remove = getRemoveFields(current, update) assert.Len(t, remove, 0) update = &map[string]uuid.UUID{ "b": uuid.New(), } (*update)["a"] = current["a"] remove = getRemoveFields(current, update) assert.Len(t, remove, 0) (*update)["a"] = uuid.New() remove = getRemoveFields(current, update) assert.Len(t, remove, 1) assert.ElementsMatch(t, []uuid.UUID{current["a"]}, remove) (*update)["a"] = uuid.Nil (*update)["c"] = current["a"] remove = getRemoveFields(current, update) assert.Len(t, remove, 1) assert.ElementsMatch(t, []uuid.UUID{current["a"]}, remove) } func TestGetAddFields(t *testing.T) { current := map[string]uuid.UUID{} var update *map[string]uuid.UUID add := getAddFields(current, update) assert.Len(t, add, 0) current["a"] = uuid.New() add = getAddFields(current, update) assert.Len(t, add, 0) update = &map[string]uuid.UUID{} add = getAddFields(current, update) assert.Len(t, add, 0) (*update)["a"] = current["a"] add = getAddFields(current, update) assert.Len(t, add, 0) (*update)["a"] = uuid.New() add = getAddFields(current, update) assert.Len(t, add, 1) assert.Equal(t, map[string]uuid.UUID{ "a": (*update)["a"], }, add) (*update)["a"] = current["a"] (*update)["b"] = uuid.New() add = getAddFields(current, update) assert.Len(t, add, 1) assert.Equal(t, map[string]uuid.UUID{ "b": (*update)["b"], }, add) current = map[string]uuid.UUID{} add = getAddFields(current, update) assert.Len(t, add, 2) assert.Equal(t, map[string]uuid.UUID{ "b": (*update)["b"], "a": (*update)["a"], }, add) } func TestNormalizeSetFieldsToDB(t *testing.T) { t.Run("valid key", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{ Collector: collector.New(cfg), }, } current := map[string]uuid.UUID{} fields := map[string]uuid.UUID{ "example_key": uuid.New(), } pool.ExpectQuery("name: AllQueriesExist :one").WithArgs([]uuid.UUID{fields["example_key"]}).WillReturnRows( pgxmock.NewRows([]string{"all_exist"}). AddRow(true), ) dbparams, err := svc.normalizeSetFieldsToDB(ctx, current, &fields) require.NoError(t, err) assert.EqualExportedValues(t, &map[string]uuid.UUID{ "example_key": fields["example_key"], }, dbparams) }) t.Run("duplicate values", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } current := map[string]uuid.UUID{} fields := map[string]uuid.UUID{ "example_key": uuid.New(), } fields["second_key"] = fields["example_key"] pool.ExpectQuery("name: AllQueriesExist :one").WithArgs([]uuid.UUID{fields["example_key"]}).WillReturnRows( pgxmock.NewRows([]string{"all_exist"}). AddRow(true), ) _, err = svc.normalizeSetFieldsToDB(ctx, current, &fields) assert.Error(t, err) }) t.Run("no changes", func(t *testing.T) { ctx := t.Context() cfg := &CollectorSetConfig{} svc := Service{ cfg: cfg, svc: &Services{ Collector: collector.New(cfg), }, } current := map[string]uuid.UUID{ "example_key": uuid.New(), } fields := map[string]uuid.UUID{ "example_key": current["example_key"], } val, err := svc.normalizeSetFieldsToDB(ctx, current, &fields) require.NoError(t, err) assert.Nil(t, val) }) } func TestInformSet(t *testing.T) { t.Run("with version update", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) mockSQS := queuemock.NewMockSQSClient(t) cfg.QueueClient = mockSQS cfg.ClientSyncURL = "here" svc := New(cfg, &Services{}) av := int32(2) update := dbSetParams{ ClientID: "HI", ActiveVersion: &av, } mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", update.ClientID) }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) err = svc.informSet(ctx, &update) require.NoError(t, err) }) t.Run("with no update", func(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) mockSQS := queuemock.NewMockSQSClient(t) cfg.QueueClient = mockSQS cfg.ClientSyncURL = "here" svc := New(cfg, &Services{}) update := dbSetParams{ ClientID: "hello", } err = svc.informSet(ctx, &update) require.NoError(t, err) }) } func TestNormalizeFieldsToDB(t *testing.T) { ctx := t.Context() pool, err := pgxmock.NewPool() require.NoError(t, err) cfg := &CollectorSetConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, svc: &Services{}, } fields := map[string]uuid.UUID{ "example_key": uuid.New(), } pool.ExpectQuery("name: AllQueriesExist :one").WithArgs([]uuid.UUID{fields["example_key"]}).WillReturnRows( pgxmock.NewRows([]string{"all_exist"}). AddRow(true), ) dbparams, err := svc.normalizeFieldsToDB(ctx, &fields) require.NoError(t, err) assert.EqualExportedValues(t, &map[string]uuid.UUID{ "example_key": fields["example_key"], }, dbparams) fields["second_key"] = fields["example_key"] pool.ExpectQuery("name: AllQueriesExist :one").WithArgs([]uuid.UUID{fields["example_key"]}).WillReturnRows( pgxmock.NewRows([]string{"all_exist"}). AddRow(true), ) _, err = svc.normalizeFieldsToDB(ctx, &fields) assert.Error(t, err) }