From c4e8ebff6f57d7c71db64d8bb0537d94d36073c2 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 18 Jun 2026 15:05:09 +0000 Subject: [PATCH] Add destination takeover config policy --- internal/config/config.go | 5 ++ internal/config/defaults.go | 10 ++++ internal/config/load_test.go | 79 ++++++++++++++++++++++++++++++++ internal/config/validate.go | 10 ++++ internal/config/validate_test.go | 58 +++++++++++++++++++++++ 5 files changed, 162 insertions(+) diff --git a/internal/config/config.go b/internal/config/config.go index 3e394d4..3c6d8eb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -57,6 +57,7 @@ type Destination struct { Links *Links `yaml:"links"` State StatePolicy `yaml:"state"` Reconciliation ReconciliationPolicy `yaml:"reconciliation"` + Takeover TakeoverPolicy `yaml:"takeover"` Retention RetentionPolicy `yaml:"retention"` Transfer TransferPolicy `yaml:"transfer"` } @@ -126,6 +127,10 @@ type ReconciliationPolicy struct { Mode string `yaml:"mode"` } +type TakeoverPolicy struct { + Mode string `yaml:"mode"` +} + type StatePolicy struct { Mode string `yaml:"mode"` } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 77bdad9..56d8d71 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -47,6 +47,13 @@ const ( ReconciliationModeMerge = "merge" ) +const ( + TakeoverModeSamePipeline = "same_pipeline" + TakeoverModeSameSource = "same_source" + TakeoverModeAnyManaged = "any_managed" + TakeoverModeNever = "never" +) + const ( StateModeSingleOwner = "single_owner" StateModeSharedRoot = "shared_root" @@ -95,6 +102,9 @@ func ApplyDefaults(cfg *Config) { if destination.Reconciliation.Mode == "" { destination.Reconciliation.Mode = ReconciliationModeReplace } + if destination.Takeover.Mode == "" { + destination.Takeover.Mode = TakeoverModeSamePipeline + } if destination.Transfer.OnDestinationSame == "" { destination.Transfer.OnDestinationSame = TransferActionSkip } diff --git a/internal/config/load_test.go b/internal/config/load_test.go index ddf6b03..e09c2fd 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -39,6 +39,9 @@ pipelines: 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 destination.Retention.Prune.Enabled { t.Fatal("retention.prune.enabled default = true, want false") } @@ -231,6 +234,50 @@ pipelines: } } +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) + } + } +} + func TestLoadFileAcceptsRetentionPruneConfig(t *testing.T) { cfg := loadConfig(t, ` pipelines: @@ -868,6 +915,38 @@ pipelines: `, "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") +} + func TestLoadFileRejectsInvalidValidationAction(t *testing.T) { assertLoadError(t, ` pipelines: diff --git a/internal/config/validate.go b/internal/config/validate.go index 0e78034..caa0cdf 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -76,6 +76,7 @@ func Validate(cfg Config) error { 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 = validateRetentionPolicy(errs, destinationContext+".retention", destination.Retention) errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer) } @@ -375,6 +376,15 @@ func validateReconciliationPolicy(errs ValidationErrors, context string, policy 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) + } + return errs +} + func validateRetentionPolicy(errs ValidationErrors, context string, policy RetentionPolicy) ValidationErrors { prune := policy.Prune if !prune.Enabled { diff --git a/internal/config/validate_test.go b/internal/config/validate_test.go index 5b538c2..46ea6b8 100644 --- a/internal/config/validate_test.go +++ b/internal/config/validate_test.go @@ -142,6 +142,64 @@ func TestValidateReconciliationPolicy(t *testing.T) { } } +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}, + } + 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", + Takeover: TakeoverPolicy{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 TestValidateTakeoverPolicyReportsFieldContext(t *testing.T) { + cfg := Config{Pipelines: []Pipeline{{ + ID: "reports", + Source: Backend{Backend: BackendLocal, Path: "/source"}, + Destinations: []Destination{{ + ID: "archive", + Backend: BackendLocal, + Path: "/destination", + Takeover: TakeoverPolicy{Mode: "unmanaged"}, + }}, + }}} + ApplyDefaults(&cfg) + err := Validate(cfg) + 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" + if !strings.Contains(err.Error(), want) { + t.Fatalf("Validate() error = %q, want %q", err, want) + } +} + func TestValidateStatePolicy(t *testing.T) { tests := []struct { name string