feat(plugins): restranformate without api client (#8034)

* feat(plugins): restranformate without api client * fix(zentao): fix test errors * fix(bamboo): fix e2e test * fix(jira): fix e2e test * fix(zentao): fix e2e test * fix(bamboo): fix e2e test
apache · Sep 11, 2024 · b80340b · b80340b
1 parent c61e912
commit b80340b
Show file tree

Hide file tree

Showing 58 changed files with 613 additions and 290 deletions.
diff --git a/backend/plugins/ae/impl/impl.go b/backend/plugins/ae/impl/impl.go
@@ -107,9 +107,14 @@ func (p AE) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]inter
 	if err != nil {
 		return nil, errors.Default.Wrap(err, "error getting connection for AE plugin")
 	}
-	apiClient, err := tasks.CreateApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, err
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, err
+		}
+		apiClient = newApiClient
 	}
 	return &tasks.AeTaskData{
 		Options:   &op,

diff --git a/backend/plugins/azuredevops_go/impl/impl.go b/backend/plugins/azuredevops_go/impl/impl.go
@@ -127,9 +127,14 @@ func (p Azuredevops) PrepareTaskData(taskCtx plugin.TaskContext, options map[str
 		return nil, errors.Default.Wrap(err, "failed to retrieve an Azure DevOps connection from the database using the provided connection ID")
 	}
 
-	apiClient, err := tasks.CreateApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, errors.Default.Wrap(err, "failed to retrieve an Azure DevOps connection from the database using the provided connection ID")
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, errors.Default.Wrap(err, "failed to retrieve an Azure DevOps connection from the database using the provided connection ID")
+		}
+		apiClient = newApiClient
 	}
 
 	if op.RepositoryId != "" {

diff --git a/backend/plugins/bamboo/e2e/plan_build_commits_test.go b/backend/plugins/bamboo/e2e/plan_build_commits_test.go
@@ -43,6 +43,7 @@ func TestBambooPlanBuildCommitsDataFlow(t *testing.T) {
 			},
 		},
 		ApiClient: getFakeAPIClient(),
+		EndPoint:  getFakeAPIClient().GetEndpoint(),
 	}
 
 	dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_bamboo_plan_build_commits.csv", &models.BambooPlanBuildVcsRevision{})

diff --git a/backend/plugins/bamboo/impl/impl.go b/backend/plugins/bamboo/impl/impl.go
@@ -140,10 +140,16 @@ func (p Bamboo) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
 	if err != nil {
 		return nil, errors.Default.Wrap(err, "unable to get Bamboo connection by the given connection ID")
 	}
+	endPoint := connection.GetEndpoint()
 
-	apiClient, err := tasks.NewBambooApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, errors.Default.Wrap(err, "unable to get Bamboo API client instance")
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.NewBambooApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, errors.Default.Wrap(err, "unable to get Bamboo API client instance")
+		}
+		apiClient = newApiClient
 	}
 	if op.PlanKey != "" {
 		var scope *models.BambooPlan
@@ -189,6 +195,7 @@ func (p Bamboo) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
 	return &tasks.BambooOptions{
 		Options:       op,
 		ApiClient:     apiClient,
+		EndPoint:      endPoint,
 		RegexEnricher: regexEnricher,
 	}, nil
 }

diff --git a/backend/plugins/bamboo/tasks/plan_commit_convertor.go b/backend/plugins/bamboo/tasks/plan_commit_convertor.go
@@ -66,9 +66,9 @@ func ConvertPlanVcs(taskCtx plugin.SubTaskContext) errors.Error {
 				Url:          line.Url,
 			}
 			domainPlanVcs.RepoId = repoMap[line.RepositoryId]
-			fakeRepoUrl, err := generateFakeRepoUrl(data.ApiClient.GetEndpoint(), line.RepositoryId)
+			fakeRepoUrl, err := generateFakeRepoUrl(data.EndPoint, line.RepositoryId)
 			if err != nil {
-				logger.Warn(err, "generate fake repo url, endpoint: %s, repo id: %d", data.ApiClient.GetEndpoint(), line.RepositoryId)
+				logger.Warn(err, "generate fake repo url, endpoint: %s, repo id: %d", data.EndPoint, line.RepositoryId)
 			} else {
 				domainPlanVcs.RepoUrl = fakeRepoUrl
 			}

