eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Add destination workflow config

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-19 15:18:14 +00:00
parent d23c624179
commit a95662226f
6 changed files with 101 additions and 652 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
internal/app/run_test.go
View File

@@ -15,7 +15,6 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/notify"
"gitea.maximumdirect.net/eric/distributor/internal/publish"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
@@ -222,67 +221,6 @@ func TestRunPublishesNewLocalBundle(t *testing.T) {
}
}
func TestRunSharedRootDryRunWritesNoOutputsOrState(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "", testBundleOptions{
Files: []testFile{{Path: "report.md", Data: "# Report\n"}},
})
configPath := writeSharedRootLocalConfig(t, sourceRoot, destinationRoot)
cfg, err := config.LoadFile(configPath)
if err != nil {
t.Fatalf("load config: %v", err)
}
report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true}, newBackendFactoryWithEnvironment)
if err != nil {
t.Fatalf("dry-run buildRunReportWithBackendFactory() error = %v", err)
}
if got, want := report.Actions[0].Action, string(publish.ActionPublishNew); got != want {
t.Fatalf("dry-run action = %q, want %q", got, want)
}
if _, statErr := os.Stat(filepath.Join(destinationRoot, "report.md")); !os.IsNotExist(statErr) {
t.Fatalf("output stat error = %v, want absent", statErr)
}
if _, statErr := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); !os.IsNotExist(statErr) {
t.Fatalf("state file stat error = %v, want absent", statErr)
}
}
func TestRunPublishesTwoPipelinesIntoSharedRoot(t *testing.T) {
firstSourceRoot := t.TempDir()
secondSourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, firstSourceRoot, "", testBundleOptions{
ID: "reports.first",
Files: []testFile{{Path: "first.md", Data: "# First\n"}},
})
writeSourceBundle(t, secondSourceRoot, "", testBundleOptions{
ID: "reports.second",
Files: []testFile{{Path: "second.md", Data: "# Second\n"}},
})
err := Run(context.Background(), RunOptions{ConfigPath: writeTwoPipelineSharedRootConfig(t, firstSourceRoot, secondSourceRoot, destinationRoot)})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "first.md"), "# First\n")
testutil.AssertFile(t, filepath.Join(destinationRoot, "second.md"), "# Second\n")
destinationState := readSharedRootStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if got, want := len(destinationState.Owners), 2; got != want {
t.Fatalf("owner count = %d, want %d", got, want)
}
if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports-first", "archive")); !ok {
t.Fatal("reports-first/archive owner missing")
}
if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports-second", "archive")); !ok {
t.Fatal("reports-second/archive owner missing")
}
if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "first.md,second.md"; got != want {
t.Fatalf("managed paths = %q, want %q", got, want)
}
}
func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -698,67 +636,6 @@ func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
}
}
func TestRunPreserveRelativeSameSourceTakeoverAllowsOwnerMismatch(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
sourceManifest := writeSourceBundle(t, sourceRoot, "daily/report", testBundleOptions{
ID: "reports.same",
Files: []testFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
})
testutil.WriteDestinationState(t, destinationRoot, "daily/report", sourceManifest, testutil.DestinationStateOptions{
PipelineID: "other",
})
if err := os.WriteFile(filepath.Join(destinationRoot, "daily", "report", "report.md"), []byte("# Report\nOld.\n"), 0o600); err != nil {
t.Fatalf("write old report: %v", err)
}
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{
ConfigPath: writeSameSourcePreserveRelativeConfig(t, sourceRoot, destinationRoot),
Stdout: &stdout,
})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if !strings.Contains(stdout.String(), "action=replace_takeover takeover_mode=same_source") {
t.Fatalf("stdout = %q, want same-source takeover", stdout.String())
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "daily", "report", "report.md"), "# Report\nNew.\n")
destinationState := readStateFile(t, filepath.Join(destinationRoot, "daily", "report", storage.StateFileName))
if destinationState.PipelineID != "reports" || destinationState.Source.Manifest.ID != "reports.same" {
t.Fatalf("state owner/source = %s/%s source=%s, want reports/archive reports.same", destinationState.PipelineID, destinationState.DestinationID, destinationState.Source.Manifest.ID)
}
}
func TestRunPreserveRelativeSameSourceRefusesDifferentSource(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
sourceManifest := writeSourceBundle(t, sourceRoot, "daily/report", testBundleOptions{
ID: "reports.same",
Files: []testFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
})
destinationManifest := sourceManifest
destinationManifest.ID = "reports.other"
testutil.WriteDestinationState(t, destinationRoot, "daily/report", destinationManifest, testutil.DestinationStateOptions{
PipelineID: "other",
})
if err := os.WriteFile(filepath.Join(destinationRoot, "daily", "report", "report.md"), []byte("# Report\nOld.\n"), 0o600); err != nil {
t.Fatalf("write old report: %v", err)
}
err := Run(context.Background(), RunOptions{
ConfigPath: writeSameSourcePreserveRelativeConfig(t, sourceRoot, destinationRoot),
})
if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
t.Fatalf("Run() error = %v, want fail_conflict", err)
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "daily", "report", "report.md"), "# Report\nOld.\n")
destinationState := readStateFile(t, filepath.Join(destinationRoot, "daily", "report", storage.StateFileName))
if destinationState.PipelineID != "other" || destinationState.Source.Manifest.ID != "reports.other" {
t.Fatalf("state owner/source = %s/%s source=%s, want unchanged other/archive reports.other", destinationState.PipelineID, destinationState.DestinationID, destinationState.Source.Manifest.ID)
}
}
func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -989,53 +866,6 @@ func TestRunNotifiesAfterReplacement(t *testing.T) {
}
}
func TestRunMergeReconciliationRetainsManagedOutput(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
manifest := testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{
Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
})
older := manifest
older.Created = older.Created.Add(-time.Hour)
defaultManifest := testutil.ValidManifest(testutil.BundleOptions{})
older.Files = append([]bundle.ManifestFile(nil), defaultManifest.Files...)
older.Digest = bundle.BundleDigest(older.Files)
writeDestinationState(t, destinationRoot, "", older)
if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
t.Fatalf("write old report: %v", err)
}
if err := os.WriteFile(filepath.Join(destinationRoot, "summary.txt"), []byte("old summary\n"), 0o600); err != nil {
t.Fatalf("write old summary: %v", err)
}
configPath := writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
reconciliation:
mode: merge
`)
err := Run(context.Background(), RunOptions{ConfigPath: configPath})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nNew.\n")
testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "old summary\n")
destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeMerge; got != want {
t.Fatalf("reconciliation mode = %q, want %q", got, want)
}
if got, want := len(destinationState.Outputs), 2; got != want {
t.Fatalf("state output count = %d, want %d", got, want)
}
}
func TestRunJSONIncludesGeneratedOutputMetadata(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -1663,135 +1493,6 @@ func TestRunSkipsNewerDestination(t *testing.T) {
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
}
func TestRunReplacesConflictWhenTransferPolicyAllows(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
conflict := manifest
conflict.ID = "other.source"
writeDestinationState(t, destinationRoot, "", conflict)
if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
t.Fatalf("write old output: %v", err)
}
configPath := writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
takeover:
mode: never
transfer:
on_conflict: replace
`)
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{ConfigPath: configPath, DryRun: true, Stdout: &stdout})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
output := stdout.String()
for _, want := range []string{
"action=replace_conflict",
"replace_conflict=1",
"force_replace=0",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
}
}
var jsonOut bytes.Buffer
err = Run(context.Background(), RunOptions{ConfigPath: configPath, DryRun: true, Stdout: &jsonOut, OutputFormat: OutputFormatJSON})
if err != nil {
t.Fatalf("Run() JSON error = %v", err)
}
result := decodeAppResult(t, jsonOut.String())
actions, ok := result["actions"].([]any)
if !ok || len(actions) != 1 {
t.Fatalf("actions = %#v, want one action", result["actions"])
}
action, ok := actions[0].(map[string]any)
if !ok || action["action"] != "replace_conflict" {
t.Fatalf("action = %#v, want replace_conflict", actions[0])
}
summary, ok := result["summary"].(map[string]any)
if !ok || summary["replace_conflict"] != float64(1) || summary["force_replace"] != float64(0) {
t.Fatalf("summary = %#v, want replace_conflict without force", result["summary"])
}
}
func TestRunReplacesNewerWhenTransferPolicyAllows(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
newer := manifest
newer.Created = newer.Created.Add(time.Hour)
writeDestinationState(t, destinationRoot, "", newer)
if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("newer\n"), 0o600); err != nil {
t.Fatalf("write newer output: %v", err)
}
configPath := writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
transfer:
on_destination_newer: replace
`)
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{ConfigPath: configPath, DryRun: true, Stdout: &stdout})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
output := stdout.String()
for _, want := range []string{
"action=replace_newer",
"replace_newer=1",
"force_replace=0",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
}
}
}
func TestRunTakeoverNeverFailsOnConflict(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
manifest.ID = "other.source"
writeDestinationState(t, destinationRoot, "", manifest)
configPath := writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
takeover:
mode: never
`)
err := Run(context.Background(), RunOptions{ConfigPath: configPath})
if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
t.Fatalf("Run() error = %v, want fail_conflict", err)
}
}
func TestRunFailsOnUnmanagedDestination(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -2032,69 +1733,6 @@ func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
}
func writeSharedRootLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
state:
mode: shared_root
`)
}
func writeTwoPipelineSharedRootConfig(t *testing.T, firstSourceRoot, secondSourceRoot, destinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports-first
source:
backend: local
path: `+firstSourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
state:
mode: shared_root
- id: reports-second
source:
backend: local
path: `+secondSourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
state:
mode: shared_root
`)
}
func writeSameSourcePreserveRelativeConfig(t *testing.T, sourceRoot, destinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
path_mapping:
mode: preserve_relative
takeover:
mode: same_source
`)
}
func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
t.Helper()
return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
@@ -2175,19 +1813,6 @@ func readStateFile(t *testing.T, path string) state.DistributorState {
return testutil.ReadDestinationState(t, path)
}
func readSharedRootStateFile(t *testing.T, path string) state.SharedRootState {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read shared-root state: %v", err)
}
destinationState, err := state.ParseSharedRoot(data)
if err != nil {
t.Fatalf("parse shared-root state: %v", err)
}
return destinationState
}
func outputsByPath(outputs []state.OutputFile) map[string]state.OutputFile {
byPath := make(map[string]state.OutputFile, len(outputs))
for _, output := range outputs {

9
internal/config/config.go
View File

@@ -55,11 +55,12 @@ type Destination struct {
Transform Transform `yaml:"transform"`
PathMap PathMapping `yaml:"path_mapping"`
Links *Links `yaml:"links"`
State StatePolicy `yaml:"state"`
Reconciliation ReconciliationPolicy `yaml:"reconciliation"`
Takeover TakeoverPolicy `yaml:"takeover"`
Workflow string `yaml:"workflow"`
State StatePolicy `yaml:"-"`
Reconciliation ReconciliationPolicy `yaml:"-"`
Takeover TakeoverPolicy `yaml:"-"`
Retention RetentionPolicy `yaml:"retention"`
Transfer TransferPolicy `yaml:"transfer"`
Transfer TransferPolicy `yaml:"-"`
}
type Backend struct {

8
internal/config/defaults.go
View File

@@ -42,6 +42,11 @@ const (
LinkPrimarySource = "source"
)
const (
WorkflowAdditive = "additive"
WorkflowReplacement = "replacement"
)
const (
ReconciliationModeReplace = "replace"
ReconciliationModeMerge = "merge"
@@ -96,6 +101,9 @@ func ApplyDefaults(cfg *Config) {
if destination.Links != nil && destination.Links.Primary == "" {
destination.Links.Primary = LinkPrimaryAuto
}
if destination.Workflow == "" {
destination.Workflow = WorkflowAdditive
}
if destination.State.Mode == "" {
destination.State.Mode = StateModeSingleOwner
}

203
internal/config/load_test.go
View File

@@ -30,17 +30,8 @@ pipelines:
if got, want := cfg.Pipelines[0].Validation.OnDigestMismatch, ValidationActionFail; got != want {
t.Fatalf("validation default = %q, want %q", got, want)
}
if got, want := destination.Transfer.OnDestinationOlder, TransferActionReplace; got != want {
t.Fatalf("transfer default = %q, want %q", got, want)
}
if got, want := destination.Reconciliation.Mode, ReconciliationModeReplace; got != want {
t.Fatalf("reconciliation mode default = %q, want %q", got, want)
}
if got, want := destination.State.Mode, StateModeSingleOwner; got != want {
t.Fatalf("state mode default = %q, want %q", got, want)
}
if got, want := destination.Takeover.Mode, TakeoverModeSamePipeline; got != want {
t.Fatalf("takeover mode default = %q, want %q", got, want)
if got, want := destination.Workflow, WorkflowAdditive; got != want {
t.Fatalf("workflow default = %q, want %q", got, want)
}
if destination.Retention.Prune.Enabled {
t.Fatal("retention.prune.enabled default = true, want false")
@@ -176,7 +167,7 @@ pipelines:
}
}
func TestLoadFileAcceptsExplicitReconciliationModes(t *testing.T) {
func TestLoadFileAcceptsExplicitWorkflows(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: reports
@@ -187,94 +178,19 @@ pipelines:
- id: archive
backend: local
path: /archive
reconciliation:
mode: replace
workflow: additive
- id: web
backend: local
path: /web
reconciliation:
mode: merge
workflow: replacement
`)
destinations := cfg.Pipelines[0].Destinations
if got, want := destinations[0].Reconciliation.Mode, ReconciliationModeReplace; got != want {
t.Fatalf("archive reconciliation mode = %q, want %q", got, want)
if got, want := destinations[0].Workflow, WorkflowAdditive; got != want {
t.Fatalf("archive workflow = %q, want %q", got, want)
}
if got, want := destinations[1].Reconciliation.Mode, ReconciliationModeMerge; got != want {
t.Fatalf("web reconciliation mode = %q, want %q", got, want)
}
}
func TestLoadFileAcceptsExplicitStateModes(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
state:
mode: single_owner
- id: web
backend: local
path: /web
state:
mode: shared_root
`)
destinations := cfg.Pipelines[0].Destinations
if got, want := destinations[0].State.Mode, StateModeSingleOwner; got != want {
t.Fatalf("archive state mode = %q, want %q", got, want)
}
if got, want := destinations[1].State.Mode, StateModeSharedRoot; got != want {
t.Fatalf("web state mode = %q, want %q", got, want)
}
}
func TestLoadFileAcceptsExplicitTakeoverModes(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: same-pipeline
backend: local
path: /same-pipeline
takeover:
mode: same_pipeline
- id: same-source
backend: local
path: /same-source
takeover:
mode: same_source
- id: any-managed
backend: local
path: /any-managed
takeover:
mode: any_managed
- id: never
backend: local
path: /never
takeover:
mode: never
`)
destinations := cfg.Pipelines[0].Destinations
wants := []string{
TakeoverModeSamePipeline,
TakeoverModeSameSource,
TakeoverModeAnyManaged,
TakeoverModeNever,
}
for index, want := range wants {
if got := destinations[index].Takeover.Mode; got != want {
t.Fatalf("destinations[%d].takeover.mode = %q, want %q", index, got, want)
}
if got, want := destinations[1].Workflow, WorkflowReplacement; got != want {
t.Fatalf("web workflow = %q, want %q", got, want)
}
}
@@ -899,8 +815,63 @@ pipelines:
`, "backend ftp is unsupported")
}
func TestLoadFileRejectsInvalidTransferAction(t *testing.T) {
func TestLoadFileRejectsInvalidWorkflow(t *testing.T) {
assertLoadError(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
workflow: append
`, "workflow must be additive or replacement")
}
func TestLoadFileRejectsLegacyDestinationPolicyFields(t *testing.T) {
tests := map[string]string{
"state": `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
state:
mode: single_owner
`,
"reconciliation": `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
reconciliation:
mode: replace
`,
"takeover": `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
takeover:
mode: same_pipeline
`,
"transfer": `
pipelines:
- id: reports
source:
@@ -911,40 +882,14 @@ pipelines:
backend: local
path: /archive
transfer:
on_destination_older: overwrite
`, "on_destination_older must be replace or fail")
}
func TestLoadFileRejectsInvalidTakeoverMode(t *testing.T) {
assertLoadError(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
takeover:
mode: unmanaged
`, "takeover.mode must be same_pipeline, same_source, any_managed, or never")
}
func TestLoadFileRejectsUnknownTakeoverFields(t *testing.T) {
assertLoadError(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
takeover:
surprise: true
`, "field surprise not found")
on_destination_older: replace
`,
}
for name, body := range tests {
t.Run(name, func(t *testing.T) {
assertLoadError(t, body, "field "+name+" not found")
})
}
}
func TestLoadFileRejectsInvalidValidationAction(t *testing.T) {
@@ -1020,8 +965,6 @@ func TestExampleConfigsLoad(t *testing.T) {
"../../examples/local-index.yml",
"../../examples/fan-out.yml",
"../../examples/archive-and-latest.yml",
"../../examples/merge-reconciliation.yml",
"../../examples/shared-root.yml",
"../../examples/http-upload-local.yml",
"../../examples/ssh-destination.yml",
"../../examples/s3-destination.yml",

43
internal/config/validate.go
View File

@@ -74,11 +74,8 @@ func Validate(cfg Config) error {
errs = validatePublishTransformPolicy(errs, destinationContext, destination.Publish, destination.Transform)
errs = validatePathMapping(errs, destinationContext+".path_mapping", destination.PathMap)
errs = validateLinks(errs, destinationContext+".links", destination.Links)
errs = validateStatePolicy(errs, destinationContext+".state", destination.State)
errs = validateReconciliationPolicy(errs, destinationContext+".reconciliation", destination.Reconciliation)
errs = validateTakeoverPolicy(errs, destinationContext+".takeover", destination.Takeover)
errs = validateWorkflow(errs, destinationContext+".workflow", destination.Workflow)
errs = validateRetentionPolicy(errs, destinationContext+".retention", destination.Retention)
errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer)
}
}
@@ -369,18 +366,9 @@ func validateLinks(errs ValidationErrors, context string, links *Links) Validati
return errs
}
func validateReconciliationPolicy(errs ValidationErrors, context string, policy ReconciliationPolicy) ValidationErrors {
if policy.Mode != ReconciliationModeReplace && policy.Mode != ReconciliationModeMerge {
errs = append(errs, context+".mode must be "+ReconciliationModeReplace+" or "+ReconciliationModeMerge)
}
return errs
}
func validateTakeoverPolicy(errs ValidationErrors, context string, policy TakeoverPolicy) ValidationErrors {
switch policy.Mode {
case TakeoverModeSamePipeline, TakeoverModeSameSource, TakeoverModeAnyManaged, TakeoverModeNever:
default:
errs = append(errs, context+".mode must be "+TakeoverModeSamePipeline+", "+TakeoverModeSameSource+", "+TakeoverModeAnyManaged+", or "+TakeoverModeNever)
func validateWorkflow(errs ValidationErrors, context, workflow string) ValidationErrors {
if workflow != WorkflowAdditive && workflow != WorkflowReplacement {
errs = append(errs, context+" must be "+WorkflowAdditive+" or "+WorkflowReplacement)
}
return errs
}
@@ -401,26 +389,3 @@ func validateRetentionPolicy(errs ValidationErrors, context string, policy Reten
}
return errs
}
func validateStatePolicy(errs ValidationErrors, context string, policy StatePolicy) ValidationErrors {
if policy.Mode != StateModeSingleOwner && policy.Mode != StateModeSharedRoot {
errs = append(errs, context+".mode must be "+StateModeSingleOwner+" or "+StateModeSharedRoot)
}
return errs
}
func validateTransferPolicy(errs ValidationErrors, context string, policy TransferPolicy) ValidationErrors {
if policy.OnDestinationSame != TransferActionSkip && policy.OnDestinationSame != TransferActionFail {
errs = append(errs, context+".on_destination_same must be skip or fail")
}
if policy.OnDestinationOlder != TransferActionReplace && policy.OnDestinationOlder != TransferActionFail {
errs = append(errs, context+".on_destination_older must be replace or fail")
}
if policy.OnDestinationNewer != TransferActionSkip && policy.OnDestinationNewer != TransferActionFail && policy.OnDestinationNewer != TransferActionReplace {
errs = append(errs, context+".on_destination_newer must be skip, replace, or fail")
}
if policy.OnConflict != TransferActionFail && policy.OnConflict != TransferActionReplace {
errs = append(errs, context+".on_conflict must be fail or replace")
}
return errs
}

115
internal/config/validate_test.go
View File

@@ -51,29 +51,6 @@ func TestValidateChecksPublishTransformPolicy(t *testing.T) {
}
}
func TestValidateAcceptsForceReplacementTransferActions(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{
Backend: BackendLocal,
Path: "/source",
},
Destinations: []Destination{{
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Transfer: TransferPolicy{
OnDestinationNewer: TransferActionReplace,
OnConflict: TransferActionReplace,
},
}},
}}}
ApplyDefaults(&cfg)
if err := Validate(cfg); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestValidatePathMapping(t *testing.T) {
tests := []struct {
name string
@@ -108,51 +85,15 @@ func TestValidatePathMapping(t *testing.T) {
}
}
func TestValidateReconciliationPolicy(t *testing.T) {
func TestValidateWorkflow(t *testing.T) {
tests := []struct {
name string
mode string
wantErr bool
name string
workflow string
wantErr bool
}{
{name: "replace", mode: ReconciliationModeReplace},
{name: "merge", mode: ReconciliationModeMerge},
{name: "invalid", mode: "append", wantErr: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{Backend: BackendLocal, Path: "/source"},
Destinations: []Destination{{
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Reconciliation: ReconciliationPolicy{Mode: tt.mode},
}},
}}}
ApplyDefaults(&cfg)
err := Validate(cfg)
if tt.wantErr && err == nil {
t.Fatal("Validate() error = nil, want error")
}
if !tt.wantErr && err != nil {
t.Fatalf("Validate() error = %v", err)
}
})
}
}
func TestValidateTakeoverPolicy(t *testing.T) {
tests := []struct {
name string
mode string
wantErr bool
}{
{name: "same pipeline", mode: TakeoverModeSamePipeline},
{name: "same source", mode: TakeoverModeSameSource},
{name: "any managed", mode: TakeoverModeAnyManaged},
{name: "never", mode: TakeoverModeNever},
{name: "invalid", mode: "unmanaged", wantErr: true},
{name: "additive", workflow: WorkflowAdditive},
{name: "replacement", workflow: WorkflowReplacement},
{name: "invalid", workflow: "append", wantErr: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -163,7 +104,7 @@ func TestValidateTakeoverPolicy(t *testing.T) {
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Takeover: TakeoverPolicy{Mode: tt.mode},
Workflow: tt.workflow,
}},
}}}
ApplyDefaults(&cfg)
@@ -178,7 +119,7 @@ func TestValidateTakeoverPolicy(t *testing.T) {
}
}
func TestValidateTakeoverPolicyReportsFieldContext(t *testing.T) {
func TestValidateWorkflowReportsFieldContext(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{Backend: BackendLocal, Path: "/source"},
@@ -186,7 +127,7 @@ func TestValidateTakeoverPolicyReportsFieldContext(t *testing.T) {
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Takeover: TakeoverPolicy{Mode: "unmanaged"},
Workflow: "append",
}},
}}}
ApplyDefaults(&cfg)
@@ -194,46 +135,12 @@ func TestValidateTakeoverPolicyReportsFieldContext(t *testing.T) {
if err == nil {
t.Fatal("Validate() error = nil, want error")
}
want := "pipelines[0].destinations[0].takeover.mode must be same_pipeline, same_source, any_managed, or never"
want := "pipelines[0].destinations[0].workflow must be additive or replacement"
if !strings.Contains(err.Error(), want) {
t.Fatalf("Validate() error = %q, want %q", err, want)
}
}
func TestValidateStatePolicy(t *testing.T) {
tests := []struct {
name string
mode string
wantErr bool
}{
{name: "single owner", mode: StateModeSingleOwner},
{name: "shared root", mode: StateModeSharedRoot},
{name: "invalid", mode: "shared", wantErr: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{Backend: BackendLocal, Path: "/source"},
Destinations: []Destination{{
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
State: StatePolicy{Mode: tt.mode},
}},
}}}
ApplyDefaults(&cfg)
err := Validate(cfg)
if tt.wantErr && err == nil {
t.Fatal("Validate() error = nil, want error")
}
if !tt.wantErr && err != nil {
t.Fatalf("Validate() error = %v", err)
}
})
}
}
func TestValidateRetentionPolicy(t *testing.T) {
olderThan := Duration(24 * time.Hour)
zeroDuration := Duration(0)