diff --git a/backend/plugins/circleci/tasks/pipeline_collector.go b/backend/plugins/circleci/tasks/pipeline_collector.go index fc981f99979..96c56c2f838 100644 --- a/backend/plugins/circleci/tasks/pipeline_collector.go +++ b/backend/plugins/circleci/tasks/pipeline_collector.go @@ -20,6 +20,7 @@ package tasks import ( "encoding/json" "net/http" + "time" "github.com/apache/devlake/core/errors" "github.com/apache/devlake/core/plugin" @@ -41,7 +42,10 @@ var CollectPipelinesMeta = plugin.SubTaskMeta{ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_TABLE) logger := taskCtx.GetLogger() - timeAfter := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy().TimeAfter + var timeAfter *time.Time + if syncPolicy := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy(); syncPolicy != nil { + timeAfter = syncPolicy.TimeAfter + } logger.Info("collect pipelines") collector, err := api.NewStatefulApiCollectorForFinalizableEntity(api.FinalizableApiCollectorArgs{ RawDataSubTaskArgs: *rawDataSubTaskArgs, @@ -53,25 +57,7 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { UrlTemplate: "/v2/project/{{ .Params.ProjectSlug }}/pipeline", Query: BuildQueryParamsWithPageToken, ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) { - data := CircleciPageTokenResp[[]json.RawMessage]{} - err := api.UnmarshalResponse(res, &data) - - if err != nil { - return nil, err - } - filteredItems := []json.RawMessage{} - for _, item := range data.Items { - pipelineCreatedAt, err := extractCreatedAt(item) - - if err != nil { - return nil, err - } - if pipelineCreatedAt.Before(*timeAfter) { - return filteredItems, api.ErrFinishCollect - } - filteredItems = append(filteredItems, item) - } - return filteredItems, nil + return parsePipelines(res, timeAfter) }, }, GetCreated: extractCreatedAt, @@ -83,3 +69,30 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { } return collector.Execute() } + +// parsePipelines returns the pipelines of a page. CircleCI lists pipelines +// newest first, so the collection ends at the first pipeline created before +// timeAfter. timeAfter is nil when the sync policy has no time range (e.g. an +// advanced mode blueprint), and then every pipeline is kept. +func parsePipelines(res *http.Response, timeAfter *time.Time) ([]json.RawMessage, errors.Error) { + data := CircleciPageTokenResp[[]json.RawMessage]{} + err := api.UnmarshalResponse(res, &data) + if err != nil { + return nil, err + } + if timeAfter == nil { + return data.Items, nil + } + filteredItems := []json.RawMessage{} + for _, item := range data.Items { + pipelineCreatedAt, err := extractCreatedAt(item) + if err != nil { + return nil, err + } + if pipelineCreatedAt.Before(*timeAfter) { + return filteredItems, api.ErrFinishCollect + } + filteredItems = append(filteredItems, item) + } + return filteredItems, nil +} diff --git a/backend/plugins/circleci/tasks/pipeline_collector_test.go b/backend/plugins/circleci/tasks/pipeline_collector_test.go new file mode 100644 index 00000000000..b35c6435dce --- /dev/null +++ b/backend/plugins/circleci/tasks/pipeline_collector_test.go @@ -0,0 +1,63 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package tasks + +import ( + "bytes" + "io" + "net/http" + "testing" + "time" + + "github.com/apache/devlake/helpers/pluginhelper/api" + "github.com/stretchr/testify/assert" +) + +const pipelinesPage = `{ + "items": [ + {"id": "p3", "created_at": "2026-03-01T00:00:00Z"}, + {"id": "p2", "created_at": "2026-02-01T00:00:00Z"}, + {"id": "p1", "created_at": "2026-01-01T00:00:00Z"} + ], + "next_page_token": "token" +}` + +func makePipelinesResponse() *http.Response { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewBufferString(pipelinesPage)), + Request: &http.Request{}, + } +} + +func TestParsePipelinesWithoutTimeAfter(t *testing.T) { + // An advanced mode blueprint has no timeAfter: keep every pipeline + // instead of dereferencing a nil time. + items, err := parsePipelines(makePipelinesResponse(), nil) + assert.Nil(t, err) + assert.Len(t, items, 3) +} + +func TestParsePipelinesStopsBeforeTimeAfter(t *testing.T) { + timeAfter := time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC) + items, err := parsePipelines(makePipelinesResponse(), &timeAfter) + assert.ErrorIs(t, err, api.ErrFinishCollect) + assert.Len(t, items, 2) + assert.Contains(t, string(items[0]), `"p3"`) + assert.Contains(t, string(items[1]), `"p2"`) +}