Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492package xrpc
import ( "bytes" "context" "encoding/json" "fmt" "log/slog" "net/http" "net/http/httptest" "slices" "strings" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/org_tangled" "tangled.org/core/api/tangled" "tangled.org/core/rbac" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets")
func newOrgPipelineXrpc(t *testing.T) (*Xrpc, models.PipelineId, string) { t.Helper() d, e := newTestXrpcDB(t) repoDid := syntax.DID("did:plc:orgpipelinerepo") if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: "did:plc:owner", Rkey: "repo", RepoDid: repoDid}); err != nil { t.Fatal(err) } id := models.PipelineId("pipeline_custom") repo := repoDid.String() createTestPipeline(t, d, id, tangled.Pipeline{ TriggerMetadata: &tangled.Pipeline_TriggerMetadata{Kind: "push", Repo: &tangled.Pipeline_TriggerRepo{Knot: "knot.test", RepoDid: &repo}, Push: &tangled.Pipeline_PushTriggerData{NewSha: "1111111111111111111111111111111111111111", Ref: "refs/heads/main"}}, Workflows: []*tangled.Pipeline_Workflow{{Name: "ci.yml", Raw: "when:\n - event: pull_request\n"}}, }) return &Xrpc{Logger: slog.Default(), Db: d, Enforcer: e}, id, repo}
func TestOrgCiGetAndQueryPipelines(t *testing.T) { x, id, repo := newOrgPipelineXrpc(t)
w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/?pipeline="+id.String(), nil) x.handleOrgTangledCiGetPipeline(w, r) if w.Code != http.StatusOK { t.Fatalf("get pipeline = %d: %s", w.Code, w.Body.String()) } var got org_tangled.CiPipeline if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.Id != id.String() || len(got.Workflows) != 1 { t.Fatalf("pipeline = %+v, %v", got, err) } if !slices.Equal(got.Workflows[0].Definition.Triggers, []string{"org.tangled.event.pullRequest"}) { t.Fatalf("persisted workflow triggers = %v", got.Workflows[0].Definition.Triggers) }
w = httptest.NewRecorder() r = httptest.NewRequest(http.MethodGet, "/?pipeline=missing_pipeline", nil) x.handleOrgTangledCiGetPipeline(w, r) if w.Code != http.StatusNotFound { t.Fatalf("missing pipeline = %d", w.Code) }
w = httptest.NewRecorder() r = httptest.NewRequest(http.MethodGet, "/?repo="+repo+"&triggers=org.tangled.event.push&limit=50", nil) x.handleOrgTangledCiQueryPipelines(w, r) if w.Code != http.StatusOK { t.Fatalf("query pipelines = %d: %s", w.Code, w.Body.String()) } var queried org_tangled.CiQueryPipelines_Output if err := json.Unmarshal(w.Body.Bytes(), &queried); err != nil || queried.Total != 1 || len(queried.Pipelines) != 1 { t.Fatalf("query = %+v, %v", queried, err) }
w = httptest.NewRecorder() r = httptest.NewRequest(http.MethodGet, "/?repo="+repo+"&triggers=push&limit=50", nil) x.handleOrgTangledCiQueryPipelines(w, r) if w.Code != http.StatusOK { t.Fatalf("plain trigger alias = %d: %s", w.Code, w.Body.String()) } queried = org_tangled.CiQueryPipelines_Output{} if err := json.Unmarshal(w.Body.Bytes(), &queried); err != nil || queried.Total != 0 || len(queried.Pipelines) != 0 { t.Fatalf("plain trigger alias matched = %+v, %v", queried, err) }}
func TestOrgSecretInputValidation(t *testing.T) { d, enforcer := newTestXrpcDB(t) actor := syntax.DID("did:plc:secretvalidator") repo := syntax.DID("did:plc:secretvalidationrepo") if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: actor, Rkey: "repo", RepoDid: repo}); err != nil { t.Fatal(err) } if err := enforcer.AddRepo(actor.String(), rbac.ThisServer, repo.String()); err != nil { t.Fatal(err) } vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } x := &Xrpc{Logger: slog.Default(), Db: d, Enforcer: enforcer, Vault: vault} valid := org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: "VALID_KEY", Value: "value"} cases := map[string][]byte{ "malformed json": []byte("{"), "empty key": mustJSON(t, org_tangled.SecretAddSecret_Input{Repo: repo.String(), Value: "value"}), "long key": mustJSON(t, org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: strings.Repeat("k", 51), Value: "value"}), "empty value": mustJSON(t, org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: valid.Key}), "long value": mustJSON(t, org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: valid.Key, Value: strings.Repeat("v", 4097)}), } for name, body := range cases { t.Run(name, func(t *testing.T) { r := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body)) r = r.WithContext(context.WithValue(r.Context(), ActorDid, actor)) w := httptest.NewRecorder() x.handleOrgTangledSecretAddSecret(w, r) if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "InvalidRequest") { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) } }) } stored, err := vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repo)) if err != nil || len(stored) != 0 { t.Fatalf("invalid input changed vault: %+v, %v", stored, err) }
r := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(mustJSON(t, org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: "key with spaces", Value: "value"}))) r = r.WithContext(context.WithValue(r.Context(), ActorDid, actor)) w := httptest.NewRecorder() x.handleOrgTangledSecretAddSecret(w, r) if w.Code != http.StatusOK { t.Fatalf("schema-valid key response = %d: %s", w.Code, w.Body.String()) }}
func TestOrgCiInputBounds(t *testing.T) { x, _, repo := newOrgPipelineXrpc(t) for name, path := range map[string]string{ "zero limit": "/?repo=" + repo + "&limit=0", "large limit": "/?repo=" + repo + "&limit=251", } { t.Run(name, func(t *testing.T) { w := httptest.NewRecorder() x.handleOrgTangledCiQueryPipelines(w, httptest.NewRequest(http.MethodGet, path, nil)) if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "InvalidRequest") { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) } }) } for name, path := range map[string]string{ "long commit": "/?repo=" + repo + "&commits=" + strings.Repeat("a", 65), "short commit": "/?repo=" + repo + "&commits=" + strings.Repeat("a", 39), "non-hex commit": "/?repo=" + repo + "&commits=" + strings.Repeat("z", 40), "unknown trigger": "/?repo=" + repo + "&triggers=scheduled", "invalid cursor": "/?repo=" + repo + "&cursor=nope", } { t.Run(name, func(t *testing.T) { w := httptest.NewRecorder() x.handleOrgTangledCiQueryPipelines(w, httptest.NewRequest(http.MethodGet, path, nil)) if w.Code != http.StatusOK { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) } }) } w := httptest.NewRecorder() x.handleOrgTangledCiGetPipeline(w, httptest.NewRequest(http.MethodGet, "/", nil)) if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "InvalidRequest") { t.Fatalf("missing pipeline = %d: %s", w.Code, w.Body.String()) }
for name, handlerPath := range map[string]struct { handler func(http.ResponseWriter, *http.Request) path string }{ "empty workflow": {x.handleOrgTangledCiGetWorkflowDefinition, "/?repo=" + repo}, "long workflow": {x.handleOrgTangledCiGetWorkflowDefinition, "/?repo=" + repo + "&workflow=" + strings.Repeat("w", 129)}, "long ref": {x.handleOrgTangledCiListWorkflowDefinitions, "/?repo=" + repo + "&ref=" + strings.Repeat("r", 257)}, } { t.Run(name, func(t *testing.T) { w := httptest.NewRecorder() handlerPath.handler(w, httptest.NewRequest(http.MethodGet, handlerPath.path, nil)) if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "InvalidRequest") { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) } }) }}
func mustJSON(t *testing.T, value any) []byte { t.Helper() body, err := json.Marshal(value) if err != nil { t.Fatal(err) } return body}
func TestOrgSecretsAreIsolatedByResolvedRepo(t *testing.T) { d, enforcer := newTestXrpcDB(t) actor := syntax.DID("did:plc:settingsactor") repoA := syntax.DID("did:plc:secretrepoa") repoB := syntax.DID("did:plc:secretrepob") for i, repo := range []syntax.DID{repoA, repoB} { if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: actor, Rkey: syntax.RecordKey(fmt.Sprintf("repo-%d", i)), RepoDid: repo}); err != nil { t.Fatal(err) } } if err := enforcer.AddRepo(actor.String(), rbac.ThisServer, repoA.String()); err != nil { t.Fatal(err) } vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } for _, repo := range []syntax.DID{repoA, repoB} { if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{Repo: secrets.RepoIdentifier(repo), Key: "SAME_KEY", Value: repo.String(), CreatedAt: time.Now(), CreatedBy: actor}); err != nil { t.Fatal(err) } } x := &Xrpc{Logger: slog.Default(), Db: d, Enforcer: enforcer, Vault: vault} request := func(repo syntax.DID) *httptest.ResponseRecorder { r := httptest.NewRequest(http.MethodGet, "/?repo="+repo.String(), nil) r = r.WithContext(context.WithValue(r.Context(), ActorDid, actor)) w := httptest.NewRecorder() x.handleOrgTangledSecretListSecrets(w, r) return w } if w := request(repoA); w.Code != http.StatusOK { t.Fatalf("repo A list = %d: %s", w.Code, w.Body.String()) } else { var listed org_tangled.SecretListSecrets_Output if err := json.Unmarshal(w.Body.Bytes(), &listed); err != nil || len(listed.Secrets) != 1 || listed.Secrets[0].Key != "SAME_KEY" { t.Fatalf("repo A secrets = %+v, %v", listed.Secrets, err) } } if w := request(repoB); w.Code != http.StatusUnauthorized || !strings.Contains(w.Body.String(), "AccessControl") { t.Fatalf("repo B list = %d: %s", w.Code, w.Body.String()) } removeBody, _ := json.Marshal(org_tangled.SecretRemoveSecret_Input{Repo: repoB.String(), Key: "SAME_KEY"}) r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(string(removeBody))) r = r.WithContext(context.WithValue(r.Context(), ActorDid, actor)) w := httptest.NewRecorder() x.handleOrgTangledSecretRemoveSecret(w, r) if w.Code != http.StatusUnauthorized { t.Fatalf("repo B remove = %d: %s", w.Code, w.Body.String()) } remaining, err := vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repoB)) if err != nil || len(remaining) != 1 || remaining[0].Value != repoB.String() { t.Fatalf("repo B secret changed: %+v, %v", remaining, err) }}
func TestOrgSubscribePipelineLogsValidatesPipelineAndWorkflows(t *testing.T) { x, id, _ := newOrgPipelineXrpc(t)
w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/?pipeline=3mu2xwiorc2xm", nil) x.handleOrgTangledCiSubscribePipelineLogs(w, r) if w.Code != http.StatusNotFound || !strings.Contains(w.Body.String(), "PipelineNotFound") { t.Fatalf("missing pipeline = %d: %s", w.Code, w.Body.String()) }
otherID := models.PipelineId("3mu2xwiorc2xn") repo := "did:plc:orgpipelinerepo" createTestPipeline(t, x.Db, otherID, tangled.Pipeline{ TriggerMetadata: &tangled.Pipeline_TriggerMetadata{Kind: "push", Repo: &tangled.Pipeline_TriggerRepo{RepoDid: &repo}, Push: &tangled.Pipeline_PushTriggerData{NewSha: strings.Repeat("2", 40)}}, Workflows: []*tangled.Pipeline_Workflow{{Name: "other.yml"}}, }) w = httptest.NewRecorder() r = httptest.NewRequest(http.MethodGet, "/?pipeline="+id.String()+"&workflows=other.yml", nil) x.handleOrgTangledCiSubscribePipelineLogs(w, r) if w.Code != http.StatusNotFound || !strings.Contains(w.Body.String(), "WorkflowNotFound") { t.Fatalf("missing workflow = %d: %s", w.Code, w.Body.String()) }}
func TestResolveKnownRepoDidContract(t *testing.T) { x, _, knownRepo := newOrgPipelineXrpc(t) if got, xerr, ok := x.resolveKnownRepoDid(knownRepo); !ok || got.String() != knownRepo || xerr.Tag != "" { t.Fatalf("known repo = %q, %+v, %v", got, xerr, ok) } for _, test := range []struct { name, repo, tag string }{ {"unknown", "did:plc:unknownrepo", "RepoNotFound"}, {"malformed", "not-a-did", "InvalidRequest"}, {"empty", "", "InvalidRequest"}, } { t.Run(test.name, func(t *testing.T) { if _, xerr, ok := x.resolveKnownRepoDid(test.repo); ok || xerr.Tag != test.tag { t.Fatalf("resolve = %+v, %v; want %s", xerr, ok, test.tag) } }) }}
func TestAuthorizeOrgSecretRepoRequiresTypedActor(t *testing.T) { x, _, repo := newOrgPipelineXrpc(t) r := httptest.NewRequest(http.MethodGet, "/", nil) r = r.WithContext(context.WithValue(r.Context(), ActorDid, "did:plc:not-typed")) w := httptest.NewRecorder() if _, _, ok := x.authorizeOrgSecretRepo(w, r, repo); ok { t.Fatal("string actor unexpectedly authorized") } if w.Code != http.StatusUnauthorized || !strings.Contains(w.Body.String(), "MissingActorDid") { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) }}
func TestOrgRepoScopedEndpointsReturnRepoNotFound(t *testing.T) { x, _, _ := newOrgPipelineXrpc(t) for name, handler := range map[string]func(http.ResponseWriter, *http.Request){ "query pipelines": x.handleOrgTangledCiQueryPipelines, "get workflow": x.handleOrgTangledCiGetWorkflowDefinition, "list workflows": x.handleOrgTangledCiListWorkflowDefinitions, } { t.Run(name, func(t *testing.T) { w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/?repo=did:plc:unknownrepo&workflow=ci.yml", nil) handler(w, r) if w.Code != http.StatusNotFound || !strings.Contains(w.Body.String(), "RepoNotFound") { t.Fatalf("response = %d: %s", w.Code, w.Body.String()) } }) }}
func TestOrgWorkflowDefinitionHandlers(t *testing.T) { d, e := newTestXrpcDB(t) repo := syntax.DID("did:plc:workflowrepo") if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: "did:plc:owner", Rkey: "repo", RepoDid: repo}); err != nil { t.Fatal(err) } name := "ci.yml" definition := &models.WorkflowDefinition{ ID: name, Name: name, Source: models.WorkflowDefinitionSource{File: &models.WorkflowFileSource{ Repo: repo.String(), Commit: "1111111111111111111111111111111111111111", Path: ".tangled/workflows/ci.yml", }}, Triggers: []string{"org.tangled.event.push"}, } x := &Xrpc{Logger: slog.Default(), Db: d, Enforcer: e, Trigger: &mockTrigger{workflowDefinitions: []*models.WorkflowDefinition{definition}}}
w := httptest.NewRecorder() r := httptest.NewRequest(http.MethodGet, "/?repo="+repo.String()+"&workflow="+name, nil) x.handleOrgTangledCiGetWorkflowDefinition(w, r) if w.Code != http.StatusOK { t.Fatalf("get workflow = %d: %s", w.Code, w.Body.String()) }
w = httptest.NewRecorder() r = httptest.NewRequest(http.MethodGet, "/?repo="+repo.String(), nil) x.handleOrgTangledCiListWorkflowDefinitions(w, r) if w.Code != http.StatusOK { t.Fatalf("list workflows = %d: %s", w.Code, w.Body.String()) }}
func TestOrgPipelineAdapter(t *testing.T) { source := "did:plc:source" record := &models.PipelineRecord{ ID: "3mu2xwiorc2xl", RepoDID: "did:plc:target", SourceRepo: &source, Commit: "1111111111111111111111111111111111111111", CreatedAt: time.Now().Format(time.RFC3339), Trigger: models.PipelineTrigger{Kind: "pull_request", PullRequest: &models.PipelinePullRequestTrigger{ SourceCommit: "1111111111111111111111111111111111111111", TargetBranch: "main", }}, Workflows: []*models.PipelineWorkflow{ nil, {ID: "ci.yml", Name: "ci.yml", Status: "running", Definition: &models.WorkflowDefinition{ ID: "ci.yml", Name: "ci.yml", Triggers: []string{"org.tangled.event.push"}, Source: models.WorkflowDefinitionSource{File: &models.WorkflowFileSource{ Repo: source, Commit: "1111111111111111111111111111111111111111", Path: ".tangled/workflows/ci.yml", }}, }}, }, } pipeline, err := toOrgPipeline(record) if err != nil { t.Fatal(err) } if len(pipeline.Workflows) != 1 { t.Fatalf("workflows = %+v", pipeline.Workflows) } workflow := pipeline.Workflows[0] sourceFile := workflow.Definition.Source.CiWorkflow_FileSource if workflow.Id != "ci.yml" || sourceFile.Repo != source || !slices.Equal(workflow.Definition.Triggers, []string{"org.tangled.event.push"}) { t.Fatalf("workflow = %+v", workflow) } if pipeline.Trigger.EventPullRequest == nil { t.Fatalf("trigger = %+v", pipeline.Trigger) } encoded, err := json.Marshal(pipeline) if err != nil { t.Fatal(err) } if !strings.Contains(string(encoded), `"definition"`) || strings.Contains(string(encoded), `"defintion"`) || !strings.Contains(string(encoded), `"$type":"org.tangled.event.pullRequest"`) { t.Fatalf("pipeline JSON = %s", encoded) }}
func TestNormalizeOrgTriggerKinds(t *testing.T) { got := normalizeOrgTriggerKinds([]string{"org.tangled.event.push", "org.tangled.event.pullRequest", "org.tangled.ci.trigger.manual", "push"}) want := []string{"push", "pull_request", "manual", "\x00push"} if !slices.Equal(got, want) { t.Fatalf("normalized trigger kinds = %v, want %v", got, want) }}
func TestOrgSecretConflictErrors(t *testing.T) { d, enforcer := newTestXrpcDB(t) actor := syntax.DID("did:plc:secret-errors") repo := syntax.DID("did:plc:secret-errors-repo") if err := d.AddRepo(db.Repo{Knot: "knot.test", Owner: actor, Rkey: "repo", RepoDid: repo}); err != nil { t.Fatal(err) } if err := enforcer.AddRepo(actor.String(), rbac.ThisServer, repo.String()); err != nil { t.Fatal(err) } vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } x := &Xrpc{Logger: slog.Default(), Db: d, Enforcer: enforcer, Vault: vault} request := func(handler func(http.ResponseWriter, *http.Request), value any) *httptest.ResponseRecorder { request := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(mustJSON(t, value))) request = request.WithContext(context.WithValue(request.Context(), ActorDid, actor)) response := httptest.NewRecorder() handler(response, request) return response } input := org_tangled.SecretAddSecret_Input{Repo: repo.String(), Key: "KEY", Value: "value"} if response := request(x.handleOrgTangledSecretAddSecret, input); response.Code != http.StatusOK { t.Fatalf("first add = %d: %s", response.Code, response.Body.String()) } if response := request(x.handleOrgTangledSecretAddSecret, input); response.Code != http.StatusConflict || !strings.Contains(response.Body.String(), "SecretAlreadyExists") { t.Fatalf("duplicate add = %d: %s", response.Code, response.Body.String()) } missing := org_tangled.SecretRemoveSecret_Input{Repo: repo.String(), Key: "MISSING"} if response := request(x.handleOrgTangledSecretRemoveSecret, missing); response.Code != http.StatusNotFound || !strings.Contains(response.Body.String(), "SecretNotFound") { t.Fatalf("missing remove = %d: %s", response.Code, response.Body.String()) }}
func TestOrgPipelineTriggerUnion(t *testing.T) { base := models.PipelineRecord{ ID: "pipeline", RepoDID: "did:plc:repo", Commit: strings.Repeat("1", 40), CreatedAt: time.Now().Format(time.RFC3339), Workflows: []*models.PipelineWorkflow{{ ID: "ci.yml", Name: "ci.yml", Status: "pending", Definition: &models.WorkflowDefinition{ ID: "ci.yml", Name: "ci.yml", Triggers: []string{}, Source: models.WorkflowDefinitionSource{External: &models.WorkflowExternalSource{Name: "generated", Link: "https://example.com/workflow"}}, }, }}, } ref := "refs/heads/main" cases := []struct { name string set func(*models.PipelineRecord) typeID string }{ {"push", func(record *models.PipelineRecord) { record.Trigger = models.PipelineTrigger{Kind: "push", Push: &models.PipelinePushTrigger{}} }, "org.tangled.event.push"}, {"pull_request", func(record *models.PipelineRecord) { record.Trigger = models.PipelineTrigger{Kind: "pull_request", PullRequest: &models.PipelinePullRequestTrigger{}} }, "org.tangled.event.pullRequest"}, {"manual", func(record *models.PipelineRecord) { record.Trigger = models.PipelineTrigger{Kind: "manual", Manual: &models.PipelineManualTrigger{Ref: &ref, Inputs: []*models.PipelineInputPair{{Key: "target", Value: "all"}}}} }, "org.tangled.ci.trigger.manual"}, } for _, test := range cases { t.Run(test.name, func(t *testing.T) { record := base test.set(&record) pipeline, err := toOrgPipeline(&record) if err != nil { t.Fatal(err) } encoded, err := json.Marshal(pipeline) if err != nil { t.Fatal(err) } if !strings.Contains(string(encoded), `"$type":"`+test.typeID+`"`) { t.Fatalf("pipeline JSON = %s", encoded) } if pipeline.Workflows[0].Definition.Source.CiWorkflow_ExternalSource == nil { t.Fatalf("external source was not preserved: %+v", pipeline.Workflows[0].Definition) } }) }}