A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
28 kB · 644 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645//nolint:goconstpackage controller
import ( "archive/tar" "bytes" "compress/gzip" "context" "encoding/json" "errors" "io" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "testing" "time"
"connectrpc.com/connect" agentv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1" "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1/agentv1connect" "forgejo.alexma.top/alexma233/composia/internal/core/config" "forgejo.alexma.top/alexma233/composia/internal/core/repo" "forgejo.alexma.top/alexma233/composia/internal/core/task" "forgejo.alexma.top/alexma233/composia/internal/platform/rpcutil" secretutil "forgejo.alexma.top/alexma233/composia/internal/platform/secret")
func TestBundleServiceStreamsTaskBundle(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithService(t, repoDir, "demo", "main") if err := os.WriteFile(filepath.Join(repoDir, "demo", "docker-compose.yaml"), []byte("services: {}\n"), 0o600); err != nil { t.Fatalf("write compose file: %v", err) } runGit(t, repoDir, "add", ".") runGit(t, repoDir, "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "add compose") revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal deploy task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-bundle", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) stream, err := client.GetServiceBundle(ctx, connect.NewRequest(&agentv1.GetServiceBundleRequest{TaskId: "task-bundle"})) if err != nil { t.Fatalf("get service bundle: %v", err) } defer func() { _ = stream.Close() }()
archive := bytes.Buffer{} var relativeRoot string for stream.Receive() { message := stream.Msg() if relativeRoot == "" { relativeRoot = message.GetRelativeRoot() } archive.Write(message.GetData()) } if err := stream.Err(); err != nil { t.Fatalf("receive service bundle: %v", err) } if relativeRoot != "demo" { t.Fatalf("expected relative root demo, got %q", relativeRoot) }
entries := untarGzEntries(t, archive.Bytes()) if !entries["demo/composia-meta.yaml"] || !entries["demo/docker-compose.yaml"] { t.Fatalf("missing expected bundle entries: %+v", entries) }
unauthorized, err := client.GetServiceBundle(ctx, connect.NewRequest(&agentv1.GetServiceBundleRequest{TaskId: "task-bundle", ServiceDir: "other"})) if err == nil { for unauthorized.Receive() { } err = unauthorized.Err() } if connect.CodeOf(err) != connect.CodePermissionDenied { t.Fatalf("expected unauthorized service_dir to be denied, got %v", err) }
if err := db.CompleteTask(ctx, "task-bundle", task.StatusSucceeded, time.Now().UTC(), ""); err != nil { t.Fatalf("complete task: %v", err) } notRunning, err := client.GetServiceBundle(ctx, connect.NewRequest(&agentv1.GetServiceBundleRequest{TaskId: "task-bundle"})) if err == nil { for notRunning.Receive() { } err = notRunning.Err() } if connect.CodeOf(err) != connect.CodeFailedPrecondition { t.Fatalf("expected completed task bundle to be rejected, got %v", err) }}
func TestServiceManifestForCaddySyncWithSingleServiceDir(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "demo/demo.caddy": "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo", ServiceDirs: []string{"demo"}}) if err != nil { t.Fatalf("marshal caddy sync task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-caddy-bundle", Type: task.TypeCaddySync, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create caddy sync task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) response, err := client.GetServiceManifest(ctx, connect.NewRequest(&agentv1.GetServiceManifestRequest{TaskId: "task-caddy-bundle"})) if err != nil { t.Fatal(err) } if response.Msg.GetRelativeRoot() != "demo" || len(response.Msg.GetFiles()) == 0 { t.Fatalf("unexpected manifest: %v", response.Msg) }}
func TestPersistentServiceFilesDecryptAllManagedSecrets(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") secretsCfg := writeAgeTestConfig(t, rootDir) caddyCiphertext, err := secretutil.Encrypt("demo.example.com { reverse_proxy 127.0.0.1:8080 }\n", secretsCfg) if err != nil { t.Fatalf("encrypt caddy source: %v", err) } secretCiphertext, err := secretutil.Encrypt("TOKEN=secret\n", secretsCfg) if err != nil { t.Fatalf("encrypt secret env: %v", err) } createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy.enc\n", "demo/demo.caddy.enc": string(caddyCiphertext), "demo/.secret.env.enc": string(secretCiphertext), }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %v", err) }
extraFiles, err := persistentServiceExtraFiles( &config.ControllerConfig{RepoDir: repoDir, Secrets: secretsCfg, Nodes: []config.NodeConfig{{ID: "main"}}}, revision, "demo", ) if err != nil { t.Fatalf("build caddy sync extra files: %v", err) } if extraFiles["demo/demo.caddy"] != "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n" { t.Fatalf("expected decrypted caddy source, got %+v", extraFiles) } if extraFiles["demo/.secret.env"] != "TOKEN=secret\n" { t.Fatal("persistent file verification must include every managed secret") }}
func TestPersistentServiceFilesRequireSecretsConfiguration(t *testing.T) { t.Parallel()
repoDir := filepath.Join(t.TempDir(), "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "demo/demo.caddy": "demo.example.com { reverse_proxy 127.0.0.1:8080 }\n", "demo/.secret.env.enc": "ciphertext\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %v", err) }
_, err = persistentServiceExtraFiles( &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}, revision, "demo", ) if !errors.Is(err, errSecretsNotConfigured) { t.Fatalf("expected missing secret configuration error: %v", err) }}
func TestServiceManifestForFullCaddySync(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "alpha/composia-meta.yaml": "name: alpha\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./alpha.caddy\n", "alpha/alpha.caddy": "alpha.example.com { reverse_proxy 127.0.0.1:8080 }\n", "bravo/composia-meta.yaml": "name: bravo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./bravo.caddy\n", "bravo/bravo.caddy": "bravo.example.com { reverse_proxy 127.0.0.1:9090 }\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "alpha", "bravo"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDirs: []string{"alpha", "bravo"}, FullRebuild: true}) if err != nil { t.Fatalf("marshal caddy sync task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-caddy-full-bundle", Type: task.TypeCaddySync, Source: task.SourceCLI, ServiceName: "", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create caddy sync task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) response, err := client.GetServiceManifest(ctx, connect.NewRequest(&agentv1.GetServiceManifestRequest{TaskId: "task-caddy-full-bundle", ServiceDir: "bravo"})) if err != nil { t.Fatal(err) } if response.Msg.GetRelativeRoot() != "bravo" || len(response.Msg.GetFiles()) != 2 { t.Fatalf("unexpected related manifest: %v", response.Msg) } if _, err := client.GetServiceManifest(ctx, connect.NewRequest(&agentv1.GetServiceManifestRequest{TaskId: "task-caddy-full-bundle", ServiceDir: "unrelated"})); connect.CodeOf(err) != connect.CodePermissionDenied { t.Fatalf("unauthorized service manifest: %v", err) }}
func TestBundleServiceInjectsDecryptedSecretEnv(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithService(t, repoDir, "demo", "main") secretsCfg := writeAgeTestConfig(t, rootDir) ciphertext, err := secretutil.Encrypt("TOKEN=secret\n", secretsCfg) if err != nil { t.Fatalf("encrypt secret env: %v", err) } if err := os.WriteFile(filepath.Join(repoDir, "demo", ".secret.env.enc"), ciphertext, 0o600); err != nil { t.Fatalf("write encrypted secret env: %v", err) } runGit(t, repoDir, "add", ".") runGit(t, repoDir, "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "add secret") revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal deploy task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-secret-bundle", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Secrets: secretsCfg}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) stream, err := client.GetServiceBundle(ctx, connect.NewRequest(&agentv1.GetServiceBundleRequest{TaskId: "task-secret-bundle"})) if err != nil { t.Fatalf("get service bundle: %v", err) } defer func() { _ = stream.Close() }()
archive := bytes.Buffer{} for stream.Receive() { archive.Write(stream.Msg().GetData()) } if err := stream.Err(); err != nil { t.Fatalf("receive service bundle: %v", err) } entries := untarGzContents(t, archive.Bytes()) if entries["demo/.secret.env"] != "TOKEN=secret\n" { t.Fatalf("expected decrypted secret env in bundle, got %q", entries["demo/.secret.env"]) } if _, ok := entries["demo/.secret.env.enc"]; ok { t.Fatalf("encrypted secret env must not remain in bundle entries") }}
func TestServiceTaskRuntimeForBackup(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\ndata_protect:\n data:\n - name: config\n backup:\n strategy: files.copy\n include:\n - ./config\nbackup:\n data:\n - name: config\n provider: rustic\n", "backup/composia-meta.yaml": "name: backup\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n profile: prod\n data_protect_dir: /data-protect\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "demo", "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo", DataNames: []string{"config"}}) if err != nil { t.Fatalf("marshal backup task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-backup-bundle", Type: task.TypeBackup, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create backup task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) response, err := client.GetServiceTaskRuntime(ctx, connect.NewRequest(&agentv1.GetServiceTaskRuntimeRequest{TaskId: "task-backup-bundle"})) if err != nil { t.Fatal(err) } payload := response.Msg.GetConfigJson() if payload == "" { t.Fatalf("expected backup runtime config in bundle") } if !strings.Contains(payload, `"service_name":"backup"`) || !strings.Contains(payload, `"compose_service":"rustic"`) || !strings.Contains(payload, `"profile":"prod"`) || !strings.Contains(payload, `"data_protect_dir":"/data-protect"`) || !strings.Contains(payload, `"node_id":"main"`) || !strings.Contains(payload, `"strategy":"files.copy"`) || !strings.Contains(payload, `"composia-service:demo"`) { t.Fatalf("unexpected backup runtime payload %q", payload) }}
func TestBuildBackupRuntimePayloadRejectsNodeWithoutRusticInfra(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - edge\ndata_protect:\n data:\n - name: config\n backup:\n strategy: files.copy\n include:\n - ./config\nbackup:\n data:\n - name: config\n", "backup/composia-meta.yaml": "name: backup\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %v", err) }
_, err = buildBackupRuntimePayload(&config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}, {ID: "edge"}}}, "demo", "edge", revision, serviceTaskParams{ServiceDir: "demo", DataNames: []string{"config"}}) if err == nil || !strings.Contains(err.Error(), "rustic infra service") { t.Fatalf("expected rustic target validation error, got %v", err) }}
func TestServiceManifestIncludesRelatedRusticSecrets(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithService(t, repoDir, "demo", "main") createGitRepoWithService(t, repoDir, "backup", "main") secretsCfg := writeAgeTestConfig(t, rootDir) ciphertext, err := secretutil.Encrypt("RUSTIC_PASSWORD=secret\n", secretsCfg) if err != nil { t.Fatalf("encrypt rustic secret: %v", err) } if err := os.WriteFile(filepath.Join(repoDir, "backup", ".secret.env.enc"), ciphertext, 0o600); err != nil { t.Fatalf("write encrypted rustic secret: %v", err) } if err := os.WriteFile(filepath.Join(repoDir, "demo", "composia-meta.yaml"), []byte("name: demo\nnodes:\n - main\ndata_protect:\n data:\n - name: config\n backup:\n strategy: files.copy\n include:\n - ./config\nbackup:\n data:\n - name: config\n provider: rustic\n"), 0o600); err != nil { t.Fatalf("write demo meta: %v", err) } if err := os.WriteFile(filepath.Join(repoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic\n"), 0o600); err != nil { t.Fatalf("write backup meta: %v", err) } runGit(t, repoDir, "add", ".") runGit(t, repoDir, "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "add rustic secret") revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %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, "demo", "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo", DataNames: []string{"config"}}) if err != nil { t.Fatalf("marshal backup task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-backup-override", Type: task.TypeBackup, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create backup task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Secrets: secretsCfg, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) response, err := client.GetServiceManifest(ctx, connect.NewRequest(&agentv1.GetServiceManifestRequest{TaskId: "task-backup-override", ServiceDir: "backup"})) if err != nil { t.Fatal(err) } found := false for _, file := range response.Msg.GetFiles() { if file.GetPath() == ".secret.env" { found = true } if file.GetPath() == ".composia-backup.json" { t.Fatal("manifest contains ephemeral parameters") } } if !found { t.Fatal("missing related secret hash") }}
func TestServiceTaskRuntimeUsesTaskRevision(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\ndata_protect:\n data:\n - name: config\n backup:\n strategy: files.copy\n include:\n - ./config-v1\nbackup:\n data:\n - name: config\n provider: rustic\n", "backup/composia-meta.yaml": "name: backup\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic-v1\n profile: prod-v1\n data_protect_dir: /data-protect-v1\n", }) revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %v", err) }
if err := os.WriteFile(filepath.Join(repoDir, "demo", "composia-meta.yaml"), []byte("name: demo\nnodes:\n - main\ndata_protect:\n data:\n - name: config\n backup:\n strategy: files.copy\n include:\n - ./config-v2\nbackup:\n data:\n - name: config\n provider: rustic\n"), 0o600); err != nil { t.Fatalf("update demo meta: %v", err) } if err := os.WriteFile(filepath.Join(repoDir, "backup", "composia-meta.yaml"), []byte("name: backup\nnodes:\n - main\ninfra:\n rustic:\n compose_service: rustic-v2\n profile: prod-v2\n data_protect_dir: /data-protect-v2\n"), 0o600); err != nil { t.Fatalf("update backup meta: %v", err) } runGit(t, repoDir, "add", ".") runGit(t, repoDir, "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "update runtime config")
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, "demo", "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo", DataNames: []string{"config"}}) if err != nil { t.Fatalf("marshal backup task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-backup-revision-bundle", Type: task.TypeBackup, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", Status: task.StatusRunning, RepoRevision: revision, ParamsJSON: string(paramsJSON), CreatedAt: time.Date(2026, 4, 4, 18, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("create backup task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() bundlePath, bundleHandler := agentv1connect.NewBundleServiceHandler(&bundleServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, Nodes: []config.NodeConfig{{ID: "main"}}}}, connect.WithInterceptors(interceptor)) mux.Handle(bundlePath, bundleHandler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := agentv1connect.NewBundleServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) response, err := client.GetServiceTaskRuntime(ctx, connect.NewRequest(&agentv1.GetServiceTaskRuntimeRequest{TaskId: "task-backup-revision-bundle"})) if err != nil { t.Fatal(err) } payload := response.Msg.GetConfigJson() if payload == "" { t.Fatalf("expected backup runtime config in bundle") } if !strings.Contains(payload, `"compose_service":"rustic-v1"`) || !strings.Contains(payload, `"profile":"prod-v1"`) || !strings.Contains(payload, `"data_protect_dir":"/data-protect-v1"`) || !strings.Contains(payload, `"include":["./config-v1"]`) { t.Fatalf("expected task revision runtime payload, got %q", payload) } if strings.Contains(payload, `rustic-v2`) || strings.Contains(payload, `prod-v2`) || strings.Contains(payload, `config-v2`) { t.Fatalf("runtime payload leaked live HEAD state: %q", payload) }}
func untarGzEntries(t *testing.T, content []byte) map[string]bool { t.Helper() entries := map[string]bool{} gzipReader, err := gzip.NewReader(bytes.NewReader(content)) if err != nil { t.Fatalf("open gzip content: %v", err) } defer func() { _ = gzipReader.Close() }()
tarReader := tar.NewReader(gzipReader) for { header, err := tarReader.Next() if err != nil { if err == io.EOF { return entries } t.Fatalf("read tar entry: %v", err) } entries[header.Name] = true }}
func untarGzContents(t *testing.T, content []byte) map[string]string { t.Helper() entries := map[string]string{} gzipReader, err := gzip.NewReader(bytes.NewReader(content)) if err != nil { t.Fatalf("open gzip content: %v", err) } defer func() { _ = gzipReader.Close() }()
tarReader := tar.NewReader(gzipReader) for { header, err := tarReader.Next() if err != nil { if errors.Is(err, io.EOF) { return entries } t.Fatalf("read tar entry: %v", err) } body, err := io.ReadAll(tarReader) if err != nil { t.Fatalf("read tar body: %v", err) } entries[header.Name] = string(body) }}