fix(tempo): only collect and extract the team of a team scope

backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_teams_team_71.csv

@@ -0,0 +1,3 @@
+id,params,data,url,input,created_at
+1,"{""ConnectionId"":1,""TeamId"":71}","{""self"":""https://api.tempo.io/4/teams/17"",""id"":17,""name"":""Abstract BE Team (renamed)"",""summary"":"""",""key"":""""}","https://api.tempo.io/4/teams?limit=50&offset=0&teamIds=71","{""self"":""https://api.tempo.io/4/teams/17"",""id"":17,""name"":""Abstract BE Team (renamed)"",""summary"":"""",""key"":""""}","2024-02-15 10:00:00.000"
+2,"{""ConnectionId"":1,""TeamId"":71}","{""self"":""https://api.tempo.io/4/teams/71"",""id"":71,""name"":""SmartShopper (RTL+WHS+POS)"",""summary"":""Microservice Applications"",""key"":""""}","https://api.tempo.io/4/teams?limit=50&offset=0&teamIds=71","{""self"":""https://api.tempo.io/4/teams/71"",""id"":71,""name"":""SmartShopper (RTL+WHS+POS)"",""summary"":""Microservice Applications"",""key"":""""}","2024-02-15 10:00:00.000"

backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_scopes.csv

@@ -0,0 +1,3 @@
+"connection_id","team_id","scope_config_id","name","summary","key","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+1,17,2,"Abstract BE Team","","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",1,
+1,71,3,"SmartShopper (RTL+WHS)","Microservice Applications","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",2,

backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_teams_team_71.csv

@@ -0,0 +1,3 @@
+"connection_id","team_id","scope_config_id","name","summary","key","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+1,17,2,"Abstract BE Team","","","{""ConnectionId"":1,""TeamId"":0}","_raw_tempo_api_teams",1,
+1,71,3,"SmartShopper (RTL+WHS+POS)","Microservice Applications","","{""ConnectionId"":1,""TeamId"":71}","_raw_tempo_api_teams",2,

backend/plugins/tempo/e2e/worklog_test.go

@@ -102,3 +102,32 @@ func TestTeamDataFlow(t *testing.T) {
 		),
 	)
 }
+
+func TestTeamScopeExtraction(t *testing.T) {
+	var plugin impl.Tempo
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)
+
+	taskData := &tasks.TempoTaskData{
+		Options: &tasks.TempoOptions{
+			ConnectionId: 1,
+			TeamId:       71,
+		},
+	}
+
+	// two team scopes with scope configs; only team 71 is extracted
+	dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_tempo_teams_scopes.csv", &models.TempoTeam{})
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_tempo_api_teams_team_71.csv", "_raw_tempo_api_teams")
+	dataflowTester.Subtask(tasks.ExtractTeamsMeta, taskData)
+	dataflowTester.VerifyTable(
+		models.TempoTeam{},
+		"./snapshot_tables/_tool_tempo_teams_team_71.csv",
+		e2ehelper.ColumnWithRawData(
+			"connection_id",
+			"team_id",
+			"scope_config_id",
+			"name",
+			"summary",
+			"key",
+		),
+	)
+}

backend/plugins/tempo/tasks/team_collector.go

@@ -22,7 +22,6 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
-	"time"
 
 	"github.com/apache/devlake/core/errors"
 	"github.com/apache/devlake/core/plugin"
