test: created testing

internal/server/v1/dlq/handler.go

@@ -1,6 +1,7 @@
 package dlq
 
 import (
+	"errors"
 	"log"
 	"net/http"
 
@@ -77,6 +78,10 @@ func (h *Handler) createDlqJob(w http.ResponseWriter, r *http.Request) {
 	// call service.CreateDLQJob
 	err := h.service.CreateDLQJob(ctx, reqCtx.UserEmail, &dlqJob)
 	if err != nil {
+		if errors.Is(err, ErrFirehoseNotFound) {
+			utils.WriteErrMsg(w, http.StatusNotFound, err.Error())
+			return
+		}
 		utils.WriteErr(w, err)
 		return
 	}
@@ -91,7 +96,7 @@ func (h *Handler) getDlqJob(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 
 	firehoseUrn := chi.URLParam(r, "firehoseURN")
-	// fetch entorpy resource (kind = job)
+	// fetch entropy resource (kind = job)
 	// mapToDlqJob(entropyResource) -> DqlJob
 	dlqJob, err := h.service.getDlqJob(ctx, firehoseUrn)
 	if err != nil {

internal/server/v1/dlq/handler_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/goto/dex/entropy"
@@ -198,26 +200,68 @@ func TestErrorFromFirehoseResource(t *testing.T) {
 
 func TestCreateDlqJob(t *testing.T) {
 	var (
-		method        = http.MethodPost
-		path          = fmt.Sprintf("/jobs")
-		resource_id   = "test-resource-id"
-		resource_type = "test-resource-type"
-		error_types   = "DESERIALIZATION_ERROR"
-		date          = "21-10-2022"
-		batch_size    = 0
-		num_threads   = 0
-		topic         = "test-topic"
-		jsonPayload   = fmt.Sprintf(`{
+		method       = http.MethodPost
+		path         = fmt.Sprintf("/jobs")
+		resourceId   = "test-resource-id"
+		resourceType = "test-resource-type"
+		errorTypes   = "DESERIALIZATION_ERROR"
+		date         = "21-10-2022"
+		batchSize    = 0
+		numThreads   = 0
+		topic        = "test-topic"
+		group        = ""
+		jsonPayload  = fmt.Sprintf(`{
 			"resource_id": "%s",
 			"resource_type": "%s",
 			"error_types": "%s",
 			"batch_size": %d,
 			"num_threads": %d,
 			"topic": "%s",
-			"date": "%s"
-		}`, resource_id, resource_type, error_types, batch_size, num_threads, topic, date)
+			"date": "%s",
+			"group": "%s"
+		}`, resourceId, resourceType, errorTypes, batchSize, numThreads, topic, date, group)
 	)
 
+	t.Run("Should return error firehose not Found", func(t *testing.T) {
+		// initt input
+		expectedErr := status.Error(codes.NotFound, "Not found")
+		entropyClient := new(mocks.ResourceServiceClient)
+		entropyClient.On(
+			"GetResource", mock.Anything, &entropyv1beta1.GetResourceRequest{Urn: resourceId},
+		).Return(nil, expectedErr)
+		defer entropyClient.AssertExpectations(t)
+
+		requestBody := bytes.NewReader([]byte(jsonPayload))
+
+		response := httptest.NewRecorder()
+		request := httptest.NewRequest(method, path, requestBody)
+		router := getRouter()
+		dlq.Routes(entropyClient, nil, dlq.DlqJobConfig{})(router)
+		router.ServeHTTP(response, request)
+
+		assert.Equal(t, http.StatusNotFound, response.Code)
+	})
+
+	t.Run("Should return error in firehose mapping", func(t *testing.T) {
+		// initt input
+		expectedErr := status.Error(codes.Internal, "Not found")
+		entropyClient := new(mocks.ResourceServiceClient)
+		entropyClient.On(
+			"GetResource", mock.Anything, &entropyv1beta1.GetResourceRequest{Urn: resourceId},
+		).Return(nil, expectedErr)
+		defer entropyClient.AssertExpectations(t)
+
+		requestBody := bytes.NewReader([]byte(jsonPayload))
+
+		response := httptest.NewRecorder()
+		request := httptest.NewRequest(method, path, requestBody)
+		router := getRouter()
+		dlq.Routes(entropyClient, nil, dlq.DlqJobConfig{})(router)
+		router.ServeHTTP(response, request)
+
+		assert.Equal(t, http.StatusInternalServerError, response.Code)
+	})
+
 	t.Run("Should return resource urn", func(t *testing.T) {
 		// initt input
 		namespace := "test-namespace"
@@ -295,9 +339,9 @@ func TestCreateDlqJob(t *testing.T) {
 		}
 
 		expectedEnvVars := map[string]string{
-			"DLQ_BATCH_SIZE":                          fmt.Sprintf("%d", batch_size),
-			"DLQ_NUM_THREADS":                         fmt.Sprintf("%d", num_threads),
-			"DLQ_ERROR_TYPES":                         error_types,
+			"DLQ_BATCH_SIZE":                          fmt.Sprintf("%d", batchSize),
+			"DLQ_NUM_THREADS":                         fmt.Sprintf("%d", numThreads),
+			"DLQ_ERROR_TYPES":                         errorTypes,
 			"DLQ_INPUT_DATE":                          date,
 			"DLQ_TOPIC_NAME":                          topic,
 			"METRIC_STATSD_TAGS":                      "a=b", // TBA
@@ -391,7 +435,7 @@ func TestCreateDlqJob(t *testing.T) {
 				},
 			},
 			JobLabels: map[string]string{
-				"firehose": resource_id,
+				"firehose": resourceId,
 				"topic":    topic,
 				"date":     date,
 			},
@@ -413,18 +457,20 @@ func TestCreateDlqJob(t *testing.T) {
 			Kind: entropy.ResourceKindJob,
 			Name: fmt.Sprintf(
 				"%s-%s-%s-%s",
-				resource_id,   // firehose urn
-				resource_type, // firehose / dagger
-				topic,         //
-				date,          //
+				firehoseResource.Name, // firehose urn
+				"firehose",            // firehose / dagger
+				topic,                 //
+				date,                  //
 			),
 			Project: firehoseResource.Project,
 			Labels: map[string]string{
-				"resource_id": resource_id,
-				"type":        resource_type,
-				"date":        date,
-				"topic":       topic,
-				"job_type":    "dlq",
+				"resource_id":     resourceId,
+				"resource_type":   resourceType,
+				"date":            date,
+				"topic":           topic,
+				"job_type":        "dlq",
+				"group":           group,
+				"prometheus_host": config.PrometheusHost,
 			},
 			CreatedBy: jobResource.CreatedBy,
 			UpdatedBy: jobResource.UpdatedBy,
@@ -442,7 +488,7 @@ func TestCreateDlqJob(t *testing.T) {
 		// set conditions
 		entropyClient := new(mocks.ResourceServiceClient)
 		entropyClient.On(
-			"GetResource", mock.Anything, &entropyv1beta1.GetResourceRequest{Urn: resource_id},
+			"GetResource", mock.Anything, &entropyv1beta1.GetResourceRequest{Urn: resourceId},
 		).Return(&entropyv1beta1.GetResourceResponse{
 			Resource: firehoseResource,
 		}, nil)

internal/server/v1/dlq/service.go

@@ -5,7 +5,9 @@ import (
 
 	entropyv1beta1rpc "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/entropy/v1beta1/entropyv1beta1grpc"
 	entropyv1beta1 "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/entropy/v1beta1"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 
 	"github.com/goto/dex/generated/models"
 	"github.com/goto/dex/internal/server/gcs"
@@ -36,11 +38,15 @@ func (s *Service) CreateDLQJob(ctx context.Context, userEmail string, dlqJob *mo
 	// fetch firehose details
 	def, err := s.client.GetResource(ctx, &entropyv1beta1.GetResourceRequest{Urn: dlqJob.ResourceID})
 	if err != nil {
-		return ErrFirehoseNotFound
+		st := status.Convert(err)
+		if st.Code() == codes.NotFound {
+			return ErrFirehoseNotFound
+		}
+		return err
 	}
 	// enrich DlqJob with firehose details
 	if err := enrichDlqJob(dlqJob, def.GetResource(), s.cfg); err != nil {
-		return ErrFirehoseNotFound
+		return err
 	}
 
 	// map DlqJob to entropy resource -> return entropy.Resource (kind = job)

internal/server/v1/dlq/service_test.go

@@ -9,7 +9,9 @@ import (
 	"github.com/go-openapi/strfmt"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/goto/dex/entropy"
@@ -20,6 +22,48 @@ import (
 )
 
 func TestServiceCreateDLQJob(t *testing.T) {
+	t.Run("should return ErrFirehoseNotFound if resource cannot be found in entropy", func(t *testing.T) {
+		// inputs
+		ctx := context.TODO()
+		dlqJob := models.DlqJob{
+			ResourceID:   "test-resource-id",
+			ResourceType: "firehose",
+		}
+		expectedErr := status.Error(codes.NotFound, "Not found")
+
+		// set conditions
+		entropyClient := new(mocks.ResourceServiceClient)
+		entropyClient.On(
+			"GetResource", ctx, &entropyv1beta1.GetResourceRequest{Urn: dlqJob.ResourceID},
+		).Return(nil, expectedErr)
+		defer entropyClient.AssertExpectations(t)
+		service := dlq.NewService(entropyClient, nil, dlq.DlqJobConfig{})
+
+		err := service.CreateDLQJob(ctx, "", &dlqJob)
+		assert.ErrorIs(t, err, dlq.ErrFirehoseNotFound)
+	})
+
+	t.Run("should return error when there is an error getting firehose in entropy", func(t *testing.T) {
+		// inputs
+		ctx := context.TODO()
+		dlqJob := models.DlqJob{
+			ResourceID:   "test-resource-id",
+			ResourceType: "firehose",
+		}
+		expectedErr := status.Error(codes.Internal, "Any Error")
+
+		// set conditions
+		entropyClient := new(mocks.ResourceServiceClient)
+		entropyClient.On(
+			"GetResource", ctx, &entropyv1beta1.GetResourceRequest{Urn: dlqJob.ResourceID},
+		).Return(nil, expectedErr)
+		defer entropyClient.AssertExpectations(t)
+		service := dlq.NewService(entropyClient, nil, dlq.DlqJobConfig{})
+
+		err := service.CreateDLQJob(ctx, "", &dlqJob)
+		assert.ErrorIs(t, err, expectedErr)
+	})
+
 	t.Run("should create a entropy resource with job kind", func(t *testing.T) {
 		// inputs
 		ctx := context.TODO()
@@ -66,10 +110,10 @@ func TestServiceCreateDLQJob(t *testing.T) {
 		}
 
 		dlqJob := models.DlqJob{
-			BatchSize:  int64(5),
-			Date:       "2012-10-30",
-			ErrorTypes: "DESERILIAZATION_ERROR",
-			// Group: "",
+			BatchSize:    int64(5),
+			Date:         "2012-10-30",
+			ErrorTypes:   "DESERILIAZATION_ERROR",
+			Group:        "",
 			NumThreads:   2,
 			ResourceID:   "test-resource-id",
 			ResourceType: "firehose",
@@ -144,7 +188,7 @@ func TestServiceCreateDLQJob(t *testing.T) {
 		}
 
 		jobConfig, err := utils.GoValToProtoStruct(entropy.JobConfig{
-			Replicas:  0,
+			Replicas:  1,
 			Namespace: namespace,
 			Containers: []entropy.JobContainer{
 				{
@@ -227,18 +271,20 @@ func TestServiceCreateDLQJob(t *testing.T) {
 			Kind: entropy.ResourceKindJob,
 			Name: fmt.Sprintf(
 				"%s-%s-%s-%s",
-				dlqJob.ResourceID,   // firehose urn
+				jobResource.Name,    // firehose urn
 				dlqJob.ResourceType, // firehose / dagger
 				dlqJob.Topic,        //
 				dlqJob.Date,         //
 			),
 			Project: firehoseResource.Project,
 			Labels: map[string]string{
-				"resource_id": dlqJob.ResourceID,
-				"type":        dlqJob.ResourceType,
-				"date":        dlqJob.Date,
-				"topic":       dlqJob.Topic,
-				"job_type":    "dlq",
+				"resource_id":     dlqJob.ResourceID,
+				"resource_type":   dlqJob.ResourceType,
+				"date":            dlqJob.Date,
+				"topic":           dlqJob.Topic,
+				"job_type":        "dlq",
+				"group":           dlqJob.Group,
+				"prometheus_host": config.PrometheusHost,
 			},
 			CreatedBy: jobResource.CreatedBy,
 			UpdatedBy: jobResource.UpdatedBy,
@@ -277,9 +323,17 @@ func TestServiceCreateDLQJob(t *testing.T) {
 			ResourceID:   dlqJob.ResourceID,
 			ResourceType: dlqJob.ResourceType,
 			Topic:        dlqJob.Topic,
-			NumThreads:   dlqJob.NumThreads,
-			Date:         dlqJob.Date,
-			ErrorTypes:   dlqJob.ErrorTypes,
+			Name: fmt.Sprintf(
+				"%s-%s-%s-%s",
+				firehoseResource.Name, // firehose title
+				"firehose",            // firehose / dagger
+				dlqJob.Topic,          //
+				dlqJob.Date,           //
+			),
+
+			NumThreads: dlqJob.NumThreads,
+			Date:       dlqJob.Date,
+			ErrorTypes: dlqJob.ErrorTypes,
 
 			// firehose resource
 			ContainerImage:       config.DlqJobImage,