fix err handling in service.go and add test to check error handling s…

…ucceeds

pkg/provisioning/service.go

@@ -110,6 +110,8 @@ func (s *Service) Init(ctx context.Context) error {
 		if err == nil {
 			multierr = cerrors.Errorf("pipelines with ID %q will be skipped: %w", pl.ID, ErrDuplicatedAPIPipelineID)
 			existingPipelinesFromAPI = append(existingPipelinesFromAPI, i)
+		} else {
+			multierr = cerrors.Errorf("could not get pipeline with ID %v: %w", pl.ID, err)
 		}
 	}
 	configs = s.deleteIndexes(configs, existingPipelinesFromAPI)

pkg/provisioning/service_test.go

@@ -408,17 +408,36 @@ func TestProvision_DuplicatePipelineFromAPI_Error(t *testing.T) {
 	is := is.New(t)
 	logger := log.Nop()
 	ctrl := gomock.NewController(t)
+	otherErr := cerrors.New("error1")
 
-	service, pipelineService, _, _, _ := newTestService(ctrl, logger)
+	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines1"
 
 	// duplicate pipeline found from API
-	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, nil)
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, otherErr)
 
 	pipelineService.EXPECT().List(anyCtx)
+	// pipeline doesn't exist
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+	// create pipeline
+	pipelineService.EXPECT().CreateWithInstance(anyCtx, p1.P1)
+	pipelineService.EXPECT().UpdateDLQ(anyCtx, p1.P1.ID, p1.P1.DLQ)
+	pipelineService.EXPECT().AddConnector(anyCtx, p1.P1.ID, p1.P1.ConnectorIDs[0])
+	pipelineService.EXPECT().AddConnector(anyCtx, p1.P1.ID, p1.P1.ConnectorIDs[1])
+	pipelineService.EXPECT().AddProcessor(anyCtx, p1.P1.ID, p1.P1.ProcessorIDs[0])
+
+	connService.EXPECT().CreateWithInstance(anyCtx, p1.P1C1)
+	connService.EXPECT().CreateWithInstance(anyCtx, p1.P1C2)
+	connService.EXPECT().AddProcessor(anyCtx, p1.P1C2.ID, p1.P1C2.ProcessorIDs[0])
+
+	procService.EXPECT().CreateWithInstance(anyCtx, p1.P1C2P1)
+	procService.EXPECT().CreateWithInstance(anyCtx, p1.P1P1)
+
+	// start pipeline
+	pipelineService.EXPECT().Start(anyCtx, connService, procService, plugService, p1.P1.ID)
 
 	err := service.Init(context.Background())
-	is.True(cerrors.Is(err, ErrDuplicatedAPIPipelineID))
+	is.True(cerrors.Is(err, otherErr))
 }
 
 func TestProvision_IntegrationTestServices(t *testing.T) {