diff --git a/backend/plugins/bamboo/tasks/task_data.go b/backend/plugins/bamboo/tasks/task_data.go
@@ -26,6 +26,7 @@ import (
 type BambooOptions struct {
 	Options       *models.BambooOptions
 	ApiClient     *helper.ApiAsyncClient
+	EndPoint      string
 	RegexEnricher *helper.RegexEnricher
 }
 

diff --git a/backend/plugins/bitbucket/api/blueprint_v200.go b/backend/plugins/bitbucket/api/blueprint_v200.go
@@ -54,9 +54,11 @@ func MakeDataSourcePipelinePlanV200(
 
 	// needed for the connection to populate its access tokens
 	// if AppKey authentication method is selected
-	_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
-	if err != nil {
-		return nil, nil, err
+	if !skipCollectors {
+		_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
+		if err != nil {
+			return nil, nil, err
+		}
 	}
 
 	plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, scopeDetails, connection)

diff --git a/backend/plugins/bitbucket/impl/impl.go b/backend/plugins/bitbucket/impl/impl.go
@@ -154,13 +154,25 @@ func (p Bitbucket) PrepareTaskData(taskCtx plugin.TaskContext, options map[strin
 		return nil, errors.Default.Wrap(err, "unable to get bitbucket connection by the given connection ID")
 	}
 
-	apiClient, err := tasks.CreateApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, errors.Default.Wrap(err, "unable to get bitbucket API client instance")
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, errors.Default.Wrap(err, "unable to get bitbucket API client instance")
+		}
+		apiClient = newApiClient
 	}
-	err = EnrichOptions(taskCtx, op, apiClient.ApiClient)
-	if err != nil {
-		return nil, err
+	if apiClient == nil {
+		err = EnrichOptions(taskCtx, op, nil)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		err = EnrichOptions(taskCtx, op, apiClient.ApiClient)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	regexEnricher := helper.NewRegexEnricher()
@@ -276,17 +288,19 @@ func EnrichOptions(taskCtx plugin.TaskContext,
 		}
 	} else {
 		if taskCtx.GetDal().IsErrorNotFound(err) && op.FullName != "" {
-			var repo *models.BitbucketApiRepo
-			repo, err = tasks.GetApiRepo(op, apiClient)
-			if err != nil {
-				return err
-			}
-			logger.Debug(fmt.Sprintf("Current repo: %s", repo.FullName))
-			scope := repo.ConvertApiScope()
-			scope.ConnectionId = op.ConnectionId
-			err = taskCtx.GetDal().CreateIfNotExist(scope)
-			if err != nil {
-				return err
+			if apiClient != nil {
+				var repo *models.BitbucketApiRepo
+				repo, err = tasks.GetApiRepo(op, apiClient)
+				if err != nil {
+					return err
+				}
+				logger.Debug(fmt.Sprintf("Current repo: %s", repo.FullName))
+				scope := repo.ConvertApiScope()
+				scope.ConnectionId = op.ConnectionId
+				err = taskCtx.GetDal().CreateIfNotExist(scope)
+				if err != nil {
+					return err
+				}
 			}
 		} else {
 			return errors.Default.Wrap(err, fmt.Sprintf("fail to find repo %s", op.FullName))

diff --git a/backend/plugins/bitbucket_server/api/blueprint_v200.go b/backend/plugins/bitbucket_server/api/blueprint_v200.go
@@ -52,11 +52,13 @@ func MakeDataSourcePipelinePlanV200(
 		return nil, nil, err
 	}
 
-	// needed for the connection to populate its access tokens
-	// if AppKey authentication method is selected
-	_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
-	if err != nil {
-		return nil, nil, err
+	if !skipCollectors {
+		// needed for the connection to populate its access tokens
+		// if AppKey authentication method is selected
+		_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
+		if err != nil {
+			return nil, nil, err
+		}
 	}
 
 	plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, scopeDetails, connection)

diff --git a/backend/plugins/bitbucket_server/impl/impl.go b/backend/plugins/bitbucket_server/impl/impl.go
@@ -123,13 +123,25 @@ func (p BitbucketServer) PrepareTaskData(taskCtx plugin.TaskContext, options map
 		return nil, errors.Default.Wrap(err, "unable to get bitbucket server connection by the given connection ID")
 	}
 
-	apiClient, err := tasks.CreateApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, errors.Default.Wrap(err, "unable to get bitbucket server API client instance")
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, errors.Default.Wrap(err, "unable to get bitbucket server API client instance")
+		}
+		apiClient = newApiClient
 	}
-	err = EnrichOptions(taskCtx, op, apiClient.ApiClient)
-	if err != nil {
-		return nil, err
+	if apiClient == nil {
+		err = EnrichOptions(taskCtx, op, nil)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		err = EnrichOptions(taskCtx, op, apiClient.ApiClient)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	regexEnricher := helper.NewRegexEnricher()
@@ -239,17 +251,19 @@ func EnrichOptions(taskCtx plugin.TaskContext,
 		}
 	} else {
 		if taskCtx.GetDal().IsErrorNotFound(err) && op.FullName != "" {
-			var repo *models.BitbucketServerApiRepo
-			repo, err = tasks.GetApiRepo(op, apiClient)
-			if err != nil {
-				return err
-			}
-			logger.Debug(fmt.Sprintf("Current repo: %s", repo.Slug))
-			scope := repo.ConvertApiScope().(*models.BitbucketServerRepo)
-			scope.ConnectionId = op.ConnectionId
-			err = taskCtx.GetDal().CreateIfNotExist(scope)
-			if err != nil {
-				return err
+			if apiClient != nil {
+				var repo *models.BitbucketServerApiRepo
+				repo, err = tasks.GetApiRepo(op, apiClient)
+				if err != nil {
+					return err
+				}
+				logger.Debug(fmt.Sprintf("Current repo: %s", repo.Slug))
+				scope := repo.ConvertApiScope().(*models.BitbucketServerRepo)
+				scope.ConnectionId = op.ConnectionId
+				err = taskCtx.GetDal().CreateIfNotExist(scope)
+				if err != nil {
+					return err
+				}
 			}
 		} else {
 			return errors.Default.Wrap(err, fmt.Sprintf("fail to find repo %s", op.FullName))

diff --git a/backend/plugins/feishu/impl/impl.go b/backend/plugins/feishu/impl/impl.go
@@ -109,9 +109,14 @@ func (p Feishu) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]i
 		return nil, err
 	}
 
-	apiClient, err := tasks.NewFeishuApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, err
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.NewFeishuApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, err
+		}
+		apiClient = newApiClient
 	}
 	return &tasks.FeishuTaskData{
 		Options:   &op,

diff --git a/backend/plugins/gitee/impl/impl.go b/backend/plugins/gitee/impl/impl.go
@@ -162,10 +162,15 @@ func (p Gitee) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]in
 	if err != nil {
 		return nil, err
 	}
-	apiClient, err := tasks.NewGiteeApiClient(taskCtx, connection)
 
-	if err != nil {
-		return nil, err
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.NewGiteeApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, err
+		}
+		apiClient = newApiClient
 	}
 
 	return &tasks.GiteeTaskData{

diff --git a/backend/plugins/icla/impl/impl.go b/backend/plugins/icla/impl/impl.go
@@ -71,9 +71,14 @@ func (p Icla) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]int
 		return nil, err
 	}
 
-	apiClient, err := errors.Convert01(tasks.NewIclaApiClient(taskCtx))
-	if err != nil {
-		return nil, err
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := errors.Convert01(tasks.NewIclaApiClient(taskCtx))
+		if err != nil {
+			return nil, err
+		}
+		apiClient = newApiClient
 	}
 
 	return &tasks.IclaTaskData{

diff --git a/backend/plugins/jenkins/api/blueprint_v200.go b/backend/plugins/jenkins/api/blueprint_v200.go
@@ -49,11 +49,13 @@ func MakeDataSourcePipelinePlanV200(
 		return nil, nil, err
 	}
 
-	// needed for the connection to populate its access tokens
-	// if AppKey authentication method is selected
-	_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
-	if err != nil {
-		return nil, nil, err
+	if !skipCollectors {
+		// needed for the connection to populate its access tokens
+		// if AppKey authentication method is selected
+		_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
+		if err != nil {
+			return nil, nil, err
+		}
 	}
 
 	plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, scopeDetails, connection)

diff --git a/backend/plugins/jenkins/impl/impl.go b/backend/plugins/jenkins/impl/impl.go
@@ -126,9 +126,14 @@ func (p Jenkins) PrepareTaskData(taskCtx plugin.TaskContext, options map[string]
 		return nil, err
 	}
 
-	apiClient, err := tasks.CreateApiClient(taskCtx, connection)
-	if err != nil {
-		return nil, err
+	var apiClient *helper.ApiAsyncClient
+	syncPolicy := taskCtx.SyncPolicy()
+	if !syncPolicy.SkipCollectors {
+		newApiClient, err := tasks.CreateApiClient(taskCtx, connection)
+		if err != nil {
+			return nil, err
+		}
+		apiClient = newApiClient
 	}
 
 	op.ConnectionEndpoint = connection.Endpoint
@@ -260,21 +265,23 @@ func EnrichOptions(taskCtx plugin.TaskContext,
 		}
 	}
 
-	err = api.GetJob(apiClient, op.JobPath, op.JobName, op.JobFullName, 100, func(job *models.Job, isPath bool) errors.Error {
-		log.Debug(fmt.Sprintf("Current job: %s", job.FullName))
-		op.JobPath = job.Path
-		op.URL = job.URL
-		op.Class = job.Class
-		jenkinsJob := job.ToJenkinsJob()
+	if apiClient != nil {
+		err = api.GetJob(apiClient, op.JobPath, op.JobName, op.JobFullName, 100, func(job *models.Job, isPath bool) errors.Error {
+			log.Debug(fmt.Sprintf("Current job: %s", job.FullName))
+			op.JobPath = job.Path
+			op.URL = job.URL
+			op.Class = job.Class
+			jenkinsJob := job.ToJenkinsJob()
 
-		jenkinsJob.ConnectionId = op.ConnectionId
-		jenkinsJob.ScopeConfigId = op.ScopeConfigId
+			jenkinsJob.ConnectionId = op.ConnectionId
+			jenkinsJob.ScopeConfigId = op.ScopeConfigId
 
-		err = taskCtx.GetDal().CreateIfNotExist(jenkinsJob)
-		return err
-	})
-	if err != nil {
-		return err
+			err = taskCtx.GetDal().CreateIfNotExist(jenkinsJob)
+			return err
+		})
+		if err != nil {
+			return err
+		}
 	}
 
 	if !strings.HasSuffix(op.JobPath, "/") {

diff --git a/backend/plugins/jira/api/blueprint_v200.go b/backend/plugins/jira/api/blueprint_v200.go
@@ -48,11 +48,13 @@ func MakeDataSourcePipelinePlanV200(
 		return nil, nil, err
 	}
 
-	// needed for the connection to populate its access tokens
-	// if AppKey authentication method is selected
-	_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
-	if err != nil {
-		return nil, nil, err
+	if !skipCollectors {
+		// needed for the connection to populate its access tokens
+		// if AppKey authentication method is selected
+		_, err = helper.NewApiClientFromConnection(context.TODO(), basicRes, connection)
+		if err != nil {
+			return nil, nil, err
+		}
 	}
 
 	plan, err := makeDataSourcePipelinePlanV200(subtaskMetas, scopeDetails, connection)