Refactor Tekton interaction in webhook interceptor

Use a Tekton client instead of the raw HTTP communication.

This makes the code easier to read, maintain and test. As a consequence
of the change, a huge portion of the webhook interceptor which wasn't
tested so far is now tested.

cmd/webhook-interceptor/main.go

@@ -11,13 +11,13 @@ import (
 	"time"
 
 	"github.com/opendevstack/pipeline/internal/interceptor"
+	tektonClient "github.com/opendevstack/pipeline/internal/tekton"
+	"github.com/opendevstack/pipeline/pkg/bitbucket"
 )
 
 const (
 	namespaceFile    = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
 	namespaceSuffix  = "-cd"
-	apiHostEnvVar    = "API_HOST"
-	apiHostDefault   = "openshift.default.svc.cluster.local"
 	repoBaseEnvVar   = "REPO_BASE"
 	tokenEnvVar      = "ACCESS_TOKEN"
 	taskKindEnvVar   = "ODS_TASK_KIND"
@@ -69,37 +69,40 @@ func serve() error {
 		)
 	}
 
-	apiHost := os.Getenv(apiHostEnvVar)
-	if len(apiHost) == 0 {
-		apiHost = apiHostDefault
-		log.Println(
-			"INFO:",
-			apiHostEnvVar,
-			"not set, using default value:",
-			apiHostDefault,
-		)
-	}
-
 	namespace, err := getFileContent(namespaceFile)
 	if err != nil {
-		return fmt.Errorf("%w", err)
+		return err
 	}
 
 	project := strings.TrimSuffix(namespace, namespaceSuffix)
 
-	client, err := interceptor.NewClient(apiHost, namespace)
+	// Initialize Tekton client.
+	client, err := tektonClient.NewInCluserClient(&tektonClient.ClientConfig{
+		Namespace: namespace,
+	})
 	if err != nil {
-		return fmt.Errorf("%w", err)
+		return err
 	}
 
