278 lines
10 KiB
Go
278 lines
10 KiB
Go
package config
|
|
|
|
import (
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
func TestPipelineConfigKitImportsProfileAndBaseSelection(t *testing.T) {
|
|
dir := t.TempDir()
|
|
writePipelineFixture(t, dir, "pipeline.yml", `configkit:
|
|
imports:
|
|
- path: config/workspace.yml
|
|
- path: config/artifact.yml
|
|
whisperx:
|
|
transcribe_url: https://transcription.example.test/transcribe
|
|
`)
|
|
writePipelineFixture(t, dir, "config/workspace.yml", "workspace:\n root: /srv/base\n")
|
|
writePipelineFixture(t, dir, "config/artifact.yml", `scriptorium:
|
|
artifacts:
|
|
recap:
|
|
enabled: false
|
|
output_path: artifacts/recap.md
|
|
`)
|
|
writePipelineFixture(t, dir, "profiles/testing.yml", "workspace:\n root: /srv/testing\n")
|
|
writePipelineFixture(t, dir, "profiles/unselected.yml", "unknown: [\n")
|
|
|
|
root := filepath.Join(dir, "pipeline.yml")
|
|
for _, profile := range []*string{nil, stringPointer(""), stringPointer("default")} {
|
|
cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: profile})
|
|
if err != nil {
|
|
t.Fatalf("load base selection %#v: %v", profile, err)
|
|
}
|
|
if cfg.Workspace.Root != "/srv/base" {
|
|
t.Fatalf("base workspace = %q", cfg.Workspace.Root)
|
|
}
|
|
}
|
|
selected := "testing"
|
|
cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: &selected})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if cfg.Workspace.Root != "/srv/testing" || len(EffectivePipelineImports(cfg)) != 2 {
|
|
t.Fatalf("profiled pipeline = %#v, imports=%#v", cfg.Workspace, EffectivePipelineImports(cfg))
|
|
}
|
|
profile, ok := SelectedPipelineProfile(cfg)
|
|
if !ok || profile.Name != "testing" || profile.OverlayPath != filepath.Join(dir, "profiles", "testing.yml") {
|
|
t.Fatalf("selected profile = %#v, %v", profile, ok)
|
|
}
|
|
}
|
|
|
|
func TestPipelineConfigKitRejectsDuplicateBaseDeclarations(t *testing.T) {
|
|
for _, test := range []struct {
|
|
name, first, second, clearedContainer string
|
|
}{
|
|
{name: "artifact", first: "scriptorium:\n artifacts:\n recap:\n prompt_id: recap\n", second: "scriptorium:\n artifacts:\n recap:\n output_path: artifacts/recap.md\n", clearedContainer: "scriptorium:\n artifacts: null\n"},
|
|
{name: "artifact family", first: "scriptorium:\n artifact_families:\n recap:\n for_each: party.characters\n", second: "scriptorium:\n artifact_families:\n recap:\n output_path_pattern: artifacts/{character_id}.md\n", clearedContainer: "scriptorium:\n artifact_families: null\n"},
|
|
} {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
writePipelineFixture(t, dir, "pipeline.yml", `configkit:
|
|
imports:
|
|
- path: first.yml
|
|
- path: second.yml
|
|
whisperx:
|
|
transcribe_url: https://transcription.example.test/transcribe
|
|
`)
|
|
writePipelineFixture(t, dir, "first.yml", test.first)
|
|
writePipelineFixture(t, dir, "second.yml", test.second)
|
|
writePipelineFixture(t, dir, "profiles/clear-container.yml", test.clearedContainer)
|
|
writePipelineFixture(t, dir, "profiles/clear-scriptorium.yml", "scriptorium: null\n")
|
|
|
|
for _, profile := range []string{"", "clear-container", "clear-scriptorium"} {
|
|
t.Run(profile, func(t *testing.T) {
|
|
options := PipelineLoadOptions{}
|
|
if profile != "" {
|
|
options.Profile = &profile
|
|
}
|
|
_, err := LoadPipelineWithOptions(filepath.Join(dir, "pipeline.yml"), options)
|
|
if err == nil || !strings.Contains(err.Error(), "declared by multiple base files") ||
|
|
!strings.Contains(err.Error(), "first.yml") || !strings.Contains(err.Error(), "second.yml") {
|
|
t.Fatalf("duplicate declaration error = %v", err)
|
|
}
|
|
})
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPipelineProfileDoesNotRequireUnusedBaseCredentials(t *testing.T) {
|
|
const production = "NARRATIO_PROFILE_UNUSED_PROD_KEY"
|
|
const testingKey = "NARRATIO_PROFILE_TEST_KEY"
|
|
t.Setenv(production, "")
|
|
if err := os.Unsetenv(production); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Setenv(testingKey, "synthetic-test-key")
|
|
for _, credential := range []string{testingKey, "null"} {
|
|
t.Run(credential, func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
root := writePipelineFixture(t, dir, "pipeline.yml", "whisperx:\n transcribe_url: https://transcription.example.test/transcribe\naudita:\n llm_api_key: "+production+"\n")
|
|
writePipelineFixture(t, dir, "profiles/testing.yml", "audita:\n llm_api_key: "+credential+"\n")
|
|
profile := "testing"
|
|
cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: &profile})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if credential == "null" {
|
|
if cfg.Audita.LLMAPIKey != nil {
|
|
t.Fatal("profile did not remove the base credential")
|
|
}
|
|
} else if cfg.Audita.LLMAPIKey == nil || cfg.Audita.LLMAPIKey.EnvName() != testingKey || !cfg.Audita.LLMAPIKey.IsResolved() {
|
|
t.Fatal("profile credential was not resolved")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNotariusCredentialProfileReplacementAndProvenance(t *testing.T) {
|
|
const baseKey = "NARRATIO_NOTARIUS_UNUSED_BASE_KEY"
|
|
const profileKey = "NARRATIO_NOTARIUS_PROFILE_KEY"
|
|
if err := os.Unsetenv(baseKey); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Setenv(profileKey, "synthetic-notarius-profile-key")
|
|
|
|
for _, credential := range []string{profileKey, "null"} {
|
|
t.Run(credential, func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
root := writePipelineFixture(t, dir, "pipeline.yml", "whisperx:\n transcribe_url: https://transcription.example.test/transcribe\nnotarius:\n enabled: false\n llm_api_key: "+baseKey+"\n")
|
|
overlay := writePipelineFixture(t, dir, "profiles/testing.yml", "notarius:\n llm_api_key: "+credential+"\n")
|
|
profile := "testing"
|
|
cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: &profile})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if credential == "null" {
|
|
if cfg.Notarius.LLMAPIKey != nil {
|
|
t.Fatal("profile did not remove the base credential")
|
|
}
|
|
return
|
|
}
|
|
if cfg.Notarius.LLMAPIKey == nil || cfg.Notarius.LLMAPIKey.EnvName() != profileKey || !cfg.Notarius.LLMAPIKey.IsResolved() {
|
|
t.Fatal("profile credential was not resolved")
|
|
}
|
|
sources, err := EffectivePipelineSources(cfg, ResolvedParty{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := false
|
|
for _, source := range sources {
|
|
if source.Path == "notarius.llm_api_key" {
|
|
found = source.Source == overlay
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatal("Notarius credential provenance did not identify the selected profile")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNotariusCredentialRootProvenance(t *testing.T) {
|
|
const envName = "NARRATIO_NOTARIUS_ROOT_KEY"
|
|
t.Setenv(envName, "synthetic-notarius-root-key")
|
|
dir := t.TempDir()
|
|
root := writePipelineFixture(t, dir, "pipeline.yml", "whisperx:\n transcribe_url: https://transcription.example.test/transcribe\nnotarius:\n enabled: false\n llm_api_key: "+envName+"\n")
|
|
cfg, err := LoadPipeline(root)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sources, err := EffectivePipelineSources(cfg, ResolvedParty{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, source := range sources {
|
|
if source.Path == "notarius.llm_api_key" && source.Source == root {
|
|
return
|
|
}
|
|
}
|
|
t.Fatal("Notarius credential provenance did not identify the root pipeline")
|
|
}
|
|
|
|
func TestNotariusCacheControlProfileReplacementAndNullReset(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
overlay string
|
|
wantResume bool
|
|
wantMode string
|
|
wantSource bool
|
|
wantNil bool
|
|
}{
|
|
{name: "replacement", overlay: "notarius:\n resume: true\n chunk_cache: refresh\n", wantResume: true, wantMode: "refresh", wantSource: true},
|
|
{name: "null reset", overlay: "notarius: null\n", wantNil: true},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
dir := t.TempDir()
|
|
root := writePipelineFixture(t, dir, "pipeline.yml", "whisperx:\n transcribe_url: https://transcription.example.test/transcribe\nnotarius:\n enabled: false\n resume: false\n chunk_cache: bypass\n")
|
|
overlay := writePipelineFixture(t, dir, "profiles/testing.yml", test.overlay)
|
|
profile := "testing"
|
|
cfg, err := LoadPipelineWithOptions(root, PipelineLoadOptions{Profile: &profile})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if test.wantNil {
|
|
if cfg.Notarius != nil {
|
|
t.Fatalf("Notarius = %#v, want null reset", cfg.Notarius)
|
|
}
|
|
return
|
|
}
|
|
if cfg.Notarius == nil || cfg.Notarius.Resume != test.wantResume || cfg.Notarius.ChunkCache != test.wantMode {
|
|
t.Fatalf("cache controls = resume %v, mode %q; want resume %v, mode %q", cfg.Notarius.Resume, cfg.Notarius.ChunkCache, test.wantResume, test.wantMode)
|
|
}
|
|
if !test.wantSource {
|
|
return
|
|
}
|
|
sources, err := EffectivePipelineSources(cfg, ResolvedParty{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := map[string]bool{}
|
|
for _, source := range sources {
|
|
if source.Path == "notarius.resume" || source.Path == "notarius.chunk_cache" {
|
|
found[source.Path] = source.Source == overlay
|
|
}
|
|
}
|
|
if !found["notarius.resume"] || !found["notarius.chunk_cache"] {
|
|
t.Fatalf("cache-control provenance did not identify profile: %#v", found)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPipelineConfigKitAllowsOneBaseDeclarationWithProfileOverride(t *testing.T) {
|
|
dir := t.TempDir()
|
|
writePipelineFixture(t, dir, "pipeline.yml", `configkit:
|
|
imports:
|
|
- path: artifact.yml
|
|
whisperx:
|
|
transcribe_url: https://transcription.example.test/transcribe
|
|
`)
|
|
writePipelineFixture(t, dir, "artifact.yml", "scriptorium:\n artifacts:\n recap:\n prompt_id: base\n output_path: artifacts/recap.md\n")
|
|
writePipelineFixture(t, dir, "profiles/testing.yml", "scriptorium:\n artifacts:\n recap:\n prompt_id: testing\n")
|
|
profile := "testing"
|
|
cfg, err := LoadPipelineWithOptions(filepath.Join(dir, "pipeline.yml"), PipelineLoadOptions{Profile: &profile})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := cfg.Scriptorium.Artifacts["recap"].PromptID; got != "testing" {
|
|
t.Fatalf("profile prompt = %q, want testing", got)
|
|
}
|
|
}
|
|
|
|
func TestPipelineConfigKitRejectsUnknownDirectiveAlias(t *testing.T) {
|
|
dir := t.TempDir()
|
|
path := writePipelineFixture(t, dir, "pipeline.yml", "composition:\n imports: []\n")
|
|
_, err := LoadPipeline(path)
|
|
if err == nil || !strings.Contains(err.Error(), "decode.unknown_field") {
|
|
t.Fatalf("composition alias error = %v", err)
|
|
}
|
|
}
|
|
|
|
func writePipelineFixture(t *testing.T, root, relative, contents 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(contents), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return path
|
|
}
|
|
|
|
func stringPointer(value string) *string { return &value }
|