Add explicit pipeline configuration imports
This commit is contained in:
@@ -166,22 +166,92 @@ func buildCompositionNode(node *yaml.Node, source, path string) (*compositionNod
|
||||
// list, or final keyed value may have only one base owner, regardless of
|
||||
// whether duplicate values happen to be equal.
|
||||
func mergeAdditiveComposition(base, incoming *compositionDocument) (*compositionDocument, error) {
|
||||
if err := validateCompositionDocument(base, "base"); err != nil {
|
||||
return mergeAdditiveCompositions(base, incoming)
|
||||
}
|
||||
|
||||
// mergeAdditiveCompositions validates the complete base source set before
|
||||
// merging so a conflict names every source that claims the same final path.
|
||||
func mergeAdditiveCompositions(documents ...*compositionDocument) (*compositionDocument, error) {
|
||||
if len(documents) == 0 {
|
||||
return nil, fmt.Errorf("configuration additive base merge requires at least one document")
|
||||
}
|
||||
for index, document := range documents {
|
||||
if err := validateCompositionDocument(document, fmt.Sprintf("base[%d]", index)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := validateAdditiveClaims(documents); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := validateCompositionDocument(incoming, "incoming"); err != nil {
|
||||
return nil, err
|
||||
|
||||
result := &compositionDocument{
|
||||
root: cloneCompositionNode(documents[0].root),
|
||||
sources: append([]string(nil), documents[0].sources...),
|
||||
}
|
||||
merged, err := mergeAdditiveNodes(cloneCompositionNode(base.root), incoming.root)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
for _, incoming := range documents[1:] {
|
||||
merged, err := mergeAdditiveNodes(result.root, incoming.root)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.root = merged
|
||||
result.sources = appendUniqueStrings(result.sources, incoming.sources...)
|
||||
}
|
||||
return &compositionDocument{
|
||||
root: merged,
|
||||
sources: appendUniqueStrings(
|
||||
append([]string(nil), base.sources...), incoming.sources...,
|
||||
),
|
||||
}, nil
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func validateAdditiveClaims(documents []*compositionDocument) error {
|
||||
claims := make(map[string][]*compositionNode)
|
||||
for _, document := range documents {
|
||||
appendCompositionClaims(document.root, claims)
|
||||
}
|
||||
paths := make([]string, 0, len(claims))
|
||||
for path := range claims {
|
||||
paths = append(paths, path)
|
||||
}
|
||||
sort.Slice(paths, func(i, j int) bool {
|
||||
leftDepth := compositionPathDepth(paths[i])
|
||||
rightDepth := compositionPathDepth(paths[j])
|
||||
if leftDepth != rightDepth {
|
||||
return leftDepth < rightDepth
|
||||
}
|
||||
return paths[i] < paths[j]
|
||||
})
|
||||
for _, path := range paths {
|
||||
values := claims[path]
|
||||
if len(values) < 2 {
|
||||
continue
|
||||
}
|
||||
allPopulatedMappings := true
|
||||
for _, value := range values {
|
||||
if value.kind != yaml.MappingNode || len(value.fields) == 0 {
|
||||
allPopulatedMappings = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if !allPopulatedMappings {
|
||||
return newCompositionConflict("additive base merge", path, values...)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func appendCompositionClaims(node *compositionNode, claims map[string][]*compositionNode) {
|
||||
if node == nil {
|
||||
return
|
||||
}
|
||||
if node.path != "" {
|
||||
claims[node.path] = append(claims[node.path], node)
|
||||
}
|
||||
for _, field := range node.fields {
|
||||
appendCompositionClaims(field.value, claims)
|
||||
}
|
||||
}
|
||||
|
||||
func compositionPathDepth(path string) int {
|
||||
if path == "" {
|
||||
return 0
|
||||
}
|
||||
return strings.Count(path, ".") + strings.Count(path, "[") + 1
|
||||
}
|
||||
|
||||
func mergeAdditiveNodes(base, incoming *compositionNode) (*compositionNode, error) {
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -165,6 +163,20 @@ func TestMergeAdditiveCompositionReportsAllClaimingSources(t *testing.T) {
|
||||
t.Fatalf("error = %q, want %q", err, want)
|
||||
}
|
||||
}
|
||||
|
||||
_, err = mergeAdditiveCompositions(
|
||||
mustParseComposition(t, "first.yml", "value: 1\n"),
|
||||
mustParseComposition(t, "second.yml", "value: 2\n"),
|
||||
mustParseComposition(t, "third.yml", "value: 3\n"),
|
||||
)
|
||||
if err == nil {
|
||||
t.Fatal("mergeAdditiveCompositions() error = nil")
|
||||
}
|
||||
for _, want := range []string{"value", "first.yml", "second.yml", "third.yml"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Fatalf("error = %q, want %q", err, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeOverlayCompositionRecursesMapsAndReplacesAtomicValues(t *testing.T) {
|
||||
@@ -307,21 +319,6 @@ zeta: 1
|
||||
}
|
||||
}
|
||||
|
||||
func TestPipelineCompositionEnvelopeIsNotPublicYet(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "pipeline.yml")
|
||||
if err := os.WriteFile(path, []byte(`composition:
|
||||
imports: []
|
||||
whisperx:
|
||||
transcribe_url: https://transcription.example.com/transcribe
|
||||
`), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err := LoadPipeline(path)
|
||||
if err == nil || !strings.Contains(err.Error(), "field composition not found") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want strict public-schema rejection", err)
|
||||
}
|
||||
}
|
||||
|
||||
func mustParseComposition(t *testing.T, source, input string) *compositionDocument {
|
||||
t.Helper()
|
||||
document, err := parseCompositionBytes(source, []byte(input))
|
||||
|
||||
@@ -32,6 +32,8 @@ type PipelineConfig struct {
|
||||
Scriptorium *ScriptoriumConfig `yaml:"scriptorium"`
|
||||
Notarius *NotariusConfig `yaml:"notarius"`
|
||||
Notification NotificationConfig `yaml:"notification"`
|
||||
|
||||
resolution *pipelineResolutionMetadata `yaml:"-"`
|
||||
}
|
||||
|
||||
// CampaignsConfig configures the local campaign registry.
|
||||
|
||||
@@ -14,15 +14,15 @@ import (
|
||||
|
||||
// LoadPipeline loads pipeline configuration from a YAML file with strict field checking.
|
||||
func LoadPipeline(path string) (*PipelineConfig, error) {
|
||||
var cfg PipelineConfig
|
||||
if err := decodeStrictYAML("pipeline", path, &cfg); err != nil {
|
||||
cfg, err := loadComposedPipeline(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load pipeline config: %w", err)
|
||||
}
|
||||
applyPipelineDefaults(&cfg)
|
||||
if err := resolveNotariusPaths(&cfg, path); err != nil {
|
||||
applyPipelineDefaults(cfg)
|
||||
if err := resolveNotariusPaths(cfg, path); err != nil {
|
||||
return nil, fmt.Errorf("load pipeline config: %w", err)
|
||||
}
|
||||
return &cfg, nil
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// LoadCampaign loads campaign configuration from a YAML file with strict field checking.
|
||||
|
||||
@@ -117,8 +117,13 @@ func TestNotariusStrictYAML(t *testing.T) {
|
||||
if err := os.WriteFile(path, []byte(testPipelineBaseYAML+"\n"+tt.yaml), 0o644); err != nil {
|
||||
t.Fatalf("write pipeline: %v", err)
|
||||
}
|
||||
if _, err := LoadPipeline(path); err == nil || !strings.Contains(err.Error(), "strict decode failed") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want strict decode failure", err)
|
||||
_, err := LoadPipeline(path)
|
||||
want := "strict decode failed"
|
||||
if tt.name == "duplicate reference selector" {
|
||||
want = "duplicate YAML key"
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), want) {
|
||||
t.Fatalf("LoadPipeline() error = %v, want containing %q", err, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
224
internal/config/pipeline_composition.go
Normal file
224
internal/config/pipeline_composition.go
Normal file
@@ -0,0 +1,224 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
type pipelineResolutionMetadata struct {
|
||||
rootPath string
|
||||
imports []string
|
||||
sources []string
|
||||
ownership []pipelineFieldOwnership
|
||||
}
|
||||
|
||||
type pipelineFieldOwnership struct {
|
||||
path string
|
||||
sources []string
|
||||
}
|
||||
|
||||
type pipelineCompositionEnvelope struct {
|
||||
imports []string
|
||||
}
|
||||
|
||||
func loadComposedPipeline(path string) (*PipelineConfig, error) {
|
||||
rootPath, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve root pipeline path %q: %w", path, err)
|
||||
}
|
||||
rootFile, err := os.Open(rootPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("pipeline file %q: open: %w", path, err)
|
||||
}
|
||||
rootDocument, parseErr := parseCompositionDocument(rootPath, rootFile)
|
||||
closeErr := rootFile.Close()
|
||||
if parseErr != nil {
|
||||
return nil, parseErr
|
||||
}
|
||||
if closeErr != nil {
|
||||
return nil, fmt.Errorf("pipeline file %q: close: %w", rootPath, closeErr)
|
||||
}
|
||||
|
||||
baseRoot, envelope, err := splitPipelineCompositionEnvelope(rootDocument)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
imports, err := loadPipelineImports(rootPath, envelope.imports)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
documents := make([]*compositionDocument, 0, len(imports)+1)
|
||||
documents = append(documents, baseRoot)
|
||||
for _, imported := range imports {
|
||||
documents = append(documents, imported.document)
|
||||
}
|
||||
merged, err := mergeAdditiveCompositions(documents...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rendered, err := merged.canonicalYAML()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var cfg PipelineConfig
|
||||
if err := decodeStrictYAMLFromReader("pipeline", rootPath, strings.NewReader(string(rendered)), &cfg); err != nil {
|
||||
return nil, fmt.Errorf("assembled pipeline sources %s: %w", formatCompositionSources(merged.sources), err)
|
||||
}
|
||||
records, err := merged.semanticRecords()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
metadata := &pipelineResolutionMetadata{
|
||||
rootPath: rootPath,
|
||||
sources: append([]string(nil), merged.sources...),
|
||||
}
|
||||
for _, imported := range imports {
|
||||
metadata.imports = append(metadata.imports, imported.path)
|
||||
}
|
||||
for _, record := range records {
|
||||
metadata.ownership = append(metadata.ownership, pipelineFieldOwnership{
|
||||
path: record.Path, sources: append([]string(nil), record.Sources...),
|
||||
})
|
||||
}
|
||||
cfg.resolution = metadata
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
func splitPipelineCompositionEnvelope(document *compositionDocument) (*compositionDocument, pipelineCompositionEnvelope, error) {
|
||||
if err := validateCompositionDocument(document, "root pipeline"); err != nil {
|
||||
return nil, pipelineCompositionEnvelope{}, err
|
||||
}
|
||||
root := cloneCompositionNode(document.root)
|
||||
index := compositionFieldIndex(root.fields, "composition")
|
||||
if index < 0 {
|
||||
return &compositionDocument{root: root, sources: append([]string(nil), document.sources...)}, pipelineCompositionEnvelope{}, nil
|
||||
}
|
||||
envelopeNode := root.fields[index].value
|
||||
if envelopeNode.kind != yaml.MappingNode {
|
||||
return nil, pipelineCompositionEnvelope{}, fmt.Errorf(
|
||||
"configuration source %s at composition: expected a mapping, got %s",
|
||||
formatCompositionSources(envelopeNode.sources), yamlKindName(envelopeNode.kind),
|
||||
)
|
||||
}
|
||||
|
||||
var envelope pipelineCompositionEnvelope
|
||||
for _, field := range envelopeNode.fields {
|
||||
switch field.key {
|
||||
case "imports":
|
||||
if field.value.kind != yaml.SequenceNode {
|
||||
return nil, pipelineCompositionEnvelope{}, fmt.Errorf(
|
||||
"configuration source %s at composition.imports: expected a list, got %s",
|
||||
formatCompositionSources(field.value.sources), yamlKindName(field.value.kind),
|
||||
)
|
||||
}
|
||||
for itemIndex, item := range field.value.items {
|
||||
if item.kind != yaml.ScalarNode || item.tag != "!!str" {
|
||||
return nil, pipelineCompositionEnvelope{}, fmt.Errorf(
|
||||
"configuration source %s at composition.imports[%d]: expected a string path",
|
||||
formatCompositionSources(item.sources), itemIndex,
|
||||
)
|
||||
}
|
||||
envelope.imports = append(envelope.imports, item.value)
|
||||
}
|
||||
default:
|
||||
return nil, pipelineCompositionEnvelope{}, fmt.Errorf(
|
||||
"configuration source %s at composition.%s: unknown composition field %q",
|
||||
formatCompositionSources(field.value.sources), field.key, field.key,
|
||||
)
|
||||
}
|
||||
}
|
||||
root.fields = append(root.fields[:index], root.fields[index+1:]...)
|
||||
for fieldIndex := range root.fields {
|
||||
root.fields[fieldIndex].order = fieldIndex
|
||||
}
|
||||
return &compositionDocument{root: root, sources: append([]string(nil), document.sources...)}, envelope, nil
|
||||
}
|
||||
|
||||
type loadedPipelineImport struct {
|
||||
path string
|
||||
document *compositionDocument
|
||||
info os.FileInfo
|
||||
}
|
||||
|
||||
func loadPipelineImports(rootPath string, declared []string) ([]loadedPipelineImport, error) {
|
||||
if len(declared) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
rootDir := filepath.Dir(rootPath)
|
||||
rootInfo, err := os.Stat(rootPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("inspect root pipeline file %q: %w", rootPath, err)
|
||||
}
|
||||
seenPaths := make(map[string]int, len(declared))
|
||||
loaded := make([]loadedPipelineImport, 0, len(declared))
|
||||
for index, raw := range declared {
|
||||
if strings.TrimSpace(raw) != raw || raw == "" {
|
||||
return nil, fmt.Errorf("composition.imports[%d] must be a non-empty path without surrounding whitespace", index)
|
||||
}
|
||||
normalized, err := pathsafe.NormalizeRelativeDestination(raw)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("composition.imports[%d] path %q is invalid: %w", index, raw, err)
|
||||
}
|
||||
extension := filepath.Ext(filepath.FromSlash(normalized))
|
||||
if extension != ".yml" && extension != ".yaml" {
|
||||
return nil, fmt.Errorf("composition.imports[%d] path %q must use .yml or .yaml", index, raw)
|
||||
}
|
||||
if prior, duplicate := seenPaths[normalized]; duplicate {
|
||||
return nil, fmt.Errorf(
|
||||
"composition.imports[%d] path %q duplicates composition.imports[%d] after normalization",
|
||||
index, raw, prior,
|
||||
)
|
||||
}
|
||||
seenPaths[normalized] = index
|
||||
resolved := filepath.Join(rootDir, filepath.FromSlash(normalized))
|
||||
if resolved == rootPath {
|
||||
return nil, fmt.Errorf("composition.imports[%d] path %q imports the root pipeline itself", index, raw)
|
||||
}
|
||||
|
||||
file, err := fileops.OpenConfinedRegularFile(rootDir, normalized)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open composition.imports[%d] path %q beneath root pipeline directory: %w", index, raw, err)
|
||||
}
|
||||
info, statErr := file.Stat()
|
||||
if statErr != nil {
|
||||
_ = file.Close()
|
||||
return nil, fmt.Errorf("inspect composition.imports[%d] path %q: %w", index, raw, statErr)
|
||||
}
|
||||
if os.SameFile(rootInfo, info) {
|
||||
_ = file.Close()
|
||||
return nil, fmt.Errorf("composition.imports[%d] path %q imports the root pipeline itself", index, raw)
|
||||
}
|
||||
for priorIndex, prior := range loaded {
|
||||
if os.SameFile(prior.info, info) {
|
||||
_ = file.Close()
|
||||
return nil, fmt.Errorf(
|
||||
"composition.imports[%d] path %q references the same file as composition.imports[%d] %q",
|
||||
index, raw, priorIndex, declared[priorIndex],
|
||||
)
|
||||
}
|
||||
}
|
||||
document, parseErr := parseCompositionDocument(resolved, file)
|
||||
closeErr := file.Close()
|
||||
if parseErr != nil {
|
||||
return nil, parseErr
|
||||
}
|
||||
if closeErr != nil {
|
||||
return nil, fmt.Errorf("close composition.imports[%d] path %q: %w", index, raw, closeErr)
|
||||
}
|
||||
if compositionFieldIndex(document.root.fields, "composition") >= 0 {
|
||||
return nil, fmt.Errorf(
|
||||
"imported configuration source %q declares composition; only the root pipeline may declare composition",
|
||||
resolved,
|
||||
)
|
||||
}
|
||||
loaded = append(loaded, loadedPipelineImport{path: resolved, document: document, info: info})
|
||||
}
|
||||
return loaded, nil
|
||||
}
|
||||
370
internal/config/pipeline_composition_test.go
Normal file
370
internal/config/pipeline_composition_test.go
Normal file
@@ -0,0 +1,370 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLoadPipelineCompositionImportsDisjointFieldsAndTracksOwnership(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writePipelineSource(t, dir, "pipeline.yml", `composition:
|
||||
imports:
|
||||
- conf.d/platform.yml
|
||||
- conf.d/artifacts.yml
|
||||
whisperx:
|
||||
transcribe_url: https://transcription.example.com/transcribe
|
||||
`)
|
||||
platformPath := writePipelineSource(t, dir, "conf.d/platform.yml", `workspace:
|
||||
root: /srv/narratio/work
|
||||
storage:
|
||||
backend: local
|
||||
`)
|
||||
artifactsPath := writePipelineSource(t, dir, "conf.d/artifacts.yml", `scriptorium:
|
||||
artifacts:
|
||||
player_handout:
|
||||
enabled: false
|
||||
session_recap:
|
||||
enabled: true
|
||||
prompt_id: dnd.session_recap
|
||||
output_path: artifacts/session_recap.md
|
||||
`)
|
||||
|
||||
cfg, err := LoadPipeline(rootPath)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadPipeline() error = %v", err)
|
||||
}
|
||||
if cfg.Workspace.Root != "/srv/narratio/work" || cfg.Storage.Backend != StorageBackendLocal {
|
||||
t.Fatalf("imported platform config = workspace=%q storage=%#v", cfg.Workspace.Root, cfg.Storage)
|
||||
}
|
||||
if cfg.Scriptorium == nil || len(cfg.Scriptorium.Artifacts) != 2 || cfg.Scriptorium.Artifacts["player_handout"].Enabled {
|
||||
t.Fatalf("imported artifacts = %#v", cfg.Scriptorium)
|
||||
}
|
||||
if cfg.WhisperX.TranscribeURL != "https://transcription.example.com/transcribe" {
|
||||
t.Fatalf("root field = %q", cfg.WhisperX.TranscribeURL)
|
||||
}
|
||||
if cfg.resolution == nil {
|
||||
t.Fatal("pipeline resolution metadata = nil")
|
||||
}
|
||||
wantSources := []string{absolutePath(t, rootPath), absolutePath(t, platformPath), absolutePath(t, artifactsPath)}
|
||||
if !reflect.DeepEqual(cfg.resolution.sources, wantSources) || !reflect.DeepEqual(cfg.resolution.imports, wantSources[1:]) {
|
||||
t.Fatalf("resolution sources=%#v imports=%#v, want %#v / %#v", cfg.resolution.sources, cfg.resolution.imports, wantSources, wantSources[1:])
|
||||
}
|
||||
assertPipelineFieldOwner(t, cfg, "whisperx.transcribe_url", absolutePath(t, rootPath))
|
||||
assertPipelineFieldOwner(t, cfg, "workspace.root", absolutePath(t, platformPath))
|
||||
assertPipelineFieldOwner(t, cfg, "scriptorium.artifacts.session_recap.prompt_id", absolutePath(t, artifactsPath))
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionMergesDisjointKeyedEntries(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writePipelineSource(t, dir, "pipeline.yml", `composition:
|
||||
imports: [first.yml, second.yml]
|
||||
whisperx:
|
||||
transcribe_url: https://transcription.example.com/transcribe
|
||||
scriptorium:
|
||||
artifacts:
|
||||
root_artifact:
|
||||
enabled: false
|
||||
`)
|
||||
writePipelineSource(t, dir, "first.yml", `scriptorium:
|
||||
artifacts:
|
||||
first_artifact:
|
||||
enabled: false
|
||||
`)
|
||||
writePipelineSource(t, dir, "second.yml", `scriptorium:
|
||||
artifacts:
|
||||
second_artifact:
|
||||
enabled: false
|
||||
`)
|
||||
|
||||
cfg, err := LoadPipeline(rootPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := len(cfg.Scriptorium.Artifacts); got != 3 {
|
||||
t.Fatalf("artifact count = %d, want 3: %#v", got, cfg.Scriptorium.Artifacts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionRejectsBaseConflictsWithAllSources(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
root string
|
||||
imports map[string]string
|
||||
path string
|
||||
sources []string
|
||||
}{
|
||||
{
|
||||
name: "root and import identical scalar",
|
||||
root: "whisperx:\n language: en\n",
|
||||
imports: map[string]string{"one.yml": "whisperx:\n language: en\n"},
|
||||
path: "whisperx.language", sources: []string{"pipeline.yml", "one.yml"},
|
||||
},
|
||||
{
|
||||
name: "all import claimants",
|
||||
imports: map[string]string{
|
||||
"one.yml": "workspace:\n root: /one\n",
|
||||
"two.yml": "workspace:\n root: /two\n",
|
||||
"three.yml": "workspace:\n root: /three\n",
|
||||
},
|
||||
path: "workspace.root", sources: []string{"one.yml", "two.yml", "three.yml"},
|
||||
},
|
||||
{
|
||||
name: "atomic list",
|
||||
root: "audita:\n modules: [one]\n",
|
||||
imports: map[string]string{"one.yml": "audita:\n modules: [two]\n"},
|
||||
path: "audita.modules", sources: []string{"pipeline.yml", "one.yml"},
|
||||
},
|
||||
{
|
||||
name: "kind conflict",
|
||||
root: "workspace:\n root: /work\n",
|
||||
imports: map[string]string{"one.yml": "workspace: invalid\n"},
|
||||
path: "workspace", sources: []string{"pipeline.yml", "one.yml"},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
order := make([]string, 0, len(tt.imports))
|
||||
for _, name := range []string{"one.yml", "two.yml", "three.yml"} {
|
||||
if _, ok := tt.imports[name]; ok {
|
||||
order = append(order, name)
|
||||
}
|
||||
}
|
||||
root := "composition:\n imports:\n"
|
||||
for _, name := range order {
|
||||
root += " - " + name + "\n"
|
||||
}
|
||||
root += tt.root
|
||||
rootPath := writePipelineSource(t, dir, "pipeline.yml", root)
|
||||
for name, content := range tt.imports {
|
||||
writePipelineSource(t, dir, name, content)
|
||||
}
|
||||
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil {
|
||||
t.Fatal("LoadPipeline() error = nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), tt.path) {
|
||||
t.Fatalf("error = %q, want path %q", err, tt.path)
|
||||
}
|
||||
for _, source := range tt.sources {
|
||||
if !strings.Contains(err.Error(), source) {
|
||||
t.Fatalf("error = %q, want source %q", err, source)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionRejectsUnsafeOrInvalidImports(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
imports []string
|
||||
setup func(*testing.T, string)
|
||||
want string
|
||||
}{
|
||||
{name: "empty", imports: []string{""}, want: "non-empty path"},
|
||||
{name: "surrounding whitespace", imports: []string{" one.yml "}, want: "surrounding whitespace"},
|
||||
{name: "absolute", imports: []string{"/tmp/one.yml"}, want: "invalid"},
|
||||
{name: "traversal", imports: []string{"../one.yml"}, want: "invalid"},
|
||||
{name: "unsupported extension", imports: []string{"one.json"}, want: ".yml or .yaml"},
|
||||
{name: "missing", imports: []string{"missing.yml"}, want: "open composition.imports"},
|
||||
{name: "duplicate normalized", imports: []string{"one.yml", "./one.yml"}, setup: func(t *testing.T, dir string) {
|
||||
writePipelineSource(t, dir, "one.yml", "workspace:\n root: /work\n")
|
||||
}, want: "duplicates composition.imports"},
|
||||
{name: "root self import", imports: []string{"pipeline.yml"}, want: "root pipeline itself"},
|
||||
{name: "directory", imports: []string{"directory.yml"}, setup: func(t *testing.T, dir string) {
|
||||
if err := os.Mkdir(filepath.Join(dir, "directory.yml"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}, want: "regular file"},
|
||||
{name: "symlink file", imports: []string{"link.yml"}, setup: func(t *testing.T, dir string) {
|
||||
writePipelineSource(t, dir, "target.yml", "workspace:\n root: /work\n")
|
||||
if err := os.Symlink("target.yml", filepath.Join(dir, "link.yml")); err != nil {
|
||||
t.Skipf("symlink unavailable: %v", err)
|
||||
}
|
||||
}, want: "not a regular file"},
|
||||
{name: "symlink directory", imports: []string{"linked/one.yml"}, setup: func(t *testing.T, dir string) {
|
||||
writePipelineSource(t, dir, "actual/one.yml", "workspace:\n root: /work\n")
|
||||
if err := os.Symlink("actual", filepath.Join(dir, "linked")); err != nil {
|
||||
t.Skipf("symlink unavailable: %v", err)
|
||||
}
|
||||
}, want: "not a regular directory"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
if tt.setup != nil {
|
||||
tt.setup(t, dir)
|
||||
}
|
||||
rootPath := writeImportRoot(t, dir, tt.imports)
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(strings.ToLower(err.Error()), strings.ToLower(tt.want)) {
|
||||
t.Fatalf("LoadPipeline() error = %v, want containing %q", err, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionRejectsSameFileAliasesAndImportedComposition(t *testing.T) {
|
||||
t.Run("same file through hard link", func(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("hard-link identity behavior is platform-specific")
|
||||
}
|
||||
dir := t.TempDir()
|
||||
writePipelineSource(t, dir, "one.yml", "workspace:\n root: /work\n")
|
||||
if err := os.Link(filepath.Join(dir, "one.yml"), filepath.Join(dir, "two.yml")); err != nil {
|
||||
t.Skipf("hard links unavailable: %v", err)
|
||||
}
|
||||
rootPath := writeImportRoot(t, dir, []string{"one.yml", "two.yml"})
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(err.Error(), "same file") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want same-file rejection", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("imported composition", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writeImportRoot(t, dir, []string{"nested.yml"})
|
||||
writePipelineSource(t, dir, "nested.yml", "composition:\n imports: []\n")
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(err.Error(), "only the root pipeline") || !strings.Contains(err.Error(), "nested.yml") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want imported composition rejection", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("future profile field", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writePipelineSource(t, dir, "pipeline.yml", "composition:\n profiles: {}\n")
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(err.Error(), "unknown composition field") || !strings.Contains(err.Error(), "profiles") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want profile field rejected before its implementation", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionReportsImportedParseAndSchemaSources(t *testing.T) {
|
||||
t.Run("malformed imported YAML", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writeImportRoot(t, dir, []string{"broken.yml"})
|
||||
brokenPath := writePipelineSource(t, dir, "broken.yml", "workspace: [\n")
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(err.Error(), absolutePath(t, brokenPath)) || !strings.Contains(err.Error(), "decode YAML") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want imported parse source", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("unknown imported field", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
rootPath := writeImportRoot(t, dir, []string{"unknown.yml"})
|
||||
unknownPath := writePipelineSource(t, dir, "unknown.yml", "unknown_field: true\n")
|
||||
_, err := LoadPipeline(rootPath)
|
||||
if err == nil || !strings.Contains(err.Error(), absolutePath(t, unknownPath)) || !strings.Contains(err.Error(), "strict decode failed") {
|
||||
t.Fatalf("LoadPipeline() error = %v, want assembled source-aware strict error", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestLoadPipelineCompositionKeepsRelativePathsRootBased(t *testing.T) {
|
||||
rootDir := t.TempDir()
|
||||
monolithicPath := writePipelineSource(t, rootDir, "monolithic.yml", testPipelineBaseYAML+`
|
||||
notarius:
|
||||
enabled: true
|
||||
config_path: tool/notarius.yml
|
||||
pipeline_id: dnd-session
|
||||
outputs:
|
||||
npc_registry:
|
||||
lane_id: npc-registry
|
||||
media_type: application/json
|
||||
schema_id: notarius.dnd.npc_registry
|
||||
schema_version: v1
|
||||
`)
|
||||
composedPath := writePipelineSource(t, rootDir, "pipeline.yml", `composition:
|
||||
imports: [conf.d/extraction.yml]
|
||||
`+testPipelineBaseYAML)
|
||||
writePipelineSource(t, rootDir, "conf.d/extraction.yml", `notarius:
|
||||
enabled: true
|
||||
config_path: tool/notarius.yml
|
||||
pipeline_id: dnd-session
|
||||
outputs:
|
||||
npc_registry:
|
||||
lane_id: npc-registry
|
||||
media_type: application/json
|
||||
schema_id: notarius.dnd.npc_registry
|
||||
schema_version: v1
|
||||
`)
|
||||
|
||||
monolithic, err := LoadPipeline(monolithicPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
composed, err := LoadPipeline(composedPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := filepath.Join(rootDir, "tool", "notarius.yml")
|
||||
if monolithic.Notarius.ConfigPath != want || composed.Notarius.ConfigPath != want {
|
||||
t.Fatalf("config paths = monolithic %q composed %q, want %q", monolithic.Notarius.ConfigPath, composed.Notarius.ConfigPath, want)
|
||||
}
|
||||
if monolithic.Notarius.WorkingDirectory != filepath.Dir(want) || composed.Notarius.WorkingDirectory != filepath.Dir(want) {
|
||||
t.Fatalf("working directories = %q / %q", monolithic.Notarius.WorkingDirectory, composed.Notarius.WorkingDirectory)
|
||||
}
|
||||
}
|
||||
|
||||
func writeImportRoot(t *testing.T, dir string, imports []string) string {
|
||||
t.Helper()
|
||||
var builder strings.Builder
|
||||
builder.WriteString("composition:\n imports:\n")
|
||||
for _, imported := range imports {
|
||||
builder.WriteString(" - ")
|
||||
if imported == "" {
|
||||
builder.WriteString(`""`)
|
||||
} else {
|
||||
builder.WriteString(`"` + imported + `"`)
|
||||
}
|
||||
builder.WriteByte('\n')
|
||||
}
|
||||
builder.WriteString("whisperx:\n transcribe_url: https://transcription.example.com/transcribe\n")
|
||||
return writePipelineSource(t, dir, "pipeline.yml", builder.String())
|
||||
}
|
||||
|
||||
func writePipelineSource(t *testing.T, root, relative, content string) string {
|
||||
t.Helper()
|
||||
path := filepath.Join(root, filepath.FromSlash(relative))
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
func absolutePath(t *testing.T, path string) string {
|
||||
t.Helper()
|
||||
absolute, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return absolute
|
||||
}
|
||||
|
||||
func assertPipelineFieldOwner(t *testing.T, cfg *PipelineConfig, path, source string) {
|
||||
t.Helper()
|
||||
if cfg == nil || cfg.resolution == nil {
|
||||
t.Fatal("pipeline resolution metadata is absent")
|
||||
}
|
||||
for _, ownership := range cfg.resolution.ownership {
|
||||
if ownership.path == path {
|
||||
if !reflect.DeepEqual(ownership.sources, []string{source}) {
|
||||
t.Fatalf("owner of %s = %#v, want %q", path, ownership.sources, source)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("ownership path %q not found: %#v", path, cfg.resolution.ownership)
|
||||
}
|
||||
Reference in New Issue
Block a user