Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347package models
import ( "testing"
"tangled.org/core/api/tangled" "tangled.org/core/workflow")
func TestPipelineEnvVars_PushBranch(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ NewSha: "abc123def456", OldSha: "000000000000", Ref: "refs/heads/main", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), DefaultBranch: "main", }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
// Check standard CI variable if env["CI"] != "true" { t.Errorf("Expected CI='true', got '%s'", env["CI"]) }
// Check ref variables if env["TANGLED_REF"] != "refs/heads/main" { t.Errorf("Expected TANGLED_REF='refs/heads/main', got '%s'", env["TANGLED_REF"]) } if env["TANGLED_REF_NAME"] != "main" { t.Errorf("Expected TANGLED_REF_NAME='main', got '%s'", env["TANGLED_REF_NAME"]) } if env["TANGLED_REF_TYPE"] != "branch" { t.Errorf("Expected TANGLED_REF_TYPE='branch', got '%s'", env["TANGLED_REF_TYPE"]) }
// Check SHA variables if env["TANGLED_SHA"] != "abc123def456" { t.Errorf("Expected TANGLED_SHA='abc123def456', got '%s'", env["TANGLED_SHA"]) } if env["TANGLED_COMMIT_SHA"] != "abc123def456" { t.Errorf("Expected TANGLED_COMMIT_SHA='abc123def456', got '%s'", env["TANGLED_COMMIT_SHA"]) }
// Check repo variables if env["TANGLED_REPO_KNOT"] != "example.com" { t.Errorf("Expected TANGLED_REPO_KNOT='example.com', got '%s'", env["TANGLED_REPO_KNOT"]) } if env["TANGLED_REPO_DID"] != "did:plc:user123" { t.Errorf("Expected TANGLED_REPO_DID='did:plc:user123', got '%s'", env["TANGLED_REPO_DID"]) } if env["TANGLED_REPO_NAME"] != "my-repo" { t.Errorf("Expected TANGLED_REPO_NAME='my-repo', got '%s'", env["TANGLED_REPO_NAME"]) } if env["TANGLED_REPO_DEFAULT_BRANCH"] != "main" { t.Errorf("Expected TANGLED_REPO_DEFAULT_BRANCH='main', got '%s'", env["TANGLED_REPO_DEFAULT_BRANCH"]) } if env["TANGLED_REPO_URL"] != "https://example.com/did:plc:boltless" { t.Errorf("Expected TANGLED_REPO_URL='https://example.com/did:plc:boltless', got '%s'", env["TANGLED_REPO_URL"]) }}
func TestPipelineEnvVars_PushTag(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ NewSha: "abc123def456", OldSha: "000000000000", Ref: "refs/tags/v1.2.3", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
if env["TANGLED_REF"] != "refs/tags/v1.2.3" { t.Errorf("Expected TANGLED_REF='refs/tags/v1.2.3', got '%s'", env["TANGLED_REF"]) } if env["TANGLED_REF_NAME"] != "v1.2.3" { t.Errorf("Expected TANGLED_REF_NAME='v1.2.3', got '%s'", env["TANGLED_REF_NAME"]) } if env["TANGLED_REF_TYPE"] != "tag" { t.Errorf("Expected TANGLED_REF_TYPE='tag', got '%s'", env["TANGLED_REF_TYPE"]) }}
func TestPipelineEnvVars_PullRequest(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPullRequest), PullRequest: &tangled.Pipeline_PullRequestTriggerData{ SourceBranch: "feature-branch", TargetBranch: "main", SourceSha: "pr-sha-789", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
// Check ref variables for PR if env["TANGLED_REF"] != "refs/heads/feature-branch" { t.Errorf("Expected TANGLED_REF='refs/heads/feature-branch', got '%s'", env["TANGLED_REF"]) } if env["TANGLED_REF_NAME"] != "feature-branch" { t.Errorf("Expected TANGLED_REF_NAME='feature-branch', got '%s'", env["TANGLED_REF_NAME"]) } if env["TANGLED_REF_TYPE"] != "branch" { t.Errorf("Expected TANGLED_REF_TYPE='branch', got '%s'", env["TANGLED_REF_TYPE"]) }
// Check SHA variables if env["TANGLED_SHA"] != "pr-sha-789" { t.Errorf("Expected TANGLED_SHA='pr-sha-789', got '%s'", env["TANGLED_SHA"]) } if env["TANGLED_COMMIT_SHA"] != "pr-sha-789" { t.Errorf("Expected TANGLED_COMMIT_SHA='pr-sha-789', got '%s'", env["TANGLED_COMMIT_SHA"]) }
// Check PR-specific variables if env["TANGLED_PIPELINE_SOURCE_BRANCH"] != "feature-branch" { t.Errorf("Expected TANGLED_PIPELINE_SOURCE_BRANCH='feature-branch', got '%s'", env["TANGLED_PIPELINE_SOURCE_BRANCH"]) } if env["TANGLED_PIPELINE_TARGET_BRANCH"] != "main" { t.Errorf("Expected TANGLED_PIPELINE_TARGET_BRANCH='main', got '%s'", env["TANGLED_PIPELINE_TARGET_BRANCH"]) } if env["TANGLED_PR_SOURCE_BRANCH"] != "feature-branch" { t.Errorf("Expected TANGLED_PR_SOURCE_BRANCH='feature-branch', got '%s'", env["TANGLED_PR_SOURCE_BRANCH"]) } if env["TANGLED_PR_TARGET_BRANCH"] != "main" { t.Errorf("Expected TANGLED_PR_TARGET_BRANCH='main', got '%s'", env["TANGLED_PR_TARGET_BRANCH"]) } if env["TANGLED_PR_SOURCE_SHA"] != "pr-sha-789" { t.Errorf("Expected TANGLED_PR_SOURCE_SHA='pr-sha-789', got '%s'", env["TANGLED_PR_SOURCE_SHA"]) }}
func TestPipelineEnvVars_SourceRepo(t *testing.T) { sourceRepoDid := "did:plc:fork" tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPullRequest), PullRequest: &tangled.Pipeline_PullRequestTriggerData{ SourceBranch: "feature-branch", TargetBranch: "main", SourceSha: "pr-sha-789", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "target.example.com", Did: "did:plc:user123", Repo: sp("target-repo"), RepoDid: sp("did:plc:target"), }, SourceRepo: &sourceRepoDid, } sourceRepo := &tangled.Pipeline_TriggerRepo{ Knot: "fork.example.com", Did: "did:plc:user456", Repo: sp("fork-repo"), RepoDid: &sourceRepoDid, DefaultBranch: "feature-branch", } id := PipelineId{ Knot: "target.example.com", Rkey: "123123", }
env := PipelineEnvVarsForSource(tr, id, sourceRepo)
if env["TANGLED_PIPELINE_SOURCE"] != sourceRepoDid { t.Errorf("Expected TANGLED_PIPELINE_SOURCE='%s', got '%s'", sourceRepoDid, env["TANGLED_PIPELINE_SOURCE"]) } if env["TANGLED_REPO_URL"] != "https://fork.example.com/did:plc:fork" { t.Errorf("Expected TANGLED_REPO_URL to point at source repo, got '%s'", env["TANGLED_REPO_URL"]) } if env["TANGLED_REPO_REPO_DID"] != sourceRepoDid { t.Errorf("Expected TANGLED_REPO_REPO_DID='%s', got '%s'", sourceRepoDid, env["TANGLED_REPO_REPO_DID"]) }}
func TestPipelineEnvVars_ManualWithInputs(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindManual), Manual: &tangled.Pipeline_ManualTriggerData{ Sha: "manualsha789", Inputs: []*tangled.Pipeline_Pair{ {Key: "version", Value: "1.0.0"}, {Key: "environment", Value: "production"}, }, }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
// Check manual input variables if env["TANGLED_INPUT_VERSION"] != "1.0.0" { t.Errorf("Expected TANGLED_INPUT_VERSION='1.0.0', got '%s'", env["TANGLED_INPUT_VERSION"]) } if env["TANGLED_INPUT_ENVIRONMENT"] != "production" { t.Errorf("Expected TANGLED_INPUT_ENVIRONMENT='production', got '%s'", env["TANGLED_INPUT_ENVIRONMENT"]) }
// Manual triggers carry the explicit SHA if env["TANGLED_SHA"] != "manualsha789" { t.Errorf("Expected TANGLED_SHA='manualsha789', got '%s'", env["TANGLED_SHA"]) } if env["TANGLED_COMMIT_SHA"] != "manualsha789" { t.Errorf("Expected TANGLED_COMMIT_SHA='manualsha789', got '%s'", env["TANGLED_COMMIT_SHA"]) } // No ref was supplied, so ref vars stay unset if _, ok := env["TANGLED_REF"]; ok { t.Error("Manual trigger without a ref should not have TANGLED_REF") }}
func TestPipelineEnvVars_DevMode(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ NewSha: "abc123", Ref: "refs/heads/main", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "localhost:3000", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
expectedURL := "http://localhost:3000/did:plc:boltless" if env["TANGLED_REPO_URL"] != expectedURL { t.Errorf("Expected TANGLED_REPO_URL='%s', got '%s'", expectedURL, env["TANGLED_REPO_URL"]) }}
func TestPipelineEnvVars_NilTrigger(t *testing.T) { id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(nil, id)
if env != nil { t.Error("Expected nil env for nil trigger") }}
func TestPipelineEnvVars_NilPushData(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: nil, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", Repo: sp("my-repo"), RepoDid: sp("did:plc:boltless"), }, } id := PipelineId{ Knot: "example.com", Rkey: "123123", } env := PipelineEnvVars(tr, id)
// Should still have repo variables if env["TANGLED_REPO_KNOT"] != "example.com" { t.Errorf("Expected TANGLED_REPO_KNOT='example.com', got '%s'", env["TANGLED_REPO_KNOT"]) }
// Should not have ref/sha variables if _, ok := env["TANGLED_REF"]; ok { t.Error("Should not have TANGLED_REF when push data is nil") }}
func TestPipelineEnvVars_Schedule(t *testing.T) { tr := &tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindSchedule), Schedule: &tangled.Pipeline_ScheduleTriggerData{ Sha: "scheduledsha123", Ref: "refs/heads/main", ScheduledAt: "2026-08-10T09:00:00Z", }, Repo: &tangled.Pipeline_TriggerRepo{ Knot: "example.com", Did: "did:plc:user123", RepoDid: sp("did:plc:boltless"), }, }
env := PipelineEnvVars(tr, PipelineId{Knot: "example.com", Rkey: "pipeline"}) expected := map[string]string{ "TANGLED_PIPELINE_KIND": "schedule", "TANGLED_REF": "refs/heads/main", "TANGLED_REF_NAME": "main", "TANGLED_REF_TYPE": "branch", "TANGLED_SHA": "scheduledsha123", "TANGLED_COMMIT_SHA": "scheduledsha123", "TANGLED_SCHEDULED_AT": "2026-08-10T09:00:00Z", } for key, value := range expected { if env[key] != value { t.Errorf("%s = %q, want %q", key, env[key], value) } }}