@@ -42,45 +41,18 @@ var CollectTeamsMeta = plugin.SubTaskMeta{
 func CollectTeams(taskCtx plugin.SubTaskContext) errors.Error {
 	data := taskCtx.GetData().(*TempoTaskData)
 
-	apiCollector, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
-		Ctx: taskCtx,
-		Params: models.TempoApiParams{
-			ConnectionId: data.Options.ConnectionId,
-		},
-		Table: RAW_TEAM_TABLE,
-	})
-	if err != nil {
-		return err
-	}
-
-	err = apiCollector.InitCollector(api.ApiCollectorArgs{
-		RawDataSubTaskArgs: api.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: models.TempoApiParams{
-				ConnectionId: data.Options.ConnectionId,
-			},
-			Table: RAW_TEAM_TABLE,
-		},
-		ApiClient:   data.ApiClient,
-		UrlTemplate: "teams",
-		PageSize:    50,
+	// Not a stateful (incremental) collector: GET /4/teams has no
+	// "updatedFrom" filter, so an incremental run would append the same
+	// teams to the raw table again instead of refreshing them.
+	apiCollector, err := api.NewApiCollector(api.ApiCollectorArgs{
+		RawDataSubTaskArgs: teamRawDataSubTaskArgs(taskCtx, data.Options),
+		ApiClient:          data.ApiClient,
+		UrlTemplate:        "teams",
+		PageSize:           50,
 		// No GetTotalPages: Tempo v4 pagination metadata (PageableMetadata)
 		// has no total, so pages are fetched until one comes back short.
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
-			query := url.Values{}
-			pager := reqData.Pager
-			if pager == nil {
-				pager = &api.Pager{Page: 1, Skip: 0, Size: 50}
-			}
-			query.Set("offset", strconv.Itoa(pager.Skip))
-			query.Set("limit", strconv.Itoa(pager.Size))
-
-			if apiCollector.IsIncremental() && apiCollector.GetSince() != nil {
-				since := apiCollector.GetSince()
-				query.Set("updatedFrom", since.Format(time.RFC3339))
-			}
-
-			return query, nil
+			return buildTeamQuery(data.Options, reqData.Pager), nil
 		},
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var response struct {
@@ -100,3 +72,32 @@ func CollectTeams(taskCtx plugin.SubTaskContext) errors.Error {
 
 	return apiCollector.Execute()
 }
+
+// teamRawDataSubTaskArgs scopes the raw team data and the extracted teams per
+// team, like the scope itself (TempoTeam.ScopeParams). The extracted rows are
+// the team scopes: with connection-wide params, extracting one team scope
+// deleted and re-created every team scope of the connection.
+func teamRawDataSubTaskArgs(taskCtx plugin.SubTaskContext, opts *TempoOptions) api.RawDataSubTaskArgs {
+	return api.RawDataSubTaskArgs{
+		Ctx: taskCtx,
+		Params: models.TempoApiParams{
+			ConnectionId: opts.ConnectionId,
+			TeamId:       opts.TeamId,
+		},
+		Table: RAW_TEAM_TABLE,
+	}
+}
+
+func buildTeamQuery(opts *TempoOptions, pager *api.Pager) url.Values {
+	if pager == nil {
+		pager = &api.Pager{Page: 1, Skip: 0, Size: 50}
+	}
+	query := url.Values{}
+	query.Set("offset", strconv.Itoa(pager.Skip))
+	query.Set("limit", strconv.Itoa(pager.Size))
+	if opts.TeamId != 0 {
+		// only the team of this scope, not every team of the Tempo instance
+		query.Set("teamIds", strconv.FormatInt(opts.TeamId, 10))
+	}
+	return query
+}

backend/plugins/tempo/tasks/team_collector_test.go

@@ -0,0 +1,45 @@
+/*
+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 (
+	"testing"
+
+	"github.com/apache/devlake/helpers/pluginhelper/api"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildTeamQuery(t *testing.T) {
+	pager := &api.Pager{Page: 2, Skip: 50, Size: 50}
+
+	t.Run("a team scope only fetches its own team", func(t *testing.T) {
+		q := buildTeamQuery(&TempoOptions{ConnectionId: 1, TeamId: 71}, pager)
+		assert.Equal(t, "71", q.Get("teamIds"))
+		assert.Equal(t, "50", q.Get("offset"))
+		assert.Equal(t, "50", q.Get("limit"))
+		// not a GET /4/teams parameter
+		assert.Empty(t, q.Get("updatedFrom"))
+	})
+
+	t.Run("without a team scope, fetches every team", func(t *testing.T) {
+		q := buildTeamQuery(&TempoOptions{ConnectionId: 1}, nil)
+		assert.Empty(t, q.Get("teamIds"))
+		assert.Equal(t, "0", q.Get("offset"))
+		assert.Equal(t, "50", q.Get("limit"))
+	})
+}

backend/plugins/tempo/tasks/team_extractor.go

@@ -20,6 +20,7 @@ package tasks
 import (
 	"encoding/json"
 
+	"github.com/apache/devlake/core/dal"
 	"github.com/apache/devlake/core/errors"
 	"github.com/apache/devlake/core/plugin"
 	"github.com/apache/devlake/helpers/pluginhelper/api"
@@ -46,30 +47,29 @@ type TempoTeamResponse struct {
 func ExtractTeams(taskCtx plugin.SubTaskContext) errors.Error {
 	data := taskCtx.GetData().(*TempoTaskData)
 
+	// The extracted teams are the team scopes themselves, which the extractor
+	// re-creates: keep the scope config users assigned to them.
+	clauses := []dal.Clause{dal.Where("connection_id = ?", data.Options.ConnectionId)}
+	if data.Options.TeamId != 0 {
+		clauses = append(clauses, dal.Where("team_id = ?", data.Options.TeamId))
+	}
+	var existingTeams []models.TempoTeam
+	err := taskCtx.GetDal().All(&existingTeams, clauses...)
+	if err != nil {
+		return err
+	}
+	scopeConfigIds := make(map[int64]uint64, len(existingTeams))
+	for _, team := range existingTeams {
+		scopeConfigIds[team.TeamId] = team.ScopeConfigId
+	}
+
 	extractor, err := api.NewApiExtractor(api.ApiExtractorArgs{
-		RawDataSubTaskArgs: api.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: models.TempoApiParams{
-				ConnectionId: data.Options.ConnectionId,
-			},
-			Table: RAW_TEAM_TABLE,
-		},
+		RawDataSubTaskArgs: teamRawDataSubTaskArgs(taskCtx, data.Options),
 		Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-			var apiTeam TempoTeamResponse
-			err := errors.Convert(json.Unmarshal(row.Data, &apiTeam))
-			if err != nil {
+			team, err := extractTeam(row.Data, data.Options, scopeConfigIds)
+			if err != nil || team == nil {
 				return nil, err
 			}
-
-			// Transform to tool layer model
-			team := &models.TempoTeam{
-				TeamId:  apiTeam.Id,
-				Key:     apiTeam.Key,
-				Name:    apiTeam.Name,
-				Summary: apiTeam.Summary,
-			}
-			team.ConnectionId = data.Options.ConnectionId
-
 			return []interface{}{team}, nil
 		},
 	})
@@ -80,3 +80,27 @@ func ExtractTeams(taskCtx plugin.SubTaskContext) errors.Error {
 
 	return extractor.Execute()
 }
+
+// extractTeam returns nil for a team other than the team of this scope, so
+// that collecting one team scope doesn't add every Tempo team as a scope.
+func extractTeam(data []byte, opts *TempoOptions, scopeConfigIds map[int64]uint64) (*models.TempoTeam, errors.Error) {
+	var apiTeam TempoTeamResponse
+	err := errors.Convert(json.Unmarshal(data, &apiTeam))
+	if err != nil {
+		return nil, err
+	}
+	if opts.TeamId != 0 && apiTeam.Id != opts.TeamId {
+		return nil, nil
+	}
+
+	// Transform to tool layer model
+	team := &models.TempoTeam{
+		TeamId:  apiTeam.Id,
+		Key:     apiTeam.Key,
+		Name:    apiTeam.Name,
+		Summary: apiTeam.Summary,
+	}
+	team.ConnectionId = opts.ConnectionId
+	team.ScopeConfigId = scopeConfigIds[apiTeam.Id]
+	return team, nil
+}

backend/plugins/tempo/tasks/team_extractor_test.go

@@ -0,0 +1,57 @@
+/*
+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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExtractTeam(t *testing.T) {
+	raw17 := []byte(`{"self":"https://api.tempo.io/4/teams/17","id":17,"name":"Abstract BE Team","summary":"","key":""}`)
+	raw71 := []byte(`{"self":"https://api.tempo.io/4/teams/71","id":71,"name":"SmartShopper (RTL+WHS)","summary":"Microservice Applications","key":""}`)
+	scopeConfigIds := map[int64]uint64{71: 3}
+
+	t.Run("keeps only the team of the scope", func(t *testing.T) {
+		team, err := extractTeam(raw17, &TempoOptions{ConnectionId: 1, TeamId: 71}, scopeConfigIds)
+		assert.Nil(t, err)
+		assert.Nil(t, team)
+	})
+
+	t.Run("keeps the scope config of the team scope", func(t *testing.T) {
+		team, err := extractTeam(raw71, &TempoOptions{ConnectionId: 1, TeamId: 71}, scopeConfigIds)
+		assert.Nil(t, err)
+		if assert.NotNil(t, team) {
+			assert.Equal(t, uint64(1), team.ConnectionId)
+			assert.Equal(t, int64(71), team.TeamId)
+			assert.Equal(t, "SmartShopper (RTL+WHS)", team.Name)
+			assert.Equal(t, "Microservice Applications", team.Summary)
+			assert.Equal(t, uint64(3), team.ScopeConfigId)
+		}
+	})
+
+	t.Run("without a team scope, extracts every team", func(t *testing.T) {
+		team, err := extractTeam(raw17, &TempoOptions{ConnectionId: 1}, scopeConfigIds)
+		assert.Nil(t, err)
+		if assert.NotNil(t, team) {
+			assert.Equal(t, int64(17), team.TeamId)
+			assert.Equal(t, uint64(0), team.ScopeConfigId)
+		}
+	})
+}