281 lines
11 KiB
Go
281 lines
11 KiB
Go
package app
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
|
)
|
|
|
|
const cleanupOrchestrationRunID = "20260905T010203Z-a1b2c3d4"
|
|
|
|
type cleanupPublishEvidenceStage struct {
|
|
runs *int
|
|
skipSpoolTarget bool
|
|
}
|
|
|
|
func (cleanupPublishEvidenceStage) Name() string { return "publish" }
|
|
|
|
func (s cleanupPublishEvidenceStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
|
if s.runs != nil {
|
|
*s.runs++
|
|
}
|
|
if !s.skipSpoolTarget {
|
|
spoolTarget := artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, m.Campaign, m.SessionID, m.RunID)
|
|
if err := os.MkdirAll(spoolTarget, 0o755); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := os.WriteFile(filepath.Join(spoolTarget, "audio.flac"), []byte("audio"), 0o644); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
workspaceTarget := artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, m.Campaign, m.SessionID, m.RunID)
|
|
if err := os.MkdirAll(workspaceTarget, 0o755); err != nil {
|
|
return nil, err
|
|
}
|
|
return &stage.StageResult{Metadata: cleanupPublishMetadata(m)}, nil
|
|
}
|
|
|
|
func cleanupPublishMetadata(m *manifest.Manifest) map[string]any {
|
|
return map[string]any{
|
|
"uploaded": true,
|
|
"published_run_id": m.RunID,
|
|
"remote_commit_key": artifacts.S3RunCommitKey(m.S3SessionPrefix, m.RunID),
|
|
"current_commit_pointer_key": artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix),
|
|
}
|
|
}
|
|
|
|
func TestCleanupStopsAtPublishAndNotifyFailureBoundaries(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
stages func(*int) []stage.Stage
|
|
}{
|
|
{
|
|
name: "publish failure",
|
|
stages: func(_ *int) []stage.Stage {
|
|
return []stage.Stage{failingStage{name: "publish", err: errors.New("publish failed")}, selectedCleanupStage(t)}
|
|
},
|
|
},
|
|
{
|
|
name: "notify failure",
|
|
stages: func(publishRuns *int) []stage.Stage {
|
|
return []stage.Stage{
|
|
cleanupPublishEvidenceStage{runs: publishRuns},
|
|
failingStage{name: "notify", err: errors.New("notify failed")},
|
|
selectedCleanupStage(t),
|
|
}
|
|
},
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
cfg, remote := cleanupOrchestrationConfig(t)
|
|
publishRuns := 0
|
|
_, err := executePlanWithDependencies(
|
|
context.Background(), cfg, BoundedPlan{stages: test.stages(&publishRuns)},
|
|
RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(),
|
|
)
|
|
if err == nil {
|
|
t.Fatal("execution error = nil")
|
|
}
|
|
loaded := loadCleanupOrchestrationManifest(t, cfg)
|
|
if loaded.Stages["cleanup"] != nil {
|
|
t.Fatalf("cleanup record = %#v, want no execution", loaded.Stages["cleanup"])
|
|
}
|
|
if test.name == "notify failure" {
|
|
workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID)
|
|
if _, statErr := os.Stat(workspaceTarget); statErr != nil {
|
|
t.Fatalf("workspace target removed after notify failure: %v", statErr)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestCleanupFailurePreservesPublishAndRetriesWithoutRepublishing(t *testing.T) {
|
|
cfg, remote := cleanupOrchestrationConfig(t)
|
|
badSpoolRoot := filepath.Join(t.TempDir(), "spool-file")
|
|
if err := os.WriteFile(badSpoolRoot, []byte("not a directory"), 0o644); err != nil {
|
|
t.Fatalf("write invalid spool root: %v", err)
|
|
}
|
|
cfg.Pipeline.Spool.Root = badSpoolRoot
|
|
remote.SeedObject(storage.FakeObject{Key: "sentinel/remote-state", Data: []byte("unchanged")})
|
|
publishRuns := 0
|
|
plan := BoundedPlan{stages: []stage.Stage{
|
|
cleanupPublishEvidenceStage{runs: &publishRuns, skipSpoolTarget: true},
|
|
resultStage{name: "notify"},
|
|
selectedCleanupStage(t),
|
|
}}
|
|
|
|
_, err := executePlanWithDependencies(
|
|
context.Background(), cfg, plan, RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(),
|
|
)
|
|
if err == nil {
|
|
t.Fatal("cleanup failure = nil")
|
|
}
|
|
loaded := loadCleanupOrchestrationManifest(t, cfg)
|
|
if loaded.Stages["publish"] == nil || loaded.Stages["publish"].Status != manifest.StatusSucceeded {
|
|
t.Fatalf("publish record = %#v, want succeeded", loaded.Stages["publish"])
|
|
}
|
|
if loaded.Stages["cleanup"] == nil || loaded.Stages["cleanup"].Status != manifest.StatusFailed {
|
|
t.Fatalf("cleanup record = %#v, want failed", loaded.Stages["cleanup"])
|
|
}
|
|
if got := string(remote.Objects["sentinel/remote-state"].Data); got != "unchanged" {
|
|
t.Fatalf("remote sentinel = %q, want unchanged", got)
|
|
}
|
|
if len(remote.Uploads)+len(remote.Downloads)+len(remote.Reads)+len(remote.Deletes)+len(remote.ExistsKeys) != 0 {
|
|
t.Fatalf("cleanup touched remote storage: %#v", remote)
|
|
}
|
|
|
|
workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID)
|
|
cfg.Pipeline.Spool.Root = t.TempDir()
|
|
spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, cleanupOrchestrationRunID)
|
|
if err := os.MkdirAll(spoolTarget, 0o755); err != nil {
|
|
t.Fatalf("create retry spool target: %v", err)
|
|
}
|
|
summary, err := executePlanWithDependencies(
|
|
context.Background(), cfg, plan, RunOptions{Env: &Env{ObjectStore: remote}}, cleanupOrchestrationDependencies(),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("retry cleanup: %v", err)
|
|
}
|
|
if publishRuns != 1 {
|
|
t.Fatalf("publish runs = %d, want one", publishRuns)
|
|
}
|
|
if !reflect.DeepEqual(summary.Executed, []string{"cleanup"}) || !reflect.DeepEqual(summary.Skipped, []string{"publish", "notify"}) {
|
|
t.Fatalf("retry summary = %#v", summary)
|
|
}
|
|
if _, statErr := os.Lstat(workspaceTarget); !errors.Is(statErr, os.ErrNotExist) {
|
|
t.Fatalf("workspace target stat error = %v, want absent", statErr)
|
|
}
|
|
if _, statErr := os.Lstat(spoolTarget); !errors.Is(statErr, os.ErrNotExist) {
|
|
t.Fatalf("spool target stat error = %v, want absent", statErr)
|
|
}
|
|
}
|
|
|
|
func TestRunStageCleanupUsesPublishEvidenceAndFailsClosedWithoutIt(t *testing.T) {
|
|
workspaceRoot := t.TempDir()
|
|
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
|
|
enableCleanupForCommandTest(t, pipelinePath)
|
|
cfg, err := config.Load(pipelinePath, campaignPath, sessionPath)
|
|
if err != nil {
|
|
t.Fatalf("load cleanup config: %v", err)
|
|
}
|
|
|
|
publishedRunID := "20260904T010203Z-a1b2c3d4"
|
|
m := manifest.New(cfg.Session.SessionID, time.Date(2026, 9, 4, 1, 2, 3, 0, time.UTC))
|
|
m.Campaign = cfg.Session.Campaign
|
|
m.RunID = publishedRunID
|
|
m.S3SessionPrefix = artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, m.Campaign, m.SessionID)
|
|
m.MarkStageSucceeded("publish", time.Date(2026, 9, 4, 1, 3, 0, 0, time.UTC), nil)
|
|
m.Stages["publish"].Metadata = cleanupPublishMetadata(m)
|
|
manifestPath := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID)
|
|
if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPath, m); err != nil {
|
|
t.Fatalf("save publication evidence: %v", err)
|
|
}
|
|
workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID, publishedRunID)
|
|
spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, m.Campaign, m.SessionID, publishedRunID)
|
|
for _, target := range []string{workspaceTarget, spoolTarget} {
|
|
if err := os.MkdirAll(target, 0o755); err != nil {
|
|
t.Fatalf("create cleanup target: %v", err)
|
|
}
|
|
}
|
|
|
|
var output bytes.Buffer
|
|
if err := RunStage(context.Background(), []string{
|
|
"cleanup", cfg.Session.SessionID, "--config", pipelinePath,
|
|
"--campaign-file", campaignPath, "--session", sessionPath,
|
|
}, &output); err != nil {
|
|
t.Fatalf("RunStage(cleanup): %v", err)
|
|
}
|
|
if !strings.Contains(output.String(), "stage=cleanup executed=1 skipped=0") {
|
|
t.Fatalf("output = %q, want direct cleanup execution", output.String())
|
|
}
|
|
for _, target := range []string{workspaceTarget, spoolTarget} {
|
|
if _, statErr := os.Lstat(target); !errors.Is(statErr, os.ErrNotExist) {
|
|
t.Fatalf("cleanup target %q stat error = %v, want absent", target, statErr)
|
|
}
|
|
}
|
|
cleaned := loadCleanupOrchestrationManifest(t, cfg)
|
|
if cleaned.RunID == publishedRunID {
|
|
t.Fatalf("cleanup invocation run id = %q, want a new invocation identity", cleaned.RunID)
|
|
}
|
|
if _, statErr := os.Lstat(cleaned.LocalWorkDir); !errors.Is(statErr, os.ErrNotExist) {
|
|
t.Fatalf("cleanup invocation work directory stat error = %v, want absent", statErr)
|
|
}
|
|
|
|
missingRoot := t.TempDir()
|
|
missingPipeline, missingCampaign, missingSession := writeValidConfigFiles(t, missingRoot)
|
|
enableCleanupForCommandTest(t, missingPipeline)
|
|
err = RunStage(context.Background(), []string{
|
|
"cleanup", "2026-05-03", "--config", missingPipeline,
|
|
"--campaign-file", missingCampaign, "--session", missingSession,
|
|
}, &bytes.Buffer{})
|
|
if err == nil || !strings.Contains(err.Error(), "cleanup requires a recorded publish stage") {
|
|
t.Fatalf("RunStage(cleanup) without evidence error = %v", err)
|
|
}
|
|
}
|
|
|
|
func enableCleanupForCommandTest(t *testing.T, pipelinePath string) {
|
|
t.Helper()
|
|
data, err := os.ReadFile(pipelinePath)
|
|
if err != nil {
|
|
t.Fatalf("read pipeline config: %v", err)
|
|
}
|
|
content := strings.Replace(string(data), "workspace:\n root: ", "workspace:\n cleanup_after_publish: true\n root: ", 1)
|
|
content = strings.Replace(content, "spool:\n root: ", "spool:\n delete_audio_after_publish: true\n root: ", 1)
|
|
content = strings.Replace(content, " upload_run: false", " upload_run: true", 1)
|
|
if err := os.WriteFile(pipelinePath, []byte(content), 0o644); err != nil {
|
|
t.Fatalf("write cleanup config: %v", err)
|
|
}
|
|
}
|
|
|
|
func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) {
|
|
t.Helper()
|
|
cfg := testConfig(t)
|
|
cfg.Pipeline.Spool.Root = t.TempDir()
|
|
cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
|
|
cfg.Pipeline.Workspace.CleanupAfterPublish = true
|
|
cfg.Pipeline.Publish = &config.PublishConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)}
|
|
cfg.Pipeline.Storage.Backend = config.StorageBackendS3
|
|
cfg.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "archive", RootPrefix: "dnd"}
|
|
return cfg, &storage.FakeBackend{}
|
|
}
|
|
|
|
func selectedCleanupStage(t *testing.T) stage.Stage {
|
|
t.Helper()
|
|
selected, err := BuildSingleStagePlan("cleanup")
|
|
if err != nil {
|
|
t.Fatalf("select cleanup: %v", err)
|
|
}
|
|
return selected[0]
|
|
}
|
|
|
|
func cleanupOrchestrationDependencies() invocationDependencies {
|
|
return invocationDependencies{
|
|
now: func() time.Time { return time.Date(2026, 9, 5, 1, 2, 3, 0, time.UTC) },
|
|
newRunID: func() (string, error) { return cleanupOrchestrationRunID, nil },
|
|
}
|
|
}
|
|
|
|
func loadCleanupOrchestrationManifest(t *testing.T, cfg *config.Config) *manifest.Manifest {
|
|
t.Helper()
|
|
path := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
|
|
loaded, err := (&manifest.LocalStore{}).Load(context.Background(), path)
|
|
if err != nil {
|
|
t.Fatalf("load manifest: %v", err)
|
|
}
|
|
return loaded
|
|
}
|