A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
11 kB · 242 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243package controller
import ( "context" "os" "path/filepath" "testing" "time"
"forgejo.alexma.top/alexma233/composia/internal/core/config" "forgejo.alexma.top/alexma233/composia/internal/core/task" "forgejo.alexma.top/alexma233/composia/internal/platform/store")
func TestRunScheduledTasksPassCreatesBackupTasksFromDefaultOverrideAndNone(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "app/composia-meta.yaml": "name: app\nnodes:\n - main\ndata_protect:\n data:\n - name: inherit\n backup:\n strategy: files.copy\n include:\n - ./inherit\n - name: override\n backup:\n strategy: files.copy\n include:\n - ./override\n - name: disabled\n backup:\n strategy: files.copy\n include:\n - ./disabled\nbackup:\n data:\n - name: inherit\n - name: override\n schedule: \"5 2 * * *\"\n - name: disabled\n schedule: none\n", }) if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) }
db := openControllerTestDB(t) defer func() { _ = db.Close() }() ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "app"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 9, 2, 5, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
cfg := &config.ControllerConfig{ RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Backup: &config.ControllerBackupConfig{DefaultSchedule: "5 2 * * *"}, } if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 2, 5, 20, 0, time.UTC)); err != nil { t.Fatalf("run scheduled pass: %v", err) } if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 2, 5, 40, 0, time.UTC)); err != nil { t.Fatalf("rerun scheduled pass: %v", err) }
tasks, totalCount, err := db.ListTasks(ctx, nil, []string{"app"}, nil, []string{string(task.TypeBackup)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list backup tasks: %v", err) } if totalCount != 1 || len(tasks) != 1 { t.Fatalf("expected 1 scheduled backup task, got total=%d len=%d", totalCount, len(tasks)) } detail, err := db.GetTask(ctx, tasks[0].TaskID) if err != nil { t.Fatalf("get task %s: %v", tasks[0].TaskID, err) } if detail.Record.Source != task.SourceSchedule { t.Fatalf("expected schedule source, got %+v", detail.Record) } params := mustTaskParams(t, detail.Record.ParamsJSON) if len(params.DataNames) != 2 { t.Fatalf("expected two scheduled data names, got %+v", params) } if params.DataNames[0] != "inherit" || params.DataNames[1] != "override" { t.Fatalf("unexpected scheduled data names: %+v", params.DataNames) }}
func TestRunScheduledTasksPassCreatesRepoWideRusticMaintenanceTasks(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "rustic/composia-meta.yaml": "name: rustic\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n", }) if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) }
db := openControllerTestDB(t) defer func() { _ = db.Close() }() ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "rustic"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 9, 3, 15, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
cfg := &config.ControllerConfig{ RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Rustic: &config.ControllerRusticConfig{ MainNodes: []string{"main"}, Maintenance: &config.ControllerRusticMaintenanceConfig{ ForgetSchedule: "15 3 * * *", PruneSchedule: "15 3 * * *", }, }, } if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 3, 15, 30, 0, time.UTC)); err != nil { t.Fatalf("run scheduled pass: %v", err) } if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 3, 15, 50, 0, time.UTC)); err != nil { t.Fatalf("rerun scheduled pass: %v", err) }
forgetTasks, forgetCount, err := db.ListTasks(ctx, nil, []string{"rustic"}, nil, []string{string(task.TypeRusticForget)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list forget tasks: %v", err) } if forgetCount != 1 || len(forgetTasks) != 1 { t.Fatalf("expected 1 scheduled forget task, got total=%d len=%d", forgetCount, len(forgetTasks)) } pruneTasks, pruneCount, err := db.ListTasks(ctx, nil, []string{"rustic"}, nil, []string{string(task.TypeRusticPrune)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list prune tasks: %v", err) } if pruneCount != 1 || len(pruneTasks) != 1 { t.Fatalf("expected 1 scheduled prune task, got total=%d len=%d", pruneCount, len(pruneTasks)) } for _, taskID := range []string{forgetTasks[0].TaskID, pruneTasks[0].TaskID} { detail, err := db.GetTask(ctx, taskID) if err != nil { t.Fatalf("get maintenance task %s: %v", taskID, err) } if detail.Record.Source != task.SourceSchedule { t.Fatalf("expected schedule source, got %+v", detail.Record) } if detail.Record.NodeID != "main" { t.Fatalf("expected maintenance on main, got %+v", detail.Record) } if detail.Record.RepoRevision == "" { t.Fatalf("expected repo revision on maintenance task, got %+v", detail.Record) } if detail.Record.ParamsJSON != `{"service_dir":"rustic","repo_wide":true}` { t.Fatalf("expected repo-wide params, got %q", detail.Record.ParamsJSON) } }}
func TestRunScheduledTasksPassCreatesImageCheckForDueImages(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "app/composia-meta.yaml": "name: app\nnodes:\n - main\nupdate:\n check_schedule: none\n images:\n api:\n image: ghcr.io/example/api\n check_schedule: \"5 4 * * *\"\n current:\n env:\n file: .env\n key: API_VERSION\n discovery:\n sources:\n - type: auto\n filter:\n type: semver\n worker:\n image: ghcr.io/example/worker\n check_schedule: \"10 4 * * *\"\n current:\n env:\n file: .env\n key: WORKER_VERSION\n discovery:\n sources:\n - type: auto\n filter:\n type: semver\n", }) if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) }
db := openControllerTestDB(t) defer func() { _ = db.Close() }() ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "app"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 9, 4, 5, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
cfg := &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}} if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 4, 5, 20, 0, time.UTC)); err != nil { t.Fatalf("run scheduled pass: %v", err) }
tasks, totalCount, err := db.ListTasks(ctx, nil, []string{"app"}, nil, []string{string(task.TypeImageCheck)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list image check tasks: %v", err) } if totalCount != 1 || len(tasks) != 1 { t.Fatalf("expected 1 scheduled image check task, got total=%d len=%d", totalCount, len(tasks)) } detail, err := db.GetTask(ctx, tasks[0].TaskID) if err != nil { t.Fatalf("get image check task: %v", err) } params := mustTaskParams(t, detail.Record.ParamsJSON) if len(params.ImageNames) != 1 || params.ImageNames[0] != "api" { t.Fatalf("expected due image api, got %+v", params.ImageNames) }}
func TestRunScheduledTasksPassUsesControllerUpdatesDefaultCheckSchedule(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "app/composia-meta.yaml": "name: app\nnodes:\n - main\nupdate:\n images:\n api:\n image: ghcr.io/example/api\n current:\n env:\n file: .env\n key: API_VERSION\n discovery:\n sources:\n - type: auto\n filter:\n type: semver\n", }) if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) }
db := openControllerTestDB(t) defer func() { _ = db.Close() }() ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "app"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 9, 4, 5, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
cfg := &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Updates: &config.ControllerUpdatesConfig{DefaultCheckSchedule: "5 4 * * *"}} if err := runScheduledTasksPass(ctx, db, cfg, configuredNodeIDs(cfg), newTaskQueueNotifier(), time.Date(2026, 4, 9, 4, 5, 20, 0, time.UTC)); err != nil { t.Fatalf("run scheduled pass: %v", err) }
_, totalCount, err := db.ListTasks(ctx, nil, []string{"app"}, nil, []string{string(task.TypeImageCheck)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list image check tasks: %v", err) } if totalCount != 1 { t.Fatalf("expected 1 scheduled image check task, got %d", totalCount) }}