A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
50 kB · 1198 lines
Go
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199//go:build e2e
package controller_e2e
import ( "context" "fmt" "net/http" "os" "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/platform/rpcutil")
const ( defaultControllerAddr = "http://127.0.0.1:7001" defaultAccessToken = "dev-admin-token" testNodeID = "main" testServiceName = "host-service")
type controllerClients struct { system controllerv1connect.SystemServiceClient nodeQuery controllerv1connect.NodeQueryServiceClient nodeMaintenance controllerv1connect.NodeMaintenanceServiceClient dockerQuery controllerv1connect.DockerQueryServiceClient dockerCommand controllerv1connect.DockerCommandServiceClient repoQuery controllerv1connect.RepoQueryServiceClient repoCommand controllerv1connect.RepoCommandServiceClient serviceQuery controllerv1connect.ServiceQueryServiceClient serviceCommand controllerv1connect.ServiceCommandServiceClient serviceInst controllerv1connect.ServiceInstanceServiceClient task controllerv1connect.TaskServiceClient backupQuery controllerv1connect.BackupQueryServiceClient backupCommand controllerv1connect.BackupCommandServiceClient secret controllerv1connect.SecretServiceClient}
func TestControllerE2EAuth(t *testing.T) { ctx := testContext(t) baseURL := controllerAPIBaseURL() waitForController(t, newControllerClients()) client := controllerv1connect.NewSystemServiceClient(http.DefaultClient, baseURL)
_, err := client.GetSystemStatus(ctx, connect.NewRequest(&controllerv1.GetSystemStatusRequest{})) assertConnectCode(t, err, connect.CodeUnauthenticated)
badClient := controllerv1connect.NewSystemServiceClient( http.DefaultClient, baseURL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("not-the-token")), ) _, err = badClient.GetSystemStatus(ctx, connect.NewRequest(&controllerv1.GetSystemStatusRequest{})) assertConnectCode(t, err, connect.CodeUnauthenticated)}
func TestControllerE2ESystemAndNodes(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
status, err := clients.system.GetSystemStatus(ctx, connect.NewRequest(&controllerv1.GetSystemStatusRequest{})) if err != nil { t.Fatalf("get system status: %v", err) } if status.Msg.GetVersion() == "" { t.Fatalf("expected system status version") } if status.Msg.GetConfiguredNodeCount() != 1 { t.Fatalf("configured node count = %d, want 1", status.Msg.GetConfiguredNodeCount()) } if status.Msg.GetServiceCount() == 0 { t.Fatalf("expected at least one declared service") }
config, err := clients.system.GetCurrentConfig(ctx, connect.NewRequest(&controllerv1.GetCurrentConfigRequest{})) if err != nil { t.Fatalf("get current config: %v", err) } if config.Msg.GetListenAddr() == "" { t.Fatalf("expected redacted config listen address") } if !hasNodeConfig(config.Msg.GetNodes(), testNodeID) { t.Fatalf("expected config to include node %q", testNodeID) } if !hasAccessToken(config.Msg.GetAccessTokens(), "web") { t.Fatalf("expected config to include web access token metadata") }
capabilities, err := clients.system.GetCapabilities(ctx, connect.NewRequest(&controllerv1.GetCapabilitiesRequest{})) if err != nil { t.Fatalf("get capabilities: %v", err) } if capabilities.Msg.GetGlobal() == nil || capabilities.Msg.GetGlobal().GetBackup().GetEnabled() { t.Fatalf("expected backup capability to be present and disabled in e2e fixture") }
nodes, err := clients.nodeQuery.ListNodes(ctx, connect.NewRequest(&controllerv1.ListNodesRequest{})) if err != nil { t.Fatalf("list nodes: %v", err) } if !hasNodeSummary(nodes.Msg.GetNodes(), testNodeID) { t.Fatalf("expected node list to include %q", testNodeID) }
node, err := clients.nodeQuery.GetNode(ctx, connect.NewRequest(&controllerv1.GetNodeRequest{NodeId: testNodeID})) if err != nil { t.Fatalf("get node: %v", err) } if node.Msg.GetNode().GetDisplayName() != "Main" { t.Fatalf("node display name = %q, want Main", node.Msg.GetNode().GetDisplayName()) }
_, err = clients.nodeQuery.GetNode(ctx, connect.NewRequest(&controllerv1.GetNodeRequest{NodeId: "missing-node"})) assertConnectCode(t, err, connect.CodeNotFound)
reload, err := clients.system.ReloadControllerConfig(ctx, connect.NewRequest(&controllerv1.ReloadControllerConfigRequest{})) if err != nil { t.Fatalf("reload controller config: %v", err) } if !reload.Msg.GetAccepted() { t.Fatalf("expected controller reload to be accepted") } waitForController(t, clients)}
func TestControllerE2ERepoAndServices(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
head := repoHead(t, ctx, clients) if head.GetHeadRevision() == "" { t.Fatalf("expected repo head revision") } if !head.GetCleanWorktree() { t.Fatalf("expected e2e repo worktree to be clean") }
files, err := clients.repoQuery.ListRepoFiles(ctx, connect.NewRequest(&controllerv1.ListRepoFilesRequest{})) if err != nil { t.Fatalf("list repo files: %v", err) } if !hasRepoEntry(files.Msg.GetEntries(), "host-service", true) { t.Fatalf("expected repo root to include host-service directory") }
meta, err := clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: "host-service/composia-meta.yaml"})) if err != nil { t.Fatalf("get repo file: %v", err) } if !strings.Contains(meta.Msg.GetContent(), "name: host-service") { t.Fatalf("unexpected host-service metadata:\n%s", meta.Msg.GetContent()) }
validation, err := clients.repoQuery.ValidateRepo(ctx, connect.NewRequest(&controllerv1.ValidateRepoRequest{})) if err != nil { t.Fatalf("validate repo: %v", err) } if len(validation.Msg.GetErrors()) != 0 { t.Fatalf("expected valid fixture repo, got %d errors", len(validation.Msg.GetErrors())) }
commits, err := clients.repoQuery.ListRepoCommits(ctx, connect.NewRequest(&controllerv1.ListRepoCommitsRequest{PageSize: 5})) if err != nil { t.Fatalf("list repo commits: %v", err) } if len(commits.Msg.GetCommits()) == 0 { t.Fatalf("expected seeded repo commit history") }
_, err = clients.repoCommand.SyncRepo(ctx, connect.NewRequest(&controllerv1.SyncRepoRequest{})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: "missing/file.txt"})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: "../outside"})) assertConnectCode(t, err, connect.CodeInvalidArgument)
workspaces, err := clients.serviceQuery.ListServiceWorkspaces(ctx, connect.NewRequest(&controllerv1.ListServiceWorkspacesRequest{})) if err != nil { t.Fatalf("list service workspaces: %v", err) } if !hasWorkspace(workspaces.Msg.GetWorkspaces(), "host-service", testServiceName) { t.Fatalf("expected host-service workspace") }
workspace, err := clients.serviceQuery.GetServiceWorkspace(ctx, connect.NewRequest(&controllerv1.GetServiceWorkspaceRequest{Folder: "host-service"})) if err != nil { t.Fatalf("get service workspace: %v", err) } if workspace.Msg.GetWorkspace().GetServiceName() != testServiceName { t.Fatalf("workspace service name = %q, want %q", workspace.Msg.GetWorkspace().GetServiceName(), testServiceName) }
_, err = clients.serviceQuery.GetServiceWorkspace(ctx, connect.NewRequest(&controllerv1.GetServiceWorkspaceRequest{Folder: "missing-workspace"})) assertConnectCode(t, err, connect.CodeNotFound)
services, err := clients.serviceQuery.ListServices(ctx, connect.NewRequest(&controllerv1.ListServicesRequest{PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list services: %v", err) } if !hasServiceSummary(services.Msg.GetServices(), testServiceName) { t.Fatalf("expected service list to include %q", testServiceName) }
servicePage, err := clients.serviceQuery.ListServices(ctx, connect.NewRequest(&controllerv1.ListServicesRequest{PageSize: 1, Page: 1})) if err != nil { t.Fatalf("list paged services: %v", err) } if servicePage.Msg.GetTotalCount() == 0 || len(servicePage.Msg.GetServices()) > 1 { t.Fatalf("unexpected paged service response: total=%d len=%d", servicePage.Msg.GetTotalCount(), len(servicePage.Msg.GetServices())) }
missingStatusServices, err := clients.serviceQuery.ListServices(ctx, connect.NewRequest(&controllerv1.ListServicesRequest{RuntimeStatus: "controller-e2e-missing-status", PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list services by missing status: %v", err) } if missingStatusServices.Msg.GetTotalCount() != 0 || len(missingStatusServices.Msg.GetServices()) != 0 { t.Fatalf("expected missing status service filter to return no services") }
service, err := clients.serviceQuery.GetService(ctx, connect.NewRequest(&controllerv1.GetServiceRequest{ServiceName: testServiceName})) if err != nil { t.Fatalf("get service: %v", err) } if service.Msg.GetName() != testServiceName || !containsString(service.Msg.GetNodes(), testNodeID) { t.Fatalf("unexpected service detail: %+v", service.Msg) }
imageChecks, err := clients.serviceQuery.GetServiceImageUpdateChecks(ctx, connect.NewRequest(&controllerv1.GetServiceImageUpdateChecksRequest{ServiceName: testServiceName})) if err != nil { t.Fatalf("get service image update checks: %v", err) } if len(imageChecks.Msg.GetChecks()) != 0 { t.Fatalf("expected no image update checks in fixture, got %d", len(imageChecks.Msg.GetChecks())) }
instances, err := clients.serviceInst.ListServiceInstances(ctx, connect.NewRequest(&controllerv1.ListServiceInstancesRequest{ServiceName: testServiceName})) if err != nil { t.Fatalf("list service instances: %v", err) } if !hasServiceInstance(instances.Msg.GetInstances(), testServiceName, testNodeID) { t.Fatalf("expected %s instance on %s", testServiceName, testNodeID) }
instance, err := clients.serviceInst.GetServiceInstance(ctx, connect.NewRequest(&controllerv1.GetServiceInstanceRequest{ServiceName: testServiceName, NodeId: testNodeID})) if err != nil { t.Fatalf("get service instance: %v", err) } if instance.Msg.GetInstance().GetServiceName() != testServiceName || instance.Msg.GetInstance().GetNodeId() != testNodeID { t.Fatalf("unexpected service instance: %+v", instance.Msg.GetInstance()) }
_, err = clients.serviceInst.GetServiceInstance(ctx, connect.NewRequest(&controllerv1.GetServiceInstanceRequest{ServiceName: testServiceName, NodeId: "missing-node"})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.serviceQuery.GetService(ctx, connect.NewRequest(&controllerv1.GetServiceRequest{ServiceName: "missing-service"})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.serviceCommand.MigrateService(ctx, connect.NewRequest(&controllerv1.MigrateServiceRequest{ServiceName: testServiceName, SourceNodeId: testNodeID, TargetNodeId: testNodeID})) assertConnectCode(t, err, connect.CodeInvalidArgument)}
func TestControllerE2ERepoWrites(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
baseRevision := repoHead(t, ctx, clients).GetHeadRevision() workspace := fmt.Sprintf("controller-e2e-%d", time.Now().UnixNano())
created, err := clients.repoCommand.CreateRepoDirectory(ctx, connect.NewRequest(&controllerv1.CreateRepoDirectoryRequest{ Path: workspace, BaseRevision: baseRevision, CommitMessage: "Create controller e2e workspace", })) if err != nil { t.Fatalf("create repo directory: %v", err) } if created.Msg.GetCommitId() == "" { t.Fatalf("expected create directory commit id") }
_, err = clients.repoCommand.UpdateRepoFile(ctx, connect.NewRequest(&controllerv1.UpdateRepoFileRequest{ Path: workspace + "/README.md", Content: "# Controller E2E\n", BaseRevision: baseRevision, CommitMessage: "Stale controller e2e write", })) assertConnectCode(t, err, connect.CodeFailedPrecondition)
baseRevision = created.Msg.GetCommitId() updated, err := clients.repoCommand.UpdateRepoFile(ctx, connect.NewRequest(&controllerv1.UpdateRepoFileRequest{ Path: workspace + "/README.md", Content: "# Controller E2E\n", BaseRevision: baseRevision, CommitMessage: "Write controller e2e file", })) if err != nil { t.Fatalf("update repo file: %v", err) }
file, err := clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: workspace + "/README.md"})) if err != nil { t.Fatalf("get written repo file: %v", err) } if file.Msg.GetContent() != "# Controller E2E\n" { t.Fatalf("written file content = %q", file.Msg.GetContent()) }
moved, err := clients.repoCommand.MoveRepoPath(ctx, connect.NewRequest(&controllerv1.MoveRepoPathRequest{ SourcePath: workspace + "/README.md", DestinationPath: workspace + "/NOTES.md", BaseRevision: updated.Msg.GetCommitId(), CommitMessage: "Move controller e2e file", })) if err != nil { t.Fatalf("move repo path: %v", err) }
_, err = clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: workspace + "/NOTES.md"})) if err != nil { t.Fatalf("get moved repo file: %v", err) }
deleted, err := clients.repoCommand.DeleteRepoPath(ctx, connect.NewRequest(&controllerv1.DeleteRepoPathRequest{ Path: workspace, BaseRevision: moved.Msg.GetCommitId(), CommitMessage: "Delete controller e2e workspace", })) if err != nil { t.Fatalf("delete repo path: %v", err) } if deleted.Msg.GetCommitId() == "" { t.Fatalf("expected delete commit id") }
_, err = clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: workspace + "/NOTES.md"})) assertConnectCode(t, err, connect.CodeNotFound)}
func TestControllerE2ERepoContractsAndValidation(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
recursive, err := clients.repoQuery.ListRepoFiles(ctx, connect.NewRequest(&controllerv1.ListRepoFilesRequest{Recursive: true})) if err != nil { t.Fatalf("list recursive repo files: %v", err) } if !hasRepoEntryPath(recursive.Msg.GetEntries(), "host-service/composia-meta.yaml", false) { t.Fatalf("expected recursive repo listing to include host-service/composia-meta.yaml") }
_, err = clients.repoQuery.ListRepoFiles(ctx, connect.NewRequest(&controllerv1.ListRepoFilesRequest{Path: "host-service/composia-meta.yaml"})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{})) assertConnectCode(t, err, connect.CodeInvalidArgument)
baseRevision := repoHead(t, ctx, clients).GetHeadRevision() _, err = clients.repoCommand.CreateRepoDirectory(ctx, connect.NewRequest(&controllerv1.CreateRepoDirectoryRequest{Path: "host-service", BaseRevision: baseRevision, CommitMessage: "Create existing controller e2e directory"})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.repoCommand.UpdateRepoFile(ctx, connect.NewRequest(&controllerv1.UpdateRepoFileRequest{Path: "host-service", Content: "not a file", BaseRevision: baseRevision, CommitMessage: "Write directory as file"})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.repoCommand.MoveRepoPath(ctx, connect.NewRequest(&controllerv1.MoveRepoPathRequest{SourcePath: "missing-source", DestinationPath: "missing-destination", BaseRevision: baseRevision, CommitMessage: "Move missing controller e2e path"})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.repoCommand.DeleteRepoPath(ctx, connect.NewRequest(&controllerv1.DeleteRepoPathRequest{Path: "missing-delete", BaseRevision: baseRevision, CommitMessage: "Delete missing controller e2e path"})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
workspace := fmt.Sprintf("controller-e2e-invalid-%d", time.Now().UnixNano()) cleanupWorkspace := true defer func() { if cleanupWorkspace { cleanupRepoPath(t, ctx, clients, workspace) } }()
created, err := clients.repoCommand.CreateRepoDirectory(ctx, connect.NewRequest(&controllerv1.CreateRepoDirectoryRequest{ Path: workspace, BaseRevision: baseRevision, CommitMessage: "Create invalid controller e2e workspace", })) if err != nil { t.Fatalf("create invalid repo workspace: %v", err) }
invalidMetaPath := workspace + "/composia-meta.yaml" updated, err := clients.repoCommand.UpdateRepoFile(ctx, connect.NewRequest(&controllerv1.UpdateRepoFileRequest{ Path: invalidMetaPath, Content: "name: controller-e2e-invalid\nnodes:\n - missing-node\n", BaseRevision: created.Msg.GetCommitId(), CommitMessage: "Write invalid controller e2e metadata", })) if err != nil { t.Fatalf("write invalid repo metadata: %v", err) }
validation, err := clients.repoQuery.ValidateRepo(ctx, connect.NewRequest(&controllerv1.ValidateRepoRequest{})) if err != nil { t.Fatalf("validate invalid repo: %v", err) } if !hasValidationErrorPath(validation.Msg.GetErrors(), invalidMetaPath) { t.Fatalf("expected validation error for %s, got %+v", invalidMetaPath, validation.Msg.GetErrors()) }
deleted, err := clients.repoCommand.DeleteRepoPath(ctx, connect.NewRequest(&controllerv1.DeleteRepoPathRequest{ Path: workspace, BaseRevision: updated.Msg.GetCommitId(), CommitMessage: "Delete invalid controller e2e workspace", })) if err != nil { t.Fatalf("delete invalid repo workspace: %v", err) } if deleted.Msg.GetCommitId() == "" { t.Fatalf("expected invalid workspace delete commit id") } cleanupWorkspace = false}
func TestControllerE2ESecrets(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
const secretFilePath = ".secret.env.enc" const initialContent = "TOKEN=controller-e2e-before\n" const updatedContent = "TOKEN=controller-e2e-after\n"
secret, err := clients.secret.GetSecret(ctx, connect.NewRequest(&controllerv1.GetSecretRequest{ServiceName: testServiceName, FilePath: secretFilePath})) if err != nil { t.Fatalf("get configured secret: %v", err) } if secret.Msg.GetServiceName() != testServiceName || secret.Msg.GetFilePath() != secretFilePath || secret.Msg.GetContent() != initialContent { t.Fatalf("unexpected secret response: %+v", secret.Msg) }
_, err = clients.secret.GetSecret(ctx, connect.NewRequest(&controllerv1.GetSecretRequest{ServiceName: testServiceName, FilePath: "../outside.env.enc"})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.secret.UpdateSecret(ctx, connect.NewRequest(&controllerv1.UpdateSecretRequest{ServiceName: testServiceName, FilePath: secretFilePath, Content: updatedContent})) assertConnectCode(t, err, connect.CodeInvalidArgument)
head := repoHead(t, ctx, clients) _, err = clients.secret.UpdateSecret(ctx, connect.NewRequest(&controllerv1.UpdateSecretRequest{ServiceName: testServiceName, FilePath: "/outside.env.enc", Content: updatedContent, BaseRevision: head.GetHeadRevision()})) assertConnectCode(t, err, connect.CodeInvalidArgument)
workspace := fmt.Sprintf("controller-e2e-secret-base-%d", time.Now().UnixNano()) created, err := clients.repoCommand.CreateRepoDirectory(ctx, connect.NewRequest(&controllerv1.CreateRepoDirectoryRequest{ Path: workspace, BaseRevision: head.GetHeadRevision(), CommitMessage: "Create secret base revision marker", })) if err != nil { t.Fatalf("create secret base revision marker: %v", err) } defer cleanupRepoPath(t, ctx, clients, workspace)
_, err = clients.secret.UpdateSecret(ctx, connect.NewRequest(&controllerv1.UpdateSecretRequest{ ServiceName: testServiceName, FilePath: secretFilePath, Content: updatedContent, BaseRevision: head.GetHeadRevision(), CommitMessage: "Stale controller e2e secret update", })) assertConnectCode(t, err, connect.CodeFailedPrecondition)
updated, err := clients.secret.UpdateSecret(ctx, connect.NewRequest(&controllerv1.UpdateSecretRequest{ ServiceName: testServiceName, FilePath: secretFilePath, Content: updatedContent, BaseRevision: created.Msg.GetCommitId(), CommitMessage: "Update controller e2e secret", })) if err != nil { t.Fatalf("update configured secret: %v", err) } if updated.Msg.GetCommitId() == "" { t.Fatalf("expected secret update commit id") } defer func() { head := repoHead(t, ctx, clients) _, err := clients.secret.UpdateSecret(ctx, connect.NewRequest(&controllerv1.UpdateSecretRequest{ ServiceName: testServiceName, FilePath: secretFilePath, Content: initialContent, BaseRevision: head.GetHeadRevision(), CommitMessage: "Restore controller e2e secret", })) if err != nil { t.Logf("restore controller e2e secret: %v", err) } }()
secret, err = clients.secret.GetSecret(ctx, connect.NewRequest(&controllerv1.GetSecretRequest{ServiceName: testServiceName, FilePath: secretFilePath})) if err != nil { t.Fatalf("get updated secret: %v", err) } if secret.Msg.GetContent() != updatedContent { t.Fatalf("updated secret content = %q, want %q", secret.Msg.GetContent(), updatedContent) }
file, err := clients.repoQuery.GetRepoFile(ctx, connect.NewRequest(&controllerv1.GetRepoFileRequest{Path: "host-service/" + secretFilePath})) if err != nil { t.Fatalf("get encrypted secret file: %v", err) } if strings.Contains(file.Msg.GetContent(), updatedContent) { t.Fatalf("expected encrypted secret file not to contain plaintext") }}
func TestControllerE2EFeatureContracts(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
backups, err := clients.backupQuery.ListBackups(ctx, connect.NewRequest(&controllerv1.ListBackupsRequest{PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list backups: %v", err) } if backups.Msg.GetTotalCount() < uint32(len(backups.Msg.GetBackups())) { t.Fatalf("backup total_count = %d, returned %d", backups.Msg.GetTotalCount(), len(backups.Msg.GetBackups())) }
_, err = clients.backupQuery.GetBackup(ctx, connect.NewRequest(&controllerv1.GetBackupRequest{BackupId: "missing-backup"})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.backupCommand.RestoreBackup(ctx, connect.NewRequest(&controllerv1.RestoreBackupRequest{BackupId: "missing-backup", NodeId: testNodeID})) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.task.GetTask(ctx, connect.NewRequest(&controllerv1.GetTaskRequest{})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.task.GetTask(ctx, connect.NewRequest(&controllerv1.GetTaskRequest{TaskId: "missing-task"})) assertConnectCode(t, err, connect.CodeNotFound)
taskLogStream, err := clients.task.TailTaskLogs(ctx, connect.NewRequest(&controllerv1.TailTaskLogsRequest{TaskId: "missing-task"})) assertStreamConnectCode(t, taskLogStream, err, connect.CodeNotFound)
_, err = clients.task.RunTaskAgain(ctx, sourceRequest(&controllerv1.RunTaskAgainRequest{}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.task.RunTaskAgain(ctx, sourceRequest(&controllerv1.RunTaskAgainRequest{TaskId: "missing-task"}, "web")) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.task.ResolveTaskConfirmation(ctx, sourceRequest(&controllerv1.ResolveTaskConfirmationRequest{TaskId: "missing-task"}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.task.ResolveTaskConfirmation(ctx, sourceRequest(&controllerv1.ResolveTaskConfirmationRequest{TaskId: "missing-task", Decision: controllerv1.TaskConfirmationDecision_TASK_CONFIRMATION_DECISION_APPROVE}, "web")) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.serviceCommand.RunServiceAction(ctx, sourceRequest(&controllerv1.RunServiceActionRequest{}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.serviceCommand.RunServiceAction(ctx, sourceRequest(&controllerv1.RunServiceActionRequest{ServiceName: "missing-service", Action: controllerv1.ServiceAction_SERVICE_ACTION_DEPLOY}, "web")) assertConnectCode(t, err, connect.CodeNotFound)
_, err = clients.serviceCommand.RunServiceAction(ctx, sourceRequest(&controllerv1.RunServiceActionRequest{ServiceName: testServiceName}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.serviceCommand.RunServiceAction(ctx, sourceRequest(&controllerv1.RunServiceActionRequest{ServiceName: testServiceName, Action: controllerv1.ServiceAction_SERVICE_ACTION_DNS_UPDATE}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.serviceCommand.MigrateService(ctx, sourceRequest(&controllerv1.MigrateServiceRequest{ServiceName: "missing-service", SourceNodeId: testNodeID, TargetNodeId: "other-node"}, "web")) assertConnectCode(t, err, connect.CodeNotFound)
stats, err := clients.nodeQuery.GetNodeDockerStats(ctx, connect.NewRequest(&controllerv1.GetNodeDockerStatsRequest{NodeId: testNodeID})) if err != nil { t.Fatalf("get node docker stats: %v", err) } if stats.Msg.GetStats() == nil { t.Fatalf("expected docker stats message") }
_, err = clients.nodeMaintenance.ReloadNodeCaddy(ctx, sourceRequest(&controllerv1.ReloadNodeCaddyRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.nodeMaintenance.SyncNodeCaddyFiles(ctx, sourceRequest(&controllerv1.SyncNodeCaddyFilesRequest{NodeId: testNodeID, ServiceName: testServiceName}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.nodeMaintenance.InitNodeRustic(ctx, sourceRequest(&controllerv1.InitNodeRusticRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.nodeMaintenance.ForgetNodeRustic(ctx, sourceRequest(&controllerv1.ForgetNodeRusticRequest{NodeId: testNodeID, ServiceName: testServiceName, DataName: "data"}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.nodeMaintenance.PruneNodeRustic(ctx, sourceRequest(&controllerv1.PruneNodeRusticRequest{NodeId: testNodeID, ServiceName: testServiceName, DataName: "data"}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.nodeMaintenance.PruneNodeDocker(ctx, sourceRequest(&controllerv1.PruneNodeDockerRequest{}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.nodeQuery.GetNodeDockerStats(ctx, connect.NewRequest(&controllerv1.GetNodeDockerStatsRequest{})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerQuery.ListNodeContainers(ctx, connect.NewRequest(&controllerv1.ListNodeContainersRequest{})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerQuery.ListNodeContainers(ctx, connect.NewRequest(&controllerv1.ListNodeContainersRequest{NodeId: "missing-node"})) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.dockerQuery.InspectNodeContainer(ctx, connect.NewRequest(&controllerv1.InspectNodeContainerRequest{NodeId: testNodeID})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RunContainerAction(ctx, sourceRequest(&controllerv1.RunContainerActionRequest{NodeId: testNodeID, ContainerId: "container"}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RemoveContainer(ctx, sourceRequest(&controllerv1.RemoveContainerRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
logStream, err := clients.dockerCommand.GetContainerLogs(ctx, connect.NewRequest(&controllerv1.GetContainerLogsRequest{NodeId: testNodeID})) assertStreamConnectCode(t, logStream, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.OpenContainerExec(ctx, connect.NewRequest(&controllerv1.OpenContainerExecRequest{NodeId: testNodeID, ContainerId: "container"})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RunContainerExec(ctx, connect.NewRequest(&controllerv1.RunContainerExecRequest{NodeId: testNodeID, ContainerId: "container"})) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RemoveNetwork(ctx, sourceRequest(&controllerv1.RemoveNetworkRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RemoveVolume(ctx, sourceRequest(&controllerv1.RemoveVolumeRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)
_, err = clients.dockerCommand.RemoveImage(ctx, sourceRequest(&controllerv1.RemoveImageRequest{NodeId: testNodeID}, "web")) assertConnectCode(t, err, connect.CodeInvalidArgument)}
func TestControllerE2EDockerQueriesAndExec(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients) waitForNodeOnline(t, ctx, clients, testNodeID)
containers := listContainers(t, ctx, clients) container := chooseExecContainer(t, containers)
pagedContainers, err := clients.dockerQuery.ListNodeContainers(ctx, connect.NewRequest(&controllerv1.ListNodeContainersRequest{NodeId: testNodeID, PageSize: 1, Page: 1})) if err != nil { t.Fatalf("list paged node containers: %v", err) } if pagedContainers.Msg.GetTotalCount() == 0 || len(pagedContainers.Msg.GetContainers()) != 1 { t.Fatalf("expected one paged container with non-zero total, got total=%d len=%d", pagedContainers.Msg.GetTotalCount(), len(pagedContainers.Msg.GetContainers())) }
searchedContainers, err := clients.dockerQuery.ListNodeContainers(ctx, connect.NewRequest(&controllerv1.ListNodeContainersRequest{NodeId: testNodeID, PageSize: 10, Page: 1, Search: container.GetName()})) if err != nil { t.Fatalf("search node containers: %v", err) } if !hasContainerInfo(searchedContainers.Msg.GetContainers(), container.GetId()) { t.Fatalf("expected container search for %q to include %q", container.GetName(), container.GetId()) }
containerInspect, err := clients.dockerQuery.InspectNodeContainer(ctx, connect.NewRequest(&controllerv1.InspectNodeContainerRequest{NodeId: testNodeID, ContainerId: container.GetId()})) if err != nil { t.Fatalf("inspect node container: %v", err) } if !strings.Contains(containerInspect.Msg.GetRawJson(), shortID(container.GetId())) && !strings.Contains(containerInspect.Msg.GetRawJson(), container.GetName()) { t.Fatalf("unexpected container inspect payload") }
networks, err := clients.dockerQuery.ListNodeNetworks(ctx, connect.NewRequest(&controllerv1.ListNodeNetworksRequest{NodeId: testNodeID, PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list node networks: %v", err) } if networks.Msg.GetTotalCount() == 0 || len(networks.Msg.GetNetworks()) == 0 { t.Fatalf("expected at least one docker network") } networkInspect, err := clients.dockerQuery.InspectNodeNetwork(ctx, connect.NewRequest(&controllerv1.InspectNodeNetworkRequest{NodeId: testNodeID, NetworkId: networks.Msg.GetNetworks()[0].GetId()})) if err != nil { t.Fatalf("inspect node network: %v", err) } if strings.TrimSpace(networkInspect.Msg.GetRawJson()) == "" { t.Fatalf("expected network inspect JSON") }
volumes, err := clients.dockerQuery.ListNodeVolumes(ctx, connect.NewRequest(&controllerv1.ListNodeVolumesRequest{NodeId: testNodeID, PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list node volumes: %v", err) } if len(volumes.Msg.GetVolumes()) > 0 { volumeInspect, err := clients.dockerQuery.InspectNodeVolume(ctx, connect.NewRequest(&controllerv1.InspectNodeVolumeRequest{NodeId: testNodeID, VolumeName: volumes.Msg.GetVolumes()[0].GetName()})) if err != nil { t.Fatalf("inspect node volume: %v", err) } if strings.TrimSpace(volumeInspect.Msg.GetRawJson()) == "" { t.Fatalf("expected volume inspect JSON") } }
images, err := clients.dockerQuery.ListNodeImages(ctx, connect.NewRequest(&controllerv1.ListNodeImagesRequest{NodeId: testNodeID, PageSize: 10, Page: 1})) if err != nil { t.Fatalf("list node images: %v", err) } if images.Msg.GetTotalCount() == 0 || len(images.Msg.GetImages()) == 0 { t.Fatalf("expected at least one docker image") } imageInspect, err := clients.dockerQuery.InspectNodeImage(ctx, connect.NewRequest(&controllerv1.InspectNodeImageRequest{NodeId: testNodeID, ImageId: images.Msg.GetImages()[0].GetId()})) if err != nil { t.Fatalf("inspect node image: %v", err) } if strings.TrimSpace(imageInspect.Msg.GetRawJson()) == "" { t.Fatalf("expected image inspect JSON") }
execResult, err := clients.dockerCommand.RunContainerExec(ctx, connect.NewRequest(&controllerv1.RunContainerExecRequest{ NodeId: testNodeID, ContainerId: container.GetId(), Command: []string{"/bin/sh", "-c", "printf composia-controller-e2e"}, TimeoutSeconds: 10, MaxOutputBytes: 1024, })) if err != nil { t.Fatalf("run container exec: %v", err) } if execResult.Msg.GetExitCode() != 0 || execResult.Msg.GetStdout() != "composia-controller-e2e" { t.Fatalf("unexpected exec result: exit=%d stdout=%q stderr=%q", execResult.Msg.GetExitCode(), execResult.Msg.GetStdout(), execResult.Msg.GetStderr()) }
stdinResult, err := clients.dockerCommand.RunContainerExec(ctx, connect.NewRequest(&controllerv1.RunContainerExecRequest{ NodeId: testNodeID, ContainerId: container.GetId(), Command: []string{"/bin/sh", "-c", "cat"}, Stdin: []byte("stdin-through-exec"), TimeoutSeconds: 10, MaxOutputBytes: 1024, })) if err != nil { t.Fatalf("run container exec with stdin: %v", err) } if stdinResult.Msg.GetExitCode() != 0 || stdinResult.Msg.GetStdout() != "stdin-through-exec" { t.Fatalf("unexpected stdin exec result: exit=%d stdout=%q stderr=%q", stdinResult.Msg.GetExitCode(), stdinResult.Msg.GetStdout(), stdinResult.Msg.GetStderr()) }
truncatedResult, err := clients.dockerCommand.RunContainerExec(ctx, connect.NewRequest(&controllerv1.RunContainerExecRequest{ NodeId: testNodeID, ContainerId: container.GetId(), Command: []string{"/bin/sh", "-c", "printf 1234567890"}, TimeoutSeconds: 10, MaxOutputBytes: 4, })) if err != nil { t.Fatalf("run truncated container exec: %v", err) } if truncatedResult.Msg.GetExitCode() != 0 || truncatedResult.Msg.GetStdout() != "1234" || !truncatedResult.Msg.GetStdoutTruncated() { t.Fatalf("unexpected truncated exec result: exit=%d stdout=%q truncated=%t", truncatedResult.Msg.GetExitCode(), truncatedResult.Msg.GetStdout(), truncatedResult.Msg.GetStdoutTruncated()) }
containerLogChunk := firstContainerLogChunk(t, ctx, clients, container.GetId()) if strings.TrimSpace(containerLogChunk) == "" { t.Fatalf("expected non-empty container log chunk") }}
func TestControllerE2ETasks(t *testing.T) { ctx := testContext(t) clients := newControllerClients() waitForController(t, clients)
restartReq := sourceRequest(&controllerv1.RunServiceActionRequest{ ServiceName: testServiceName, Action: controllerv1.ServiceAction_SERVICE_ACTION_RESTART, }, "web") _, err := clients.serviceCommand.RunServiceAction(ctx, restartReq) assertConnectCode(t, err, connect.CodeFailedPrecondition)
deployReq := sourceRequest(&controllerv1.RunServiceActionRequest{ ServiceName: testServiceName, Action: controllerv1.ServiceAction_SERVICE_ACTION_DEPLOY, }, "web") deploy, err := clients.serviceCommand.RunServiceAction(ctx, deployReq) if err != nil { t.Fatalf("run deploy action: %v", err) } if len(deploy.Msg.GetTasks()) != 1 || deploy.Msg.GetTasks()[0].GetTaskId() == "" { t.Fatalf("expected one deploy task, got %+v", deploy.Msg.GetTasks()) }
taskID := deploy.Msg.GetTasks()[0].GetTaskId() taskDetail := waitForTaskStatus(t, ctx, clients, taskID, controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED) if taskDetail.GetType() != controllerv1.TaskType_TASK_TYPE_DEPLOY { t.Fatalf("task type = %s, want DEPLOY", taskDetail.GetType()) } if taskDetail.GetSource() != controllerv1.TaskSource_TASK_SOURCE_WEB { t.Fatalf("task source = %s, want WEB", taskDetail.GetSource()) } if taskDetail.GetServiceName() != testServiceName || taskDetail.GetNodeId() != testNodeID { t.Fatalf("unexpected task target: service=%q node=%q", taskDetail.GetServiceName(), taskDetail.GetNodeId()) } if len(taskDetail.GetSteps()) == 0 { t.Fatalf("expected deploy task steps") }
logChunk := firstTaskLogChunk(t, ctx, clients, taskID) if strings.TrimSpace(logChunk) == "" { t.Fatalf("expected non-empty task log chunk") }
_, err = clients.task.ResolveTaskConfirmation(ctx, sourceRequest(&controllerv1.ResolveTaskConfirmationRequest{TaskId: taskID, Decision: controllerv1.TaskConfirmationDecision_TASK_CONFIRMATION_DECISION_APPROVE}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
_, err = clients.task.CreateMigrationRollback(ctx, sourceRequest(&controllerv1.CreateMigrationRollbackRequest{TaskId: taskID, DeploySource: true}, "web")) assertConnectCode(t, err, connect.CodeFailedPrecondition)
filtered, err := clients.task.ListTasks(ctx, connect.NewRequest(&controllerv1.ListTasksRequest{ Status: []controllerv1.TaskStatus{controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED}, ServiceName: []string{testServiceName}, Type: []controllerv1.TaskType{controllerv1.TaskType_TASK_TYPE_DEPLOY}, PageSize: 10, Page: 1, })) if err != nil { t.Fatalf("list filtered tasks: %v", err) } if !hasTaskSummary(filtered.Msg.GetTasks(), taskID) { t.Fatalf("expected filtered tasks to include %q", taskID) }
excluded, err := clients.task.ListTasks(ctx, connect.NewRequest(&controllerv1.ListTasksRequest{ ExcludeStatus: []controllerv1.TaskStatus{controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED}, ExcludeServiceName: []string{"not-" + testServiceName}, ExcludeType: []controllerv1.TaskType{controllerv1.TaskType_TASK_TYPE_RESTART}, PageSize: 20, Page: 1, })) if err != nil { t.Fatalf("list excluded tasks: %v", err) } if hasTaskSummary(excluded.Msg.GetTasks(), taskID) { t.Fatalf("expected exclude_status to remove succeeded task %q", taskID) }
rerun, err := clients.task.RunTaskAgain(ctx, sourceRequest(&controllerv1.RunTaskAgainRequest{TaskId: taskID}, "cli")) if err != nil { t.Fatalf("run task again: %v", err) } if rerun.Msg.GetTaskId() == "" || rerun.Msg.GetTaskId() == taskID { t.Fatalf("unexpected rerun task id %q", rerun.Msg.GetTaskId()) }
rerunDetail := waitForTaskStatus(t, ctx, clients, rerun.Msg.GetTaskId(), controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED) if rerunDetail.GetAttemptOfTaskId() != taskID { t.Fatalf("rerun attempt_of_task_id = %q, want %q", rerunDetail.GetAttemptOfTaskId(), taskID) } if rerunDetail.GetSource() != controllerv1.TaskSource_TASK_SOURCE_CLI { t.Fatalf("rerun source = %s, want CLI", rerunDetail.GetSource()) }}
func newControllerClients() controllerClients { baseURL := controllerAPIBaseURL() opts := []connect.ClientOption{connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor(accessToken()))} return controllerClients{ system: controllerv1connect.NewSystemServiceClient(http.DefaultClient, baseURL, opts...), nodeQuery: controllerv1connect.NewNodeQueryServiceClient(http.DefaultClient, baseURL, opts...), nodeMaintenance: controllerv1connect.NewNodeMaintenanceServiceClient(http.DefaultClient, baseURL, opts...), dockerQuery: controllerv1connect.NewDockerQueryServiceClient(http.DefaultClient, baseURL, opts...), dockerCommand: controllerv1connect.NewDockerCommandServiceClient(http.DefaultClient, baseURL, opts...), repoQuery: controllerv1connect.NewRepoQueryServiceClient(http.DefaultClient, baseURL, opts...), repoCommand: controllerv1connect.NewRepoCommandServiceClient(http.DefaultClient, baseURL, opts...), serviceQuery: controllerv1connect.NewServiceQueryServiceClient(http.DefaultClient, baseURL, opts...), serviceCommand: controllerv1connect.NewServiceCommandServiceClient(http.DefaultClient, baseURL, opts...), serviceInst: controllerv1connect.NewServiceInstanceServiceClient(http.DefaultClient, baseURL, opts...), task: controllerv1connect.NewTaskServiceClient(http.DefaultClient, baseURL, opts...), backupQuery: controllerv1connect.NewBackupQueryServiceClient(http.DefaultClient, baseURL, opts...), backupCommand: controllerv1connect.NewBackupCommandServiceClient(http.DefaultClient, baseURL, opts...), secret: controllerv1connect.NewSecretServiceClient(http.DefaultClient, baseURL, opts...), }}
func controllerAPIBaseURL() string { addr := strings.TrimSpace(os.Getenv("COMPOSIA_E2E_CONTROLLER_ADDR")) if addr == "" { addr = defaultControllerAddr } return rpcutil.JoinBaseURL(addr, rpcutil.ControllerAPIBasePath)}
func accessToken() string { token := strings.TrimSpace(os.Getenv("COMPOSIA_E2E_ACCESS_TOKEN")) if token == "" { return defaultAccessToken } return token}
func testContext(t *testing.T) context.Context { t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) t.Cleanup(cancel) return ctx}
func waitForController(t *testing.T, clients controllerClients) { t.Helper() deadline := time.Now().Add(60 * time.Second) var lastErr error for time.Now().Before(deadline) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) _, lastErr = clients.system.GetSystemStatus(ctx, connect.NewRequest(&controllerv1.GetSystemStatusRequest{})) cancel() if lastErr == nil { return } time.Sleep(2 * time.Second) } t.Fatalf("controller did not become ready: %v", lastErr)}
func repoHead(t *testing.T, ctx context.Context, clients controllerClients) *controllerv1.GetRepoHeadResponse { t.Helper() head, err := clients.repoQuery.GetRepoHead(ctx, connect.NewRequest(&controllerv1.GetRepoHeadRequest{})) if err != nil { t.Fatalf("get repo head: %v", err) } return head.Msg}
func cleanupRepoPath(t *testing.T, ctx context.Context, clients controllerClients, path string) { t.Helper() head, err := clients.repoQuery.GetRepoHead(ctx, connect.NewRequest(&controllerv1.GetRepoHeadRequest{})) if err != nil { t.Logf("cleanup get repo head for %q: %v", path, err) return } _, err = clients.repoCommand.DeleteRepoPath(ctx, connect.NewRequest(&controllerv1.DeleteRepoPathRequest{ Path: path, BaseRevision: head.Msg.GetHeadRevision(), CommitMessage: "Cleanup controller e2e workspace", })) if err == nil { return } if code := connect.CodeOf(err); code == connect.CodeFailedPrecondition || code == connect.CodeNotFound { return } t.Logf("cleanup delete repo path %q: %v", path, err)}
func waitForTaskStatus(t *testing.T, ctx context.Context, clients controllerClients, taskID string, want controllerv1.TaskStatus) *controllerv1.GetTaskResponse { t.Helper() deadline := time.Now().Add(60 * time.Second) var last *controllerv1.GetTaskResponse for time.Now().Before(deadline) { resp, err := clients.task.GetTask(ctx, connect.NewRequest(&controllerv1.GetTaskRequest{TaskId: taskID})) if err != nil { t.Fatalf("get task %s: %v", taskID, err) } last = resp.Msg if last.GetStatus() == want { return last } if isTerminalTaskStatus(last.GetStatus()) { t.Fatalf("task %s reached %s, want %s: %s", taskID, last.GetStatus(), want, last.GetErrorSummary()) } time.Sleep(1 * time.Second) } if last == nil { t.Fatalf("task %s did not become visible", taskID) } t.Fatalf("task %s status = %s after timeout, want %s", taskID, last.GetStatus(), want) return nil}
func firstTaskLogChunk(t *testing.T, ctx context.Context, clients controllerClients, taskID string) string { t.Helper() streamCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() stream, err := clients.task.TailTaskLogs(streamCtx, connect.NewRequest(&controllerv1.TailTaskLogsRequest{TaskId: taskID})) if err != nil { t.Fatalf("tail task logs: %v", err) } defer func() { _ = stream.Close() }() if !stream.Receive() { t.Fatalf("expected task log chunk, got err=%v", stream.Err()) } return stream.Msg().GetContent()}
func firstContainerLogChunk(t *testing.T, ctx context.Context, clients controllerClients, containerID string) string { t.Helper() streamCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() stream, err := clients.dockerCommand.GetContainerLogs(streamCtx, connect.NewRequest(&controllerv1.GetContainerLogsRequest{NodeId: testNodeID, ContainerId: containerID, Tail: "20"})) if err != nil { t.Fatalf("get container logs: %v", err) } defer func() { _ = stream.Close() }() if !stream.Receive() { t.Fatalf("expected container log chunk, got err=%v", stream.Err()) } return stream.Msg().GetContent()}
func waitForNodeOnline(t *testing.T, ctx context.Context, clients controllerClients, nodeID string) { t.Helper() deadline := time.Now().Add(60 * time.Second) for time.Now().Before(deadline) { resp, err := clients.nodeQuery.GetNode(ctx, connect.NewRequest(&controllerv1.GetNodeRequest{NodeId: nodeID})) if err != nil { t.Fatalf("get node while waiting online: %v", err) } if resp.Msg.GetNode().GetIsOnline() { return } time.Sleep(1 * time.Second) } t.Fatalf("node %q did not become online", nodeID)}
func listContainers(t *testing.T, ctx context.Context, clients controllerClients) []*controllerv1.ContainerInfo { t.Helper() resp, err := clients.dockerQuery.ListNodeContainers(ctx, connect.NewRequest(&controllerv1.ListNodeContainersRequest{NodeId: testNodeID, PageSize: 50, Page: 1})) if err != nil { t.Fatalf("list node containers: %v", err) } if resp.Msg.GetTotalCount() == 0 || len(resp.Msg.GetContainers()) == 0 { t.Fatalf("expected at least one docker container") } return resp.Msg.GetContainers()}
func chooseExecContainer(t *testing.T, containers []*controllerv1.ContainerInfo) *controllerv1.ContainerInfo { t.Helper() for _, container := range containers { name := container.GetName() if container.GetState() == "running" && strings.Contains(name, "composia-e2e-exec") { return container } } t.Fatalf("expected running composia-e2e-exec fixture container") return nil}
func shortID(id string) string { if len(id) < 12 { return id } return id[:12]}
func sourceRequest[T any](msg *T, source string) *connect.Request[T] { req := connect.NewRequest(msg) req.Header().Set("X-Composia-Source", source) return req}
func assertStreamConnectCode[T any](t *testing.T, stream *connect.ServerStreamForClient[T], err error, code connect.Code) { t.Helper() if err != nil { assertConnectCode(t, err, code) return } defer func() { _ = stream.Close() }() if stream.Receive() { t.Fatalf("expected stream connect code %s, got message", code) } assertConnectCode(t, stream.Err(), code)}
func assertConnectCode(t *testing.T, err error, code connect.Code) { t.Helper() if err == nil { t.Fatalf("expected connect code %s, got nil", code) } if got := connect.CodeOf(err); got != code { t.Fatalf("connect code = %s, want %s: %v", got, code, err) }}
func isTerminalTaskStatus(status controllerv1.TaskStatus) bool { switch status { case controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED, controllerv1.TaskStatus_TASK_STATUS_FAILED, controllerv1.TaskStatus_TASK_STATUS_CANCELLED: return true default: return false }}
func hasNodeConfig(nodes []*controllerv1.NodeConfigSummary, nodeID string) bool { for _, node := range nodes { if node.GetId() == nodeID { return true } } return false}
func hasAccessToken(tokens []*controllerv1.AccessTokenSummary, name string) bool { for _, token := range tokens { if token.GetName() == name && token.GetEnabled() { return true } } return false}
func hasNodeSummary(nodes []*controllerv1.NodeSummary, nodeID string) bool { for _, node := range nodes { if node.GetNodeId() == nodeID { return true } } return false}
func hasRepoEntry(entries []*controllerv1.RepoFileEntry, name string, isDir bool) bool { for _, entry := range entries { if entry.GetName() == name && entry.GetIsDir() == isDir { return true } } return false}
func hasRepoEntryPath(entries []*controllerv1.RepoFileEntry, path string, isDir bool) bool { for _, entry := range entries { if entry.GetPath() == path && entry.GetIsDir() == isDir { return true } } return false}
func hasValidationErrorPath(errors []*controllerv1.RepoValidationError, path string) bool { for _, validationError := range errors { if validationError.GetPath() == path { return true } } return false}
func hasWorkspace(workspaces []*controllerv1.ServiceWorkspaceSummary, folder, serviceName string) bool { for _, workspace := range workspaces { if workspace.GetFolder() == folder && workspace.GetServiceName() == serviceName && workspace.GetIsDeclared() { return true } } return false}
func hasServiceSummary(services []*controllerv1.ServiceSummary, name string) bool { for _, service := range services { if service.GetName() == name && service.GetIsDeclared() { return true } } return false}
func hasServiceInstance(instances []*controllerv1.ServiceInstanceSummary, serviceName, nodeID string) bool { for _, instance := range instances { if instance.GetServiceName() == serviceName && instance.GetNodeId() == nodeID && instance.GetIsDeclared() { return true } } return false}
func hasContainerInfo(containers []*controllerv1.ContainerInfo, containerID string) bool { for _, container := range containers { if container.GetId() == containerID { return true } } return false}
func hasTaskSummary(tasks []*controllerv1.TaskSummary, taskID string) bool { for _, item := range tasks { if item.GetTaskId() == taskID { return true } } return false}
func containsString(values []string, want string) bool { for _, value := range values { if value == want { return true } } return false}