-	server := interceptor.NewServer(client, interceptor.ServerConfig{
-		Namespace:  namespace,
-		Project:    project,
-		RepoBase:   repoBase,
-		Token:      token,
-		TaskKind:   taskKind,
-		TaskSuffix: taskSuffix,
+	// Initialize Bitbucket client.
+	bitbucketClient := bitbucket.NewClient(&bitbucket.ClientConfig{
+		APIToken: token,
+		BaseURL:  strings.TrimSuffix(repoBase, "/scm"),
+	})
+
+	server, err := interceptor.NewServer(interceptor.ServerConfig{
+		Namespace:       namespace,
+		Project:         project,
+		RepoBase:        repoBase,
+		Token:           token,
+		TaskKind:        taskKind,
+		TaskSuffix:      taskSuffix,
+		BitbucketClient: bitbucketClient,
+		TektonClient:    client,
 	})
+	if err != nil {
+		return err
+	}
 
 	log.Println("Ready to accept requests")
 

internal/interceptor/server.go

@@ -3,6 +3,7 @@ package interceptor
 import (
 	"crypto/sha1"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"log"
@@ -14,11 +15,11 @@ import (
 	"time"
 
 	intrepo "github.com/opendevstack/pipeline/internal/repository"
+	tektonClient "github.com/opendevstack/pipeline/internal/tekton"
 	"github.com/opendevstack/pipeline/pkg/bitbucket"
 	"github.com/opendevstack/pipeline/pkg/config"
 	tekton "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"sigs.k8s.io/yaml"
 )
 
 const (
@@ -29,18 +30,20 @@ const (
 	repositoryLabel = labelPrefix + "repository"
 	// Label specifying the Git ref (e.g. branch) related to the pipeline.
 	gitRefLabel = labelPrefix + "git-ref"
+	// letterBytes contains letters to use for random strings.
+	letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
 )
 
 // Server represents this service, and is a global.
 type Server struct {
-	OpenShiftClient Client
+	TektonClient    tektonClient.ClientInterface
 	Namespace       string
 	Project         string
 	RepoBase        string
 	Token           string
 	TaskKind        tekton.TaskKind
 	TaskSuffix      string
-	BitbucketClient *bitbucket.Client
+	BitbucketClient bitbucketInterface
 }
 
 // ServerConfig configures a server.
@@ -58,11 +61,14 @@ type ServerConfig struct {
 	TaskKind string
 	// TaskSuffic is the suffix applied to tasks (version information).
 	TaskSuffix string
+	// TektonClient is a tekton client
+	TektonClient tektonClient.ClientInterface
+	// BitbucketClient is a Bitbucket client
+	BitbucketClient bitbucketInterface
 }
 
 type PipelineData struct {
 	Name            string `json:"name"`
-	ResourceVersion int    `json:"resourceVersion"`
 	Project         string `json:"project"`
 	Component       string `json:"component"`
 	Repository      string `json:"repository"`
@@ -81,26 +87,39 @@ type PipelineData struct {
 	PullRequestBase string `json:"prBase"`
 }
 
+type bitbucketInterface interface {
+	bitbucket.CommitClientInterface
+	bitbucket.RawClientInterface
+}
+
 func init() {
 	rand.Seed(time.Now().UnixNano())
 }
 
 // NewServer returns a new server.
-func NewServer(client Client, serverConfig ServerConfig) *Server {
-	bitbucketClient := bitbucket.NewClient(&bitbucket.ClientConfig{
-		APIToken: serverConfig.Token,
-		BaseURL:  strings.TrimSuffix(serverConfig.RepoBase, "/scm"),
-	})
+func NewServer(serverConfig ServerConfig) (*Server, error) {
+	if serverConfig.Namespace == "" {
+		return nil, errors.New("namespace is required")
+	}
+	if serverConfig.Token == "" {
+		return nil, errors.New("token is required")
+	}
+	if serverConfig.TektonClient == nil {
+		return nil, errors.New("tekton client is required")
+	}
+	if serverConfig.BitbucketClient == nil {
+		return nil, errors.New("bitbucket client is required")
+	}
 	return &Server{
-		OpenShiftClient: client,
+		TektonClient:    serverConfig.TektonClient,
 		Namespace:       serverConfig.Namespace,
 		Project:         serverConfig.Project,
 		RepoBase:        serverConfig.RepoBase,
 		Token:           serverConfig.Token,
 		TaskKind:        tekton.TaskKind(serverConfig.TaskKind),
 		TaskSuffix:      serverConfig.TaskSuffix,
-		BitbucketClient: bitbucketClient,
-	}
+		BitbucketClient: serverConfig.BitbucketClient,
+	}, nil
 }
 
 type repository struct {
@@ -135,14 +154,12 @@ type requestBitbucket struct {
 	} `json:"comment"`
 }
 
-// HandleRoot handles all requests to this service.
+// HandleRoot handles all requests to this service. It performs the following:
+// - extract pipeline data from request body
+// - extend body with calculated pipeline information
+// - create/update pipeline that will be triggerd by event listener
 func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 
-	// read request body into Go object
-	// extract pipeline data
-	// extend body with data
-	// create/update pipeline
-
 	requestID := randStringBytes(6)
 	log.Println(requestID, "---START---")
 
@@ -223,28 +240,20 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	)
 
 	pipelineName := makePipelineName(component, gitRef)
-	resourceVersion, err := s.OpenShiftClient.GetPipelineResourceVersion(pipelineName)
-	if err != nil {
-		msg := "Could not retrieve pipeline resourceVersion"
-		log.Println(requestID, fmt.Sprintf("%s: %s", msg, err))
-		http.Error(w, msg, 500)
-		return
-	}
 
 	pData := PipelineData{
-		Name:            pipelineName,
-		Project:         project,
-		Component:       component,
-		Repository:      repo,
-		GitRef:          gitRef,
-		GitFullRef:      gitFullRef,
-		ResourceVersion: resourceVersion,
-		RepoBase:        s.RepoBase,
-		GitURI:          gitURI,
-		Namespace:       s.Namespace,
-		PVC:             "ods-pipeline",
-		TriggerEvent:    req.EventKey,
-		Comment:         commentText,
+		Name:         pipelineName,
+		Project:      project,
+		Component:    component,
+		Repository:   repo,
+		GitRef:       gitRef,
+		GitFullRef:   gitFullRef,
+		RepoBase:     s.RepoBase,
+		GitURI:       gitURI,
+		Namespace:    s.Namespace,
+		PVC:          "ods-pipeline",
+		TriggerEvent: req.EventKey,
+		Comment:      commentText,
 	}
 
 	if len(commitSHA) == 0 {
@@ -295,27 +304,22 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	pData.Environment = selectEnvironmentFromMapping(odsConfig.BranchToEnvironmentMapping, pData.GitRef)
 	pData.Version = odsConfig.Version
 
-	rendered, err := renderPipeline(odsConfig, pData, s.TaskKind, s.TaskSuffix)
-	if err != nil {
-		msg := "Could not render pipeline definition"
-		log.Println(requestID, fmt.Sprintf("%s: %s", msg, err))
-		http.Error(w, msg, 500)
-		return
-	}
+	tknPipeline := assemblePipeline(odsConfig, pData, s.TaskKind, s.TaskSuffix)
 
-	jsonBytes, err := yaml.YAMLToJSON(rendered)
+	_, err = s.TektonClient.GetPipeline(r.Context(), pData.Name, metav1.GetOptions{})
 	if err != nil {
-		msg := "could not convert YAML to JSON"
+		_, err := s.TektonClient.CreatePipeline(r.Context(), tknPipeline, metav1.CreateOptions{})
+		if err != nil {
+			msg := fmt.Sprintf("cannot create pipeline %s", tknPipeline.Name)
+			log.Println(requestID, fmt.Sprintf("%s: %s", msg, err))
+			http.Error(w, msg, http.StatusInternalServerError)
+			return
+		}
+	} else {
+		_, err := s.TektonClient.UpdatePipeline(r.Context(), tknPipeline, metav1.UpdateOptions{})
+		msg := fmt.Sprintf("cannot update pipeline %s", tknPipeline.Name)
 		log.Println(requestID, fmt.Sprintf("%s: %s", msg, err))
-		http.Error(w, msg, 500)
-		return
-	}
-
-	createStatusCode, createErr := s.OpenShiftClient.ApplyPipeline(jsonBytes, pData)
-	if createErr != nil {
-		msg := "Could not create/update pipeline"
-		log.Println(requestID, fmt.Sprintf("%s [%d]: %s", msg, createStatusCode, createErr))
-		http.Error(w, msg, createStatusCode)
+		http.Error(w, msg, http.StatusInternalServerError)
 		return
 	}
 
@@ -337,22 +341,29 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 
 func selectEnvironmentFromMapping(mapping []config.BranchToEnvironmentMapping, branch string) string {
 	for _, bem := range mapping {
-		// exact match
-		if bem.Branch == branch {
+		if mappingBranchMatch(bem.Branch, branch) {
 			return bem.Environment
 		}
-		// prefix match like "release/*", also catches "*"
-		if strings.HasSuffix(bem.Branch, "*") {
-			branchPrefix := strings.TrimSuffix(bem.Branch, "*")
-			if strings.HasPrefix(branch, branchPrefix) {
-				return bem.Environment
-			}
-		}
 	}
 	return ""
 }
 
-func getCommitSHA(bitbucketClient *bitbucket.Client, project, repository, gitFullRef string) (string, error) {
+func mappingBranchMatch(mappingBranch, testBranch string) bool {
+	// exact match
+	if mappingBranch == testBranch {
+		return true
+	}
+	// prefix match like "release/*", also catches "*"
+	if strings.HasSuffix(mappingBranch, "*") {
+		branchPrefix := strings.TrimSuffix(mappingBranch, "*")
+		if strings.HasPrefix(testBranch, branchPrefix) {
+			return true
+		}
+	}
+	return false
+}
+
+func getCommitSHA(bitbucketClient bitbucket.CommitClientInterface, project, repository, gitFullRef string) (string, error) {
 	commitList, err := bitbucketClient.CommitList(project, repository, bitbucket.CommitListParams{
 		Until: gitFullRef,
 	})
@@ -412,10 +423,10 @@ func makePipelineName(component string, branch string) string {
 		s := fmt.Sprintf("%x", bs)
 		pipeline = fmt.Sprintf("%s-%s", shortenedPipeline, s[0:7])
 	}
-	return pipeline
+	return strings.ToLower(pipeline)
 }
 
-func renderPipeline(odsConfig *config.ODS, data PipelineData, taskKind tekton.TaskKind, taskSuffix string) ([]byte, error) {
+func assemblePipeline(odsConfig *config.ODS, data PipelineData, taskKind tekton.TaskKind, taskSuffix string) *tekton.Pipeline {
 
 	var tasks []tekton.PipelineTask
 	tasks = append(tasks, tekton.PipelineTask{
@@ -515,10 +526,9 @@ func renderPipeline(odsConfig *config.ODS, data PipelineData, taskKind tekton.Ta
 		},
 	})
 
-	p := tekton.Pipeline{
+	p := &tekton.Pipeline{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:            data.Name,
-			ResourceVersion: strconv.Itoa(data.ResourceVersion),
+			Name: data.Name,
 			Labels: map[string]string{
 				repositoryLabel: data.Repository,
 				gitRefLabel:     data.GitRef,
@@ -613,16 +623,15 @@ func renderPipeline(odsConfig *config.ODS, data PipelineData, taskKind tekton.Ta
 			Finally: finallyTasks,
 		},
 	}
-
-	return yaml.Marshal(p)
+	return p
 }
 
 type prInfo struct {
 	ID   int
 	Base string
 }
 
-func extractPullRequestInfo(bitbucketClient *bitbucket.Client, projectKey, repositorySlug, gitCommit string) (prInfo, error) {
+func extractPullRequestInfo(bitbucketClient bitbucket.CommitClientInterface, projectKey, repositorySlug, gitCommit string) (prInfo, error) {
 	var i prInfo
 
 	prPage, err := bitbucketClient.CommitPullRequestList(projectKey, repositorySlug, gitCommit)
@@ -642,7 +651,7 @@ func extractPullRequestInfo(bitbucketClient *bitbucket.Client, projectKey, repos
 	return i, nil
 }
 
-func shouldSkip(bitbucketClient *bitbucket.Client, projectKey, repositorySlug, gitCommit string) bool {
+func shouldSkip(bitbucketClient bitbucket.CommitClientInterface, projectKey, repositorySlug, gitCommit string) bool {
 	c, err := bitbucketClient.CommitGet(projectKey, repositorySlug, gitCommit)
 	if err != nil {
 		return false
@@ -665,3 +674,12 @@ func isCiSkipInCommitMessage(message string) bool {
 		strings.Contains(subject, "[skipci]") ||
 		strings.Contains(subject, "***noci***")
 }
+
+// randStringBytes creates a random string of length n.
+func randStringBytes(n int) string {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = letterBytes[rand.Intn(len(letterBytes))]
+	}
+	return string(b)
+}

internal/interceptor/server_test.go

@@ -10,11 +10,16 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"unicode"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/opendevstack/pipeline/internal/projectpath"
+	tektonClient "github.com/opendevstack/pipeline/internal/tekton"
 	"github.com/opendevstack/pipeline/internal/testfile"
+	"github.com/opendevstack/pipeline/pkg/bitbucket"
 	"github.com/opendevstack/pipeline/pkg/config"
+	tekton "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/yaml"
 )
 
@@ -28,7 +33,6 @@ func TestRenderPipeline(t *testing.T) {
 		GitRef:          "main",
 		GitFullRef:      "refs/heads/main",
 		GitSHA:          "ef8755f06ee4b28c96a847a95cb8ec8ed6ddd1ca",
-		ResourceVersion: 0,
 		RepoBase:        "https://bitbucket.acme.org",
 		GitURI:          "https://bitbucket.acme.org/scm/foo/bar.git",
 		Namespace:       "foo-cd",
@@ -64,7 +68,6 @@ func TestExtensions(t *testing.T) {
 		GitRef:          "main",
 		GitFullRef:      "refs/heads/main",
 		GitSHA:          "ef8755f06ee4b28c96a847a95cb8ec8ed6ddd1ca",
-		ResourceVersion: 0,
 		RepoBase:        "https://bitbucket.acme.org",
 		GitURI:          "https://bitbucket.acme.org/scm/foo/bar.git",
 		PVC:             "pipeline-bar",
@@ -107,6 +110,45 @@ func TestIsCiSkipInCommitMessage(t *testing.T) {
 	}
 }
 
+func TestMakePipelineName(t *testing.T) {
+	tests := map[string]struct {
+		component string
+		branch    string
+		expected  string
+	}{
+		"branch contains non-alphanumeric characters": {
+			component: "comp",
+			branch:    "bugfix/prj-529-bar-6-baz",
+			expected:  "comp-bugfix-prj-529-bar-6-baz",
+		},
+		"branch contains uppercase characters": {
+			component: "comp",
+			branch:    "PRJ-529-bar-6-baz",
+			expected:  "comp-prj-529-bar-6-baz",
+		},
+		"branch name is too long": {
+			component: "comp",
+			branch:    "bugfix/some-arbitarily-long-branch-name-that-should-be-way-shorter",
+			expected:  "comp-bugfix-some-arbitarily-long-branch-name-th-87136df",
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			got := makePipelineName(tc.component, tc.branch)
+			if tc.expected != got {
+				t.Fatalf(
+					"Want '%s', got '%s' for (component='%s', branch='%s')",
+					tc.expected,
+					got,
+					tc.component,
+					tc.branch,
+				)
+			}
+		})
+	}
+}
+
 func TestSelectEnvironmentFromMapping(t *testing.T) {
 	tests := []struct {
 		mapping []config.BranchToEnvironmentMapping
@@ -170,50 +212,169 @@ func fatalIfErr(t *testing.T, err error) {
 	}
 }
 
-type mockClient struct {
-}
-
-func (c *mockClient) GetPipelineResourceVersion(name string) (int, error) {
-	return 200, nil
-}
-func (c *mockClient) ApplyPipeline(pipelineBody []byte, data PipelineData) (int, error) {
-	return 200, nil
-}
-
-func testServer() (*httptest.Server, *mockClient) {
-	mc := &mockClient{}
-	server := NewServer(mc, ServerConfig{
-		Namespace: "bar-cd",
-		Project:   "bar",
-		Token:     "",
-		TaskKind:  "ClusterTask",
-		RepoBase:  "https://domain.com",
+func testServer(tc tektonClient.ClientInterface, bc bitbucketInterface) (*httptest.Server, error) {
+	server, err := NewServer(ServerConfig{
+		Namespace:       "bar-cd",
+		Project:         "bar",
+		Token:           "test",
+		TaskKind:        "ClusterTask",
+		RepoBase:        "https://domain.com",
+		TektonClient:    tc,
+		BitbucketClient: bc,
 	})
-	return httptest.NewServer(http.HandlerFunc(server.HandleRoot)), mc
+	if err != nil {
+		return nil, err
+	}
+	return httptest.NewServer(http.HandlerFunc(server.HandleRoot)), nil
 }
 
 func TestServer(t *testing.T) {
-	ts, _ := testServer()
-	defer ts.Close()
 
-	tests := []struct {
+	tests := map[string]struct {
 		requestBodyFixture string
+		tektonClient       *tektonClient.TestClient
+		bitbucketClient    *bitbucket.TestClient
 		wantStatus         int
 		wantBody           string
+		check              func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient)
 	}{
-		{
+		"invalid JSON is not processed": {
+			requestBodyFixture: "interceptor/payload-invalid.json",
+			wantStatus:         http.StatusBadRequest,
+			wantBody:           "cannot parse JSON: invalid character '\\n' in string literal",
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) > 0 || len(tc.UpdatedPipelines) > 0 {
+					t.Fatal("no pipeline should have been created/updated")
+				}
+			},
+		},
+		"unsupported events are not processed": {
 			requestBodyFixture: "interceptor/payload-unknown-event.json",
 			wantStatus:         http.StatusBadRequest,
 			wantBody:           "Unsupported event key: repo:ref_changed",
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) > 0 || len(tc.UpdatedPipelines) > 0 {
+					t.Fatal("no pipeline should have been created/updated")
+				}
+			},
 		},
-		{
+		"tags are not processed": {
 			requestBodyFixture: "interceptor/payload-tag.json",
 			wantStatus:         http.StatusTeapot,
 			wantBody:           "Skipping change ref type TAG, only BRANCH is supported",
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) > 0 || len(tc.UpdatedPipelines) > 0 {
+					t.Fatal("no pipeline should have been created/updated")
+				}
+			},
+		},
+		"commits with skip message are not processed": {
+			requestBodyFixture: "interceptor/payload.json",
+			bitbucketClient: &bitbucket.TestClient{
+				Commits: []bitbucket.Commit{
+					{
+						// commit referenced in payload
+						ID:      "0e183aa3bc3c6deb8f40b93fb2fc4354533cf62f",
+						Message: "Update readme [ci skip]",
+					},
+				},
+			},
+			wantStatus: http.StatusTeapot,
+			wantBody:   "Commit should be skipped",
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) > 0 || len(tc.UpdatedPipelines) > 0 {
+					t.Fatal("no pipeline should have been created/updated")
+				}
+			},
+		},
+		"pushes into new branch creates a pipeline": {
+			requestBodyFixture: "interceptor/payload.json",
+			bitbucketClient: &bitbucket.TestClient{
+				Files: map[string][]byte{
+					"ods.yaml": readTestdataFile(t, "fixtures/interceptor/ods.yaml"),
+				},
+			},
+			wantStatus: http.StatusOK,
+			wantBody:   string(readTestdataFile(t, "golden/interceptor/extended-payload.json")),
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) != 1 || len(tc.UpdatedPipelines) != 0 {
+					t.Fatal("exactly one pipeline should have been created")
+				}
+			},
+		},
+		"pushes into an existing branch updates a pipeline": {
+			requestBodyFixture: "interceptor/payload.json",
+			bitbucketClient: &bitbucket.TestClient{
+				Files: map[string][]byte{
+					"ods.yaml": readTestdataFile(t, "fixtures/interceptor/ods.yaml"),
+				},
+			},
+			tektonClient: &tektonClient.TestClient{
+				Pipelines: []*tekton.Pipeline{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							// generated pipeline name
+							Name: "ods-pipeline-master",
+						},
+					},
+				},
+			},
+			wantStatus: http.StatusOK,
+			wantBody:   string(readTestdataFile(t, "golden/interceptor/extended-payload.json")),
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) != 0 || len(tc.UpdatedPipelines) != 1 {
+					t.Fatal("exactly one pipeline should have been updated")
+				}
+			},
+		},
+		"PR open events update a pipeline": {
+			requestBodyFixture: "interceptor/payload-pr-opened.json",
+			bitbucketClient: &bitbucket.TestClient{
+				Files: map[string][]byte{
+					"ods.yaml": readTestdataFile(t, "fixtures/interceptor/ods.yaml"),
+				},
+				PullRequests: []bitbucket.PullRequest{
+					{
+						Open: true,
+						ID:   1,
+						ToRef: bitbucket.Ref{
+							ID: "refs/heads/master",
+						},
+					},
+				},
+			},
+			tektonClient: &tektonClient.TestClient{
+				Pipelines: []*tekton.Pipeline{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							// generated pipeline name
+							Name: "ods-pipeline-feature-foo",
+						},
+					},
+				},
+			},
+			wantStatus: http.StatusOK,
+			wantBody:   string(readTestdataFile(t, "golden/interceptor/extended-payload-pr-opened.json")),
+			check: func(t *testing.T, tc *tektonClient.TestClient, bc *bitbucket.TestClient) {
+				if len(tc.CreatedPipelines) != 0 || len(tc.UpdatedPipelines) != 1 {
+					t.Fatal("exactly one pipeline should have been updated")
+				}
+			},
 		},
 	}
-	for i, tc := range tests {
-		t.Run(fmt.Sprintf("mapping #%d", i), func(t *testing.T) {
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			if tc.tektonClient == nil {
+				tc.tektonClient = &tektonClient.TestClient{}
+			}
+			if tc.bitbucketClient == nil {
+				tc.bitbucketClient = &bitbucket.TestClient{}
+			}
+			ts, err := testServer(tc.tektonClient, tc.bitbucketClient)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer ts.Close()
 			filename := filepath.Join(projectpath.Root, "test/testdata/fixtures", tc.requestBodyFixture)
 			f, err := os.Open(filename)
 			if err != nil {
@@ -231,10 +392,35 @@ func TestServer(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			gotBody := strings.TrimSpace(string(gotBodyBytes))
-			if tc.wantBody != gotBody {
-				t.Fatalf("Got body: %v, want: %v", gotBody, tc.wantBody)
+			gotBody := removeSpace(string(gotBodyBytes))
+			if diff := cmp.Diff(removeSpace(tc.wantBody), gotBody); diff != "" {
+				t.Fatalf("body mismatch (-want +got):\n%s", diff)
+			}
+			if tc.check != nil {
+				tc.check(t, tc.tektonClient, tc.bitbucketClient)
 			}
 		})
 	}
 }
+
+func renderPipeline(odsConfig *config.ODS, data PipelineData, taskKind tekton.TaskKind, taskSuffix string) ([]byte, error) {
+	p := assemblePipeline(odsConfig, data, taskKind, taskSuffix)
+	return yaml.Marshal(p)
+}
+
+func removeSpace(str string) string {
+	return strings.Map(func(r rune) rune {
+		if unicode.IsSpace(r) {
+			return -1
+		}
+		return r
+	}, str)
+}
+
+func readTestdataFile(t *testing.T, filename string) []byte {
+	b, err := ioutil.ReadFile(filepath.Join(projectpath.Root, "test/testdata", filename))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}

internal/tekton/client.go

@@ -0,0 +1,81 @@
+package tekton
+
+import (
+	"errors"
+
+	"github.com/opendevstack/pipeline/pkg/logging"
+	tektonClient "github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
+	v1beta1 "github.com/tektoncd/pipeline/pkg/client/clientset/versioned/typed/pipeline/v1beta1"
+	"k8s.io/client-go/rest"
+)
+
+// Client represents a Tekton client, wrapping
+// github.com/tektoncd/pipeline/pkg/client/clientset/versioned.Clientset
+type Client struct {
+	clientConfig *ClientConfig
+}
+
+// ClientConfig configures a Tekton client.
+type ClientConfig struct {
+	// Kubernetes namespace.
+	Namespace string
+	// Logger is the logger to send logging messages to.
+	Logger logging.LeveledLoggerInterface
+	// HTTP client used to download assets.
+	TektonClient *tektonClient.Clientset
+}
+
+type ClientInterface interface {
+	ClientPipelineInterface
+}
+
+// NewInCluserClient initializes a Tekton client from within a cluster.
+func NewInCluserClient(clientConfig *ClientConfig) (*Client, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+	// create the Tekton clientset
+	tektonClientSet, err := tektonClient.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	clientConfig.TektonClient = tektonClientSet
+	return NewClient(clientConfig)
+}
+
+// NewClient initializes a Tekton client.
+func NewClient(clientConfig *ClientConfig) (*Client, error) {
+	if clientConfig.Namespace == "" {
+		return nil, errors.New("namespace is required")
+	}
+
+	if clientConfig.TektonClient == nil {
+		return nil, errors.New("tekton client is required")
+	}
+
+	// Be careful not to pass a variable of type *logging.LeveledLogger
+	// holding a nil value. If you pass nil for Logger, make sure it is of
+	// logging.LeveledLoggerInterface type.
+	if clientConfig.Logger == nil {
+		clientConfig.Logger = &logging.LeveledLogger{Level: logging.LevelError}
+	}
+
+	return &Client{clientConfig: clientConfig}, nil
+}
+
+func (c *Client) logger() logging.LeveledLoggerInterface {
+	return c.clientConfig.Logger
+}
+
+func (c *Client) namespace() string {
+	return c.clientConfig.Namespace
+}
+
+func (c *Client) tektonV1beta1Client() v1beta1.TektonV1beta1Interface {
+	return c.clientConfig.TektonClient.TektonV1beta1()
+}
+
+func (c *Client) pipelinesClient() v1beta1.PipelineInterface {
+	return c.tektonV1beta1Client().Pipelines(c.namespace())
+}

internal/tekton/pipeline.go

@@ -0,0 +1,29 @@
+package tekton
+
+import (
+	"context"
+
+	tekton "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type ClientPipelineInterface interface {
+	GetPipeline(ctxt context.Context, name string, options metav1.GetOptions) (*tekton.Pipeline, error)
+	CreatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.CreateOptions) (*tekton.Pipeline, error)
+	UpdatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.UpdateOptions) (*tekton.Pipeline, error)
+}
+
+func (c *Client) GetPipeline(ctxt context.Context, name string, options metav1.GetOptions) (*tekton.Pipeline, error) {
+	c.logger().Debugf("Get pipeline %s", name)
+	return c.pipelinesClient().Get(ctxt, name, options)
+}
+
+func (c *Client) CreatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.CreateOptions) (*tekton.Pipeline, error) {
+	c.logger().Debugf("Create pipeline %s", pipeline.Name)
+	return c.pipelinesClient().Create(ctxt, pipeline, options)
+}
+
+func (c *Client) UpdatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.UpdateOptions) (*tekton.Pipeline, error) {
+	c.logger().Debugf("Update pipeline %s", pipeline.Name)
+	return c.pipelinesClient().Update(ctxt, pipeline, options)
+}

internal/tekton/test_client.go

@@ -0,0 +1,38 @@
+package tekton
+
+import (
+	"context"
+	"fmt"
+
+	tekton "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// TestClient returns mocked pipelines.
+type TestClient struct {
+	// Pipelines is the pool of pipelines which can be retrieved.
+	Pipelines []*tekton.Pipeline
+	// CreatedPipelines is a slice of created pipeline names.
+	CreatedPipelines []string
+	// UpdatedPipelines is a slice of updated pipeline names.
+	UpdatedPipelines []string
+}
+
+func (c *TestClient) GetPipeline(ctxt context.Context, name string, options metav1.GetOptions) (*tekton.Pipeline, error) {
+	for _, p := range c.Pipelines {
+		if p.Name == name {
+			return p, nil
+		}
+	}
+	return nil, fmt.Errorf("pipeline %s not found", name)
+}
+
+func (c *TestClient) CreatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.CreateOptions) (*tekton.Pipeline, error) {
+	c.CreatedPipelines = append(c.CreatedPipelines, pipeline.Name)
+	return pipeline, nil
+}
+
+func (c *TestClient) UpdatePipeline(ctxt context.Context, pipeline *tekton.Pipeline, options metav1.UpdateOptions) (*tekton.Pipeline, error) {
+	c.UpdatedPipelines = append(c.UpdatedPipelines, pipeline.Name)
+	return pipeline, nil
+}

pkg/bitbucket/commits.go

@@ -109,6 +109,12 @@ type CommitListParams struct {
 	Until string `json:"until"`
 }
 
+type CommitClientInterface interface {
+	CommitList(projectKey string, repositorySlug string, params CommitListParams) (*CommitPage, error)
+	CommitGet(projectKey, repositorySlug, commitID string) (*Commit, error)
+	CommitPullRequestList(projectKey, repositorySlug, commitID string) (*PullRequestPage, error)
+}
+
 // CommitList retrieves a page of commits from a given starting commit or "between" two commits. If no explicit commit is specified, the tip of the repository's default branch is assumed. commits may be identified by branch or tag name or by ID. A path may be supplied to restrict the returned commits to only those which affect that path.
 // The authenticated user must have REPO_READ permission for the specified repository to call this resource.
 // https://docs.atlassian.com/bitbucket-server/rest/7.13.0/bitbucket-rest.html#idp222

pkg/bitbucket/test_client.go

@@ -7,8 +7,10 @@ import (
 
 // TestClient returns mocked branches and tags.
 type TestClient struct {
-	Branches []Branch
-	Tags     []Tag
+	Branches     []Branch
+	Tags         []Tag
+	Commits      []Commit
+	PullRequests []PullRequest
 	// Files contains byte slices for filenames
 	Files map[string][]byte
 }
@@ -44,3 +46,20 @@ func (c *TestClient) RawGet(project, repository, filename, gitFullRef string) ([
 	}
 	return nil, fmt.Errorf("%s not found", filename)
 }
+
+func (c *TestClient) CommitList(projectKey string, repositorySlug string, params CommitListParams) (*CommitPage, error) {
+	return &CommitPage{Values: c.Commits}, nil
+}
+
+func (c *TestClient) CommitGet(projectKey, repositorySlug, commitID string) (*Commit, error) {
+	for _, co := range c.Commits {
+		if co.ID == commitID {
+			return &co, nil
+		}
+	}
+	return nil, fmt.Errorf("no commit %s", commitID)
+}
+
+func (c *TestClient) CommitPullRequestList(projectKey, repositorySlug, commitID string) (*PullRequestPage, error) {
+	return &PullRequestPage{Values: c.PullRequests}, nil
+}

test/testdata/fixtures/interceptor/payload-invalid.json

@@ -0,0 +1,11 @@
+{
+    "eventKey": "repo:ref_changed",
+    "date": "2021-01-15T13:29:05+0000",
+    "actor": {
+        "name": "max.mustermann@acme.org",
+        "emailAddress": "max.mustermann@acme.org",
+        "id": 4653,
+        "displayName": "Mustermann,Max ACME",
+        "active": true,
+        "slug": "max.mustermann_acme.org",
+        "broken-on-purpose...

test/testdata/fixtures/interceptor/payload-pr-opened.json

@@ -0,0 +1,93 @@
+{  
+    "eventKey":"pr:opened",
+    "date":"2017-09-19T09:58:11+1000",
+    "actor":{  
+      "name":"max.mustermann@acme.org",
+      "emailAddress":"max.mustermann@acme.org",
+      "id":1,
+      "displayName":"Mustermann,Max ACME",
+      "active":true,
+      "slug":"max.mustermann_acme.org",
+      "type":"NORMAL"
+    },
+    "pullRequest":{  
+      "id":1,
+      "version":0,
+      "title":"a new file added",
+      "state":"OPEN",
+      "open":true,
+      "closed":false,
+      "createdDate":1505779091796,
+      "updatedDate":1505779091796,
+      "fromRef":{  
+        "id":"refs/heads/feature/foo",
+        "displayId":"feature/foo",
+        "latestCommit":"ef8755f06ee4b28c96a847a95cb8ec8ed6ddd1ca",
+        "repository":{  
+          "slug":"ods-pipeline",
+          "id":84,
+          "name":"ods-pipeline",
+          "scmId":"git",
+          "state":"AVAILABLE",
+          "statusMessage":"Available",
+          "forkable":true,
+          "project":{  
+            "key":"FOO",
+            "id":84,
+            "name":"Max Mustermann Playground",
+            "public":false,
+            "type":"NORMAL"
+          },
+          "public":false
+        }
+      },
+      "toRef":{  
+        "id":"refs/heads/master",
+        "displayId":"master",
+        "latestCommit":"178864a7d521b6f5e720b386b2c2b0ef8563e0dc",
+        "repository":{  
+          "slug":"ods-pipeline",
+          "id":84,
+          "name":"ods-pipeline",
+          "scmId":"git",
+          "state":"AVAILABLE",
+          "statusMessage":"Available",
+          "forkable":true,
+          "project":{  
+            "key":"FOO",
+            "id":84,
+            "name":"Max Mustermann Playground",
+            "public":false,
+            "type":"NORMAL"
+          },
+          "public":false
+        }
+      },
+      "locked":false,
+      "author":{  
+        "user":{  
+          "name":"max.mustermann@acme.org",
+          "emailAddress":"max.mustermann@acme.org",
+          "id":1,
+          "displayName":"Mustermann,Max ACME",
+          "active":true,
+          "slug":"max.mustermann_acme.org",
+          "type":"NORMAL"
+        },
+        "role":"AUTHOR",
+        "approved":false,
+        "status":"UNAPPROVED"
+      },
+      "reviewers":[  
+
+      ],
+      "participants":[  
+
+      ],
+      "links":{  
+        "self":[  
+          null
+        ]
+      }
+    }
+  }

test/testdata/golden/interceptor/extended-payload-pr-opened.json

@@ -0,0 +1,108 @@
+{
+    "actor":{
+        "active":true,
+        "displayName":"Mustermann,Max ACME",
+        "emailAddress":"max.mustermann@acme.org",
+        "id":1,
+        "name":"max.mustermann@acme.org",
+        "slug":"max.mustermann_acme.org",
+        "type":"NORMAL"
+    },
+    "date":"2017-09-19T09:58:11+1000",
+    "eventKey":"pr:opened",
+    "extensions": {
+        "name": "ods-pipeline-feature-foo",
+        "project": "foo",
+        "component": "ods-pipeline",
+        "repository": "ods-pipeline",
+        "environment": "",
+        "version": "",
+        "gitRef": "feature/foo",
+        "gitFullRef": "refs/heads/feature/foo",
+        "gitSha": "ef8755f06ee4b28c96a847a95cb8ec8ed6ddd1ca",
+        "repoBase": "https://domain.com",
+        "gitURI": "https://domain.com/foo/ods-pipeline.git",
+        "namespace": "bar-cd",
+        "pvc": "ods-pipeline",
+        "trigger-event": "pr:opened",
+        "comment": "",
+        "prKey": 1,
+        "prBase": "refs/heads/master"
+    },
+    "pullRequest":{
+        "author":{
+            "approved":false,
+            "role":"AUTHOR",
+            "status":"UNAPPROVED",
+            "user":{
+                "active":true,
+                "displayName":"Mustermann,Max ACME",
+                "emailAddress":"max.mustermann@acme.org",
+                "id":1,
+                "name":"max.mustermann@acme.org",
+                "slug":"max.mustermann_acme.org",
+                "type":"NORMAL"
+            }
+        },
+        "closed":false,
+        "createdDate":1505779091796,
+        "fromRef":{
+            "displayId":"feature/foo",
+            "id":"refs/heads/feature/foo",
+            "latestCommit":"ef8755f06ee4b28c96a847a95cb8ec8ed6ddd1ca",
+            "repository":{
+                "forkable":true,
+                "id":84,
+                "name":"ods-pipeline",
+                "project":{
+                    "id":84,
+                    "key":"FOO",
+                    "name":"Max Mustermann Playground",
+                    "public":false,
+                    "type":"NORMAL"
+                },
+                "public":false,
+                "scmId":"git",
+                "slug":"ods-pipeline",
+                "state":"AVAILABLE",
+                "statusMessage":"Available"
+            }
+          },
+        "id":1,
+        "links":{
+            "self":[
+                null
+            ]
+        },
+        "locked":false,
+        "open":true,
+        "participants":[],
+        "reviewers":[],
+        "state":"OPEN",
+        "title":"a new file added",
+        "toRef":{
+            "displayId":"master",
+            "id":"refs/heads/master",
+            "latestCommit":"178864a7d521b6f5e720b386b2c2b0ef8563e0dc",
+            "repository":{
+                "forkable":true,
+                "id":84,
+                "name":"ods-pipeline",
+                "project":{
+                    "id":84,
+                    "key":"FOO",
+                    "name":"Max Mustermann Playground",
+                    "public":false,
+                    "type":"NORMAL"
+                },
+                "public":false,
+                "scmId":"git",
+                "slug":"ods-pipeline",
+                "state":"AVAILABLE",
+                "statusMessage":"Available"
+            }
+        },
+        "updatedDate":1505779091796,
+        "version":0
+    }
+  }

test/testdata/golden/interceptor/extended-payload.json

@@ -0,0 +1,93 @@
+{
+    "actor": {
+        "active": true,
+        "displayName": "Mustermann,Max ACME",
+        "emailAddress": "max.mustermann@acme.org",
+        "id": 4653,
+        "links": {
+            "self": [
+                {
+                    "href": "https://bitbucket.acme.org/users/max.mustermann_acme.org"
+                }
+            ]
+        },
+        "name": "max.mustermann@acme.org",
+        "slug": "max.mustermann_acme.org",
+        "type": "NORMAL"
+    },
+    "changes": [
+        {
+            "fromHash": "dc85ccd8bb912006162e0d1d9f48e1f2d7210c9c",
+            "ref": {
+                "displayId": "master",
+                "id": "refs/heads/master",
+                "type": "BRANCH"
+            },
+            "refId": "refs/heads/master",
+            "toHash": "0e183aa3bc3c6deb8f40b93fb2fc4354533cf62f",
+            "type": "UPDATE"
+        }
+    ],
+    "date": "2021-01-15T13:29:05+0000",
+    "eventKey": "repo:refs_changed",
+    "extensions": {
+        "name": "ods-pipeline-master",
+        "project": "foo",
+        "component": "ods-pipeline",
+        "repository": "ods-pipeline",
+        "environment": "",
+        "version": "",
+        "gitRef": "master",
+        "gitFullRef": "refs/heads/master",
+        "gitSha": "0e183aa3bc3c6deb8f40b93fb2fc4354533cf62f",
+        "repoBase": "https://domain.com",
+        "gitURI": "https://domain.com/foo/ods-pipeline.git",
+        "namespace": "bar-cd",
+        "pvc": "ods-pipeline",
+        "trigger-event": "repo:refs_changed",
+        "comment": "",
+        "prKey": 0,
+        "prBase": ""
+    },
+    "repository": {
+        "forkable": true,
+        "id": 8733,
+        "links": {
+            "clone": [
+                {
+                    "href": "https://bitbucket.acme.org/scm/foo/ods-pipeline.git",
+                    "name": "http"
+                },
+                {
+                    "href": "ssh://git@bitbucket.acme.org:7999/foo/ods-pipeline.git",
+                    "name": "ssh"
+                }
+            ],
+            "self": [
+                {
+                    "href": "https://bitbucket.acme.org/projects/FOO/repos/ods-pipeline/browse"
+                }
+            ]
+        },
+        "name": "ods-pipeline",
+        "project": {
+            "id": 6603,
+            "key": "FOO",
+            "links": {
+                "self": [
+                    {
+                        "href": "https://bitbucket.acme.org/projects/FOO"
+                    }
+                ]
+            },
+            "name": "Max Mustermann Playground",
+            "public": false,
+            "type": "NORMAL"
+        },
+        "public": false,
+        "scmId": "git",
+        "slug": "ods-pipeline",
+        "state": "AVAILABLE",
+        "statusMessage": "Available"
+    }
+}

test/testdata/golden/interceptor/payload.json

@@ -74,7 +74,6 @@
     "extensions": {
         "name": "bar-main",
         "namespace": "foo-cd",
-        "resourceVersion": 0,
         "project": "foo",
         "component": "bar",
         "repository": "foo-bar",

test/testdata/golden/interceptor/pipeline.yaml

@@ -6,7 +6,6 @@ metadata:
     pipeline.opendevstack.org/git-ref: main
     pipeline.opendevstack.org/repository: foo-bar
   name: bar-main
-  resourceVersion: "0"
 spec:
   description: ODS
   finally: