9 Commits

40 changed files with 1284 additions and 483 deletions

View File

@@ -22,7 +22,9 @@ Each file requires `path`, `sha256`, and `size`. Digests must use lowercase `sha
## Validation
Bundle validation checks source path safety, duplicate file paths, reserved paths, file existence, regular-file type, file size, per-file SHA-256, and the top-level bundle digest.
`ValidateManifest` owns normalized source manifest semantics: schema version, id, digest format, timestamp presence, file list presence, source path safety, duplicate file paths, reserved paths, file digest format, non-negative file sizes, and the top-level bundle digest.
Storage-backed bundle validation additionally checks file existence, regular-file type, file size, and per-file SHA-256.
The bundle digest is SHA-256 of a deterministic JSON array of file records in manifest order with fields `path`, `sha256`, and `size`.

View File

@@ -6,7 +6,7 @@
## Inputs and outputs
Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, publish policy, transfer policy, destination bundle path, and existing destination state.
Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, publish policy, transform policy, transformer resolver, transfer policy, destination bundle path, and existing destination state.
Output is a plan with an action, reason, and selected source or generated outputs. Execution writes selected source files, generated files, and `.distributor.json` for publish or replacement actions.
@@ -18,7 +18,7 @@ Supported actions are publish new, replace older destination, skip same source,
The current implementation publishes source files and Markdown-to-HTML sidecar outputs. Remote backend execution is not implemented.
The package uses `internal/state` for destination comparison and `internal/storage` for IO. It does not parse CLI flags or load config files.
The package uses `internal/state` for destination comparison, `internal/storage` for IO, and the shared `internal/config` publish/transform policy helper for request validation. It resolves transforms through a narrow resolver supplied by the caller; concrete transform registration is owned by the app layer. It does not parse CLI flags or load config files.
## Safety

View File

@@ -14,7 +14,7 @@ Entries report a logical path, type, and size when available. Entry types are `f
Core packages should depend on `internal/storage`, not adapter packages. Adapter-specific path handling stays behind backend implementations.
The local adapter lives in `internal/adapters/local`. The fake backend lives in `internal/storage/fake` for tests and is not registered for runtime use.
The local adapter lives in `internal/adapters/local`. Runtime backend construction is wired through the app-level backend factory and storage registry. The fake backend lives in `internal/storage/fake` for tests and is not registered for runtime use.
## Paths

View File

@@ -18,6 +18,8 @@ Generated HTML is deterministic for the same source content and transform config
Transforms do not publish files, mutate source bundles, or write destination state. Publish planning selects and writes transform outputs.
The app layer owns default transform registration. The transform package does not import concrete transform implementations.
## Tests
Before changing transform behavior, inspect tests under `internal/transform`.

View File

@@ -234,22 +234,10 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(bundlePath); err != nil {
return err
}
targets := make([]string, 0, len(managedOutputPaths)+1)
for _, outputPath := range managedOutputPaths {
target, err := storage.Join(bundlePath, outputPath)
if err != nil {
return err
}
targets = append(targets, target)
}
statePath, err := storage.StatePath(bundlePath)
targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths)
if err != nil {
return err
}
targets = append(targets, statePath)
for _, logicalPath := range targets {
nativePath, err := b.nativePath(logicalPath, false)

View File

@@ -145,16 +145,20 @@ func TestBackendManagedDeletion(t *testing.T) {
backend := newBackend(t)
mustWrite(t, backend, "bundle/report.html", "html")
mustWrite(t, backend, "bundle/keep.txt", "keep")
mustWrite(t, backend, "bundle/.distributor.json", "{}")
statePath, err := storage.StatePath("bundle")
if err != nil {
t.Fatalf("StatePath() error = %v", err)
}
mustWrite(t, backend, statePath, "{}")
err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
err = backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
if err != nil {
t.Fatalf("DeleteManagedBundle() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.html"); !storage.IsNotFound(err) {
t.Fatalf("managed output stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/.distributor.json"); !storage.IsNotFound(err) {
if _, err := backend.Stat(context.Background(), statePath); !storage.IsNotFound(err) {
t.Fatalf("state stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/keep.txt"); err != nil {

View File

@@ -1,14 +1,10 @@
package app
import "errors"
const Name = "distributor"
// Version can be replaced at build time with -ldflags "-X .../internal/app.Version=<value>".
var Version = "dev"
var ErrNotImplemented = errors.New("not implemented")
func VersionString() string {
return Name + " " + Version
}

45
internal/app/backends.go Normal file
View File

@@ -0,0 +1,45 @@
package app
import (
"context"
"fmt"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
const storagePathKey = "path"
type backendFactory struct {
registry *storage.Registry
}
func newBackendFactory() *backendFactory {
registry := storage.NewRegistry()
_ = registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
return local.New(cfg[storagePathKey])
})
return &backendFactory{registry: registry}
}
func (f *backendFactory) openSource(ctx context.Context, source config.Backend) (storage.Backend, error) {
if source.Backend != config.BackendLocal {
return nil, fmt.Errorf("source backend %s is not implemented for execution", source.Backend)
}
return f.registry.Open(ctx, source.Backend, storage.OpenConfig{storagePathKey: source.Path})
}
func (f *backendFactory) openDestination(ctx context.Context, destination config.Destination) (storage.Backend, error) {
if destination.Backend != config.BackendLocal {
return nil, fmt.Errorf("backend %s is not implemented for execution", destination.Backend)
}
return f.registry.Open(ctx, destination.Backend, storage.OpenConfig{storagePathKey: destination.Path})
}
func (f *backendFactory) openLocalPath(ctx context.Context, path string) (storage.Backend, error) {
return f.registry.Open(ctx, config.BackendLocal, storage.OpenConfig{storagePathKey: path})
}

View File

@@ -0,0 +1,72 @@
package app
import (
"context"
"strings"
"testing"
"gitea.maximumdirect.net/eric/distributor/internal/config"
)
func TestBackendFactoryOpensLocalSource(t *testing.T) {
factory := newBackendFactory()
backend, err := factory.openSource(context.Background(), config.Backend{
Backend: config.BackendLocal,
Path: t.TempDir(),
})
if err != nil {
t.Fatalf("openSource() error = %v", err)
}
if backend == nil {
t.Fatal("openSource() backend = nil")
}
}
func TestBackendFactoryOpensLocalDestination(t *testing.T) {
factory := newBackendFactory()
backend, err := factory.openDestination(context.Background(), config.Destination{
Backend: config.BackendLocal,
Path: t.TempDir(),
})
if err != nil {
t.Fatalf("openDestination() error = %v", err)
}
if backend == nil {
t.Fatal("openDestination() backend = nil")
}
}
func TestBackendFactoryOpensDirectLocalPath(t *testing.T) {
factory := newBackendFactory()
backend, err := factory.openLocalPath(context.Background(), t.TempDir())
if err != nil {
t.Fatalf("openLocalPath() error = %v", err)
}
if backend == nil {
t.Fatal("openLocalPath() backend = nil")
}
}
func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) {
factory := newBackendFactory()
_, err := factory.openSource(context.Background(), config.Backend{
Backend: config.BackendSSH,
URI: "ssh://reports@example.com:22",
Path: "/reports",
})
if err == nil || !strings.Contains(err.Error(), "source backend ssh is not implemented for execution") {
t.Fatalf("openSource() error = %v, want not implemented", err)
}
}
func TestBackendFactoryRejectsUnsupportedDestination(t *testing.T) {
factory := newBackendFactory()
_, err := factory.openDestination(context.Background(), config.Destination{
Backend: config.BackendS3,
Endpoint: "https://s3.example.com",
Bucket: "reports",
})
if err == nil || !strings.Contains(err.Error(), "backend s3 is not implemented for execution") {
t.Fatalf("openDestination() error = %v, want not implemented", err)
}
}

View File

@@ -5,8 +5,8 @@ import (
"fmt"
"io"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
type InspectOptions struct {
@@ -18,7 +18,7 @@ func Inspect(ctx context.Context, options InspectOptions) error {
if options.Path == "" {
return fmt.Errorf("inspect command requires a path")
}
backend, err := local.New(options.Path)
backend, err := newBackendFactory().openLocalPath(ctx, options.Path)
if err != nil {
return err
}
@@ -40,7 +40,7 @@ func writeInspection(w io.Writer, bundles []bundle.Bundle) error {
if _, err := fmt.Fprintf(
w,
"- path=%s id=%s created=%s digest=%s files=%d\n",
displayBundlePath(sourceBundle.RootRelativePath),
storage.DisplayPath(sourceBundle.RootRelativePath),
sourceBundle.Manifest.ID,
sourceBundle.Manifest.Created.Format("2006-01-02T15:04:05Z07:00"),
sourceBundle.Manifest.Digest,
@@ -56,10 +56,3 @@ func writeInspection(w io.Writer, bundles []bundle.Bundle) error {
}
return nil
}
func displayBundlePath(path string) string {
if path == "" {
return "."
}
return path
}

View File

@@ -1,5 +0,0 @@
package app
type Pipeline struct {
ID string
}

View File

@@ -7,11 +7,11 @@ import (
"io"
"strings"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
"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/storage"
)
type RunOptions struct {
@@ -44,18 +44,17 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
}
summary := runSummary{dryRun: options.DryRun}
var failures runFailures
backends := newBackendFactory()
transforms := newTransformRegistry()
if options.Stdout != nil {
if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
return err
}
}
for _, pipeline := range cfg.Pipelines {
if pipeline.Source.Backend != config.BackendLocal {
return fmt.Errorf("pipeline %s source backend %s is not implemented for execution", pipeline.ID, pipeline.Source.Backend)
}
sourceBackend, err := local.New(pipeline.Source.Path)
sourceBackend, err := backends.openSource(ctx, pipeline.Source)
if err != nil {
return err
return fmt.Errorf("pipeline %s: %w", pipeline.ID, err)
}
bundles, err := bundle.Discover(ctx, sourceBackend, "")
if err != nil {
@@ -68,18 +67,9 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
}
for _, sourceBundle := range bundles {
for _, destination := range pipeline.Destinations {
if destination.Backend != config.BackendLocal {
err := fmt.Errorf("backend %s is not implemented for execution", destination.Backend)
failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
if options.Stdout != nil {
writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err)
}
continue
}
destinationBackend, err := local.New(destination.Path)
destinationBackend, err := backends.openDestination(ctx, destination)
if err != nil {
failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
if options.Stdout != nil {
writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err)
@@ -95,6 +85,7 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
DestinationBundlePath: sourceBundle.RootRelativePath,
Publish: *destination.Publish,
Transform: destination.Transform,
Transformers: transforms,
Transfer: destination.Transfer,
DistributorVersion: Version,
}
@@ -106,20 +97,20 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
writePlanLine(options.Stdout, plan, err)
}
if err != nil {
failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
}
summary.recordPlan(plan.Action)
if !options.DryRun {
if err := publish.Execute(ctx, req, plan); err != nil {
failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
}
if shouldNotify(plan.Action) {
if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil {
failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
}
@@ -148,17 +139,17 @@ func writePlanLine(w io.Writer, plan publish.Plan, planErr error) {
if destinationID == "" {
destinationID = "unknown"
}
fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), destinationID, planErr.Error())
fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, planErr.Error())
return
}
fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%s reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, plan.Action, outputSummary(plan.Outputs), plan.Reason)
fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, plan.Action, outputSummary(plan.Outputs), plan.Reason)
}
func writeErrorLine(w io.Writer, bundlePath, destinationID string, err error) {
if w == nil {
return
}
fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(bundlePath), destinationID, err.Error())
fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", storage.DisplayPath(bundlePath), destinationID, err.Error())
}
func outputSummary(outputs []publish.Output) string {

View File

@@ -3,7 +3,6 @@ package app
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
@@ -14,6 +13,8 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/notify"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
func TestRunDryRunPrintsConfigSummary(t *testing.T) {
@@ -63,7 +64,7 @@ func TestRunPublishesNewLocalBundle(t *testing.T) {
if _, err := os.Stat(filepath.Join(destinationRoot, "manifest.json")); !os.IsNotExist(err) {
t.Fatalf("destination manifest stat error = %v, want not exist", err)
}
destinationState := readStateFile(t, filepath.Join(destinationRoot, ".distributor.json"))
destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if destinationState.PipelineID != "reports" || destinationState.DestinationID != "archive" {
t.Fatalf("state identity = %s/%s", destinationState.PipelineID, destinationState.DestinationID)
}
@@ -81,7 +82,7 @@ func TestRunNotifiesAfterPublication(t *testing.T) {
writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
notifier := &recordingNotifier{
check: func() {
if _, err := os.Stat(filepath.Join(destinationRoot, ".distributor.json")); err != nil {
if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil {
t.Fatalf("state stat during notify: %v", err)
}
},
@@ -217,7 +218,7 @@ func TestRunPublishesHTMLOnly(t *testing.T) {
if _, err := os.Stat(filepath.Join(destinationRoot, "report.md")); !os.IsNotExist(err) {
t.Fatalf("report.md stat error = %v, want not exist", err)
}
destinationState := readStateFile(t, filepath.Join(destinationRoot, ".distributor.json"))
destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if got, want := len(destinationState.Outputs), 1; got != want {
t.Fatalf("state output count = %d, want %d", got, want)
}
@@ -239,7 +240,7 @@ func TestRunPublishesSourceAndHTML(t *testing.T) {
assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
assertFileContains(t, filepath.Join(destinationRoot, "report.html"), "<p>Sunny.</p>")
assertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
destinationState := readStateFile(t, filepath.Join(destinationRoot, ".distributor.json"))
destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
if got, want := len(destinationState.Outputs), 3; got != want {
t.Fatalf("state output count = %d, want %d", got, want)
}
@@ -435,61 +436,20 @@ type testFile struct {
func writeSourceBundle(t *testing.T, root, relative string, opts testBundleOptions) bundle.Manifest {
t.Helper()
if opts.ID == "" {
opts.ID = "weather.daily.brentwood.2026-05-30"
extraFiles := make([]testutil.SourceFile, 0, len(opts.ExtraFiles))
for _, file := range opts.ExtraFiles {
extraFiles = append(extraFiles, testutil.SourceFile{Path: file.Path, Data: file.Data})
}
if opts.Created.IsZero() {
opts.Created = time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC)
}
bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
t.Fatalf("mkdir bundle: %v", err)
}
files := []struct {
path string
data string
}{
{path: "report.md", data: "# Report\nSunny.\n"},
{path: "summary.txt", data: "Summary\n"},
}
for _, extra := range opts.ExtraFiles {
files = append(files, struct {
path string
data string
}{path: extra.Path, data: extra.Data})
}
manifestFiles := make([]bundle.ManifestFile, 0, len(files))
for _, file := range files {
if err := os.WriteFile(filepath.Join(bundleRoot, filepath.FromSlash(file.path)), []byte(file.data), 0o600); err != nil {
t.Fatalf("write source file: %v", err)
}
manifestFiles = append(manifestFiles, bundle.ManifestFile{
Path: file.path,
SHA256: bundle.FileDigest([]byte(file.data)),
Size: int64(len(file.data)),
})
}
manifest := bundle.Manifest{
SchemaVersion: 1,
ID: opts.ID,
Created: opts.Created,
Files: manifestFiles,
}
manifest.Digest = bundle.BundleDigest(manifest.Files)
data, err := json.MarshalIndent(manifest, "", " ")
if err != nil {
t.Fatalf("marshal manifest: %v", err)
}
data = append(data, '\n')
if err := os.WriteFile(filepath.Join(bundleRoot, "manifest.json"), data, 0o600); err != nil {
t.Fatalf("write manifest: %v", err)
}
return manifest
return testutil.WriteSourceBundle(t, root, relative, testutil.BundleOptions{
ID: opts.ID,
Created: opts.Created,
ExtraFiles: extraFiles,
})
}
func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
t.Helper()
return writeLocalConfigWithPolicy(t, sourceRoot, destinationRoot, true, false)
return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
}
func writeLocalConfigWithPolicy(t *testing.T, sourceRoot, destinationRoot string, publishSource, publishHTML bool) string {
@@ -520,20 +480,7 @@ pipelines:
func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive-one
backend: local
path: `+firstDestination+`
- id: archive-two
backend: local
path: `+secondDestination+`
`)
return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)
}
func writeConfigFile(t *testing.T, body string) string {
@@ -547,42 +494,12 @@ func writeConfigFile(t *testing.T, body string) string {
func writeDestinationState(t *testing.T, root, relative string, manifest bundle.Manifest) {
t.Helper()
bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
t.Fatalf("mkdir destination: %v", err)
}
destinationState := state.DistributorState{
SchemaVersion: state.SchemaVersion,
PipelineID: "reports",
DestinationID: "archive",
PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC),
Source: state.SourceState{Manifest: manifest},
Outputs: []state.OutputFile{
{Path: "report.md", Kind: state.OutputKindSource, SourcePath: "report.md", SHA256: manifest.Files[0].SHA256, Size: manifest.Files[0].Size},
{Path: "summary.txt", Kind: state.OutputKindSource, SourcePath: "summary.txt", SHA256: manifest.Files[1].SHA256, Size: manifest.Files[1].Size},
},
}
data, err := json.MarshalIndent(destinationState, "", " ")
if err != nil {
t.Fatalf("marshal state: %v", err)
}
data = append(data, '\n')
if err := os.WriteFile(filepath.Join(bundleRoot, ".distributor.json"), data, 0o600); err != nil {
t.Fatalf("write state: %v", err)
}
testutil.WriteDestinationState(t, root, relative, manifest, testutil.DestinationStateOptions{})
}
func readStateFile(t *testing.T, path string) state.DistributorState {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read state: %v", err)
}
destinationState, err := state.Parse(data)
if err != nil {
t.Fatalf("parse state: %v", err)
}
return destinationState
return testutil.ReadDestinationState(t, path)
}
func assertFile(t *testing.T, path, want string) {

View File

@@ -0,0 +1,12 @@
package app
import (
"gitea.maximumdirect.net/eric/distributor/internal/transform"
"gitea.maximumdirect.net/eric/distributor/internal/transform/markdown"
)
func newTransformRegistry() *transform.Registry {
registry := transform.NewRegistry()
_ = registry.Register(transform.MarkdownToHTML, markdown.New())
return registry
}

View File

@@ -5,7 +5,6 @@ import (
"fmt"
"io"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
)
@@ -18,7 +17,7 @@ func Validate(ctx context.Context, options ValidateOptions) error {
if options.Path == "" {
return fmt.Errorf("validate command requires a path")
}
backend, err := local.New(options.Path)
backend, err := newBackendFactory().openLocalPath(ctx, options.Path)
if err != nil {
return err
}

View File

@@ -35,7 +35,7 @@ func Discover(ctx context.Context, backend storage.Backend, sourceRoot string) (
}
sort.Strings(roots)
if len(roots) == 0 {
return nil, fmt.Errorf("no bundles found under %q", displayRoot(sourceRoot))
return nil, fmt.Errorf("no bundles found under %q", storage.DisplayPath(sourceRoot))
}
if err := rejectNestedRoots(roots); err != nil {
return nil, err
@@ -57,7 +57,7 @@ func rejectNestedRoots(roots []string) error {
for index, root := range roots {
for _, candidate := range roots[index+1:] {
if isAncestor(root, candidate) {
return fmt.Errorf("nested manifest %q under bundle %q", displayRoot(candidate), displayRoot(root))
return fmt.Errorf("nested manifest %q under bundle %q", storage.DisplayPath(candidate), storage.DisplayPath(root))
}
}
}

View File

@@ -59,9 +59,6 @@ func ParseManifest(data []byte) (Manifest, error) {
return Manifest{}, fmt.Errorf("manifest schema_version is required")
}
manifest.SchemaVersion = *raw.SchemaVersion
if manifest.SchemaVersion != 1 {
return Manifest{}, fmt.Errorf("manifest schema_version must be 1")
}
if raw.ID == nil || *raw.ID == "" {
return Manifest{}, fmt.Errorf("manifest id is required")
}
@@ -69,9 +66,6 @@ func ParseManifest(data []byte) (Manifest, error) {
if raw.Digest == nil || *raw.Digest == "" {
return Manifest{}, fmt.Errorf("manifest digest is required")
}
if err := ValidateDigest(*raw.Digest); err != nil {
return Manifest{}, fmt.Errorf("manifest digest: %w", err)
}
manifest.Digest = *raw.Digest
if raw.Created == nil || *raw.Created == "" {
return Manifest{}, fmt.Errorf("manifest created is required")
@@ -85,18 +79,16 @@ func ParseManifest(data []byte) (Manifest, error) {
return Manifest{}, fmt.Errorf("manifest files is required")
}
seen := make(map[string]struct{}, len(raw.Files))
for index, rawFile := range raw.Files {
file, err := parseManifestFile(index, rawFile)
if err != nil {
return Manifest{}, err
}
if _, exists := seen[file.Path]; exists {
return Manifest{}, fmt.Errorf("manifest files[%d].path duplicates %q", index, file.Path)
}
seen[file.Path] = struct{}{}
manifest.Files = append(manifest.Files, file)
}
if err := ValidateManifest(manifest); err != nil {
return Manifest{}, fmt.Errorf("manifest %w", err)
}
return manifest, nil
}
@@ -104,21 +96,12 @@ func parseManifestFile(index int, raw rawManifestFile) (ManifestFile, error) {
if raw.Path == nil || *raw.Path == "" {
return ManifestFile{}, fmt.Errorf("manifest files[%d].path is required", index)
}
if err := ValidateSourcePath(*raw.Path); err != nil {
return ManifestFile{}, fmt.Errorf("manifest files[%d].path: %w", index, err)
}
if raw.SHA256 == nil || *raw.SHA256 == "" {
return ManifestFile{}, fmt.Errorf("manifest files[%d].sha256 is required", index)
}
if err := ValidateDigest(*raw.SHA256); err != nil {
return ManifestFile{}, fmt.Errorf("manifest files[%d].sha256: %w", index, err)
}
if raw.Size == nil {
return ManifestFile{}, fmt.Errorf("manifest files[%d].size is required", index)
}
if *raw.Size < 0 {
return ManifestFile{}, fmt.Errorf("manifest files[%d].size must be non-negative", index)
}
return ManifestFile{
Path: *raw.Path,
SHA256: *raw.SHA256,

View File

@@ -4,6 +4,9 @@ import (
"os"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
func TestParseManifestValid(t *testing.T) {
@@ -71,7 +74,7 @@ func TestParseManifestRejectsUnsafeFilePaths(t *testing.T) {
`"path": "/report.md"`,
`"path": "nested/../report.md"`,
`"path": "manifest.json"`,
`"path": ".distributor.json"`,
`"path": "` + storage.StateFileName + `"`,
}
for _, replacement := range tests {
t.Run(replacement, func(t *testing.T) {
@@ -90,6 +93,75 @@ func TestParseManifestRejectsDuplicatePaths(t *testing.T) {
assertErrorContains(t, err, "duplicates")
}
func TestValidateManifestAcceptsValidFixture(t *testing.T) {
manifest := validFixtureManifest(t)
if err := ValidateManifest(manifest); err != nil {
t.Fatalf("ValidateManifest() error = %v", err)
}
}
func TestValidateManifestRejectsInvalidManifest(t *testing.T) {
tests := map[string]func(Manifest) Manifest{
"schema version": func(manifest Manifest) Manifest {
manifest.SchemaVersion = 2
return manifest
},
"empty id": func(manifest Manifest) Manifest {
manifest.ID = ""
return manifest
},
"bad digest": func(manifest Manifest) Manifest {
manifest.Digest = "SHA256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe"
return manifest
},
"zero created": func(manifest Manifest) Manifest {
manifest.Created = time.Time{}
return manifest
},
"empty files": func(manifest Manifest) Manifest {
manifest.Files = nil
manifest.Digest = BundleDigest(manifest.Files)
return manifest
},
"unsafe path": func(manifest Manifest) Manifest {
manifest.Files[0].Path = "../report.md"
manifest.Digest = BundleDigest(manifest.Files)
return manifest
},
"duplicate path": func(manifest Manifest) Manifest {
manifest.Files[1].Path = manifest.Files[0].Path
manifest.Digest = BundleDigest(manifest.Files)
return manifest
},
"negative size": func(manifest Manifest) Manifest {
manifest.Files[0].Size = -1
manifest.Digest = BundleDigest(manifest.Files)
return manifest
},
"digest mismatch": func(manifest Manifest) Manifest {
manifest.Digest = "sha256:0000000000000000000000000000000000000000000000000000000000000000"
return manifest
},
}
for name, mutate := range tests {
t.Run(name, func(t *testing.T) {
err := ValidateManifest(mutate(validFixtureManifest(t)))
if err == nil {
t.Fatal("ValidateManifest() error = nil, want error")
}
})
}
}
func validFixtureManifest(t *testing.T) Manifest {
t.Helper()
manifest, err := ParseManifest(readFixture(t, "testdata/valid_bundle/manifest.json"))
if err != nil {
t.Fatalf("ParseManifest() error = %v", err)
}
return manifest
}
func readFixture(t *testing.T, path string) []byte {
t.Helper()
data, err := os.ReadFile(path)

View File

@@ -12,12 +12,50 @@ func ValidateSourcePath(path string) error {
return err
}
switch path {
case ManifestName, ".distributor.json":
case ManifestName, storage.StateFileName:
return fmt.Errorf("%q is reserved", path)
}
return nil
}
func ValidateManifest(manifest Manifest) error {
if manifest.SchemaVersion != 1 {
return fmt.Errorf("schema_version must be 1")
}
if manifest.ID == "" {
return fmt.Errorf("id is required")
}
if err := ValidateDigest(manifest.Digest); err != nil {
return fmt.Errorf("digest: %w", err)
}
if manifest.Created.IsZero() {
return fmt.Errorf("created is required")
}
if len(manifest.Files) == 0 {
return fmt.Errorf("files is required")
}
seen := make(map[string]struct{}, len(manifest.Files))
for index, file := range manifest.Files {
if err := ValidateSourcePath(file.Path); err != nil {
return fmt.Errorf("files[%d].path: %w", index, err)
}
if err := ValidateDigest(file.SHA256); err != nil {
return fmt.Errorf("files[%d].sha256: %w", index, err)
}
if file.Size < 0 {
return fmt.Errorf("files[%d].size must be non-negative", index)
}
if _, exists := seen[file.Path]; exists {
return fmt.Errorf("files[%d].path duplicates %q", index, file.Path)
}
seen[file.Path] = struct{}{}
}
if actual := BundleDigest(manifest.Files); actual != manifest.Digest {
return fmt.Errorf("digest mismatch: got %s want %s", actual, manifest.Digest)
}
return nil
}
func Validate(ctx context.Context, backend storage.Backend, bundleRoot string) (Bundle, error) {
return validateAt(ctx, backend, bundleRoot, bundleRoot)
}
@@ -36,31 +74,31 @@ func validateAt(ctx context.Context, backend storage.Backend, bundleRoot, relati
}
manifest, err := ParseManifest(manifestData)
if err != nil {
return Bundle{}, fmt.Errorf("bundle %q: %w", displayRoot(relativeRoot), err)
return Bundle{}, fmt.Errorf("bundle %q: %w", storage.DisplayPath(relativeRoot), err)
}
for index, manifestFile := range manifest.Files {
filePath, err := storage.Join(bundleRoot, manifestFile.Path)
if err != nil {
return Bundle{}, fmt.Errorf("bundle %q file %q: %w", displayRoot(relativeRoot), manifestFile.Path, err)
return Bundle{}, fmt.Errorf("bundle %q file %q: %w", storage.DisplayPath(relativeRoot), manifestFile.Path, err)
}
entry, err := backend.Stat(ctx, filePath)
if err != nil {
return Bundle{}, fmt.Errorf("bundle %q file %q stat: %w", displayRoot(relativeRoot), manifestFile.Path, err)
return Bundle{}, fmt.Errorf("bundle %q file %q stat: %w", storage.DisplayPath(relativeRoot), manifestFile.Path, err)
}
if entry.Type != storage.EntryTypeFile {
return Bundle{}, fmt.Errorf("bundle %q file %q must be a regular file", displayRoot(relativeRoot), manifestFile.Path)
return Bundle{}, fmt.Errorf("bundle %q file %q must be a regular file", storage.DisplayPath(relativeRoot), manifestFile.Path)
}
if entry.Size != manifestFile.Size {
return Bundle{}, fmt.Errorf("bundle %q file %q size mismatch: got %d want %d", displayRoot(relativeRoot), manifestFile.Path, entry.Size, manifestFile.Size)
return Bundle{}, fmt.Errorf("bundle %q file %q size mismatch: got %d want %d", storage.DisplayPath(relativeRoot), manifestFile.Path, entry.Size, manifestFile.Size)
}
data, err := backend.ReadFile(ctx, filePath)
if err != nil {
return Bundle{}, fmt.Errorf("bundle %q file %q read: %w", displayRoot(relativeRoot), manifestFile.Path, err)
return Bundle{}, fmt.Errorf("bundle %q file %q read: %w", storage.DisplayPath(relativeRoot), manifestFile.Path, err)
}
actualDigest := FileDigest(data)
if actualDigest != manifestFile.SHA256 {
return Bundle{}, fmt.Errorf("bundle %q file %q sha256 mismatch: got %s want %s", displayRoot(relativeRoot), manifestFile.Path, actualDigest, manifestFile.SHA256)
return Bundle{}, fmt.Errorf("bundle %q file %q sha256 mismatch: got %s want %s", storage.DisplayPath(relativeRoot), manifestFile.Path, actualDigest, manifestFile.SHA256)
}
manifest.Files[index].SHA256 = actualDigest
manifest.Files[index].Size = int64(len(data))
@@ -68,7 +106,7 @@ func validateAt(ctx context.Context, backend storage.Backend, bundleRoot, relati
actualBundleDigest := BundleDigest(manifest.Files)
if actualBundleDigest != manifest.Digest {
return Bundle{}, fmt.Errorf("bundle %q digest mismatch: got %s want %s", displayRoot(relativeRoot), actualBundleDigest, manifest.Digest)
return Bundle{}, fmt.Errorf("bundle %q digest mismatch: got %s want %s", storage.DisplayPath(relativeRoot), actualBundleDigest, manifest.Digest)
}
return Bundle{
@@ -76,10 +114,3 @@ func validateAt(ctx context.Context, backend storage.Backend, bundleRoot, relati
Manifest: manifest,
}, nil
}
func displayRoot(root string) string {
if root == "" {
return "."
}
return root
}

View File

@@ -2,6 +2,7 @@ package bundle
import (
"context"
"encoding/json"
"strings"
"testing"
@@ -32,8 +33,10 @@ func TestValidateRejectsMissingFile(t *testing.T) {
func TestValidateRejectsSizeMismatch(t *testing.T) {
backend := validFakeBundle(t)
manifest := strings.Replace(string(readFixture(t, "testdata/valid_bundle/manifest.json")), `"size": 8`, `"size": 9`, 1)
writeFakeFile(t, backend, "manifest.json", manifest)
manifest := validFixtureManifest(t)
manifest.Files[1].Size = 9
manifest.Digest = BundleDigest(manifest.Files)
writeManifest(t, backend, manifest)
_, err := Validate(context.Background(), backend, "")
assertErrorContains(t, err, "size mismatch")
}
@@ -79,6 +82,15 @@ func writeFakeFile(t *testing.T, backend *fake.Backend, path, data string) {
}
}
func writeManifest(t *testing.T, backend *fake.Backend, manifest Manifest) {
t.Helper()
data, err := json.MarshalIndent(manifest, "", " ")
if err != nil {
t.Fatalf("MarshalIndent() error = %v", err)
}
writeFakeFile(t, backend, ManifestName, string(append(data, '\n')))
}
func deleteFakeFile(t *testing.T, backend *fake.Backend, path string) {
t.Helper()
err := backend.DeleteManagedBundle(context.Background(), "", []string{path}, storage.DeleteOptions{IgnoreMissing: true})

View File

@@ -13,14 +13,10 @@ func inspectCommand(ctx context.Context, args []string, stdout, stderr io.Writer
printInspectHelp(stdout)
return exitOK
}
if len(args) > 1 {
fmt.Fprintf(stderr, "%s: inspect accepts at most one path\n", app.Name)
path, ok := parseOptionalPathArg(stderr, "inspect", args)
if !ok {
return exitUsage
}
var path string
if len(args) == 1 {
path = args[0]
}
if err := app.Inspect(ctx, app.InspectOptions{Path: path, Stdout: stdout}); err != nil {
return fail(stderr, err)
}

View File

@@ -2,7 +2,6 @@ package cli
import (
"context"
"errors"
"fmt"
"io"
"strings"
@@ -67,10 +66,6 @@ func hasHelp(args []string) bool {
}
func fail(stderr io.Writer, err error) int {
if errors.Is(err, app.ErrNotImplemented) {
fmt.Fprintf(stderr, "%s: %s\n", app.Name, err)
return exitError
}
fmt.Fprintf(stderr, "%s: %s\n", app.Name, err)
return exitError
}
@@ -82,3 +77,22 @@ func rejectExtraArgs(stderr io.Writer, command string, args []string) bool {
fmt.Fprintf(stderr, "%s: %s does not accept arguments: %s\n", app.Name, command, strings.Join(args, " "))
return true
}
func parseOptionalPathArg(stderr io.Writer, command string, args []string) (string, bool) {
if len(args) > 1 {
fmt.Fprintf(stderr, "%s: %s accepts at most one path\n", app.Name, command)
return "", false
}
if len(args) == 0 {
return "", true
}
return args[0], true
}
func rejectPositionalArgs(stderr io.Writer, command string, args []string) bool {
if len(args) == 0 {
return false
}
fmt.Fprintf(stderr, "%s: %s does not accept positional arguments: %v\n", app.Name, command, args)
return true
}

View File

@@ -7,6 +7,9 @@ import (
"path/filepath"
"strings"
"testing"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
func TestExecuteRootHelp(t *testing.T) {
@@ -54,6 +57,51 @@ func TestExecuteValidate(t *testing.T) {
}
}
func TestExecuteValidateArgs(t *testing.T) {
validPath := filepath.Join("..", "bundle", "testdata", "valid_bundle")
tests := []struct {
name string
args []string
wantCode int
wantStdout string
wantStderr string
}{
{
name: "zero args",
args: []string{"validate"},
wantCode: exitError,
wantStderr: "requires a path",
},
{
name: "one arg",
args: []string{"validate", validPath},
wantCode: exitOK,
wantStdout: "Validated 1 bundle(s)",
},
{
name: "two args",
args: []string{"validate", validPath, validPath},
wantCode: exitUsage,
wantStderr: "accepts at most one path",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var stdout, stderr bytes.Buffer
code := Execute(context.Background(), tt.args, &stdout, &stderr)
if code != tt.wantCode {
t.Fatalf("exit code = %d, want %d; stderr = %q", code, tt.wantCode, stderr.String())
}
if tt.wantStdout != "" && !strings.Contains(stdout.String(), tt.wantStdout) {
t.Fatalf("stdout = %q, want substring %q", stdout.String(), tt.wantStdout)
}
if tt.wantStderr != "" && !strings.Contains(stderr.String(), tt.wantStderr) {
t.Fatalf("stderr = %q, want substring %q", stderr.String(), tt.wantStderr)
}
})
}
}
func TestExecuteInspect(t *testing.T) {
var stdout, stderr bytes.Buffer
@@ -67,24 +115,55 @@ func TestExecuteInspect(t *testing.T) {
}
}
func TestExecuteInspectArgs(t *testing.T) {
validPath := filepath.Join("..", "bundle", "testdata", "valid_bundle")
tests := []struct {
name string
args []string
wantCode int
wantStdout string
wantStderr string
}{
{
name: "zero args",
args: []string{"inspect"},
wantCode: exitError,
wantStderr: "requires a path",
},
{
name: "one arg",
args: []string{"inspect", validPath},
wantCode: exitOK,
wantStdout: "id=weather.daily.brentwood.2026-05-30",
},
{
name: "two args",
args: []string{"inspect", validPath, validPath},
wantCode: exitUsage,
wantStderr: "accepts at most one path",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var stdout, stderr bytes.Buffer
code := Execute(context.Background(), tt.args, &stdout, &stderr)
if code != tt.wantCode {
t.Fatalf("exit code = %d, want %d; stderr = %q", code, tt.wantCode, stderr.String())
}
if tt.wantStdout != "" && !strings.Contains(stdout.String(), tt.wantStdout) {
t.Fatalf("stdout = %q, want substring %q", stdout.String(), tt.wantStdout)
}
if tt.wantStderr != "" && !strings.Contains(stderr.String(), tt.wantStderr) {
t.Fatalf("stderr = %q, want substring %q", stderr.String(), tt.wantStderr)
}
})
}
}
func TestExecuteRunDryRun(t *testing.T) {
sourceRoot := t.TempDir()
writeCLIBundle(t, sourceRoot)
configPath := filepath.Join(t.TempDir(), "config.yml")
err := os.WriteFile(configPath, []byte(`
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+t.TempDir()+`
`), 0o600)
if err != nil {
t.Fatalf("write config: %v", err)
}
testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, t.TempDir())
var stdout, stderr bytes.Buffer
@@ -101,25 +180,24 @@ pipelines:
}
}
func TestExecuteRunRejectsExtraPositionalArgs(t *testing.T) {
var stdout, stderr bytes.Buffer
code := Execute(context.Background(), []string{"run", "--config", "config.yml", "extra"}, &stdout, &stderr)
if code != exitUsage {
t.Fatalf("exit code = %d, want %d", code, exitUsage)
}
if !strings.Contains(stderr.String(), "does not accept positional arguments") {
t.Fatalf("stderr = %q, want positional argument error", stderr.String())
}
}
func TestExecuteRunPublishes(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeCLIBundle(t, sourceRoot)
configPath := filepath.Join(t.TempDir(), "config.yml")
err := os.WriteFile(configPath, []byte(`
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
`), 0o600)
if err != nil {
t.Fatalf("write config: %v", err)
}
testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
var stdout, stderr bytes.Buffer
@@ -128,7 +206,7 @@ pipelines:
if code != exitOK {
t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
}
if _, err := os.Stat(filepath.Join(destinationRoot, ".distributor.json")); err != nil {
if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil {
t.Fatalf("state stat error = %v", err)
}
}
@@ -145,37 +223,3 @@ func TestUnknownCommandIsUsageError(t *testing.T) {
t.Fatalf("stderr = %q, want unknown command error", stderr.String())
}
}
func writeCLIBundle(t *testing.T, root string) {
t.Helper()
for _, file := range []struct {
path string
data string
}{
{"manifest.json", `{
"schema_version": 1,
"id": "weather.daily.brentwood.2026-05-30",
"digest": "sha256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe",
"created": "2026-05-30T11:10:00Z",
"files": [
{
"path": "report.md",
"sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6",
"size": 16
},
{
"path": "summary.txt",
"sha256": "sha256:3cbb36aca330b3bd113955dfbada0adb7a5f95ad9f678bd61f175406c6a37e95",
"size": 8
}
]
}
`},
{"report.md", "# Report\nSunny.\n"},
{"summary.txt", "Summary\n"},
} {
if err := os.WriteFile(filepath.Join(root, file.path), []byte(file.data), 0o600); err != nil {
t.Fatalf("write bundle file: %v", err)
}
}
}

View File

@@ -22,8 +22,7 @@ func runCommand(ctx context.Context, args []string, stdout, stderr io.Writer) in
if err := flags.Parse(args); err != nil {
return exitUsage
}
if flags.NArg() > 0 {
fmt.Fprintf(stderr, "%s: run does not accept positional arguments: %v\n", app.Name, flags.Args())
if rejectPositionalArgs(stderr, "run", flags.Args()) {
return exitUsage
}

View File

@@ -13,14 +13,10 @@ func validateCommand(ctx context.Context, args []string, stdout, stderr io.Write
printValidateHelp(stdout)
return exitOK
}
if len(args) > 1 {
fmt.Fprintf(stderr, "%s: validate accepts at most one path\n", app.Name)
path, ok := parseOptionalPathArg(stderr, "validate", args)
if !ok {
return exitUsage
}
var path string
if len(args) == 1 {
path = args[0]
}
if err := app.Validate(ctx, app.ValidateOptions{Path: path, Stdout: stdout}); err != nil {
return fail(stderr, err)
}

View File

@@ -57,8 +57,7 @@ func Validate(cfg Config) error {
}
errs = validateBackend(errs, destinationContext, destination.Backend, destination.Path, destination.URI, destination.Endpoint, destination.Bucket)
errs = validatePublishPolicy(errs, destinationContext+".publish", destination.Publish)
errs = validateTransform(errs, destinationContext+".transform", destination.Publish, destination.Transform)
errs = validatePublishTransformPolicy(errs, destinationContext, destination.Publish, destination.Transform)
errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer)
}
}
@@ -104,36 +103,40 @@ func validateValidationPolicy(errs ValidationErrors, context string, policy Vali
return errs
}
func validatePublishPolicy(errs ValidationErrors, context string, policy *PublishPolicy) ValidationErrors {
func validatePublishTransformPolicy(errs ValidationErrors, context string, policy *PublishPolicy, transform Transform) ValidationErrors {
if policy == nil {
errs = append(errs, context+" is required")
errs = append(errs, context+".publish is required")
return errs
}
if !policy.Source && !policy.HTML {
errs = append(errs, context+" must enable source or html")
if err := ValidatePublishTransformPolicy(*policy, transform); err != nil {
errs = append(errs, context+"."+err.Error())
}
return errs
}
func validateTransform(errs ValidationErrors, context string, publish *PublishPolicy, transform Transform) ValidationErrors {
publishesHTML := publish != nil && publish.HTML
if transform.MarkdownToHTML == nil {
if publishesHTML {
errs = append(errs, context+".markdown_to_html is required when publish.html is true")
}
return errs
func ValidatePublishTransformPolicy(publish PublishPolicy, transform Transform) error {
if !publish.Source && !publish.HTML {
return fmt.Errorf("publish must enable source or html")
}
if publishesHTML && !transform.MarkdownToHTML.Enabled {
errs = append(errs, context+".markdown_to_html.enabled must be true when publish.html is true")
if publish.HTML && transform.MarkdownToHTML == nil {
return fmt.Errorf("transform.markdown_to_html is required when publish.html is true")
}
if transform.MarkdownToHTML == nil {
return nil
}
if publish.HTML && !transform.MarkdownToHTML.Enabled {
return fmt.Errorf("transform.markdown_to_html.enabled must be true when publish.html is true")
}
if publish.HTML && transform.MarkdownToHTML.Mode != TransformModeSidecar {
return fmt.Errorf("transform.markdown_to_html.mode must be %s", TransformModeSidecar)
}
if transform.MarkdownToHTML.Enabled && transform.MarkdownToHTML.Mode != TransformModeSidecar {
errs = append(errs, context+".markdown_to_html.mode must be "+TransformModeSidecar)
return fmt.Errorf("transform.markdown_to_html.mode must be %s", TransformModeSidecar)
}
if !transform.MarkdownToHTML.Enabled && transform.MarkdownToHTML.Mode != "" && transform.MarkdownToHTML.Mode != TransformModeSidecar {
errs = append(errs, context+".markdown_to_html.mode must be "+TransformModeSidecar)
return fmt.Errorf("transform.markdown_to_html.mode must be %s", TransformModeSidecar)
}
return errs
return nil
}
func validateTransferPolicy(errs ValidationErrors, context string, policy TransferPolicy) ValidationErrors {

View File

@@ -0,0 +1,141 @@
package config
import "testing"
func TestValidatePublishTransformPolicy(t *testing.T) {
tests := publishTransformPolicyCases()
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := ValidatePublishTransformPolicy(tt.publish, tt.transform)
if tt.wantErr && err == nil {
t.Fatal("ValidatePublishTransformPolicy() error = nil, want error")
}
if !tt.wantErr && err != nil {
t.Fatalf("ValidatePublishTransformPolicy() error = %v", err)
}
})
}
}
func TestValidateChecksPublishTransformPolicy(t *testing.T) {
tests := publishTransformPolicyCases()
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",
Publish: &tt.publish,
Transform: tt.transform,
}},
}}}
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)
}
})
}
}
type publishTransformPolicyCase struct {
name string
publish PublishPolicy
transform Transform
wantErr bool
}
func publishTransformPolicyCases() []publishTransformPolicyCase {
return []publishTransformPolicyCase{
{
name: "source only allowed",
publish: PublishPolicy{Source: true},
},
{
name: "html only sidecar allowed",
publish: PublishPolicy{HTML: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: true,
Mode: TransformModeSidecar,
}},
},
{
name: "source and html sidecar allowed",
publish: PublishPolicy{Source: true, HTML: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: true,
Mode: TransformModeSidecar,
}},
},
{
name: "no outputs rejected",
publish: PublishPolicy{},
wantErr: true,
},
{
name: "html without transform rejected",
publish: PublishPolicy{HTML: true},
wantErr: true,
},
{
name: "html with disabled transform rejected",
publish: PublishPolicy{HTML: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: false,
Mode: TransformModeSidecar,
}},
wantErr: true,
},
{
name: "html with wrong mode rejected",
publish: PublishPolicy{HTML: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: true,
Mode: "inline",
}},
wantErr: true,
},
{
name: "enabled markdown wrong mode rejected",
publish: PublishPolicy{Source: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: true,
Mode: "inline",
}},
wantErr: true,
},
{
name: "disabled markdown empty mode allowed",
publish: PublishPolicy{Source: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: false,
}},
},
{
name: "disabled markdown sidecar mode allowed",
publish: PublishPolicy{Source: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: false,
Mode: TransformModeSidecar,
}},
},
{
name: "disabled markdown wrong mode rejected",
publish: PublishPolicy{Source: true},
transform: Transform{MarkdownToHTML: &MarkdownToHTML{
Enabled: false,
Mode: "inline",
}},
wantErr: true,
},
}
}

View File

@@ -5,18 +5,17 @@ import (
"fmt"
"io"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
func TestExecuteCleansUpAfterWriteFailure(t *testing.T) {
sourceBackend := fake.New()
destinationBackend := &failingBackend{Backend: fake.New(), failPath: "summary.txt"}
sourceBundle := writeFakeSourceBundle(t, sourceBackend)
sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{})
req := Request{
PipelineID: "reports",
DestinationID: "archive",
@@ -63,29 +62,3 @@ func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader
}
return b.Backend.WriteFrom(ctx, path, r, opts)
}
func writeFakeSourceBundle(t *testing.T, backend *fake.Backend) bundle.Bundle {
t.Helper()
files := []struct {
path string
data string
}{
{path: "report.md", data: "# Report\nSunny.\n"},
{path: "summary.txt", data: "Summary\n"},
}
manifestFiles := make([]bundle.ManifestFile, 0, len(files))
for _, file := range files {
if _, err := backend.WriteFile(context.Background(), file.path, []byte(file.data), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
manifestFiles = append(manifestFiles, bundle.ManifestFile{Path: file.path, SHA256: bundle.FileDigest([]byte(file.data)), Size: int64(len(file.data))})
}
manifest := bundle.Manifest{
SchemaVersion: 1,
ID: "weather.daily.brentwood.2026-05-30",
Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
Files: manifestFiles,
}
manifest.Digest = bundle.BundleDigest(manifest.Files)
return bundle.Bundle{Manifest: manifest}
}

View File

@@ -19,7 +19,11 @@ func PlanOutputs(ctx context.Context, req Request) ([]Output, error) {
outputs = append(outputs, sourceOutputs...)
}
if req.Publish.HTML {
generatedOutputs, err := markdownTransformer().Generate(ctx, transform.Request{
transformer, err := resolveTransformer(req.Transformers, transform.MarkdownToHTML)
if err != nil {
return nil, err
}
generatedOutputs, err := transformer.Generate(ctx, transform.Request{
SourceBundle: req.SourceBundle,
SourceBackend: req.SourceBackend,
})
@@ -47,6 +51,17 @@ func PlanOutputs(ctx context.Context, req Request) ([]Output, error) {
return outputs, nil
}
func resolveTransformer(resolver TransformerResolver, name string) (transform.Transformer, error) {
if resolver == nil {
return nil, fmt.Errorf("transformer resolver is required for %s", name)
}
transformer, ok := resolver.Get(name)
if !ok {
return nil, fmt.Errorf("transformer %s is not registered", name)
}
return transformer, nil
}
func PlanSourceOutputs(req Request) ([]Output, error) {
outputs := make([]Output, 0, len(req.SourceBundle.Manifest.Files))
for _, file := range req.SourceBundle.Manifest.Files {

View File

@@ -3,41 +3,35 @@ package publish
import (
"context"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
"gitea.maximumdirect.net/eric/distributor/internal/transform"
)
func TestPlanOutputsRejectsCollision(t *testing.T) {
sourceBackend := fake.New()
if _, err := sourceBackend.WriteFile(context.Background(), "report.md", []byte("# Report\n"), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile report.md error = %v", err)
}
if _, err := sourceBackend.WriteFile(context.Background(), "report.html", []byte("<p>source html</p>\n"), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile report.html error = %v", err)
}
reportDigest := bundle.FileDigest([]byte("# Report\n"))
htmlDigest := bundle.FileDigest([]byte("<p>source html</p>\n"))
files := []bundle.ManifestFile{
{Path: "report.md", SHA256: reportDigest, Size: 9},
{Path: "report.html", SHA256: htmlDigest, Size: 19},
}
sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
Files: []testutil.SourceFile{
{Path: "report.md", Data: "# Report\n"},
{Path: "report.html", Data: "<p>source html</p>\n"},
},
})
_, err := PlanOutputs(context.Background(), Request{
SourceBackend: sourceBackend,
SourceBundle: bundle.Bundle{
Manifest: bundle.Manifest{
SchemaVersion: 1,
ID: "bundle",
Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
Digest: bundle.BundleDigest(files),
Files: files,
},
},
Publish: config.PublishPolicy{Source: true, HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
SourceBundle: sourceBundle,
Publish: config.PublishPolicy{Source: true, HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
Transformers: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
Path: "report.html",
SourcePath: "report.md",
Transform: transform.MarkdownToHTML,
Data: []byte("<p>Report</p>\n"),
SHA256: bundle.FileDigest([]byte("<p>Report</p>\n")),
Size: int64(len("<p>Report</p>\n")),
}}}},
})
if err == nil {
t.Fatal("PlanSourceOutputs() error = nil, want collision")
@@ -46,48 +40,83 @@ func TestPlanOutputsRejectsCollision(t *testing.T) {
func TestPlanOutputsRejectsHTMLWithoutMarkdown(t *testing.T) {
sourceBackend := fake.New()
if _, err := sourceBackend.WriteFile(context.Background(), "summary.txt", []byte("Summary\n"), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile summary.txt error = %v", err)
}
files := []bundle.ManifestFile{{Path: "summary.txt", SHA256: bundle.FileDigest([]byte("Summary\n")), Size: 8}}
sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
Files: []testutil.SourceFile{{Path: "summary.txt", Data: "Summary\n"}},
})
_, err := PlanOutputs(context.Background(), Request{
SourceBackend: sourceBackend,
SourceBundle: bundle.Bundle{Manifest: bundle.Manifest{
SchemaVersion: 1,
ID: "bundle",
Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
Digest: bundle.BundleDigest(files),
Files: files,
}},
Publish: config.PublishPolicy{HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
SourceBundle: sourceBundle,
Publish: config.PublishPolicy{HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
Transformers: testResolver{
transform.MarkdownToHTML: testTransformer{},
},
})
if err == nil {
t.Fatal("PlanOutputs() error = nil, want no markdown failure")
}
}
func TestPlanOutputsRejectsHTMLWithoutTransformerResolver(t *testing.T) {
_, err := PlanOutputs(context.Background(), Request{
Publish: config.PublishPolicy{HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
})
if err == nil {
t.Fatal("PlanOutputs() error = nil, want resolver failure")
}
}
func TestPlanOutputsRejectsMissingMarkdownTransformer(t *testing.T) {
_, err := PlanOutputs(context.Background(), Request{
Publish: config.PublishPolicy{HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
Transformers: testResolver{},
})
if err == nil {
t.Fatal("PlanOutputs() error = nil, want missing transformer failure")
}
}
func TestPlanOutputsUsesRegisteredTransformer(t *testing.T) {
data := []byte("<p>Generated</p>\n")
outputs, err := PlanOutputs(context.Background(), Request{
Publish: config.PublishPolicy{HTML: true},
Transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
Transformers: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
Path: "report.html",
SourcePath: "report.md",
Transform: transform.MarkdownToHTML,
Data: data,
SHA256: bundle.FileDigest(data),
Size: int64(len(data)),
}}}},
})
if err != nil {
t.Fatalf("PlanOutputs() error = %v", err)
}
if got, want := len(outputs), 1; got != want {
t.Fatalf("output count = %d, want %d", got, want)
}
if outputs[0].DestinationPath != "report.html" || string(outputs[0].Data) != string(data) {
t.Fatalf("output = %#v", outputs[0])
}
}
func TestBuildRejectsHTMLWithoutTransform(t *testing.T) {
sourceBackend := fake.New()
destinationBackend := fake.New()
if _, err := sourceBackend.WriteFile(context.Background(), "report.md", []byte("# Report\n"), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile report.md error = %v", err)
}
files := []bundle.ManifestFile{{Path: "report.md", SHA256: bundle.FileDigest([]byte("# Report\n")), Size: 9}}
sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}},
})
_, err := Build(context.Background(), Request{
PipelineID: "reports",
DestinationID: "archive",
SourceBackend: sourceBackend,
DestinationBackend: destinationBackend,
DestinationBundlePath: "",
SourceBundle: bundle.Bundle{Manifest: bundle.Manifest{
SchemaVersion: 1,
ID: "bundle",
Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
Digest: bundle.BundleDigest(files),
Files: files,
}},
Publish: config.PublishPolicy{HTML: true},
SourceBundle: sourceBundle,
Publish: config.PublishPolicy{HTML: true},
Transfer: config.TransferPolicy{
OnDestinationSame: config.TransferActionSkip,
OnDestinationOlder: config.TransferActionReplace,
@@ -99,3 +128,128 @@ func TestBuildRejectsHTMLWithoutTransform(t *testing.T) {
t.Fatal("Build() error = nil, want missing transform error")
}
}
func TestValidateRequestChecksPublishTransformPolicy(t *testing.T) {
tests := []struct {
name string
publish config.PublishPolicy
transform config.Transform
wantErr bool
}{
{
name: "source only allowed",
publish: config.PublishPolicy{Source: true},
},
{
name: "html only sidecar allowed",
publish: config.PublishPolicy{HTML: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: true,
Mode: config.TransformModeSidecar,
}},
},
{
name: "source and html sidecar allowed",
publish: config.PublishPolicy{Source: true, HTML: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: true,
Mode: config.TransformModeSidecar,
}},
},
{
name: "no outputs rejected",
publish: config.PublishPolicy{},
wantErr: true,
},
{
name: "html without transform rejected",
publish: config.PublishPolicy{HTML: true},
wantErr: true,
},
{
name: "html with disabled transform rejected",
publish: config.PublishPolicy{HTML: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: false,
Mode: config.TransformModeSidecar,
}},
wantErr: true,
},
{
name: "html with wrong mode rejected",
publish: config.PublishPolicy{HTML: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: true,
Mode: "inline",
}},
wantErr: true,
},
{
name: "enabled markdown wrong mode rejected",
publish: config.PublishPolicy{Source: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: true,
Mode: "inline",
}},
wantErr: true,
},
{
name: "disabled markdown empty mode allowed",
publish: config.PublishPolicy{Source: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: false,
}},
},
{
name: "disabled markdown sidecar mode allowed",
publish: config.PublishPolicy{Source: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: false,
Mode: config.TransformModeSidecar,
}},
},
{
name: "disabled markdown wrong mode rejected",
publish: config.PublishPolicy{Source: true},
transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{
Enabled: false,
Mode: "inline",
}},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateRequest(Request{
PipelineID: "reports",
DestinationID: "archive",
SourceBackend: fake.New(),
DestinationBackend: fake.New(),
Publish: tt.publish,
Transform: tt.transform,
})
if tt.wantErr && err == nil {
t.Fatal("validateRequest() error = nil, want error")
}
if !tt.wantErr && err != nil {
t.Fatalf("validateRequest() error = %v", err)
}
})
}
}
type testResolver map[string]transform.Transformer
func (r testResolver) Get(name string) (transform.Transformer, bool) {
transformer, ok := r[name]
return transformer, ok
}
type testTransformer struct {
outputs []transform.Output
err error
}
func (t testTransformer) Generate(context.Context, transform.Request) ([]transform.Output, error) {
return t.outputs, t.err
}

