401 lines
16 KiB
Go
401 lines
16 KiB
Go
package publish
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/state"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/storage"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
|
|
)
|
|
|
|
var (
|
|
planCreatedAt = time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)
|
|
planUpdatedAt = time.Date(2026, 6, 1, 9, 30, 0, 0, time.UTC)
|
|
)
|
|
|
|
func TestBuildAdditivePublishesCatalogOutputs(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionPublishNew || plan.Workflow != config.WorkflowAdditive {
|
|
t.Fatalf("plan action=%s workflow=%s, want publish_new additive", plan.Action, plan.Workflow)
|
|
}
|
|
if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete))
|
|
}
|
|
for _, output := range plan.CatalogOutputsToWrite {
|
|
if output.PipelineID != "reports" || output.DestinationID != "archive" {
|
|
t.Fatalf("catalog output owner = %s/%s", output.PipelineID, output.DestinationID)
|
|
}
|
|
if output.Source.ID != req.SourceBundle.Manifest.ID || output.Source.Digest != req.SourceBundle.Manifest.Digest || !output.Source.Created.Equal(req.SourceBundle.Manifest.Created) {
|
|
t.Fatalf("catalog output source = %#v", output.Source)
|
|
}
|
|
if output.Kind == state.OutputKindSource && output.SourcePath != "" {
|
|
t.Fatalf("source catalog output source_path = %q, want empty", output.SourcePath)
|
|
}
|
|
if !output.CreatedAt.Equal(planUpdatedAt) || !output.UpdatedAt.Equal(planUpdatedAt) {
|
|
t.Fatalf("catalog output times = %s/%s", output.CreatedAt, output.UpdatedAt)
|
|
}
|
|
}
|
|
if _, err := destinationBackend.Stat(context.Background(), storage.StateFileName); !storage.IsNotFound(err) {
|
|
t.Fatalf("destination state stat error = %v, want missing", err)
|
|
}
|
|
}
|
|
|
|
func TestBuildAdditiveOverwritesManagedAndRetainsOthers(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
existing := baseCatalog(req)
|
|
existing.Outputs = []state.CatalogOutputFile{
|
|
catalogOutput(req, "reports", "archive", "report.md", state.OutputKindSource, planCreatedAt),
|
|
catalogOutput(req, "reports", "web", "old.txt", state.OutputKindSource, planCreatedAt),
|
|
}
|
|
writeCatalogState(t, destinationBackend, "", existing)
|
|
testutil.WriteFakeFile(t, destinationBackend, "report.md", "old")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionUpsertAdditive {
|
|
t.Fatalf("plan action = %s, want %s", plan.Action, ActionUpsertAdditive)
|
|
}
|
|
if len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("delete outputs = %#v, want none", plan.CatalogOutputsToDelete)
|
|
}
|
|
if len(plan.CatalogOutputsToRetain) != 1 || plan.CatalogOutputsToRetain[0].Path != "old.txt" {
|
|
t.Fatalf("retained outputs = %#v, want old.txt", plan.CatalogOutputsToRetain)
|
|
}
|
|
written, ok := state.FindCatalogOutputByPath(plan.CatalogOutputsToWrite, "report.md")
|
|
if !ok {
|
|
t.Fatalf("written outputs = %#v, want report.md", plan.CatalogOutputsToWrite)
|
|
}
|
|
if !written.CreatedAt.Equal(planCreatedAt) || !written.UpdatedAt.Equal(planUpdatedAt) {
|
|
t.Fatalf("report.md times = %s/%s, want created preserved and updated now", written.CreatedAt, written.UpdatedAt)
|
|
}
|
|
}
|
|
|
|
func TestBuildReplacementDeletesCurrentOwnerAndRetainsOtherOwners(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement)
|
|
existing := baseCatalog(req)
|
|
existing.Outputs = []state.CatalogOutputFile{
|
|
catalogOutput(req, "reports", "archive", "report.md", state.OutputKindSource, planCreatedAt),
|
|
catalogOutput(req, "reports", "archive", "stale.txt", state.OutputKindSource, planCreatedAt),
|
|
catalogOutput(req, "reports", "web", "shared.txt", state.OutputKindSource, planCreatedAt),
|
|
}
|
|
writeCatalogState(t, destinationBackend, "", existing)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionReplaceCatalog {
|
|
t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceCatalog)
|
|
}
|
|
if len(plan.CatalogOutputsToDelete) != 1 || plan.CatalogOutputsToDelete[0].Path != "stale.txt" {
|
|
t.Fatalf("delete outputs = %#v, want stale.txt", plan.CatalogOutputsToDelete)
|
|
}
|
|
if len(plan.CatalogOutputsToRetain) != 1 || plan.CatalogOutputsToRetain[0].Path != "shared.txt" {
|
|
t.Fatalf("retained outputs = %#v, want shared.txt", plan.CatalogOutputsToRetain)
|
|
}
|
|
}
|
|
|
|
func TestBuildTransfersManagedPathOwnership(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement)
|
|
existing := baseCatalog(req)
|
|
existing.Outputs = []state.CatalogOutputFile{
|
|
catalogOutput(req, "reports", "web", "report.md", state.OutputKindSource, planCreatedAt),
|
|
}
|
|
writeCatalogState(t, destinationBackend, "", existing)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
written, ok := state.FindCatalogOutputByPath(plan.CatalogOutputsToWrite, "report.md")
|
|
if !ok {
|
|
t.Fatalf("written outputs = %#v, want report.md", plan.CatalogOutputsToWrite)
|
|
}
|
|
if written.PipelineID != "reports" || written.DestinationID != "archive" {
|
|
t.Fatalf("written owner = %s/%s, want reports/archive", written.PipelineID, written.DestinationID)
|
|
}
|
|
if !written.CreatedAt.Equal(planCreatedAt) || !written.UpdatedAt.Equal(planUpdatedAt) {
|
|
t.Fatalf("written times = %s/%s", written.CreatedAt, written.UpdatedAt)
|
|
}
|
|
if len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("retain=%#v delete=%#v, want no old record for overwritten path", plan.CatalogOutputsToRetain, plan.CatalogOutputsToDelete)
|
|
}
|
|
}
|
|
|
|
func TestBuildRejectsUnmanagedPlannedPathCollision(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
writeCatalogState(t, destinationBackend, "", baseCatalog(req))
|
|
testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err == nil {
|
|
t.Fatal("Build() error = nil, want unmanaged collision")
|
|
}
|
|
if plan.Action != ActionFailUnmanaged || !strings.Contains(err.Error(), "not managed by catalog state") {
|
|
t.Fatalf("plan action=%s error=%v, want unmanaged catalog collision", plan.Action, err)
|
|
}
|
|
}
|
|
|
|
func TestBuildForceReplacesUnmanagedPlannedPathCollision(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
req.Force = true
|
|
existing := baseCatalog(req)
|
|
writeCatalogState(t, destinationBackend, "", existing)
|
|
testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot)
|
|
}
|
|
if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete))
|
|
}
|
|
}
|
|
|
|
func TestBuildRejectsNoStateNonEmptyDestinationWithoutForce(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
testutil.WriteFakeFile(t, destinationBackend, "unplanned.txt", "unmanaged")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err == nil {
|
|
t.Fatal("Build() error = nil, want unmanaged destination")
|
|
}
|
|
if plan.Action != ActionFailUnmanaged || !strings.Contains(err.Error(), "destination has content but no distributor state") {
|
|
t.Fatalf("plan action=%s error=%v, want unmanaged destination", plan.Action, err)
|
|
}
|
|
}
|
|
|
|
func TestBuildForceReplacesNoStateNonEmptyDestination(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
req.Force = true
|
|
testutil.WriteFakeFile(t, destinationBackend, "unplanned.txt", "unmanaged")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot)
|
|
}
|
|
if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete))
|
|
}
|
|
}
|
|
|
|
func TestBuildPlansSupersededLegacyAdditive(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
legacyState := testutil.DestinationState(req.SourceBundle.Manifest, testutil.DestinationStateOptions{})
|
|
writeJSONState(t, destinationBackend, "", legacyState)
|
|
testutil.WriteFakeFile(t, destinationBackend, "report.md", "legacy")
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.SupersededLegacy == nil || plan.SupersededLegacy.SchemaVersion != state.SchemaVersion {
|
|
t.Fatalf("superseded legacy = %#v", plan.SupersededLegacy)
|
|
}
|
|
if plan.Action != ActionUpsertAdditive || plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s clear=%t, want additive overwrite without clear", plan.Action, plan.ClearDestinationRoot)
|
|
}
|
|
}
|
|
|
|
func TestBuildPlansSupersededLegacyReplacementClear(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement)
|
|
legacyState := testutil.DestinationState(req.SourceBundle.Manifest, testutil.DestinationStateOptions{})
|
|
writeJSONState(t, destinationBackend, "", legacyState)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionReplaceCatalog || !plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s clear=%t, want replacement clear", plan.Action, plan.ClearDestinationRoot)
|
|
}
|
|
}
|
|
|
|
func TestBuildRejectsInvalidOrFutureState(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
data string
|
|
}{
|
|
{name: "invalid json", data: `{"schema_version":`},
|
|
{name: "future schema", data: `{"schema_version":99}`},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
testutil.WriteFakeFile(t, destinationBackend, storage.StateFileName, tt.data)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err == nil {
|
|
t.Fatal("Build() error = nil, want conflict")
|
|
}
|
|
if plan.Action != ActionFailConflict {
|
|
t.Fatalf("plan action = %s, want %s", plan.Action, ActionFailConflict)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBuildForceReplacesInvalidOrFutureState(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
data string
|
|
}{
|
|
{name: "invalid json", data: `{"schema_version":`},
|
|
{name: "future schema", data: `{"schema_version":99}`},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive)
|
|
req.Force = true
|
|
testutil.WriteFakeFile(t, destinationBackend, storage.StateFileName, tt.data)
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot)
|
|
}
|
|
if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 {
|
|
t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete))
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBuildForceDoesNotChangeValidCatalogActions(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
workflow string
|
|
want Action
|
|
}{
|
|
{name: "additive", workflow: config.WorkflowAdditive, want: ActionUpsertAdditive},
|
|
{name: "replacement", workflow: config.WorkflowReplacement, want: ActionReplaceCatalog},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
_, destinationBackend, req := catalogPlanRequest(t, tt.workflow)
|
|
req.Force = true
|
|
writeCatalogState(t, destinationBackend, "", baseCatalog(req))
|
|
|
|
plan, err := Build(context.Background(), req)
|
|
if err != nil {
|
|
t.Fatalf("Build() error = %v", err)
|
|
}
|
|
if plan.Action != tt.want || plan.Force || plan.ClearDestinationRoot {
|
|
t.Fatalf("plan action=%s force=%t clear=%t, want %s without forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestValidateRequestRejectsInvalidWorkflow(t *testing.T) {
|
|
sourceBackend, destinationBackend, req := catalogPlanRequest(t, "append")
|
|
req.SourceBackend = sourceBackend
|
|
req.DestinationBackend = destinationBackend
|
|
|
|
err := validateRequest(req)
|
|
if err == nil {
|
|
t.Fatal("validateRequest() error = nil, want invalid workflow")
|
|
}
|
|
if !strings.Contains(err.Error(), "destination.workflow") {
|
|
t.Fatalf("validateRequest() error = %v, want workflow context", err)
|
|
}
|
|
}
|
|
|
|
func catalogPlanRequest(t *testing.T, workflow string) (*fake.Backend, *fake.Backend, Request) {
|
|
t.Helper()
|
|
sourceBackend := fake.New()
|
|
destinationBackend := fake.New()
|
|
sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{})
|
|
return sourceBackend, destinationBackend, Request{
|
|
PipelineID: "reports",
|
|
DestinationID: "archive",
|
|
SourceBundle: sourceBundle,
|
|
SourceBackend: sourceBackend,
|
|
DestinationBackend: destinationBackend,
|
|
DestinationBundlePath: "",
|
|
PathMapping: config.PathMappingPreserveRelative,
|
|
Publish: config.PublishPolicy{Source: true},
|
|
Workflow: workflow,
|
|
DistributorVersion: "test",
|
|
Now: planUpdatedAt,
|
|
}
|
|
}
|
|
|
|
func baseCatalog(req Request) state.CatalogState {
|
|
return state.CatalogState{
|
|
SchemaVersion: state.CatalogSchemaVersion,
|
|
DistributorVersion: "previous",
|
|
CreatedAt: planCreatedAt,
|
|
UpdatedAt: planCreatedAt,
|
|
State: state.StatePolicy{Mode: state.StateModeCatalog},
|
|
Outputs: []state.CatalogOutputFile{},
|
|
}
|
|
}
|
|
|
|
func catalogOutput(req Request, pipelineID, destinationID, path, kind string, createdAt time.Time) state.CatalogOutputFile {
|
|
sourceSHA := req.SourceBundle.Manifest.Files[0].SHA256
|
|
sourceSize := req.SourceBundle.Manifest.Files[0].Size
|
|
for _, file := range req.SourceBundle.Manifest.Files {
|
|
if file.Path == path {
|
|
sourceSHA = file.SHA256
|
|
sourceSize = file.Size
|
|
break
|
|
}
|
|
}
|
|
output := state.CatalogOutputFile{
|
|
Path: path,
|
|
PipelineID: pipelineID,
|
|
DestinationID: destinationID,
|
|
Source: state.CatalogSourceIdentity{
|
|
ID: req.SourceBundle.Manifest.ID,
|
|
Digest: req.SourceBundle.Manifest.Digest,
|
|
Created: req.SourceBundle.Manifest.Created,
|
|
},
|
|
Kind: kind,
|
|
SHA256: sourceSHA,
|
|
Size: sourceSize,
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: createdAt,
|
|
}
|
|
return output
|
|
}
|
|
|
|
func writeCatalogState(t *testing.T, backend *fake.Backend, relative string, catalog state.CatalogState) {
|
|
t.Helper()
|
|
writeJSONState(t, backend, relative, catalog)
|
|
}
|
|
|
|
func writeJSONState(t *testing.T, backend *fake.Backend, relative string, value any) {
|
|
t.Helper()
|
|
data, err := json.MarshalIndent(value, "", " ")
|
|
if err != nil {
|
|
t.Fatalf("marshal state: %v", err)
|
|
}
|
|
statePath, err := storage.StatePath(relative)
|
|
if err != nil {
|
|
t.Fatalf("state path: %v", err)
|
|
}
|
|
testutil.WriteFakeFile(t, backend, statePath, string(append(data, '\n')))
|
|
}
|