A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
26 kB · 641 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642package controller
import ( "context" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "testing" "time"
"connectrpc.com/connect" controllerv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/controller/v1" "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/controller/v1/controllerv1connect" "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" "forgejo.alexma.top/alexma233/composia/internal/platform/store")
func TestNodeQueryServiceListNodes(t *testing.T) { t.Parallel()
stateDir := filepath.Join(t.TempDir(), "state") if err := os.MkdirAll(stateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) }
db, err := store.Open(stateDir) if err != nil { t.Fatalf("open sqlite db: %v", err) } defer func() { _ = db.Close() }()
ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main", "node-2"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{ NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC), }); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil //nolint:goconst })
path, handler := controllerv1connect.NewNodeQueryServiceHandler( &nodeQueryServer{ db: db, cfg: &config.ControllerConfig{ Nodes: []config.NodeConfig{ {ID: "main", DisplayName: "Main", Enabled: boolPtr(true)}, {ID: "node-2"}, }, }, }, connect.WithInterceptors(interceptor), )
mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeQueryServiceClient( httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken)), )
response, err := client.ListNodes(context.Background(), connect.NewRequest(&controllerv1.ListNodesRequest{})) if err != nil { t.Fatalf("list nodes: %v", err) } if len(response.Msg.GetNodes()) != 2 { t.Fatalf("expected 2 nodes, got %d", len(response.Msg.GetNodes())) } if response.Msg.GetNodes()[0].GetNodeId() != "main" || !response.Msg.GetNodes()[0].GetIsOnline() { t.Fatalf("unexpected first node: %+v", response.Msg.GetNodes()[0]) } if response.Msg.GetNodes()[0].GetDisplayName() != "Main" { t.Fatalf("expected explicit display name Main, got %q", response.Msg.GetNodes()[0].GetDisplayName()) } if response.Msg.GetNodes()[1].GetDisplayName() != "node-2" { t.Fatalf("expected fallback display name node-2, got %q", response.Msg.GetNodes()[1].GetDisplayName()) } if response.Msg.GetNodes()[1].GetEnabled() != true { t.Fatalf("expected nil enabled to default true") } if response.Msg.GetNodes()[1].GetIsOnline() { t.Fatalf("expected node-2 to be offline") }}
func TestNodeQueryServiceGetNodeReturnsMinimalSummary(t *testing.T) { t.Parallel()
stateDir := filepath.Join(t.TempDir(), "state") if err := os.MkdirAll(stateDir, 0o750); err != nil { t.Fatalf("create state dir: %v", err) }
db, err := store.Open(stateDir) if err != nil { t.Fatalf("open sqlite db: %v", err) } 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, "caddy"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil })
path, handler := controllerv1connect.NewNodeQueryServiceHandler( &nodeQueryServer{db: db, cfg: &config.ControllerConfig{Nodes: []config.NodeConfig{{ID: "main", DisplayName: "Main"}}}}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeQueryServiceClient( httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken)), )
response, err := client.GetNode(ctx, connect.NewRequest(&controllerv1.GetNodeRequest{NodeId: "main"})) if err != nil { t.Fatalf("get node: %v", err) } if response.Msg.GetNode().GetNodeId() != "main" || response.Msg.GetNode().GetDisplayName() != "Main" { t.Fatalf("unexpected node response: %+v", response.Msg.GetNode()) } if !response.Msg.GetNode().GetIsOnline() { t.Fatalf("expected node to be online") }}
func TestNodeMaintenanceServiceReloadNodeCaddyCreatesTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "edge-proxy/composia-meta.yaml": "name: edge-proxy\nnodes:\n - main\ninfra:\n caddy:\n compose_service: caddy\n config_dir: /etc/caddy\n", }) if err := os.MkdirAll(logDir, 0o750); err != nil { t.Fatalf("create log dir: %v", err) } 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, "edge-proxy"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.ReloadNodeCaddy(ctx, connect.NewRequest(&controllerv1.ReloadNodeCaddyRequest{NodeId: "main"})) if err != nil { t.Fatalf("reload node caddy: %v", err) } if response.Msg.GetTaskId() == "" { t.Fatalf("expected task_id in response") } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeCaddyReload { t.Fatalf("expected caddy_reload task type, got %q", detail.Record.Type) } if detail.Record.ServiceName != "edge-proxy" || detail.Record.NodeID != "main" { t.Fatalf("unexpected created task record: %+v", detail.Record) } if detail.Record.TriggeredBy != "test-client" { //nolint:goconst t.Fatalf("expected triggered_by test-client, got %q", detail.Record.TriggeredBy) }}
func TestNodeMaintenanceServiceSyncNodeCaddyFilesCreatesSyncTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") 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", }) 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, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.SyncNodeCaddyFiles(ctx, connect.NewRequest(&controllerv1.SyncNodeCaddyFilesRequest{NodeId: "main", ServiceName: "demo"})) if err != nil { t.Fatalf("sync node caddy files: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeCaddySync { t.Fatalf("expected caddy_sync task type, got %q", detail.Record.Type) } params := mustTaskParams(t, detail.Record.ParamsJSON) if params.ServiceDir != "demo" || len(params.ServiceDirs) != 1 || params.ServiceDirs[0] != "demo" || params.FullRebuild { //nolint:goconst t.Fatalf("unexpected caddy sync params: %+v", params) } revision, err := repo.CurrentRevision(repoDir) if err != nil { t.Fatalf("read current revision: %v", err) } if detail.Record.RepoRevision != revision { t.Fatalf("unexpected caddy sync repo revision %q", detail.Record.RepoRevision) }}
func TestNodeMaintenanceServiceSyncNodeCaddyFilesFullRebuildCreatesMultiServiceTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") 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", "charlie/composia-meta.yaml": "name: charlie\nnodes:\n - main\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, "alpha", "bravo", "charlie"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.SyncNodeCaddyFiles(ctx, connect.NewRequest(&controllerv1.SyncNodeCaddyFilesRequest{NodeId: "main", FullRebuild: true})) if err != nil { t.Fatalf("sync node caddy files full rebuild: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeCaddySync { t.Fatalf("expected caddy_sync task type, got %q", detail.Record.Type) } params := mustTaskParams(t, detail.Record.ParamsJSON) if params.ServiceDir != "" || !params.FullRebuild || len(params.ServiceDirs) != 2 || params.ServiceDirs[0] != "alpha" || params.ServiceDirs[1] != "bravo" { t.Fatalf("unexpected full rebuild caddy sync params: %+v", params) } if detail.Record.ServiceName != "" { t.Fatalf("unexpected caddy sync service name %q", detail.Record.ServiceName) }}
func TestNodeMaintenanceServicePruneNodeDockerCreatesTaskLogAndNotifiesQueue(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") if err := os.MkdirAll(repoDir, 0o750); err != nil { t.Fatalf("create repo dir: %v", err) } 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 := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) notifier := newTaskQueueNotifier() waitCh := notifier.Subscribe() defer notifier.Unsubscribe(waitCh) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, taskQueue: notifier, taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.PruneNodeDocker(ctx, connect.NewRequest(&controllerv1.PruneNodeDockerRequest{NodeId: "main", Target: "images_all"})) if err != nil { t.Fatalf("prune node docker: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypePrune { t.Fatalf("expected prune task type, got %q", detail.Record.Type) } expectedLogPath := filepath.Join(logDir, "tasks", response.Msg.GetTaskId()+".log") if detail.Record.LogPath != expectedLogPath { t.Fatalf("expected prune task log path %q, got %q", expectedLogPath, detail.Record.LogPath) } if _, err := os.Stat(expectedLogPath); err != nil { t.Fatalf("expected prune task log file to exist: %v", err) } select { case <-waitCh: default: t.Fatalf("expected prune task creation to notify task queue") } if detail.Record.ParamsJSON != `{"target":"images_all"}` { t.Fatalf("unexpected prune params %q", detail.Record.ParamsJSON) }}
func TestNodeMaintenanceServicePruneNodeDockerRejectsMissingOrUnknownTarget(t *testing.T) { t.Parallel()
server := &nodeMaintenanceServer{} for _, target := range []string{"", " ", "image", "../../images"} { _, err := server.PruneNodeDocker(context.Background(), connect.NewRequest(&controllerv1.PruneNodeDockerRequest{NodeId: "main", Target: target})) if err == nil { t.Fatalf("expected prune target %q to fail", target) } if connect.CodeOf(err) != connect.CodeInvalidArgument { t.Fatalf("expected invalid argument for target %q, got %v", target, err) } }}
func TestNodeMaintenanceServicePruneNodeRusticCreatesRusticPruneTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "backup/composia-meta.yaml": "name: backup\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, "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Rustic: &config.ControllerRusticConfig{MainNodes: []string{"main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.PruneNodeRustic(ctx, connect.NewRequest(&controllerv1.PruneNodeRusticRequest{ServiceName: "demo", DataName: "db"})) if err != nil { t.Fatalf("prune node rustic: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeRusticPrune { t.Fatalf("expected rustic_prune task type, got %q", detail.Record.Type) } if detail.Record.ServiceName != "backup" || detail.Record.NodeID != "main" { t.Fatalf("unexpected created task record: %+v", detail.Record) } if detail.Record.RepoRevision == "" { t.Fatalf("expected repo revision on rustic prune task, got %+v", detail.Record) } if !strings.Contains(detail.Record.ParamsJSON, `"service_dir":"backup"`) || !strings.Contains(detail.Record.ParamsJSON, `"service_name":"demo"`) || !strings.Contains(detail.Record.ParamsJSON, `"data_name":"db"`) { t.Fatalf("unexpected rustic prune params %q", detail.Record.ParamsJSON) }}
func TestNodeMaintenanceServiceInitNodeRusticCreatesRusticInitTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "backup/composia-meta.yaml": "name: backup\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, "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Rustic: &config.ControllerRusticConfig{MainNodes: []string{"main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.InitNodeRustic(ctx, connect.NewRequest(&controllerv1.InitNodeRusticRequest{})) if err != nil { t.Fatalf("init node rustic: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeRusticInit { t.Fatalf("expected rustic_init task type, got %q", detail.Record.Type) } if detail.Record.ServiceName != "backup" || detail.Record.NodeID != "main" { t.Fatalf("unexpected created task record: %+v", detail.Record) } if detail.Record.RepoRevision == "" { t.Fatalf("expected repo revision on rustic init task, got %+v", detail.Record) } if !strings.Contains(detail.Record.ParamsJSON, `"service_dir":"backup"`) { t.Fatalf("unexpected rustic init params %q", detail.Record.ParamsJSON) }}
func TestNodeMaintenanceServiceForgetNodeRusticCreatesRusticForgetTask(t *testing.T) { t.Parallel()
rootDir := t.TempDir() repoDir := filepath.Join(rootDir, "repo") logDir := filepath.Join(rootDir, "logs") createGitRepoWithContent(t, repoDir, map[string]string{ "backup/composia-meta.yaml": "name: backup\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, "backup"); err != nil { t.Fatalf("sync declared services: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 12, 0, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != testAccessToken { return "", assertError("unexpected token") } return "test-client", nil }) path, handler := controllerv1connect.NewNodeMaintenanceServiceHandler( &nodeMaintenanceServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}, Rustic: &config.ControllerRusticConfig{MainNodes: []string{"main"}}}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor), ) mux := http.NewServeMux() mux.Handle(path, handler) httpServer := httptest.NewServer(mux) defer httpServer.Close()
client := controllerv1connect.NewNodeMaintenanceServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(testAccessToken))) response, err := client.ForgetNodeRustic(ctx, connect.NewRequest(&controllerv1.ForgetNodeRusticRequest{ServiceName: "demo", DataName: "db"})) if err != nil { t.Fatalf("forget node rustic: %v", err) } detail, err := db.GetTask(ctx, response.Msg.GetTaskId()) if err != nil { t.Fatalf("get created task: %v", err) } if detail.Record.Type != task.TypeRusticForget { t.Fatalf("expected rustic_forget task type, got %q", detail.Record.Type) } if detail.Record.ServiceName != "backup" || detail.Record.NodeID != "main" { t.Fatalf("unexpected created task record: %+v", detail.Record) } if detail.Record.RepoRevision == "" { t.Fatalf("expected repo revision on rustic forget task, got %+v", detail.Record) } if !strings.Contains(detail.Record.ParamsJSON, `"service_dir":"backup"`) || !strings.Contains(detail.Record.ParamsJSON, `"service_name":"demo"`) || !strings.Contains(detail.Record.ParamsJSON, `"data_name":"db"`) { t.Fatalf("unexpected rustic forget params %q", detail.Record.ParamsJSON) }}
func boolPtr(value bool) *bool { return &value}