View File

@@ -9,7 +9,6 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/transform"
markdowntransform "gitea.maximumdirect.net/eric/distributor/internal/transform/markdown"
)
type Action string
@@ -32,10 +31,15 @@ type Request struct {
DestinationBundlePath string
Publish config.PublishPolicy
Transform config.Transform
Transformers TransformerResolver
Transfer config.TransferPolicy
DistributorVersion string
}
type TransformerResolver interface {
Get(name string) (transform.Transformer, bool)
}
type Plan struct {
PipelineID string
DestinationID string
@@ -102,21 +106,12 @@ func validateRequest(req Request) error {
if req.DestinationBackend == nil {
return fmt.Errorf("destination backend is required")
}
if !req.Publish.Source && !req.Publish.HTML {
return fmt.Errorf("publish source or html must be enabled")
}
if req.Publish.HTML {
if req.Transform.MarkdownToHTML == nil || !req.Transform.MarkdownToHTML.Enabled || req.Transform.MarkdownToHTML.Mode != config.TransformModeSidecar {
return fmt.Errorf("publish html requires markdown_to_html transform enabled with sidecar mode")
}
if err := config.ValidatePublishTransformPolicy(req.Publish, req.Transform); err != nil {
return fmt.Errorf("publish/transform policy: %w", err)
}
return nil
}
func markdownTransformer() transform.Transformer {
return markdowntransform.New()
}
func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy) (Action, string) {
switch comparison.Outcome {
case state.OutcomeDestinationAbsent:
@@ -144,10 +139,3 @@ func actionForComparison(comparison state.Comparison, transfer config.TransferPo
return ActionFailConflict, "unsupported comparison outcome"
}
}
func displayPath(path string) string {
if path == "" {
return "."
}
return path
}

View File

@@ -13,7 +13,7 @@ func ensureDestinationEmpty(ctx context.Context, backend storage.Backend, bundle
return err
}
if hasAny {
return fmt.Errorf("destination bundle path %q is not empty after managed cleanup", displayPath(bundlePath))
return fmt.Errorf("destination bundle path %q is not empty after managed cleanup", storage.DisplayPath(bundlePath))
}
return nil
}

View File

@@ -70,6 +70,60 @@ func TestParseRejectsInvalidEmbeddedManifest(t *testing.T) {
assertStateErrorContains(t, err, "source.manifest")
}
func TestValidateRejectsInvalidEmbeddedManifest(t *testing.T) {
tests := map[string]func(bundle.Manifest) bundle.Manifest{
"schema version": func(manifest bundle.Manifest) bundle.Manifest {
manifest.SchemaVersion = 2
return manifest
},
"empty id": func(manifest bundle.Manifest) bundle.Manifest {
manifest.ID = ""
return manifest
},
"bad digest": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Digest = "SHA256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe"
return manifest
},
"zero created": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Created = time.Time{}
return manifest
},
"empty files": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Files = nil
manifest.Digest = bundle.BundleDigest(manifest.Files)
return manifest
},
"unsafe path": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Files[0].Path = "../report.md"
manifest.Digest = bundle.BundleDigest(manifest.Files)
return manifest
},
"duplicate path": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Files[1].Path = manifest.Files[0].Path
manifest.Digest = bundle.BundleDigest(manifest.Files)
return manifest
},
"negative size": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Files[0].Size = -1
manifest.Digest = bundle.BundleDigest(manifest.Files)
return manifest
},
"digest mismatch": func(manifest bundle.Manifest) bundle.Manifest {
manifest.Digest = "sha256:0000000000000000000000000000000000000000000000000000000000000000"
return manifest
},
}
for name, mutate := range tests {
t.Run(name, func(t *testing.T) {
source := mutate(validManifest(t))
state := *withState(t, validManifest(t), func(*DistributorState) {})
state.Source.Manifest = source
err := Validate(state)
assertStateErrorContains(t, err, "source.manifest")
})
}
}
func TestParseRejectsInvalidOutputMetadata(t *testing.T) {
source := validManifest(t)
tests := map[string]func(*DistributorState){

View File

@@ -45,41 +45,7 @@ func Validate(s DistributorState) error {
}
func validateEmbeddedManifest(manifest bundle.Manifest) error {
if manifest.SchemaVersion != 1 {
return fmt.Errorf("schema_version must be 1")
}
if manifest.ID == "" {
return fmt.Errorf("id is required")
}
if err := bundle.ValidateDigest(manifest.Digest); err != nil {
return fmt.Errorf("digest: %w", err)
}
if manifest.Created.IsZero() {
return fmt.Errorf("created is required")
}
if len(manifest.Files) == 0 {
return fmt.Errorf("files is required")
}
seen := make(map[string]struct{}, len(manifest.Files))
for index, file := range manifest.Files {
if err := bundle.ValidateSourcePath(file.Path); err != nil {
return fmt.Errorf("files[%d].path: %w", index, err)
}
if err := bundle.ValidateDigest(file.SHA256); err != nil {
return fmt.Errorf("files[%d].sha256: %w", index, err)
}
if file.Size < 0 {
return fmt.Errorf("files[%d].size must be non-negative", index)
}
if _, exists := seen[file.Path]; exists {
return fmt.Errorf("files[%d].path duplicates %q", index, file.Path)
}
seen[file.Path] = struct{}{}
}
if actual := bundle.BundleDigest(manifest.Files); actual != manifest.Digest {
return fmt.Errorf("digest mismatch: got %s want %s", actual, manifest.Digest)
}
return nil
return bundle.ValidateManifest(manifest)
}
func validateOutput(index int, output OutputFile) error {

View File

@@ -181,22 +181,10 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(bundlePath); err != nil {
return err
}
targets := make([]string, 0, len(managedOutputPaths)+1)
for _, outputPath := range managedOutputPaths {
target, err := storage.Join(bundlePath, outputPath)
if err != nil {
return err
}
targets = append(targets, target)
}
statePath, err := storage.StatePath(bundlePath)
targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths)
if err != nil {
return err
}
targets = append(targets, statePath)
for _, target := range targets {
if _, ok := b.dirs[target]; ok {

View File

@@ -116,21 +116,28 @@ func TestBackendManagedDeletion(t *testing.T) {
backend := New()
mustWrite(t, backend, "bundle/report.html", "html")
mustWrite(t, backend, "bundle/keep.txt", "keep")
mustWrite(t, backend, "bundle/.distributor.json", "{}")
statePath, err := storage.StatePath("bundle")
if err != nil {
t.Fatalf("StatePath() error = %v", err)
}
mustWrite(t, backend, statePath, "{}")
err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
err = backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
if err != nil {
t.Fatalf("DeleteManagedBundle() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.html"); !storage.IsNotFound(err) {
t.Fatalf("managed output stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/.distributor.json"); !storage.IsNotFound(err) {
if _, err := backend.Stat(context.Background(), statePath); !storage.IsNotFound(err) {
t.Fatalf("state stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/keep.txt"); err != nil {
t.Fatalf("unlisted file stat error = %v", err)
}
if err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{""}, storage.DeleteOptions{}); !storage.IsInvalidPath(err) {
t.Fatalf("DeleteManagedBundle invalid output error = %v, want invalid path", err)
}
}
func TestBackendHasAnyAndWalkStop(t *testing.T) {

View File

@@ -6,7 +6,7 @@ import (
"strings"
)
const stateFileName = ".distributor.json"
const StateFileName = ".distributor.json"
func ValidatePath(value string) error {
if value == "" {
@@ -37,9 +37,36 @@ func Join(base, child string) (string, error) {
func StatePath(bundlePath string) (string, error) {
if bundlePath == "" {
return stateFileName, nil
return StateFileName, nil
}
return Join(bundlePath, stateFileName)
return Join(bundlePath, StateFileName)
}
func DisplayPath(path string) string {
if path == "" {
return "."
}
return path
}
func ManagedBundleTargets(bundlePath string, managedOutputPaths []string) ([]string, error) {
if err := ValidatePrefix(bundlePath); err != nil {
return nil, err
}
targets := make([]string, 0, len(managedOutputPaths)+1)
for _, outputPath := range managedOutputPaths {
target, err := Join(bundlePath, outputPath)
if err != nil {
return nil, err
}
targets = append(targets, target)
}
statePath, err := StatePath(bundlePath)
if err != nil {
return nil, err
}
targets = append(targets, statePath)
return targets, nil
}
func SortEntries(entries []Entry) {

View File

@@ -49,6 +49,81 @@ func TestValidatePrefixAllowsRoot(t *testing.T) {
}
}
func TestStatePath(t *testing.T) {
tests := map[string]string{
"": StateFileName,
"bundle": "bundle/" + StateFileName,
}
for bundlePath, want := range tests {
t.Run(bundlePath, func(t *testing.T) {
got, err := StatePath(bundlePath)
if err != nil {
t.Fatalf("StatePath(%q) error = %v", bundlePath, err)
}
if got != want {
t.Fatalf("StatePath(%q) = %q, want %q", bundlePath, got, want)
}
})
}
}
func TestDisplayPath(t *testing.T) {
tests := map[string]string{
"": ".",
"bundle": "bundle",
}
for path, want := range tests {
t.Run(path, func(t *testing.T) {
if got := DisplayPath(path); got != want {
t.Fatalf("DisplayPath(%q) = %q, want %q", path, got, want)
}
})
}
}
func TestManagedBundleTargets(t *testing.T) {
tests := []struct {
name string
bundlePath string
outputs []string
want []string
}{
{
name: "root",
outputs: []string{"report.html", "assets/style.css"},
want: []string{"report.html", "assets/style.css", StateFileName},
},
{
name: "nested",
bundlePath: "daily",
outputs: []string{"report.html"},
want: []string{"daily/report.html", "daily/" + StateFileName},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := ManagedBundleTargets(tt.bundlePath, tt.outputs)
if err != nil {
t.Fatalf("ManagedBundleTargets() error = %v", err)
}
if len(got) != len(tt.want) {
t.Fatalf("targets = %v, want %v", got, tt.want)
}
for i := range got {
if got[i] != tt.want[i] {
t.Fatalf("targets = %v, want %v", got, tt.want)
}
}
})
}
}
func TestManagedBundleTargetsRejectsInvalidOutputPath(t *testing.T) {
if _, err := ManagedBundleTargets("bundle", []string{"../outside"}); !IsInvalidPath(err) {
t.Fatalf("ManagedBundleTargets() error = %v, want invalid path", err)
}
}
func TestListSortsEntries(t *testing.T) {
backend := walkBackend{
entries: []Entry{

View File

@@ -0,0 +1,256 @@
package testutil
import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
)
const DefaultBundleID = "weather.daily.brentwood.2026-05-30"
var DefaultCreated = time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC)
type SourceFile struct {
Path string
Data string
}
type BundleOptions struct {
ID string
Created time.Time
Files []SourceFile
ExtraFiles []SourceFile
}
type DestinationStateOptions struct {
PipelineID string
DestinationID string
DistributorVersion string
PublishedAt time.Time
}
func DefaultSourceFiles() []SourceFile {
return []SourceFile{
{Path: "report.md", Data: "# Report\nSunny.\n"},
{Path: "summary.txt", Data: "Summary\n"},
}
}
func ValidManifest(opts BundleOptions) bundle.Manifest {
files := sourceFiles(opts)
manifestFiles := make([]bundle.ManifestFile, 0, len(files))
for _, file := range files {
data := []byte(file.Data)
manifestFiles = append(manifestFiles, bundle.ManifestFile{
Path: file.Path,
SHA256: bundle.FileDigest(data),
Size: int64(len(data)),
})
}
manifest := bundle.Manifest{
SchemaVersion: 1,
ID: defaultBundleID(opts.ID),
Created: defaultCreated(opts.Created),
Files: manifestFiles,
}
manifest.Digest = bundle.BundleDigest(manifest.Files)
return manifest
}
func WriteSourceBundle(t testing.TB, root, relative string, opts BundleOptions) bundle.Manifest {
t.Helper()
bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
t.Fatalf("mkdir source bundle: %v", err)
}
manifest := ValidManifest(opts)
for _, file := range sourceFiles(opts) {
path := filepath.Join(bundleRoot, filepath.FromSlash(file.Path))
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
t.Fatalf("mkdir source file parent: %v", err)
}
if err := os.WriteFile(path, []byte(file.Data), 0o600); err != nil {
t.Fatalf("write source file: %v", err)
}
}
writeJSONFile(t, filepath.Join(bundleRoot, bundle.ManifestName), manifest)
return manifest
}
func WriteFakeSourceBundle(t testing.TB, backend *fake.Backend, relative string, opts BundleOptions) bundle.Bundle {
t.Helper()
manifest := ValidManifest(opts)
for _, file := range sourceFiles(opts) {
path := joinStoragePath(t, relative, file.Path)
if _, err := backend.WriteFile(context.Background(), path, []byte(file.Data), storage.WriteOptions{}); err != nil {
t.Fatalf("write fake source file: %v", err)
}
}
data, err := json.MarshalIndent(manifest, "", " ")
if err != nil {
t.Fatalf("marshal manifest: %v", err)
}
manifestPath := joinStoragePath(t, relative, bundle.ManifestName)
if _, err := backend.WriteFile(context.Background(), manifestPath, append(data, '\n'), storage.WriteOptions{}); err != nil {
t.Fatalf("write fake manifest: %v", err)
}
return bundle.Bundle{RootRelativePath: relative, Manifest: manifest}
}
func WriteMinimalLocalConfig(t testing.TB, sourceRoot, destinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
`)
}
func WriteFanoutLocalConfig(t testing.TB, sourceRoot, firstDestinationRoot, secondDestinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive-one
backend: local
path: `+firstDestinationRoot+`
- id: archive-two
backend: local
path: `+secondDestinationRoot+`
`)
}
func WriteDestinationState(t testing.TB, root, relative string, manifest bundle.Manifest, opts DestinationStateOptions) state.DistributorState {
t.Helper()
bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
t.Fatalf("mkdir destination bundle: %v", err)
}
destinationState := DestinationState(manifest, opts)
writeJSONFile(t, filepath.Join(bundleRoot, storage.StateFileName), destinationState)
return destinationState
}
func DestinationState(manifest bundle.Manifest, opts DestinationStateOptions) state.DistributorState {
return state.DistributorState{
SchemaVersion: state.SchemaVersion,
DistributorVersion: opts.DistributorVersion,
PipelineID: defaultString(opts.PipelineID, "reports"),
DestinationID: defaultString(opts.DestinationID, "archive"),
PublishedAt: defaultPublishedAt(opts.PublishedAt),
Source: state.SourceState{Manifest: manifest},
Outputs: sourceOutputs(manifest),
}
}
func ReadDestinationState(t testing.TB, path string) state.DistributorState {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read destination state: %v", err)
}
destinationState, err := state.Parse(data)
if err != nil {
t.Fatalf("parse destination state: %v", err)
}
return destinationState
}
func sourceFiles(opts BundleOptions) []SourceFile {
files := opts.Files
if files == nil {
files = DefaultSourceFiles()
} else {
files = append([]SourceFile(nil), files...)
}
files = append(files, opts.ExtraFiles...)
return files
}
func sourceOutputs(manifest bundle.Manifest) []state.OutputFile {
outputs := make([]state.OutputFile, 0, len(manifest.Files))
for _, file := range manifest.Files {
outputs = append(outputs, state.OutputFile{
Path: file.Path,
Kind: state.OutputKindSource,
SourcePath: file.Path,
SHA256: file.SHA256,
Size: file.Size,
})
}
return outputs
}
func writeConfigFile(t testing.TB, body string) string {
t.Helper()
path := filepath.Join(t.TempDir(), "config.yml")
if err := os.WriteFile(path, []byte(strings.TrimSpace(body)+"\n"), 0o600); err != nil {
t.Fatalf("write config: %v", err)
}
return path
}
func writeJSONFile(t testing.TB, path string, value any) {
t.Helper()
data, err := json.MarshalIndent(value, "", " ")
if err != nil {
t.Fatalf("marshal json: %v", err)
}
data = append(data, '\n')
if err := os.WriteFile(path, data, 0o600); err != nil {
t.Fatalf("write json file: %v", err)
}
}
func joinStoragePath(t testing.TB, root, path string) string {
t.Helper()
joined, err := storage.Join(root, path)
if err != nil {
t.Fatalf("join storage path: %v", err)
}
return joined
}
func defaultBundleID(value string) string {
return defaultString(value, DefaultBundleID)
}
func defaultString(value, fallback string) string {
if value != "" {
return value
}
return fallback
}
func defaultCreated(value time.Time) time.Time {
if value.IsZero() {
return DefaultCreated
}
return value
}
func defaultPublishedAt(value time.Time) time.Time {
if value.IsZero() {
return time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)
}
return value
}

View File

@@ -9,6 +9,7 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
"gitea.maximumdirect.net/eric/distributor/internal/transform"
)
@@ -92,22 +93,12 @@ func TestGenerateDeterministicOutput(t *testing.T) {
func markdownFixture(t *testing.T, markdown string) (*fake.Backend, bundle.Bundle) {
t.Helper()
backend := fake.New()
if _, err := backend.WriteFile(context.Background(), "report.md", []byte(markdown), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
if _, err := backend.WriteFile(context.Background(), "summary.txt", []byte("Summary\n"), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
files := []bundle.ManifestFile{
{Path: "report.md", SHA256: bundle.FileDigest([]byte(markdown)), Size: int64(len(markdown))},
{Path: "summary.txt", SHA256: bundle.FileDigest([]byte("Summary\n")), Size: 8},
}
manifest := bundle.Manifest{
SchemaVersion: 1,
ID: "bundle",
Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
Files: files,
}
manifest.Digest = bundle.BundleDigest(manifest.Files)
return backend, bundle.Bundle{Manifest: manifest}
sourceBundle := testutil.WriteFakeSourceBundle(t, backend, "", testutil.BundleOptions{
ID: "bundle",
Files: []testutil.SourceFile{
{Path: "report.md", Data: markdown},
{Path: "summary.txt", Data: "Summary\n"},
},
})
return backend, sourceBundle
}