jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/momocks (pull request #150)

Browse Source 
Decrease Mocks

* feature/nomocks

* nonet

* assertsaws

* assert
This commit is contained in:
Michael McGuinness
2025-05-23 00:20:01 +00:00
parent 36ff8096f3
commit 33d68b7e04
42 changed files with 617 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/clientSyncRunner/runner_test.go
+12 -19
View File
@@ -2,36 +2,37 @@ package clientsyncrunner_test
import (
"fmt"
"regexp"
"testing"
clientsyncrunner "queryorchestration/api/clientSyncRunner"
clientsync "queryorchestration/internal/client/sync"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/documentsync"
"queryorchestration/internal/test"
queuemock "queryorchestration/mocks/queue"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type ClientSyncConfig struct {
runner.BaseConfig[clientsyncrunner.Body]
documentsync.DocSyncConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestQueryRunner(t *testing.T) {
func TestClientSyncRunner(t *testing.T) {
t.Parallel()
cfg := &ClientSyncConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.DocumentSyncURL = "/i/am/here"
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.DocumentSyncURL = test.CreateQueue(t, cfg, test.DocSyncRunnerName)
svc := clientsync.New(cfg)
@@ -54,15 +55,7 @@ func TestQueryRunner(t *testing.T) {
ClientID: "client_id",
}
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(t.Context(), bod))
test.AssertMessageBody(t, cfg, cfg.GetDocumentSyncURL(), regexp.MustCompile(fmt.Sprintf(`{"id":"%s"}`, docId.String())))
}
api/docCleanRunner/runner_test.go
+8 -17
View File
@@ -3,6 +3,7 @@ package doccleanrunner_test
import (
"fmt"
"io"
"regexp"
"strings"
"testing"
"time"
@@ -15,10 +16,8 @@ import (
"queryorchestration/internal/serviceconfig/queue/documenttexttrigger"
"queryorchestration/internal/test"
objectstoremock "queryorchestration/mocks/objectstore"
queuemock "queryorchestration/mocks/queue"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
@@ -34,14 +33,14 @@ type DocCleanConfig struct {
func TestDocCleanRunner(t *testing.T) {
t.Parallel()
cfg := &DocCleanConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.DocumentTextTriggerURL = test.CreateQueue(t, cfg, test.DocTextRunnerName)
mockStore := objectstoremock.NewMockS3Client(t)
cfg.StoreClient = mockStore
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.DocumentTextTriggerURL = "/i/am/here"
runner := doccleanrunner.New(&doccleanrunner.Services{
Clean: documentclean.New(cfg),
@@ -74,16 +73,6 @@ func TestDocCleanRunner(t *testing.T) {
})
require.NoError(t, err)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.DocumentTextTriggerURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
mimeType := "application/pdf"
mockStore.EXPECT().
HeadObject(
@@ -114,6 +103,8 @@ func TestDocCleanRunner(t *testing.T) {
DocumentID: docId,
}
assert.True(t, runner.Process(t.Context(), bod))
test.AssertMessageBody(t, cfg, cfg.GetDocumentTextTriggerURL(), regexp.MustCompile(fmt.Sprintf("{\"id\":\"%s\"}", docId.String())))
}
const pdfHelloWorld = `%PDF-1.4
api/docInitRunner/runner_test.go
+10 -17
View File
@@ -1,6 +1,7 @@
package docinitrunner_test
import (
"regexp"
"testing"
"time"
@@ -9,29 +10,29 @@ import (
documentinit "queryorchestration/internal/document/init"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/documentsync"
"queryorchestration/internal/test"
queuemock "queryorchestration/mocks/queue"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type DocInitConfig struct {
runner.BaseConfig[docinitrunner.Body]
documentsync.DocSyncConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestDocInitRunner(t *testing.T) {
t.Parallel()
cfg := &DocInitConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.DocumentSyncURL = test.CreateQueue(t, cfg, test.DocSyncRunnerName)
runner := docinitrunner.New(&docinitrunner.Services{
Document: documentinit.New(cfg),
@@ -54,15 +55,7 @@ func TestDocInitRunner(t *testing.T) {
Hash: "hash",
}
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.DocumentSyncURL
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(t.Context(), doc))
test.AssertMessageBody(t, cfg, cfg.GetDocumentSyncURL(), regexp.MustCompile(`{"id":".+"}`))
}
api/docSyncRunner/runner_test.go
+11 -20
View File
@@ -1,8 +1,8 @@
package docsyncrunner_test
import (
"context"
"fmt"
"regexp"
"testing"
"time"
@@ -13,31 +13,30 @@ import (
documentsync "queryorchestration/internal/document/sync"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/documentclean"
"queryorchestration/internal/test"
queuemock "queryorchestration/mocks/queue"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type DocSyncConfig struct {
runner.BaseConfig[docsyncrunner.Body]
documentclean.DocCleanConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestDocSyncRunner(t *testing.T) {
t.Parallel()
cfg := &DocSyncConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
ctx := context.Background()
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.DocumentCleanURL = test.CreateQueue(t, cfg, test.DocCleanRunnerName)
runner := docsyncrunner.New(&docsyncrunner.Services{
Document: documentsync.New(cfg, &documentsync.Services{
@@ -78,19 +77,11 @@ func TestDocSyncRunner(t *testing.T) {
})
require.NoError(t, err)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.DocumentCleanURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId)
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
doc := docsyncrunner.Body{
DocumentID: docId,
}
assert.True(t, runner.Process(ctx, doc))
assert.True(t, runner.Process(t.Context(), doc))
test.AssertMessageBody(t, cfg, cfg.GetDocumentCleanURL(), regexp.MustCompile(fmt.Sprintf("{\"id\":\"%s\"}", docId)))
}
api/docTextRunner/runner_test.go
+8 -16
View File
@@ -3,6 +3,7 @@ package doctextrunner_test
import (
"fmt"
"io"
"regexp"
"strings"
"testing"
"time"
@@ -17,11 +18,9 @@ import (
"queryorchestration/internal/serviceconfig/textract"
"queryorchestration/internal/test"
objectstoremock "queryorchestration/mocks/objectstore"
queuemock "queryorchestration/mocks/queue"
textractmock "queryorchestration/mocks/textract"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/sqs"
awstextract "github.com/aws/aws-sdk-go-v2/service/textract"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
@@ -40,16 +39,16 @@ type DocTextConfig struct {
func TestDocTextRunner(t *testing.T) {
t.Parallel()
cfg := &DocTextConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.QuerySyncURL = test.CreateQueue(t, cfg, test.QuerySyncRunnerName)
mockStore := objectstoremock.NewMockS3Client(t)
cfg.StoreClient = mockStore
mockTextract := textractmock.NewMockTextractClient(t)
cfg.TextractClient = mockTextract
mockQueue := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockQueue
cfg.QuerySyncURL = "ex"
runner := doctextrunner.New(&doctextrunner.Services{
Text: documenttext.New(cfg),
@@ -133,20 +132,13 @@ func TestDocTextRunner(t *testing.T) {
mock.Anything,
).
Return(&s3.PutObjectOutput{}, nil)
mockQueue.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
doc := doctextrunner.Body{
DocumentID: docId,
}
assert.True(t, runner.Process(t.Context(), doc))
test.AssertMessageBody(t, cfg, cfg.GetQuerySyncURL(), regexp.MustCompile(fmt.Sprintf("{\"id\":\"%s\"}", docId.String())))
}
const pdfHelloWorld = `%PDF-1.4
api/queryAPI/client_test.go
+12 -22
View File
@@ -8,8 +8,6 @@ import (
"testing"
queryapi "queryorchestration/api/queryAPI"
"queryorchestration/internal/client"
clientupdate "queryorchestration/internal/client/update"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/serviceconfig/queue/clientsync"
@@ -27,13 +25,11 @@ type ClientConfig struct {
func TestCreateClient(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Client: client.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
body := queryapi.ClientCreate{
Name: "example_name",
@@ -57,13 +53,11 @@ func TestCreateClient(t *testing.T) {
func TestGetClient(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Client: client.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
id := "client_id"
@@ -92,15 +86,11 @@ func TestGetClient(t *testing.T) {
func TestUpdateClient(t *testing.T) {
t.Parallel()
cfg := &ClientConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
ClientUpdate: clientupdate.New(cfg, &clientupdate.Services{
Client: client.New(cfg),
}),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
id := "client_id"
api/queryAPI/collector_test.go
+15 -34
View File
@@ -3,31 +3,28 @@ package queryapi_test
import (
"fmt"
"net/http"
"regexp"
"testing"
queryapi "queryorchestration/api/queryAPI"
"queryorchestration/internal/collector"
collectorset "queryorchestration/internal/collector/set"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/test"
queuemock "queryorchestration/mocks/queue"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
func TestSetCollector(t *testing.T) {
t.Parallel()
cfg := &ClientConfig{}
cfg.ClientSyncURL = "example"
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.ClientSyncURL = test.CreateQueue(t, cfg, test.ClientSyncRunnerName)
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
id := "clientid"
@@ -53,37 +50,21 @@ func TestSetCollector(t *testing.T) {
}
ctx, rec := createContextWithBody(t, body)
cons := queryapi.NewControllers(&queryapi.Services{
CollectorSet: collectorset.New(cfg, &collectorset.Services{
Collector: collector.New(cfg),
}),
})
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.GetClientSyncURL() && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id)
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
err = cons.SetCollectorByClientId(ctx, id)
require.NoError(t, err)
assert.Equal(t, http.StatusOK, rec.Code)
assert.Empty(t, rec.Body.String())
test.AssertMessageBody(t, cfg, cfg.GetClientSyncURL(), regexp.MustCompile(fmt.Sprintf("{\"id\":\"%s\"}", id)))
}
func TestGetCollectorByClientId(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Collector: collector.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
id := "clientid"
api/queryAPI/documents_test.go
+8 -21
View File
@@ -6,9 +6,6 @@ import (
queryapi "queryorchestration/api/queryAPI"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/document"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/serviceconfig/queue/clientsync"
"queryorchestration/internal/test"
"github.com/stretchr/testify/assert"
@@ -17,16 +14,11 @@ import (
func TestListDocumentsByClientId(t *testing.T) {
t.Parallel()
cfg := &struct {
serviceconfig.BaseConfig
clientsync.ClientSyncConfig
}{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Document: document.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_id",
@@ -54,16 +46,11 @@ func TestListDocumentsByClientId(t *testing.T) {
func TestGetDocument(t *testing.T) {
t.Parallel()
cfg := &struct {
serviceconfig.BaseConfig
clientsync.ClientSyncConfig
}{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Document: document.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_id",
api/queryAPI/query_test.go
+81 -71
View File
@@ -3,37 +3,41 @@ package queryapi_test
import (
"fmt"
"net/http"
"regexp"
"testing"
"time"
queryapi "queryorchestration/api/queryAPI"
"queryorchestration/internal/client"
clientupdate "queryorchestration/internal/client/update"
"queryorchestration/internal/collector"
collectorset "queryorchestration/internal/collector/set"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/document"
"queryorchestration/internal/export"
"queryorchestration/internal/query"
"queryorchestration/internal/query/result"
resultprocessor "queryorchestration/internal/query/result/processor"
querytest "queryorchestration/internal/query/test"
queryupdate "queryorchestration/internal/query/update"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/clientsync"
"queryorchestration/internal/serviceconfig/queue/queryversionsync"
"queryorchestration/internal/test"
queuemock "queryorchestration/mocks/queue"
"github.com/jackc/pgx/v5/pgtype"
"github.com/stretchr/testify/require"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
)
func TestCreateQuery(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Query: query.New(cfg),
@@ -53,13 +57,11 @@ func TestCreateQuery(t *testing.T) {
func TestListQueries(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Query: query.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
ctx, rec := createContext(t)
@@ -83,13 +85,11 @@ func TestListQueries(t *testing.T) {
func TestGetQuery(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Query: query.New(cfg),
})
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
ctx, rec := createContext(t)
@@ -109,21 +109,15 @@ func TestGetQuery(t *testing.T) {
func TestUpdateQuery(t *testing.T) {
t.Parallel()
cfg := &struct {
serviceconfig.BaseConfig
queryversionsync.QueryVersionSyncConfig
}{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.QueryVersionSyncURL = "here"
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
QueryUpdate: queryupdate.New(cfg, &queryupdate.Services{
Query: query.New(cfg),
}),
})
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.QueryVersionSyncURL = test.CreateQueue(t, cfg, test.QueryVersionSyncRunnerName)
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
id, err := cfg.GetDBQueries().CreateQuery(t.Context(), repository.QuerytypeJsonExtractor)
require.NoError(t, err)
@@ -136,64 +130,55 @@ func TestUpdateQuery(t *testing.T) {
}
ctx, rec := createContextWithBody(t, body)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.QueryVersionSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id.String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
err = cons.UpdateQuery(ctx, id)
require.NoError(t, err)
assert.Equal(t, http.StatusOK, rec.Code)
assert.Empty(t, rec.Body.String())
test.AssertMessageBody(t, cfg, cfg.GetQueryVersionSyncURL(), regexp.MustCompile(fmt.Sprintf(`{"id":"%s"}`, id)))
}
type QueryUpdateConfig struct {
type ControllerConfig struct {
serviceconfig.BaseConfig
queryversionsync.QueryVersionSyncConfig
clientsync.ClientSyncConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestTestQuery(t *testing.T) {
t.Parallel()
cfg := &QueryUpdateConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
func createControllerServices(cfg *ControllerConfig) *queryapi.Services {
docsvc := document.New(cfg)
col := collector.New(cfg)
que := query.New(cfg)
queup := queryupdate.New(cfg, &queryupdate.Services{
Query: que,
})
cons := queryapi.NewControllers(&queryapi.Services{
Collector: col,
Query: que,
QueryUpdate: queup,
QueryTest: querytest.New(cfg, &querytest.Services{
Document: docsvc,
cli := client.New(cfg)
return &queryapi.Services{
Export: export.New(),
Collector: col,
CollectorSet: collectorset.New(cfg, &collectorset.Services{
Collector: col,
}),
Query: que,
QueryUpdate: queryupdate.New(cfg, &queryupdate.Services{
Query: que,
}),
QueryTest: querytest.New(cfg, &querytest.Services{
Result: result.New(cfg, &result.Services{
Query: que,
}),
Collector: col,
Document: docsvc,
}),
})
Client: cli,
ClientUpdate: clientupdate.New(cfg, &clientupdate.Services{
Client: cli,
}),
Document: docsvc,
}
}
contextQueryId, err := que.Create(t.Context(), &resultprocessor.Create{
Type: resultprocessor.TypeContextFull,
})
require.NoError(t, err)
err = cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_id",
Name: "client_name",
})
require.NoError(t, err)
func createDocumentWithText(t testing.TB, cfg *ControllerConfig, clientId string) (uuid.UUID, uuid.UUID, uuid.UUID) {
docId, err := cfg.GetDBQueries().CreateDocument(t.Context(), &repository.CreateDocumentParams{
Clientid: "client_id",
Clientid: clientId,
Hash: "hash",
})
require.NoError(t, err)
@@ -230,6 +215,31 @@ func TestTestQuery(t *testing.T) {
Version: 1,
})
require.NoError(t, err)
return docId, cleanId, textId
}
func TestTestQuery(t *testing.T) {
t.Parallel()
cfg := &ControllerConfig{}
test.CreateDB(t, cfg)
svc := createControllerServices(cfg)
cons := queryapi.NewControllers(svc)
contextQueryId, err := svc.Query.Create(t.Context(), &resultprocessor.Create{
Type: resultprocessor.TypeContextFull,
})
require.NoError(t, err)
_, err = svc.Client.Create(t.Context(), client.CreateParams{
ID: "client_id",
Name: "client_name",
})
require.NoError(t, err)
docId, _, textId := createDocumentWithText(t, cfg, "client_id")
strVal := `{"mykey": "example_value", "oldkey": "old_value"}`
_, err = cfg.GetDBQueries().AddResult(t.Context(), &repository.AddResultParams{
Queryid: contextQueryId,
@@ -240,7 +250,7 @@ func TestTestQuery(t *testing.T) {
require.NoError(t, err)
c := `{"path": "oldkey"}`
queryId, err := que.Create(t.Context(), &resultprocessor.Create{
queryId, err := svc.Query.Create(t.Context(), &resultprocessor.Create{
Type: resultprocessor.TypeJsonExtractor,
Config: &c,
RequiredQueryIDs: &[]uuid.UUID{
@@ -249,7 +259,7 @@ func TestTestQuery(t *testing.T) {
})
require.NoError(t, err)
c = `{"path": "mykey"}`
err = queup.Update(t.Context(), &resultprocessor.Update{
err = svc.QueryUpdate.Update(t.Context(), &resultprocessor.Update{
ID: queryId,
Config: &c,
})
api/queryAPI/status_test.go
+2 -4
View File
@@ -17,8 +17,7 @@ import (
func TestGetClientStatus(t *testing.T) {
t.Parallel()
cfg := &ClientConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Client: client.New(cfg),
@@ -50,8 +49,7 @@ func TestGetClientStatus(t *testing.T) {
func BenchmarkGetClientStatus(b *testing.B) {
cfg := &ClientConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
cons := queryapi.NewControllers(&queryapi.Services{
Client: client.New(cfg),
api/queryRunner/runner_test.go
+100 -83
View File
@@ -1,46 +1,42 @@
package queryrunner_test
import (
"context"
"fmt"
"regexp"
"testing"
"time"
queryrunner "queryorchestration/api/queryRunner"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/query"
"queryorchestration/internal/query/result"
resultprocessor "queryorchestration/internal/query/result/processor"
resultset "queryorchestration/internal/query/result/set"
resultsync "queryorchestration/internal/query/result/sync"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
queryc "queryorchestration/internal/serviceconfig/queue/query"
queuemock "queryorchestration/mocks/queue"
"queryorchestration/internal/test"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/pashagolub/pgxmock/v3"
"github.com/jackc/pgx/v5/pgtype"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type QueryConfig struct {
runner.BaseConfig[queryrunner.Body]
queryc.QueryConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestQueryRunner(t *testing.T) {
ctx := context.Background()
pool, err := pgxmock.NewPool()
require.NoError(t, err)
cfg := &QueryConfig{}
cfg.DBPool = pool
cfg.DBQueries = repository.New(pool)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.QueryURL = "/i/am/here"
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.QueryURL = test.CreateQueue(t, cfg, test.QueryRunnerName)
que := query.New(cfg)
runner := queryrunner.New(&queryrunner.Services{
@@ -53,74 +49,95 @@ func TestQueryRunner(t *testing.T) {
}),
})
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_one",
Name: "name_one",
})
require.NoError(t, err)
docId, err := cfg.GetDBQueries().CreateDocument(t.Context(), &repository.CreateDocumentParams{
Clientid: "client_one",
Hash: "hash",
})
require.NoError(t, err)
fill := "fill"
cleanId, err := cfg.GetDBQueries().AddDocumentClean(t.Context(), &repository.AddDocumentCleanParams{
Documentid: docId,
Bucket: &fill,
Key: &fill,
Hash: &fill,
Mimetype: repository.NullCleanmimetype{
Valid: true,
Cleanmimetype: repository.CleanmimetypeApplicationPdf,
},
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddDocumentCleanEntry(t.Context(), &repository.AddDocumentCleanEntryParams{
Cleanid: cleanId,
Version: 1,
})
require.NoError(t, err)
textId, err := cfg.GetDBQueries().AddDocumentText(t.Context(), &repository.AddDocumentTextParams{
Cleanid: cleanId,
Bucket: fill,
Key: fill,
Hash: fill,
Createdat: pgtype.Timestamp{
Time: time.Now(),
Valid: true,
},
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddDocumentTextEntry(t.Context(), &repository.AddDocumentTextEntryParams{
Textid: textId,
Version: 1,
})
require.NoError(t, err)
contextQueryId, err := cfg.GetDBQueries().CreateQuery(t.Context(), repository.QuerytypeContextFull)
require.NoError(t, err)
version, err := cfg.GetDBQueries().AddLatestQueryVersion(t.Context(), contextQueryId)
require.NoError(t, err)
err = cfg.GetDBQueries().AddActiveQueryVersion(t.Context(), &repository.AddActiveQueryVersionParams{
Versionid: version,
Queryid: contextQueryId,
})
require.NoError(t, err)
queryId, err := cfg.GetDBQueries().CreateQuery(t.Context(), repository.QuerytypeJsonExtractor)
require.NoError(t, err)
version, err = cfg.GetDBQueries().AddLatestQueryVersion(t.Context(), queryId)
require.NoError(t, err)
err = cfg.GetDBQueries().AddActiveQueryVersion(t.Context(), &repository.AddActiveQueryVersionParams{
Versionid: version,
Queryid: queryId,
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddRequiredQuery(t.Context(), &repository.AddRequiredQueryParams{
Queryid: queryId,
Requiredqueryid: contextQueryId,
Addedversion: version,
})
require.NoError(t, err)
collectorOneVersion, err := cfg.GetDBQueries().AddLatestCollectorVersion(t.Context(), "client_one")
require.NoError(t, err)
err = cfg.GetDBQueries().SetActiveCollectorVersion(t.Context(), &repository.SetActiveCollectorVersionParams{
Versionid: version,
Clientid: "client_one",
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddCollectorQuery(t.Context(), &repository.AddCollectorQueryParams{
Queryid: queryId,
Clientid: "client_one",
Name: "EXAMPLE_ONE",
Addedversion: collectorOneVersion,
})
require.NoError(t, err)
doc := queryrunner.Body{
DocumentID: uuid.New(),
QueryID: uuid.New(),
DocumentID: docId,
QueryID: contextQueryId,
}
qcfg := "{\"path\":\"examplekey\"}"
reqQuery := uuid.New()
query := &resultprocessor.Query{
ID: doc.QueryID,
Version: 2,
RequiredQueryIDs: &[]uuid.UUID{reqQuery},
Config: &qcfg,
}
params := &resultset.Set{
DocumentID: doc.DocumentID,
}
assert.True(t, runner.Process(t.Context(), doc))
textEntryId := uuid.New()
pool.ExpectQuery("name: GetTextEntryByDocId :one").WithArgs(params.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "hash", "cleanId", "extractionVersion"}).
AddRow(textEntryId, params.DocumentID, "buket", "/i/am/here", "example", uuid.New(), int64(543)),
)
pool.ExpectQuery("name: GetQuery :one").WithArgs(query.ID).WillReturnRows(
pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
AddRow(query.ID, repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), *query.RequiredQueryIDs),
)
pool.ExpectQuery("name: GetQueryWithVersion :one").WithArgs(&query.ID, &query.Version).WillReturnRows(
pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}).
AddRow(query.ID, repository.QuerytypeJsonExtractor, query.Version, query.Version, []byte(*query.Config), *query.RequiredQueryIDs),
)
requiredResultId := uuid.New()
strVal := "{\"examplekey\":\"example_value\"}"
pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(&query.ID, &query.Version, &params.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
AddRow(&requiredResultId, (*query.RequiredQueryIDs)[0], repository.QuerytypeContextFull, &strVal),
)
pool.ExpectBegin()
resultId := uuid.New()
pool.ExpectQuery("name: AddResult :one").WithArgs(query.ID, pgxmock.AnyArg(), textEntryId, query.Version).
WillReturnRows(
pgxmock.NewRows([]string{"id"}).
AddRow(resultId),
)
pool.ExpectQuery("name: ListQueryRequirementValues :many").WithArgs(&query.ID, &query.Version, &params.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "type", "value"}).
AddRow(&requiredResultId, (*query.RequiredQueryIDs)[0], repository.QuerytypeContextFull, &strVal),
)
pool.ExpectExec("name: AddResultDependency :exec").WithArgs(resultId, requiredResultId).
WillReturnResult(pgxmock.NewResult("", 1))
pool.ExpectCommit()
pool.ExpectQuery("name: ListQueryDirectDependentsByDocumentID :many").WithArgs(&query.ID, &doc.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"queryId"}).
AddRow(&reqQuery),
)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", params.DocumentID.String(), (*query.RequiredQueryIDs)[0].String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(ctx, doc))
test.AssertMessageBody(t, cfg, cfg.GetQueryURL(), regexp.MustCompile(fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", docId, queryId)))
}
api/querySyncRunner/runner_test.go
+87 -48
View File
@@ -1,81 +1,120 @@
package querysyncrunner_test
import (
"context"
"fmt"
"regexp"
"testing"
"time"
querysyncrunner "queryorchestration/api/querySyncRunner"
"queryorchestration/internal/database/repository"
resultsync "queryorchestration/internal/query/result/sync"
querysync "queryorchestration/internal/query/sync"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
queryc "queryorchestration/internal/serviceconfig/queue/query"
queuemock "queryorchestration/mocks/queue"
"queryorchestration/internal/test"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/pashagolub/pgxmock/v3"
"github.com/jackc/pgx/v5/pgtype"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type QuerySyncConfig struct {
runner.BaseConfig[querysyncrunner.Body]
queryc.QueryConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestQueryRunner(t *testing.T) {
ctx := context.Background()
pool, err := pgxmock.NewPool()
require.NoError(t, err)
func TestQuerySyncRunner(t *testing.T) {
cfg := &QuerySyncConfig{}
cfg.DBPool = pool
cfg.DBQueries = repository.New(pool)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.QueryURL = "/i/am/here"
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
svc := querysync.New(cfg, &querysync.Services{
ResultSync: resultsync.New(cfg),
})
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.QueryURL = test.CreateQueue(t, cfg, test.QueryRunnerName)
runner := querysyncrunner.New(&querysyncrunner.Services{
QuerySync: svc,
QuerySync: querysync.New(cfg, &querysync.Services{
ResultSync: resultsync.New(cfg),
}),
})
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_one",
Name: "name_one",
})
require.NoError(t, err)
docId, err := cfg.GetDBQueries().CreateDocument(t.Context(), &repository.CreateDocumentParams{
Clientid: "client_one",
Hash: "hash",
})
require.NoError(t, err)
fill := "fill"
cleanId, err := cfg.GetDBQueries().AddDocumentClean(t.Context(), &repository.AddDocumentCleanParams{
Documentid: docId,
Bucket: &fill,
Key: &fill,
Hash: &fill,
Mimetype: repository.NullCleanmimetype{
Valid: true,
Cleanmimetype: repository.CleanmimetypeApplicationPdf,
},
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddDocumentCleanEntry(t.Context(), &repository.AddDocumentCleanEntryParams{
Cleanid: cleanId,
Version: 1,
})
require.NoError(t, err)
textId, err := cfg.GetDBQueries().AddDocumentText(t.Context(), &repository.AddDocumentTextParams{
Cleanid: cleanId,
Bucket: fill,
Key: fill,
Hash: fill,
Createdat: pgtype.Timestamp{
Time: time.Now(),
Valid: true,
},
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddDocumentTextEntry(t.Context(), &repository.AddDocumentTextEntryParams{
Textid: textId,
Version: 1,
})
require.NoError(t, err)
queryId, err := cfg.GetDBQueries().CreateQuery(t.Context(), repository.QuerytypeContextFull)
require.NoError(t, err)
version, err := cfg.GetDBQueries().AddLatestQueryVersion(t.Context(), queryId)
require.NoError(t, err)
err = cfg.GetDBQueries().AddActiveQueryVersion(t.Context(), &repository.AddActiveQueryVersionParams{
Versionid: version,
Queryid: queryId,
})
require.NoError(t, err)
collectorOneVersion, err := cfg.GetDBQueries().AddLatestCollectorVersion(t.Context(), "client_one")
require.NoError(t, err)
err = cfg.GetDBQueries().SetActiveCollectorVersion(t.Context(), &repository.SetActiveCollectorVersionParams{
Versionid: version,
Clientid: "client_one",
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddCollectorQuery(t.Context(), &repository.AddCollectorQueryParams{
Queryid: queryId,
Clientid: "client_one",
Name: "EXAMPLE_ONE",
Addedversion: collectorOneVersion,
})
require.NoError(t, err)
doc := querysyncrunner.Body{
DocumentID: uuid.New(),
DocumentID: docId,
}
reqId := uuid.New()
qs := []uuid.UUID{
reqId,
}
assert.True(t, runner.Process(t.Context(), doc))
pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(doc.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"isextracted"}).
AddRow(true),
)
pool.ExpectQuery("name: ListUnsyncedNoDepsQueriesByDocId :many").WithArgs(&doc.DocumentID).
WillReturnRows(
pgxmock.NewRows([]string{"id"}).
AddRow(&reqId),
)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.QueryURL && *in.MessageBody == fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", doc.DocumentID.String(), qs[0].String())
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(ctx, doc))
test.AssertMessageBody(t, cfg, cfg.GetQueryURL(), regexp.MustCompile(fmt.Sprintf("{\"document_id\":\"%s\",\"query_id\":\"%s\"}", docId, queryId)))
}
api/queryVersionSyncRunner/runner_test.go
+66 -52
View File
@@ -1,42 +1,36 @@
package queryversionsyncrunner_test
import (
"context"
"fmt"
"regexp"
"testing"
queryversionsyncrunner "queryorchestration/api/queryVersionSyncRunner"
"queryorchestration/internal/database/repository"
queryversionsync "queryorchestration/internal/query/versionsync"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/clientsync"
queuemock "queryorchestration/mocks/queue"
"queryorchestration/internal/test"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/pashagolub/pgxmock/v3"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type QuerySyncConfig struct {
runner.BaseConfig[queryversionsyncrunner.Body]
clientsync.ClientSyncConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestQueryRunner(t *testing.T) {
ctx := context.Background()
pool, err := pgxmock.NewPool()
require.NoError(t, err)
func TestQueryVersionSyncRunner(t *testing.T) {
cfg := &QuerySyncConfig{}
cfg.DBPool = pool
cfg.DBQueries = repository.New(pool)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.ClientSyncURL = "/i/am/here"
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.ClientSyncURL = test.CreateQueue(t, cfg, test.ClientSyncRunnerName)
svc := queryversionsync.New(cfg)
@@ -44,42 +38,62 @@ func TestQueryRunner(t *testing.T) {
Sync: svc,
})
queryId, err := cfg.GetDBQueries().CreateQuery(t.Context(), repository.QuerytypeContextFull)
require.NoError(t, err)
version, err := cfg.GetDBQueries().AddLatestQueryVersion(t.Context(), queryId)
require.NoError(t, err)
err = cfg.GetDBQueries().AddActiveQueryVersion(t.Context(), &repository.AddActiveQueryVersionParams{
Versionid: version,
Queryid: queryId,
})
require.NoError(t, err)
err = cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_one",
Name: "name_one",
})
require.NoError(t, err)
collectorOneVersion, err := cfg.GetDBQueries().AddLatestCollectorVersion(t.Context(), "client_one")
require.NoError(t, err)
err = cfg.GetDBQueries().SetActiveCollectorVersion(t.Context(), &repository.SetActiveCollectorVersionParams{
Versionid: version,
Clientid: "client_one",
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddCollectorQuery(t.Context(), &repository.AddCollectorQueryParams{
Queryid: queryId,
Clientid: "client_one",
Name: "EXAMPLE_ONE",
Addedversion: collectorOneVersion,
})
require.NoError(t, err)
err = cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_two",
Name: "name_two",
})
require.NoError(t, err)
collectorTwoVersion, err := cfg.GetDBQueries().AddLatestCollectorVersion(t.Context(), "client_two")
require.NoError(t, err)
err = cfg.GetDBQueries().SetActiveCollectorVersion(t.Context(), &repository.SetActiveCollectorVersionParams{
Versionid: version,
Clientid: "client_two",
})
require.NoError(t, err)
err = cfg.GetDBQueries().AddCollectorQuery(t.Context(), &repository.AddCollectorQueryParams{
Queryid: queryId,
Clientid: "client_two",
Name: "EXAMPLE_TWO",
Addedversion: collectorTwoVersion,
})
require.NoError(t, err)
doc := queryversionsyncrunner.Body{
QueryID: uuid.New(),
QueryID: queryId,
}
clientOne := "hello"
clientTwo := "bye"
clientIds := []string{
clientOne,
clientTwo,
}
assert.True(t, runner.Process(t.Context(), doc))
pool.ExpectQuery("name: ListQueryClientIDs :many").WithArgs(&doc.QueryID).
WillReturnRows(
pgxmock.NewRows([]string{"clientId"}).
AddRow(clientOne).
AddRow(clientTwo),
)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", clientIds[0])
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.ClientSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", clientIds[1])
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(ctx, doc))
test.AssertMessageBodies(t, cfg, cfg.GetClientSyncURL(), []*regexp.Regexp{
regexp.MustCompile(`^{"id":"client_one"}$`),
regexp.MustCompile(`^{"id":"client_two"}$`),
})
}
api/storeEventRunner/runner_test.go
+20 -50
View File
@@ -1,43 +1,37 @@
package storeeventrunner
import (
"context"
"regexp"
"testing"
"time"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/document"
documentstore "queryorchestration/internal/document/store"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue"
"queryorchestration/internal/serviceconfig/queue/documentinit"
queuemock "queryorchestration/mocks/queue"
"queryorchestration/internal/test"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/google/uuid"
"github.com/pashagolub/pgxmock/v3"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
type DocInitConfig struct {
runner.BaseConfig[S3EventNotification]
documentinit.DocInitConfig
queue.QueueConfig
objectstore.ObjectStoreConfig
}
func TestDocInitRunner(t *testing.T) {
ctx := context.Background()
pool, err := pgxmock.NewPool()
require.NoError(t, err)
func TestStoreEventRunner(t *testing.T) {
cfg := &DocInitConfig{}
cfg.DBPool = pool
cfg.DBQueries = repository.New(pool)
mockSQS := queuemock.NewMockSQSClient(t)
cfg.QueueClient = mockSQS
cfg.DocInitURL = "hi"
test.CreateDB(t, cfg)
acfg := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
cfg.DocInitURL = test.CreateQueue(t, cfg, test.DocInitRunnerName)
runner := New(&Services{
documentstore.New(cfg),
@@ -49,11 +43,7 @@ func TestDocInitRunner(t *testing.T) {
Location: objectstore.Import,
ClientID: clientId,
CreatedAt: time.Now().UTC(),
}
docinfo := document.DocumentSummary{
ID: uuid.New(),
ClientID: clientId,
Hash: "example_hash",
EntityID: uuid.New(),
}
doc := S3EventNotification{
Records: []S3EventRecord{
@@ -65,40 +55,20 @@ func TestDocInitRunner(t *testing.T) {
},
Object: S3Object{
Key: location.String(),
ETag: docinfo.Hash,
ETag: "example_hash",
},
},
},
},
}
pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, clientId).WillReturnRows(
pgxmock.NewRows([]string{"id"}),
)
pool.ExpectBegin()
pool.ExpectQuery("name: CreateDocument :one").WithArgs(clientId, docinfo.Hash).
WillReturnRows(
pgxmock.NewRows([]string{"id"}).
AddRow(docinfo.ID),
)
pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(docinfo.ID, bucketName, location.String()).
WillReturnResult(pgxmock.NewResult("", 1))
pool.ExpectCommit()
pool.ExpectQuery("name: GetDocumentSummary :one").WithArgs(docinfo.ID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "clientId", "hash"}).
AddRow(docinfo.ID, docinfo.ClientID, "example"),
)
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: clientId,
Name: "name",
})
require.NoError(t, err)
mockSQS.EXPECT().
SendMessage(
mock.Anything,
mock.MatchedBy(func(in *sqs.SendMessageInput) bool {
return *in.QueueUrl == cfg.DocInitURL
}),
mock.Anything,
).
Return(&sqs.SendMessageOutput{}, nil)
assert.True(t, runner.Process(t.Context(), doc))
assert.True(t, runner.Process(ctx, doc))
test.AssertMessageBody(t, cfg, cfg.GetDocInitURL(), regexp.MustCompile(`^\{"bucket":"bucketName","key":"hi\/import\/.+\/0\/.+","hash":"example_hash"\}$`))
}
internal/database/repository/clean_test.go
+1 -2
View File
@@ -20,8 +20,7 @@ func TestClean(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/client_test.go
+1 -2
View File
@@ -20,8 +20,7 @@ func TestClientQueries(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/collector_test.go
+3 -6
View File
@@ -22,8 +22,7 @@ func TestCollector(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -225,8 +224,7 @@ func BenchmarkListCollectorQueries(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
@@ -248,8 +246,7 @@ func BenchmarkGetCollectorByClientID(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/document_test.go
+9 -18
View File
@@ -21,8 +21,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -42,8 +41,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -73,8 +71,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -109,8 +106,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -156,8 +152,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -189,8 +184,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -223,8 +217,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -255,8 +248,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -297,8 +289,7 @@ func TestDocument(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/query_test.go
+5 -10
View File
@@ -21,8 +21,7 @@ func TestQueries(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -231,8 +230,7 @@ func TestQueryDependencyTree(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -450,8 +448,7 @@ func TestQueriesList(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -506,8 +503,7 @@ func TestListQueryClients(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -595,8 +591,7 @@ func BenchmarkListQueryDirectDependentsByDocId(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/result_test.go
+6 -12
View File
@@ -23,8 +23,7 @@ func TestResults(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -51,8 +50,7 @@ func TestResultValues(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -266,8 +264,7 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
ctx := t.Context()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -382,8 +379,7 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -479,8 +475,7 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -507,8 +502,7 @@ func BenchmarkListUnsynced(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/sync_test.go
+24 -48
View File
@@ -23,8 +23,7 @@ func TestListClientDocumentIDs(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -201,8 +200,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -223,8 +221,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -253,8 +250,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -289,8 +285,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -326,8 +321,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -369,8 +363,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -410,8 +403,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -441,8 +433,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -472,8 +463,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -517,8 +507,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -578,8 +567,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -617,8 +605,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -664,8 +651,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -706,8 +692,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -774,8 +759,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -800,8 +784,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -832,8 +815,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -864,8 +846,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -903,8 +884,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -934,8 +914,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -964,8 +943,7 @@ func TestClientSync(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -1127,8 +1105,7 @@ func BenchmarkIsClientSynced(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
@@ -1147,8 +1124,7 @@ func BenchmarkGetDocumentExternal(b *testing.B) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(b)
test.CreateDB(b, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(b, cfg)
queries := cfg.GetDBQueries()
internal/database/repository/text_test.go
+3 -6
View File
@@ -23,8 +23,7 @@ func TestTextExtraction(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -131,8 +130,7 @@ func TestTextTextractPart(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
@@ -261,8 +259,7 @@ func TestTextOutPart(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
queries := cfg.GetDBQueries()
internal/query/get_test.go
+1 -2
View File
@@ -56,8 +56,7 @@ func TestGet(t *testing.T) {
func TestGetWithVersion(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_id",
internal/query/result/get_test.go
+3 -6
View File
@@ -34,8 +34,7 @@ func TestGetValueWithVersion(t *testing.T) {
t.Run("no entry", func(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
svc := New(cfg, &Services{})
@@ -52,8 +51,7 @@ func TestGetValueWithVersion(t *testing.T) {
t.Run("existing entry", func(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
svc := New(cfg, &Services{})
@@ -148,8 +146,7 @@ func TestGetValueWithVersion(t *testing.T) {
t.Run("not existing version", func(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
svc := New(cfg, &Services{})
internal/query/result/process_test.go
+1 -2
View File
@@ -22,8 +22,7 @@ import (
func TestProcess(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
Clientid: "client_id",
internal/server/api/listener_test.go
+1 -2
View File
@@ -28,8 +28,7 @@ func TestNewAPI(t *testing.T) {
cfg := &BaseConfig{}
_ = serviceconfig.InitializeConfig(cfg)
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{
test.CreateDBWithParams(t, cfg, &test.CreateDatabaseConfig{
NoMigrations: true,
})
internal/server/runner/listener_test.go
+3 -4
View File
@@ -28,13 +28,12 @@ func TestNewRunner(t *testing.T) {
}
cfg := &TestConfig{}
_ = serviceconfig.InitializeConfig(cfg)
net := test.GetNetwork(t)
var wg sync.WaitGroup
wg.Add(1)
go func() {
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{
test.CreateDBWithParams(t, cfg, &test.CreateDatabaseConfig{
NoMigrations: true,
})
@@ -44,10 +43,10 @@ func TestNewRunner(t *testing.T) {
wg.Done()
}()
a := test.CreateAWSContainer(t, cfg, net)
a := test.CreateAWSContainer(t, cfg)
test.SetQueueClient(t, t.Context(), cfg, a.ExternalEndpoint)
cfg.QueueURL = test.CreateQueue(t, t.Context(), cfg, "queueName")
cfg.QueueURL = test.CreateQueue(t, cfg, "queueName")
cfg.SetSQSEndpoint(a.ExternalEndpoint)
wg.Wait()
internal/server/server_test.go
+1 -2
View File
@@ -22,9 +22,8 @@ func TestNewServer(t *testing.T) {
cfg := &server.BaseConfig{}
_ = serviceconfig.InitializeConfig(cfg)
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{
test.CreateDBWithParams(t, cfg, &test.CreateDatabaseConfig{
NoMigrations: true,
})
internal/serviceconfig/database/pool_test.go
+1 -2
View File
@@ -21,8 +21,7 @@ func TestSetDBPool(t *testing.T) {
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
err := cfg.SetDBPool(ctx)
require.NoError(t, err)
internal/serviceconfig/database/transaction_test.go
+2 -4
View File
@@ -17,8 +17,7 @@ func TestExecuteTransaction(t *testing.T) {
t.Run("create client", func(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
err := cfg.ExecuteDBTransaction(t.Context(), func(ctx context.Context, q *repository.Queries) error {
err := q.CreateClient(ctx, &repository.CreateClientParams{
@@ -39,8 +38,7 @@ func TestExecuteTransaction(t *testing.T) {
t.Run("error in transaction", func(t *testing.T) {
t.Parallel()
cfg := &serviceconfig.BaseConfig{}
net := test.GetNetwork(t)
test.CreateDB(t, cfg, net, &test.CreateDatabaseConfig{})
test.CreateDB(t, cfg)
err := cfg.ExecuteDBTransaction(t.Context(), func(ctx context.Context, q *repository.Queries) error {
err := q.CreateClient(ctx, &repository.CreateClientParams{
internal/serviceconfig/objectstore/config_test.go
+1 -2
View File
@@ -95,8 +95,7 @@ func TestCalculateAndCompareETag(t *testing.T) {
cfg := &StoreConfig{}
net := test.GetNetwork(t)
acfg := test.CreateAWSContainer(t, cfg, net)
acfg := test.CreateAWSContainer(t, cfg)
test.SetStoreClient(t, ctx, cfg, acfg.ExternalEndpoint)
test.CreateBucket(t, ctx, cfg)
internal/serviceconfig/queue/receive.go
+15 -4
View File
@@ -7,16 +7,27 @@ import (
)
type ReceiveParams struct {
QueueURL string
Attributes []string
QueueURL string
Attributes []string
VisibilityTimeout int32
WaitTimeSeconds int32
}
func (c *QueueConfig) ReceiveFromQueue(ctx context.Context, params *ReceiveParams) (*sqs.ReceiveMessageOutput, error) {
return c.QueueClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
p := &sqs.ReceiveMessageInput{
QueueUrl: &params.QueueURL,
MaxNumberOfMessages: 1,
WaitTimeSeconds: 10,
VisibilityTimeout: 10,
MessageAttributeNames: params.Attributes,
})
}
if params.VisibilityTimeout > 0 {
p.VisibilityTimeout = params.VisibilityTimeout
}
if params.WaitTimeSeconds > 0 {
p.WaitTimeSeconds = params.WaitTimeSeconds
}
return c.QueueClient.ReceiveMessage(ctx, p)
}
internal/test/aws.go
+3 -1
View File
@@ -30,10 +30,12 @@ const (
awsPort = 4566
)
func CreateAWSContainer(t testing.TB, cfg AWSConfigProvider, network string) *AWSContainerConfig {
func CreateAWSContainer(t testing.TB, cfg AWSConfigProvider) *AWSContainerConfig {
port, err := nat.NewPort("tcp", strconv.Itoa(awsPort))
require.NoError(t, err)
network := GetNetwork(t)
cfg.SetAWSKeyID("test")
cfg.SetAWSSecretKey("test")
cfg.SetAWSSessionToken("")
internal/test/database.go
+7 -1
View File
@@ -23,7 +23,11 @@ const (
dbPort = 5432
)
func CreateDB(t testing.TB, cfg serviceconfig.ConfigProvider, network string, dcfg *CreateDatabaseConfig) {
func CreateDB(t testing.TB, cfg serviceconfig.ConfigProvider) {
CreateDBWithParams(t, cfg, &CreateDatabaseConfig{})
}
func CreateDBWithParams(t testing.TB, cfg serviceconfig.ConfigProvider, dcfg *CreateDatabaseConfig) {
port, err := nat.NewPort("tcp", strconv.Itoa(dbPort))
require.NoError(t, err)
@@ -32,6 +36,8 @@ func CreateDB(t testing.TB, cfg serviceconfig.ConfigProvider, network string, dc
cfg.SetDBName(GetAlias(t, "postgres"))
cfg.SetDBNoSSL(true)
network := GetNetwork(t)
req := testcontainers.ContainerRequest{
Image: "postgres:17.2-alpine3.21",
Name: "postgres_test_queryorchestration",
internal/test/ecosystem.go
+4 -4
View File
@@ -136,7 +136,7 @@ func CreateFullDependenciesWithParams(t testing.TB, cfg FullDependenciesConfig,
wg.Add(1)
go func() {
deps.AWSConfig = CreateAWSContainer(t, cfg, network)
deps.AWSConfig = CreateAWSContainer(t, cfg)
SetQueueClient(t, t.Context(), cfg, deps.AWSConfig.ExternalEndpoint)
@@ -154,7 +154,7 @@ func CreateFullDependenciesWithParams(t testing.TB, cfg FullDependenciesConfig,
for _, runner := range *runs {
wg.Add(1)
go func() {
queue := CreateQueue(t, t.Context(), cfg, runner)
queue := CreateQueue(t, cfg, runner)
lock.Lock()
deps.QueueURLs[runner] = queue
@@ -173,10 +173,10 @@ func CreateFullDependenciesWithParams(t testing.TB, cfg FullDependenciesConfig,
wg.Done()
}()
mockServer := CreateMockServer(t, network)
mockServer := CreateMockServer(t)
deps.MockServer = mockServer
CreateDB(t, cfg, network, &CreateDatabaseConfig{
CreateDBWithParams(t, cfg, &CreateDatabaseConfig{
NoMigrations: true,
})
internal/test/mockserver.go
+3 -1
View File
@@ -57,10 +57,12 @@ const (
mockServerPort = 1080
)
func CreateMockServer(t testing.TB, network string) *MockServer {
func CreateMockServer(t testing.TB) *MockServer {
port, err := nat.NewPort("tcp", strconv.Itoa(mockServerPort))
require.NoError(t, err)
network := GetNetwork(t)
req := testcontainers.ContainerRequest{
Image: "mockserver/mockserver:latest",
Name: "mockserver_test_queryorchestration",
internal/test/mockserver_test.go
+1 -2
View File
@@ -55,9 +55,8 @@ func TestWaitForMockEndpoint(t *testing.T) {
if testing.Short() {
t.SkipNow()
}
net := GetNetwork(t)
server := CreateMockServer(t, net)
server := CreateMockServer(t)
body := strings.NewReader(`{"team":"hello"}`)
req, err := http.NewRequest("GET", string(server.External), body)
internal/test/objectstore_test.go
+1 -2
View File
@@ -29,8 +29,7 @@ func TestCreateBucket(t *testing.T) {
cfg := &StoreConfig{}
net := GetNetwork(t)
acfg := CreateAWSContainer(t, cfg, net)
acfg := CreateAWSContainer(t, cfg)
SetStoreClient(t, ctx, cfg, acfg.ExternalEndpoint)
CreateBucket(t, ctx, cfg)
internal/test/queue.go
+64 -11
View File
@@ -3,6 +3,7 @@ package test
import (
"context"
"fmt"
"log"
"log/slog"
"regexp"
"testing"
@@ -50,22 +51,22 @@ func GetQueueName(t testing.TB, name RunnerName) string {
return GetAlias(t, string(name))
}
func CreateQueue(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, name RunnerName) string {
func CreateQueue(t testing.TB, cfg serviceconfig.ConfigProvider, name RunnerName) string {
t.Helper()
queueName := GetQueueName(t, name)
queueM, err := cfg.GetQueueClient().CreateQueue(ctx, &sqs.CreateQueueInput{
queueM, err := cfg.GetQueueClient().CreateQueue(t.Context(), &sqs.CreateQueueInput{
QueueName: aws.String(queueName),
})
require.NoError(t, err)
slog.Info("create queue", "name", queueName, "url", *queueM.QueueUrl)
err = cfg.PingQueueByURL(ctx, *queueM.QueueUrl)
err = cfg.PingQueueByURL(t.Context(), *queueM.QueueUrl)
require.NoError(t, err)
return *queueM.QueueUrl
}
func AssertMessage(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
func AssertMessage(t testing.TB, cfg serviceconfig.ConfigProvider, params *queue.ReceiveParams) types.Message {
t.Helper()
timeout := time.After(30 * time.Second)
tick := time.NewTicker(500 * time.Millisecond)
@@ -78,7 +79,7 @@ func AssertMessage(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigPr
case <-tick.C:
slog.Info("receiving from queue")
result, err := cfg.ReceiveFromQueue(ctx, params)
result, err := cfg.ReceiveFromQueue(t.Context(), params)
if err != nil {
continue
} else if len(result.Messages) < 1 {
@@ -86,24 +87,24 @@ func AssertMessage(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigPr
}
return result.Messages[0]
case <-ctx.Done():
t.Fatal(ctx.Err())
case <-t.Context().Done():
t.Fatal(t.Context().Err())
}
}
}
func AssertMessageBody(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, body *regexp.Regexp) {
func AssertMessageBody(t testing.TB, cfg serviceconfig.ConfigProvider, url string, body *regexp.Regexp) {
t.Helper()
message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
message := AssertMessage(t, cfg, &queue.ReceiveParams{
QueueURL: url,
})
assert.Regexp(t, body, *message.Body)
}
func AssertMessageAttr(t testing.TB, ctx context.Context, cfg serviceconfig.ConfigProvider, url string, name string, value *regexp.Regexp) {
func AssertMessageAttr(t testing.TB, cfg serviceconfig.ConfigProvider, url string, name string, value *regexp.Regexp) {
t.Helper()
message := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
message := AssertMessage(t, cfg, &queue.ReceiveParams{
QueueURL: url,
Attributes: []string{name},
})
@@ -111,3 +112,55 @@ func AssertMessageAttr(t testing.TB, ctx context.Context, cfg serviceconfig.Conf
assert.NotNil(t, message.MessageAttributes[name])
assert.Regexp(t, value, *(message.MessageAttributes[name]).StringValue)
}
func AssertMessageBodies(t testing.TB, cfg serviceconfig.ConfigProvider, url string, bodies []*regexp.Regexp) {
t.Helper()
timeout := time.After(30 * time.Second)
tick := time.NewTicker(2 * time.Second)
defer tick.Stop()
for {
select {
case <-timeout:
t.Fatal("assert timeout")
case <-tick.C:
slog.Info("receiving from queue")
result, err := cfg.ReceiveFromQueue(t.Context(), &queue.ReceiveParams{
QueueURL: url,
VisibilityTimeout: 1,
WaitTimeSeconds: 1,
})
if err != nil {
continue
} else if len(result.Messages) < 1 {
continue
}
found := true
for _, m := range result.Messages {
foundBody := false
for _, b := range bodies {
log.Print(*m.Body)
if b.MatchString(*m.Body) {
foundBody = true
break
}
}
if !foundBody {
found = false
break
}
}
if !found {
continue
}
return
case <-t.Context().Done():
t.Fatal(t.Context().Err())
}
}
}
internal/test/queue_test.go
+11 -18
View File
@@ -31,11 +31,10 @@ func TestCreateQueue(t *testing.T) {
cfg := &TestConfig{}
net := GetNetwork(t)
a := CreateAWSContainer(t, cfg, net)
a := CreateAWSContainer(t, cfg)
SetQueueClient(t, ctx, cfg, a.ExternalEndpoint)
url := CreateQueue(t, ctx, cfg, "myname")
url := CreateQueue(t, cfg, "myname")
assert.Equal(t, "http://localstack:4566/queue/us-east-1/000000000000/mynametestcreatequeue", url)
}
@@ -48,11 +47,9 @@ func TestAssertMessageWait(t *testing.T) {
cfg := &TestConfig{}
net := GetNetwork(t)
a := CreateAWSContainer(t, cfg, net)
a := CreateAWSContainer(t, cfg)
SetQueueClient(t, ctx, cfg, a.ExternalEndpoint)
url := CreateQueue(t, ctx, cfg, "myname")
url := CreateQueue(t, cfg, "myname")
err := cfg.SendToQueue(ctx, &queue.SendParams{
QueueURL: url,
@@ -60,7 +57,7 @@ func TestAssertMessageWait(t *testing.T) {
})
require.NoError(t, err)
msg := AssertMessage(t, ctx, cfg, &queue.ReceiveParams{
msg := AssertMessage(t, cfg, &queue.ReceiveParams{
QueueURL: url,
})
assert.NotNil(t, msg)
@@ -75,11 +72,9 @@ func TestAssertMessageBodyWait(t *testing.T) {
cfg := &TestConfig{}
net := GetNetwork(t)
a := CreateAWSContainer(t, cfg, net)
a := CreateAWSContainer(t, cfg)
SetQueueClient(t, ctx, cfg, a.ExternalEndpoint)
url := CreateQueue(t, ctx, cfg, "myname")
url := CreateQueue(t, cfg, "myname")
err := cfg.SendToQueue(ctx, &queue.SendParams{
QueueURL: url,
@@ -87,7 +82,7 @@ func TestAssertMessageBodyWait(t *testing.T) {
})
require.NoError(t, err)
AssertMessageBody(t, ctx, cfg, url, regexp.MustCompile("\"body\""))
AssertMessageBody(t, cfg, url, regexp.MustCompile("\"body\""))
}
func TestAssertMessageAttrWait(t *testing.T) {
@@ -99,11 +94,9 @@ func TestAssertMessageAttrWait(t *testing.T) {
cfg := &TestConfig{}
net := GetNetwork(t)
a := CreateAWSContainer(t, cfg, net)
a := CreateAWSContainer(t, cfg)
SetQueueClient(t, ctx, cfg, a.ExternalEndpoint)
url := CreateQueue(t, ctx, cfg, "myname")
url := CreateQueue(t, cfg, "myname")
name := "name"
value := "value"
@@ -120,7 +113,7 @@ func TestAssertMessageAttrWait(t *testing.T) {
})
require.NoError(t, err)
AssertMessageAttr(t, ctx, cfg, url, name, regexp.MustCompile(value))
AssertMessageAttr(t, cfg, url, name, regexp.MustCompile(value))
}
func TestGetQueueURL(t *testing.T) {
test/queryAPI/collectorservice_test.go
+1 -1
View File
@@ -98,5 +98,5 @@ func TestCollectorService(t *testing.T) {
assert.Len(t, collRes.JSON200.Fields, 1)
assert.ElementsMatch(t, fields, collRes.JSON200.Fields)
test.AssertMessageBody(t, ctx, cfg, c.Dependencies.QueueURLs[test.ClientSyncRunnerName], regexp.MustCompile(`{"id":".+"}`))
test.AssertMessageBody(t, cfg, c.Dependencies.QueueURLs[test.ClientSyncRunnerName], regexp.MustCompile(`{"id":".+"}`))
}
test/queryAPI/queryservice_test.go
+1 -1
View File
@@ -88,7 +88,7 @@ func TestQueryAPI(t *testing.T) {
require.NoError(t, err)
assert.NotNil(t, res)
test.AssertMessageBody(t, ctx, cfg, c.Dependencies.QueueURLs[test.QueryVersionSyncRunnerName], regexp.MustCompile(`{"id":".+"}`))
test.AssertMessageBody(t, cfg, c.Dependencies.QueueURLs[test.QueryVersionSyncRunnerName], regexp.MustCompile(`{"id":".+"}`))
queryRes, err := client.GetQueryWithResponse(ctx, jsonID)
require.NoError(t, err)