package endtoend_test import ( "context" "fmt" "os" "path" "strings" "testing" "time" "queryorchestration/internal/serviceconfig" "queryorchestration/internal/serviceconfig/objectstore" documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" "queryorchestration/internal/test" queryservice "queryorchestration/pkg/queryService" "github.com/aws/aws-sdk-go-v2/service/s3" awstypes "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/oapi-codegen/runtime/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) type ProcessConfig struct { serviceconfig.BaseConfig documentcleanc.DocCleanConfig objectstore.ObjectStoreConfig } func TestProcess(t *testing.T) { ctx := context.Background() cfg := &ProcessConfig{} test.SetCfgProvider(t, cfg) cfg.SetBasePath(path.Join(os.Getenv("PWD"), "..")) network, ncleanup := test.CreateNetwork(t, ctx) defer ncleanup() acfg, clean := test.CreateAWSContainer(t, ctx, &test.CreateAWSConfig{ Cfg: cfg, Network: network, }) defer clean() err := cfg.SetQueueClient(ctx) require.NoError(t, err) test.CreateStoreClient(t, ctx, cfg, acfg.ExternalEndpoint) dociniturl := test.CreateQueue(t, ctx, cfg, test.DocInitRunner) docsyncurl := test.CreateQueue(t, ctx, cfg, test.DocSyncRunner) doccleanurl := test.CreateQueue(t, ctx, cfg, test.DocCleanRunner) doctexturl := test.CreateQueue(t, ctx, cfg, test.DocTextRunner) querysyncurl := test.CreateQueue(t, ctx, cfg, test.QuerySyncRunner) queryurl := test.CreateQueue(t, ctx, cfg, test.QueryRunner) clientsyncurl := test.CreateQueue(t, ctx, cfg, test.ClientSyncRunner) queryversionsyncurl := test.CreateQueue(t, ctx, cfg, test.QueryVersionSyncRunner) bucketName := "docinitbucket" test.CreateBucket(t, ctx, cfg, bucketName) arn := fmt.Sprintf("arn:aws:sqs:%s:000000000000:%s", cfg.AWSRegion, test.DocInitRunner) _, err = cfg.StoreClient.PutBucketNotificationConfiguration(ctx, &s3.PutBucketNotificationConfigurationInput{ Bucket: &bucketName, NotificationConfiguration: &awstypes.NotificationConfiguration{ QueueConfigurations: []awstypes.QueueConfiguration{ { QueueArn: &arn, Events: []awstypes.Event{ awstypes.EventS3ObjectCreated, }, }, }, }, }) require.NoError(t, err) net, cleanup := test.CreateRunnersAndServicesNetwork(t, ctx, &test.EcosystemNetworkConfig{ Cfg: cfg, Network: network, AWSContainer: acfg, Runners: []*test.RunnerNetworkConfig{ { Name: test.DocInitRunner, QueueURL: &dociniturl, Env: map[string]string{ "DOCUMENT_SYNC_URL": docsyncurl, }, }, { Name: test.DocSyncRunner, QueueURL: &docsyncurl, Env: map[string]string{ "DOCUMENT_CLEAN_URL": doccleanurl, }, }, { Name: test.DocCleanRunner, QueueURL: &doccleanurl, Env: map[string]string{ "DOCUMENT_TEXT_URL": doctexturl, }, }, { Name: test.DocTextRunner, QueueURL: &doctexturl, Env: map[string]string{ "QUERY_SYNC_URL": querysyncurl, }, }, { Name: test.QuerySyncRunner, QueueURL: &querysyncurl, Env: map[string]string{ "QUERY_URL": queryurl, }, }, { Name: test.QueryRunner, QueueURL: &queryurl, Env: map[string]string{ "QUERY_URL": queryurl, }, }, { Name: test.ClientSyncRunner, QueueURL: &clientsyncurl, Env: map[string]string{ "DOCUMENT_SYNC_URL": docsyncurl, }, }, { Name: test.QueryVersionSyncRunner, QueueURL: &queryversionsyncurl, Env: map[string]string{ "CLIENT_SYNC_URL": clientsyncurl, }, }, }, Services: []*test.ServiceNetworkConfig{ { Name: test.QueryService, Env: map[string]string{ "CLIENT_SYNC_URL": clientsyncurl, "QUERY_VERSION_SYNC_URL": queryversionsyncurl, }, }, }, }) defer cleanup() qService, err := queryservice.NewClientWithResponses(net.Services[test.QueryService].URI) require.NoError(t, err) clientRes, err := qService.CreateClientWithResponse(ctx, queryservice.ClientCreate{ Name: "example_name", }) assert.NoError(t, err) canSync := true _, err = qService.UpdateClientWithResponse(ctx, clientRes.JSON201.Id, queryservice.ClientUpdate{ CanSync: &canSync, }) assert.NoError(t, err) contextQueryRes, err := qService.CreateQueryWithResponse(ctx, queryservice.QueryCreate{ Type: queryservice.CONTEXTFULL, }) assert.NoError(t, err) jcfg := `{"path":"keyone"}` jsonQueryRes, err := qService.CreateQueryWithResponse(ctx, queryservice.QueryCreate{ Type: queryservice.JSONEXTRACTOR, Config: &jcfg, RequiredQueries: &[]types.UUID{contextQueryRes.JSON201.Id}, }) assert.NoError(t, err) newActiveVersion := int32(2) _, err = qService.UpdateCollectorByClientIdWithResponse(ctx, clientRes.JSON201.Id, queryservice.CollectorUpdate{ ActiveVersion: &newActiveVersion, Fields: &[]queryservice.CollectorField{ { Name: "JSON_QUERY", QueryId: jsonQueryRes.JSON201.Id, }, }, }) assert.NoError(t, err) WaitForClientStatus(t, ctx, qService, clientRes.JSON201.Id, queryservice.INSYNC) location := fmt.Sprintf("%s/%s", clientRes.JSON201.Id, "object_name") body := strings.NewReader(pdfHelloWorld) _, err = cfg.StoreClient.PutObject(ctx, &s3.PutObjectInput{ Bucket: &bucketName, Key: &location, Body: body, }) assert.NoError(t, err) WaitForClientStatus(t, ctx, qService, clientRes.JSON201.Id, queryservice.NOTSYNCED) WaitForClientStatus(t, ctx, qService, clientRes.JSON201.Id, queryservice.INSYNC) docs, err := qService.ListDocumentsByClientIdWithResponse(ctx, clientRes.JSON201.Id) assert.NoError(t, err) assert.Len(t, *docs.JSON200, 1) doc := (*docs.JSON200)[0] assert.Equal(t, bucketName, doc.Bucket) assert.Equal(t, location, doc.Key) testRes, err := qService.TestQueryWithResponse(ctx, jsonQueryRes.JSON201.Id, queryservice.QueryTestRequest{ QueryVersion: 1, DocumentId: doc.Id, }) assert.NoError(t, err) assert.Equal(t, "valueone", testRes.JSON200.Value) aV := int32(2) jcfg = `{"path":"keytwo"}` res, err := qService.UpdateQueryWithResponse(ctx, jsonQueryRes.JSON201.Id, queryservice.QueryUpdate{ ActiveVersion: &aV, Config: &jcfg, }) assert.NoError(t, err) assert.NotNil(t, res) WaitForClientStatus(t, ctx, qService, clientRes.JSON201.Id, queryservice.NOTSYNCED) WaitForClientStatus(t, ctx, qService, clientRes.JSON201.Id, queryservice.INSYNC) testRes, err = qService.TestQueryWithResponse(ctx, jsonQueryRes.JSON201.Id, queryservice.QueryTestRequest{ QueryVersion: 2, DocumentId: doc.Id, }) assert.NoError(t, err) assert.Equal(t, "valuetwo", testRes.JSON200.Value) } func WaitForClientStatus(t testing.TB, ctx context.Context, service *queryservice.ClientWithResponses, id types.UUID, status queryservice.ClientStatus) { t.Helper() timeout := time.After(30 * time.Second) ticker := time.NewTicker(500 * time.Millisecond) defer ticker.Stop() for { select { case <-timeout: require.NoError(t, fmt.Errorf("Timeout waiting for client status to become %s", status)) case <-ticker.C: jRes, err := service.GetStatusByClientIdWithResponse(ctx, id) if err != nil { assert.NoError(t, err) } if jRes.JSON200.Status == status { assert.Equal(t, status, jRes.JSON200.Status) return } } } } const pdfHelloWorld = `%PDF-1.4 %���� 1 0 obj << /Type /Catalog /Pages 2 0 R /Version /1.4 >> endobj 2 0 obj << /Type /Pages /Kids [3 0 R] /Count 1 >> endobj 3 0 obj << /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Resources << /Font << /F1 << /Type /Font /Subtype /Type1 /BaseFont /Helvetica >> >> >> /Contents 4 0 R >> endobj 4 0 obj << /Length 44 >> stream BT /F1 24 Tf 100 700 Td (Hello World!) Tj ET endstream endobj xref 0 5 0000000000 65535 f 0000000015 00000 n 0000000086 00000 n 0000000151 00000 n 0000000376 00000 n trailer << /Size 5 /Root 1 0 R >> startxref 472 %%EOF`