202 lines
7.4 KiB
Go
202 lines
7.4 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"reflect"
|
|
"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 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
|
|
}
|