Function TestAsyncMessageHandler_Handle

pkg/dequeuer/async_handler_test.go:38–89

(t *testing.T)

)

func TestAsyncMessageHandler_Handle(t *testing.T) {
	t.Parallel()

	log := newLogger(t)
	defer func() { _ = log.Sync() }()

	awsClient := testAWSClient(t)

	requestID := random.String(8)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, requestID, r.Header.Get(CortexRequestIDHeader))
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("{}"))
	}))

	var requestEventsCount int
	eventHandler := NewRequestEventHandlerFunc(func(event RequestEvent) {
		requestEventsCount++
	})

	asyncHandler := NewAsyncMessageHandler(AsyncMessageHandlerConfig{
		ClusterUID: "cortex-test",
		Bucket:     _testBucket,
		APIName:    "async-test",
		TargetURL:  server.URL,
	}, awsClient, eventHandler, log)

	_, err := awsClient.S3().CreateBucket(&s3.CreateBucketInput{
		Bucket: aws.String(_testBucket),
	})
	require.NoError(t, err)

	err = awsClient.UploadStringToS3("{}", asyncHandler.config.Bucket, async.PayloadPath(asyncHandler.storagePath, requestID))
	require.NoError(t, err)

	err = awsClient.UploadStringToS3("{}", asyncHandler.config.Bucket, async.HeadersPath(asyncHandler.storagePath, requestID))
	require.NoError(t, err)

	err = asyncHandler.Handle(&sqs.Message{
		Body:      aws.String(requestID),
		MessageId: aws.String(requestID),
	})
	require.NoError(t, err)

	_, err = awsClient.ReadStringFromS3(
		_testBucket,
		fmt.Sprintf("%s/%s/status/%s", asyncHandler.storagePath, requestID, async.StatusCompleted),
	)
	require.NoError(t, err)
	require.Equal(t, 1, requestEventsCount)
}

func TestAsyncMessageHandler_Handle_Errors(t *testing.T) {
	t.Parallel()