package collectorset_test import ( "context" "fmt" "testing" "queryorchestration/internal/collector" collectorset "queryorchestration/internal/collector/set" "queryorchestration/internal/database" "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 TestSet(t *testing.T) { ctx := context.Background() 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 := collectorset.New(cfg, &collectorset.Services{ Collector: collector.New(cfg), }) t.Run("valid", func(t *testing.T) { current := collector.Collector{ ClientID: uuid.New(), ActiveVersion: 1, LatestVersion: 4, } av := int32(2) mv := int64(1) update := collectorset.SetParams{ ActiveVersion: &av, MinCleanVersion: &mv, } pool.ExpectQuery("name: GetCollectorByClientID :one").WithArgs(database.MustToDBUUID(current.ClientID)). WillReturnRows( pgxmock.NewRows([]string{"clientId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). AddRow(database.MustToDBUUID(current.ClientID), current.MinCleanVersion, current.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), ) pool.ExpectBeginTx(pgx.TxOptions{}) rv := current.LatestVersion + 1 pool.ExpectQuery("name: AddLatestCollectorVersion :one").WithArgs(database.MustToDBUUID(current.ClientID)).WillReturnRows( pgxmock.NewRows([]string{"version"}). AddRow(rv), ) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorCleanVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), rv, *update.MinCleanVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.ClientID.String()) }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) err = svc.SetByClientId(ctx, current.ClientID, &update) assert.NoError(t, err) }) t.Run("no collector", func(t *testing.T) { current := collector.Collector{ ClientID: uuid.New(), } av := int32(1) mv := int64(1) update := collectorset.SetParams{ ActiveVersion: &av, MinCleanVersion: &mv, } pool.ExpectQuery("name: GetCollectorByClientID :one").WithArgs(database.MustToDBUUID(current.ClientID)). WillReturnRows( pgxmock.NewRows([]string{"clientId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). AddRow(database.MustToDBUUID(current.ClientID), current.MinCleanVersion, current.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), ) pool.ExpectBeginTx(pgx.TxOptions{}) rv := current.LatestVersion + 1 pool.ExpectQuery("name: AddLatestCollectorVersion :one").WithArgs(database.MustToDBUUID(current.ClientID)).WillReturnRows( pgxmock.NewRows([]string{"version"}). AddRow(rv), ) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), int32(1)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorCleanVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), rv, *update.MinCleanVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.ClientID.String()) }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) err = svc.SetByClientId(ctx, current.ClientID, &update) assert.NoError(t, err) }) } func TestSetByExternalId(t *testing.T) { ctx := context.Background() 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 := collectorset.New(cfg, &collectorset.Services{ Collector: collector.New(cfg), }) externalId := "client_id" t.Run("valid", func(t *testing.T) { current := collector.Collector{ ClientID: uuid.New(), ActiveVersion: 1, LatestVersion: 4, } av := int32(2) mv := int64(1) update := collectorset.SetParams{ ActiveVersion: &av, MinCleanVersion: &mv, } pool.ExpectQuery("name: GetClientByExternalId :one").WithArgs(externalId).WillReturnRows( pgxmock.NewRows([]string{"id", "externalId", "name", "can_sync"}). AddRow(database.MustToDBUUID(current.ClientID), externalId, "name", true), ) pool.ExpectQuery("name: GetCollectorByClientID :one").WithArgs(database.MustToDBUUID(current.ClientID)). WillReturnRows( pgxmock.NewRows([]string{"clientId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). AddRow(database.MustToDBUUID(current.ClientID), current.MinCleanVersion, current.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), ) pool.ExpectBeginTx(pgx.TxOptions{}) rv := current.LatestVersion + 1 pool.ExpectQuery("name: AddLatestCollectorVersion :one").WithArgs(database.MustToDBUUID(current.ClientID)).WillReturnRows( pgxmock.NewRows([]string{"version"}). AddRow(rv), ) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), int32(2)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorCleanVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), rv, *update.MinCleanVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.ClientID.String()) }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) err = svc.SetByClientExternalId(ctx, externalId, &update) assert.NoError(t, err) }) t.Run("no collector", func(t *testing.T) { current := collector.Collector{ ClientID: uuid.New(), } av := int32(1) mv := int64(1) update := collectorset.SetParams{ ActiveVersion: &av, MinCleanVersion: &mv, } pool.ExpectQuery("name: GetClientByExternalId :one").WithArgs(externalId).WillReturnRows( pgxmock.NewRows([]string{"id", "externalId", "name", "can_sync"}). AddRow(database.MustToDBUUID(current.ClientID), externalId, "name", true), ) pool.ExpectQuery("name: GetCollectorByClientID :one").WithArgs(database.MustToDBUUID(current.ClientID)). WillReturnRows( pgxmock.NewRows([]string{"clientId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). AddRow(database.MustToDBUUID(current.ClientID), current.MinCleanVersion, current.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), ) pool.ExpectBeginTx(pgx.TxOptions{}) rv := current.LatestVersion + 1 pool.ExpectQuery("name: AddLatestCollectorVersion :one").WithArgs(database.MustToDBUUID(current.ClientID)).WillReturnRows( pgxmock.NewRows([]string{"version"}). AddRow(rv), ) pool.ExpectExec("name: SetActiveCollectorVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), int32(1)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: SetCollectorCleanVersion :exec").WithArgs(database.MustToDBUUID(current.ClientID), rv, *update.MinCleanVersion). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( mock.Anything, mock.MatchedBy(func(in *sqs.SendMessageInput) bool { return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.ClientID.String()) }), mock.Anything, ). Return(&sqs.SendMessageOutput{}, nil) err = svc.SetByClientExternalId(ctx, externalId, &update) assert.NoError(t, err) }) }