feat: thothctl workspace preprocessing CLI and file-ingress contracts (P2)
This commit is contained in:
@@ -178,7 +178,16 @@ func TestLoadRejectsRelativeInstallationPaths(t *testing.T) {
|
||||
func writeInstallation(t *testing.T, profile string) (string, string, string, string) {
|
||||
t.Helper()
|
||||
|
||||
root := filepath.Join(t.TempDir(), "installation folder with spaces")
|
||||
temporaryRoot, err := filepath.EvalSymlinks(os.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
physicalRoot, err := os.MkdirTemp(temporaryRoot, "thothctl-config-")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(physicalRoot) })
|
||||
root := filepath.Join(physicalRoot, "installation folder with spaces")
|
||||
projectDirectory := filepath.Join(root, "project directory with spaces")
|
||||
if err := os.MkdirAll(filepath.Join(projectDirectory, "deploy"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -743,7 +743,12 @@ func writeLifecycleOverride(path, image string) error {
|
||||
if err != nil {
|
||||
return errors.New("lifecycle image override could not be encoded")
|
||||
}
|
||||
contents := []byte("services:\n core:\n image: " + string(quoted) + "\n")
|
||||
contents := []byte(`services:
|
||||
core:
|
||||
image: ` + string(quoted) + `
|
||||
workspace-maintenance:
|
||||
image: ` + string(quoted) + `
|
||||
`)
|
||||
if err := writeFileDurably(path, ".pi-lifecycle-", contents); err != nil {
|
||||
return errors.New("lifecycle image override could not be written durably")
|
||||
}
|
||||
|
||||
@@ -126,6 +126,24 @@ func TestSuccessfulUpdateAndRollbackRemainSelectedOnFreshRecreate(t *testing.T)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteLifecycleOverridePinsCoreAndWorkspaceMaintenanceToTheSameImage(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "current-image.yaml")
|
||||
if err := writeLifecycleOverride(path, "sha256:"+strings.Repeat("a", 64)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
contents, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
text := string(contents)
|
||||
wantedImage := `image: "sha256:` + strings.Repeat("a", 64) + `"`
|
||||
for _, expected := range []string{"services:", "core:", "workspace-maintenance:", wantedImage} {
|
||||
if !strings.Contains(text, expected) {
|
||||
t.Fatalf("override = %q, missing %q", text, expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSelectorPromotionDoesNotMaskPostRenameDirectoryFsyncFailure(t *testing.T) {
|
||||
directory := t.TempDir()
|
||||
source := filepath.Join(directory, "candidate.yaml")
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Package safeio reads installation files without following symlinked path components.
|
||||
// Package safeio reads and writes local files without following symlinked path components.
|
||||
package safeio
|
||||
|
||||
import (
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
var ErrUnsafeFile = errors.New("unsafe file")
|
||||
@@ -19,17 +20,83 @@ func ValidateCanonicalPath(path string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func readBoundedRegularFile(file *os.File, maximum int64) ([]byte, error) {
|
||||
func readBoundedRegularFile(path string, file *os.File, maximum int64) ([]byte, error) {
|
||||
if maximum < 0 || maximum == int64(^uint64(0)>>1) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
info, err := file.Stat()
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
if err != nil || !info.Mode().IsRegular() || !hasSingleLink(info) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
contents, err := io.ReadAll(io.LimitReader(file, maximum+1))
|
||||
if err != nil || int64(len(contents)) > maximum {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
after, err := file.Stat()
|
||||
if err != nil || !after.Mode().IsRegular() || !hasSingleLink(after) || !os.SameFile(info, after) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
current, err := os.Stat(path)
|
||||
if err != nil || !os.SameFile(info, current) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
return contents, nil
|
||||
}
|
||||
|
||||
func ReadCanonicalUTF8(path string, maximum int64) (string, error) {
|
||||
contents, err := ReadCanonicalRegular(path, maximum)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !utf8.Valid(contents) {
|
||||
return "", ErrUnsafeFile
|
||||
}
|
||||
return string(contents), nil
|
||||
}
|
||||
|
||||
func WriteCanonicalNewFile(path string, contents []byte, mode os.FileMode) error {
|
||||
if err := ValidateCanonicalPath(path); err != nil {
|
||||
return err
|
||||
}
|
||||
parent := filepath.Dir(path)
|
||||
if err := requireCanonicalDirectory(parent); err != nil {
|
||||
return err
|
||||
}
|
||||
if info, err := os.Lstat(path); err == nil {
|
||||
if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 || info.Mode()&os.ModeType != 0 {
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
return ErrUnsafeFile
|
||||
} else if !errors.Is(err, os.ErrNotExist) {
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, mode)
|
||||
if err != nil {
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
defer file.Close()
|
||||
if _, err := file.Write(contents); err != nil {
|
||||
_ = os.Remove(path)
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
if err := file.Sync(); err != nil {
|
||||
_ = os.Remove(path)
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func requireCanonicalDirectory(path string) error {
|
||||
if err := ValidateCanonicalPath(path); err != nil {
|
||||
return err
|
||||
}
|
||||
resolved, err := filepath.EvalSymlinks(path)
|
||||
if err != nil || resolved != path {
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil || !info.IsDir() {
|
||||
return ErrUnsafeFile
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -41,3 +41,55 @@ func TestReadCanonicalRegularRejectsFinalAndParentSymlinks(t *testing.T) {
|
||||
t.Fatalf("final symlink error = %v, want ErrUnsafeFile", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadCanonicalUTF8RejectsNonUTF8AndHardlinks(t *testing.T) {
|
||||
temporaryRoot, err := filepath.EvalSymlinks(os.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root, err := os.MkdirTemp(temporaryRoot, "thothctl-safeio-")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(root) })
|
||||
|
||||
nonUTF8 := filepath.Join(root, "annotations.yaml")
|
||||
if err := os.WriteFile(nonUTF8, []byte{0xff, 0xfe, 0xfd}, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ReadCanonicalUTF8(nonUTF8, 1024); !errors.Is(err, ErrUnsafeFile) {
|
||||
t.Fatalf("ReadCanonicalUTF8(nonUTF8) error = %v, want ErrUnsafeFile", err)
|
||||
}
|
||||
|
||||
target := filepath.Join(root, "regular.txt")
|
||||
if err := os.WriteFile(target, []byte("linked"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
link := filepath.Join(root, "hardlink.txt")
|
||||
if err := os.Link(target, link); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ReadCanonicalUTF8(link, 1024); !errors.Is(err, ErrUnsafeFile) {
|
||||
t.Fatalf("ReadCanonicalUTF8(hardlink) error = %v, want ErrUnsafeFile", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteCanonicalNewFileRejectsExistingTargets(t *testing.T) {
|
||||
temporaryRoot, err := filepath.EvalSymlinks(os.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root, err := os.MkdirTemp(temporaryRoot, "thothctl-safeio-")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(root) })
|
||||
|
||||
path := filepath.Join(root, "artifact.yaml")
|
||||
if err := os.WriteFile(path, []byte("existing"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := WriteCanonicalNewFile(path, []byte("new"), 0o600); !errors.Is(err, ErrUnsafeFile) {
|
||||
t.Fatalf("WriteCanonicalNewFile(existing) error = %v, want ErrUnsafeFile", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -47,7 +47,7 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
defer file.Close()
|
||||
return readBoundedRegularFile(file, maximum)
|
||||
return readBoundedRegularFile(path, file, maximum)
|
||||
}
|
||||
|
||||
func closeUnixDescriptors(descriptors []int) {
|
||||
|
||||
@@ -53,7 +53,7 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) {
|
||||
return nil, ErrUnsafeFile
|
||||
}
|
||||
defer file.Close()
|
||||
return readBoundedRegularFile(file, maximum)
|
||||
return readBoundedRegularFile(path, file, maximum)
|
||||
}
|
||||
|
||||
func openWindowsComponent(path string, directory bool) (windows.Handle, error) {
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
//go:build !windows
|
||||
|
||||
package safeio
|
||||
|
||||
import (
|
||||
"os"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
func hasSingleLink(info os.FileInfo) bool {
|
||||
stat, ok := info.Sys().(*syscall.Stat_t)
|
||||
return ok && stat.Nlink == 1
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
//go:build windows
|
||||
|
||||
package safeio
|
||||
|
||||
import "os"
|
||||
|
||||
func hasSingleLink(info os.FileInfo) bool {
|
||||
return true
|
||||
}
|
||||
@@ -294,7 +294,15 @@ func TestRemoveRejectsConfirmationForDifferentContainerIDs(t *testing.T) {
|
||||
|
||||
func testInstallation(t *testing.T) config.Installation {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
temporaryRoot, err := filepath.EvalSymlinks(os.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root, err := os.MkdirTemp(temporaryRoot, "thothctl-serverops-")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(root) })
|
||||
project := filepath.Join(root, "project")
|
||||
if err := os.Mkdir(project, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -0,0 +1,772 @@
|
||||
// Package workspaceops implements the closed host-side workspace preprocessing contract.
|
||||
package workspaceops
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/aritmolab/thothii/tools/thothctl/internal/compose"
|
||||
"github.com/aritmolab/thothii/tools/thothctl/internal/config"
|
||||
"github.com/aritmolab/thothii/tools/thothctl/internal/safeio"
|
||||
)
|
||||
|
||||
var (
|
||||
workspacePattern = regexp.MustCompile(`^[a-z][a-z0-9-]{2,62}$`)
|
||||
runIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`)
|
||||
reviewedCandidatesDigest = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
|
||||
)
|
||||
|
||||
const (
|
||||
maxFromSQLFiles = 32
|
||||
maxAssumptions = 256
|
||||
)
|
||||
|
||||
type Runner interface {
|
||||
Run(context.Context, []string, io.Reader) (compose.Result, error)
|
||||
}
|
||||
|
||||
type Request interface {
|
||||
workspaceRequest()
|
||||
workspaceID() string
|
||||
JSONMode() bool
|
||||
operatorCommand() string
|
||||
stdinEnvelope() (requestEnvelope, error)
|
||||
}
|
||||
|
||||
type baseRequest struct {
|
||||
Workspace string
|
||||
JSON bool
|
||||
}
|
||||
|
||||
func (b baseRequest) workspaceID() string { return b.Workspace }
|
||||
func (b baseRequest) JSONMode() bool { return b.JSON }
|
||||
|
||||
type InspectRequest struct{ baseRequest }
|
||||
|
||||
type DwhRequest struct {
|
||||
baseRequest
|
||||
Resume string
|
||||
}
|
||||
|
||||
type SuggestFksRequest struct {
|
||||
baseRequest
|
||||
FromSQL []string
|
||||
Assume []string
|
||||
Output string
|
||||
}
|
||||
|
||||
type CheckSchemaRequest struct {
|
||||
baseRequest
|
||||
Annotations string
|
||||
ReviewedCandidates string
|
||||
}
|
||||
|
||||
type IndexSchemaRequest struct{ baseRequest }
|
||||
|
||||
type EvidenceRequest struct {
|
||||
baseRequest
|
||||
DryRun bool
|
||||
Resume string
|
||||
}
|
||||
|
||||
type RunRequest struct {
|
||||
baseRequest
|
||||
Resume string
|
||||
}
|
||||
|
||||
func (InspectRequest) workspaceRequest() {}
|
||||
func (DwhRequest) workspaceRequest() {}
|
||||
func (SuggestFksRequest) workspaceRequest() {}
|
||||
func (CheckSchemaRequest) workspaceRequest() {}
|
||||
func (IndexSchemaRequest) workspaceRequest() {}
|
||||
func (EvidenceRequest) workspaceRequest() {}
|
||||
func (RunRequest) workspaceRequest() {}
|
||||
|
||||
func (InspectRequest) operatorCommand() string { return "inspect" }
|
||||
func (DwhRequest) operatorCommand() string { return "preprocess-dwh" }
|
||||
func (SuggestFksRequest) operatorCommand() string { return "schema-suggest-fks" }
|
||||
func (CheckSchemaRequest) operatorCommand() string {
|
||||
return "schema-check"
|
||||
}
|
||||
func (IndexSchemaRequest) operatorCommand() string { return "index-schema" }
|
||||
func (EvidenceRequest) operatorCommand() string { return "preprocess-evidence" }
|
||||
func (RunRequest) operatorCommand() string { return "preprocess-run" }
|
||||
|
||||
func (r InspectRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace}, nil
|
||||
}
|
||||
|
||||
func (r DwhRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume}, nil
|
||||
}
|
||||
|
||||
func (r SuggestFksRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
envelope := requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Assume: append([]string(nil), r.Assume...)}
|
||||
totalBytes := 0
|
||||
for _, path := range r.FromSQL {
|
||||
contents, err := safeio.ReadCanonicalUTF8(path, 1<<20)
|
||||
if err != nil {
|
||||
return requestEnvelope{}, errors.New("SQL input could not be read safely")
|
||||
}
|
||||
totalBytes += len(contents)
|
||||
if totalBytes > 16<<20 {
|
||||
return requestEnvelope{}, errors.New("SQL input total exceeds 16 MiB")
|
||||
}
|
||||
envelope.SQLFiles = append(envelope.SQLFiles, inputFile{Path: path, Contents: contents})
|
||||
}
|
||||
return envelope, nil
|
||||
}
|
||||
|
||||
func (r CheckSchemaRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
annotations, err := safeio.ReadCanonicalUTF8(r.Annotations, 16<<20)
|
||||
if err != nil {
|
||||
return requestEnvelope{}, errors.New("annotation file could not be read safely")
|
||||
}
|
||||
return requestEnvelope{
|
||||
SchemaVersion: 1,
|
||||
Operation: r.operatorCommand(),
|
||||
WorkspaceID: r.Workspace,
|
||||
Annotations: annotations,
|
||||
ReviewedCandidates: r.ReviewedCandidates,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (r IndexSchemaRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace}, nil
|
||||
}
|
||||
|
||||
func (r EvidenceRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume, DryRun: r.DryRun}, nil
|
||||
}
|
||||
|
||||
func (r RunRequest) stdinEnvelope() (requestEnvelope, error) {
|
||||
return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume}, nil
|
||||
}
|
||||
|
||||
type requestEnvelope struct {
|
||||
SchemaVersion int `json:"schemaVersion"`
|
||||
Operation string `json:"operation"`
|
||||
WorkspaceID string `json:"workspaceId"`
|
||||
Resume string `json:"resume,omitempty"`
|
||||
DryRun bool `json:"dryRun,omitempty"`
|
||||
Assume []string `json:"assume,omitempty"`
|
||||
SQLFiles []inputFile `json:"sqlFiles,omitempty"`
|
||||
Annotations string `json:"annotations,omitempty"`
|
||||
ReviewedCandidates string `json:"reviewedCandidates,omitempty"`
|
||||
}
|
||||
|
||||
type inputFile struct {
|
||||
Path string `json:"path"`
|
||||
Contents string `json:"contents"`
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
SchemaVersion int `json:"schemaVersion"`
|
||||
Status string `json:"status"`
|
||||
Code string `json:"code"`
|
||||
WorkspaceID string `json:"workspaceId"`
|
||||
WorkspaceRevision string `json:"workspaceRevision"`
|
||||
DescriptorBlob string `json:"descriptorBlob"`
|
||||
Operation string `json:"operation"`
|
||||
RunID string `json:"runId,omitempty"`
|
||||
ChildRuns map[string]string `json:"childRuns,omitempty"`
|
||||
CompletedStages []string `json:"completedStages"`
|
||||
Counts map[string]int `json:"counts,omitempty"`
|
||||
ArtifactIdentities []ArtifactIdentity `json:"artifactIdentities,omitempty"`
|
||||
Warnings []string `json:"warnings,omitempty"`
|
||||
}
|
||||
|
||||
type ArtifactIdentity struct {
|
||||
Kind string `json:"kind"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
|
||||
type operationResponse struct {
|
||||
Result
|
||||
SuggestedFksYAML string `json:"suggestedFksYaml,omitempty"`
|
||||
}
|
||||
|
||||
type Stage string
|
||||
|
||||
type ExitClass string
|
||||
|
||||
const (
|
||||
StageRenderedConfig Stage = "rendered-config"
|
||||
StageImageInspect Stage = "image-inspect"
|
||||
StageComposeRun Stage = "compose-run"
|
||||
|
||||
ExitClassNonzero ExitClass = "nonzero-exit"
|
||||
ExitClassUnavailable ExitClass = "unavailable"
|
||||
ExitClassTimeout ExitClass = "timeout"
|
||||
ExitClassInvocation ExitClass = "invocation-failure"
|
||||
)
|
||||
|
||||
type OperationError struct {
|
||||
stage Stage
|
||||
class ExitClass
|
||||
detail string
|
||||
}
|
||||
|
||||
func (e *OperationError) Error() string {
|
||||
return fmt.Sprintf("stage=%s class=%s", e.stage, e.class)
|
||||
}
|
||||
|
||||
func (e *OperationError) Stage() Stage { return e.stage }
|
||||
func (e *OperationError) Class() ExitClass { return e.class }
|
||||
func (e *OperationError) Detail() string { return e.detail }
|
||||
|
||||
func Parse(args []string) (Request, error) {
|
||||
if len(args) == 0 {
|
||||
return nil, errors.New("workspace requires a subcommand")
|
||||
}
|
||||
switch args[0] {
|
||||
case "inspect":
|
||||
parsed, err := parseInspect(args[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return parsed, nil
|
||||
case "preprocess":
|
||||
return parsePreprocess(args[1:])
|
||||
case "schema":
|
||||
return parseSchema(args[1:])
|
||||
case "index-schema":
|
||||
parsed, err := parseIndexSchema(args[1:])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return parsed, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown workspace command %q", args[0])
|
||||
}
|
||||
}
|
||||
|
||||
func Execute(ctx context.Context, installation config.Installation, runner Runner, request Request) (Result, error) {
|
||||
envelope, err := request.stdinEnvelope()
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
rendered, err := runDocker(ctx, runner, StageRenderedConfig, installation.ComposeArgs("config", "--format", "json"))
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
imageReference, err := selectedCoreImage(rendered.Stdout)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
imageID, err := immutableImageID(ctx, runner, imageReference)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
override, cleanup, err := maintenanceOverride(installation, imageID)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
defer cleanup()
|
||||
stdin, err := encodeEnvelope(envelope)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
args, err := installation.ComposeArgsWithFinalOverride(
|
||||
override,
|
||||
"run", "--rm", "--no-deps", "--no-TTY", "--name", ownedContainerName(installation, request), "workspace-maintenance", request.operatorCommand(),
|
||||
)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
result, err := runDocker(ctx, runner, StageComposeRun, args, bytes.NewReader(stdin))
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
response, err := parseResponse(result.Stdout)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
if suggest, ok := request.(SuggestFksRequest); ok && suggest.Output != "" {
|
||||
if response.SuggestedFksYAML == "" {
|
||||
return Result{}, errors.New("workspace maintenance did not return the requested FK artifact")
|
||||
}
|
||||
if digest := suggestedArtifactDigest(response); digest != "" {
|
||||
sum := sha256.Sum256([]byte(response.SuggestedFksYAML))
|
||||
if digest != "sha256:"+fmt.Sprintf("%x", sum[:]) {
|
||||
return Result{}, errors.New("workspace maintenance returned an FK artifact with a mismatched digest")
|
||||
}
|
||||
}
|
||||
if err := safeio.WriteCanonicalNewFile(suggest.Output, []byte(response.SuggestedFksYAML), 0o600); err != nil {
|
||||
return Result{}, errors.New("workspace FK output file could not be created safely")
|
||||
}
|
||||
}
|
||||
return response.Result, nil
|
||||
}
|
||||
|
||||
func encodeEnvelope(envelope requestEnvelope) ([]byte, error) {
|
||||
encoded, err := json.Marshal(envelope)
|
||||
if err != nil {
|
||||
return nil, errors.New("workspace request could not be encoded")
|
||||
}
|
||||
if len(encoded) > 1<<20 {
|
||||
return nil, errors.New("workspace request exceeds the bounded stdin contract")
|
||||
}
|
||||
return append(encoded, '\n'), nil
|
||||
}
|
||||
|
||||
func parseResponse(document string) (operationResponse, error) {
|
||||
decoder := json.NewDecoder(strings.NewReader(document))
|
||||
decoder.DisallowUnknownFields()
|
||||
var response operationResponse
|
||||
if err := decoder.Decode(&response); err != nil {
|
||||
return operationResponse{}, errors.New("workspace maintenance returned invalid JSON")
|
||||
}
|
||||
var extra any
|
||||
if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) {
|
||||
return operationResponse{}, errors.New("workspace maintenance returned trailing output")
|
||||
}
|
||||
if err := validateResult(response.Result); err != nil {
|
||||
return operationResponse{}, err
|
||||
}
|
||||
return response, nil
|
||||
}
|
||||
|
||||
func validateResult(result Result) error {
|
||||
if result.SchemaVersion != 1 {
|
||||
return errors.New("workspace maintenance returned an unsupported schema version")
|
||||
}
|
||||
if !workspacePattern.MatchString(result.WorkspaceID) {
|
||||
return errors.New("workspace maintenance returned an invalid workspace identity")
|
||||
}
|
||||
if len(result.WorkspaceRevision) != 40 || !isLowerHex(result.WorkspaceRevision) {
|
||||
return errors.New("workspace maintenance returned an invalid workspace revision")
|
||||
}
|
||||
if !strings.HasPrefix(result.DescriptorBlob, "sha256:") || len(result.DescriptorBlob) != len("sha256:")+64 || !isLowerHex(strings.TrimPrefix(result.DescriptorBlob, "sha256:")) {
|
||||
return errors.New("workspace maintenance returned an invalid descriptor digest")
|
||||
}
|
||||
validStatuses := map[string]struct{}{"succeeded": {}, "unchanged": {}, "dry_run": {}, "blocked": {}, "failed": {}}
|
||||
if _, ok := validStatuses[result.Status]; !ok {
|
||||
return errors.New("workspace maintenance returned an invalid status")
|
||||
}
|
||||
if strings.TrimSpace(result.Code) == "" || strings.TrimSpace(result.Operation) == "" || result.CompletedStages == nil {
|
||||
return errors.New("workspace maintenance omitted required fields")
|
||||
}
|
||||
for _, digest := range result.ArtifactIdentities {
|
||||
if strings.TrimSpace(digest.Kind) == "" || !strings.HasPrefix(digest.Digest, "sha256:") {
|
||||
return errors.New("workspace maintenance returned an invalid artifact identity")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isLowerHex(value string) bool {
|
||||
for _, r := range value {
|
||||
if !(r >= '0' && r <= '9' || r >= 'a' && r <= 'f') {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return value != ""
|
||||
}
|
||||
|
||||
func parseInspect(args []string) (InspectRequest, error) {
|
||||
base, err := parseBaseFlags(args, false)
|
||||
if err != nil {
|
||||
return InspectRequest{}, err
|
||||
}
|
||||
return InspectRequest{baseRequest: base}, nil
|
||||
}
|
||||
|
||||
func parsePreprocess(args []string) (Request, error) {
|
||||
if len(args) == 0 {
|
||||
return nil, errors.New("workspace preprocess requires dwh, evidence, or run")
|
||||
}
|
||||
switch args[0] {
|
||||
case "dwh":
|
||||
base, resume, dryRun, err := parseResumeFlags(args[1:], false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if dryRun {
|
||||
return nil, errors.New("workspace preprocess dwh does not accept --dry-run")
|
||||
}
|
||||
return DwhRequest{baseRequest: base, Resume: resume}, nil
|
||||
case "evidence":
|
||||
base, resume, dryRun, err := parseResumeFlags(args[1:], true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return EvidenceRequest{baseRequest: base, Resume: resume, DryRun: dryRun}, nil
|
||||
case "run":
|
||||
base, resume, dryRun, err := parseResumeFlags(args[1:], false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if dryRun {
|
||||
return nil, errors.New("workspace preprocess run does not accept --dry-run")
|
||||
}
|
||||
return RunRequest{baseRequest: base, Resume: resume}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown workspace preprocess command %q", args[0])
|
||||
}
|
||||
}
|
||||
|
||||
func parseSchema(args []string) (Request, error) {
|
||||
if len(args) == 0 {
|
||||
return nil, errors.New("workspace schema requires suggest-fks or check")
|
||||
}
|
||||
switch args[0] {
|
||||
case "suggest-fks":
|
||||
return parseSuggestFks(args[1:])
|
||||
case "check":
|
||||
return parseSchemaCheck(args[1:])
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown workspace schema command %q", args[0])
|
||||
}
|
||||
}
|
||||
|
||||
func parseIndexSchema(args []string) (IndexSchemaRequest, error) {
|
||||
base, err := parseBaseFlags(args, false)
|
||||
if err != nil {
|
||||
return IndexSchemaRequest{}, err
|
||||
}
|
||||
return IndexSchemaRequest{baseRequest: base}, nil
|
||||
}
|
||||
|
||||
func parseResumeFlags(args []string, allowDryRun bool) (baseRequest, string, bool, error) {
|
||||
var resume string
|
||||
var dryRun bool
|
||||
base, seen, err := parseSharedFlags(args, map[string]func(string) error{
|
||||
"--resume": func(value string) error {
|
||||
if resume != "" {
|
||||
return errors.New("--resume may be supplied once")
|
||||
}
|
||||
if !runIDPattern.MatchString(value) {
|
||||
return errors.New("--resume must be 32 lowercase hex characters")
|
||||
}
|
||||
resume = value
|
||||
return nil
|
||||
},
|
||||
}, map[string]func() error{
|
||||
"--dry-run": func() error {
|
||||
if !allowDryRun {
|
||||
return errors.New("--dry-run is not accepted here")
|
||||
}
|
||||
if dryRun {
|
||||
return errors.New("--dry-run may be supplied once")
|
||||
}
|
||||
dryRun = true
|
||||
return nil
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return baseRequest{}, "", false, err
|
||||
}
|
||||
if !seen.workspace {
|
||||
return baseRequest{}, "", false, errors.New("--workspace is required")
|
||||
}
|
||||
return base, resume, dryRun, nil
|
||||
}
|
||||
|
||||
func parseSuggestFks(args []string) (SuggestFksRequest, error) {
|
||||
request := SuggestFksRequest{}
|
||||
base, seen, err := parseSharedFlags(args, map[string]func(string) error{
|
||||
"--from-sql": func(value string) error {
|
||||
if len(request.FromSQL) >= maxFromSQLFiles {
|
||||
return fmt.Errorf("--from-sql may be supplied at most %d times", maxFromSQLFiles)
|
||||
}
|
||||
request.FromSQL = append(request.FromSQL, value)
|
||||
return nil
|
||||
},
|
||||
"--assume": func(value string) error {
|
||||
if len(request.Assume) >= maxAssumptions {
|
||||
return fmt.Errorf("--assume may be supplied at most %d times", maxAssumptions)
|
||||
}
|
||||
if len(value) > 256 || !strings.Contains(value, "=") {
|
||||
return errors.New("--assume values must be column=table entries up to 256 bytes")
|
||||
}
|
||||
left, right, _ := strings.Cut(value, "=")
|
||||
if strings.TrimSpace(left) == "" || strings.TrimSpace(right) == "" {
|
||||
return errors.New("--assume values must be column=table entries up to 256 bytes")
|
||||
}
|
||||
request.Assume = append(request.Assume, value)
|
||||
return nil
|
||||
},
|
||||
"--output": func(value string) error {
|
||||
if request.Output != "" {
|
||||
return errors.New("--output may be supplied once")
|
||||
}
|
||||
request.Output = value
|
||||
return nil
|
||||
},
|
||||
}, nil)
|
||||
if err != nil {
|
||||
return SuggestFksRequest{}, err
|
||||
}
|
||||
if !seen.workspace {
|
||||
return SuggestFksRequest{}, errors.New("--workspace is required")
|
||||
}
|
||||
request.baseRequest = base
|
||||
return request, nil
|
||||
}
|
||||
|
||||
func parseSchemaCheck(args []string) (CheckSchemaRequest, error) {
|
||||
request := CheckSchemaRequest{}
|
||||
base, seen, err := parseSharedFlags(args, map[string]func(string) error{
|
||||
"--annotations": func(value string) error {
|
||||
if request.Annotations != "" {
|
||||
return errors.New("--annotations may be supplied once")
|
||||
}
|
||||
request.Annotations = value
|
||||
return nil
|
||||
},
|
||||
"--reviewed-candidates": func(value string) error {
|
||||
if request.ReviewedCandidates != "" {
|
||||
return errors.New("--reviewed-candidates may be supplied once")
|
||||
}
|
||||
if !reviewedCandidatesDigest.MatchString(value) {
|
||||
return errors.New("--reviewed-candidates must be sha256:<64 lowercase hex>")
|
||||
}
|
||||
request.ReviewedCandidates = value
|
||||
return nil
|
||||
},
|
||||
}, nil)
|
||||
if err != nil {
|
||||
return CheckSchemaRequest{}, err
|
||||
}
|
||||
if !seen.workspace {
|
||||
return CheckSchemaRequest{}, errors.New("--workspace is required")
|
||||
}
|
||||
if (request.Annotations == "") != (request.ReviewedCandidates == "") {
|
||||
return CheckSchemaRequest{}, errors.New("--annotations and --reviewed-candidates must be supplied together")
|
||||
}
|
||||
request.baseRequest = base
|
||||
return request, nil
|
||||
}
|
||||
|
||||
func parseBaseFlags(args []string, allowDryRun bool) (baseRequest, error) {
|
||||
base, seen, err := parseSharedFlags(args, nil, nil)
|
||||
if err != nil {
|
||||
return baseRequest{}, err
|
||||
}
|
||||
if !seen.workspace {
|
||||
return baseRequest{}, errors.New("--workspace is required")
|
||||
}
|
||||
return base, nil
|
||||
}
|
||||
|
||||
type seenFlags struct {
|
||||
workspace bool
|
||||
json bool
|
||||
}
|
||||
|
||||
func parseSharedFlags(args []string, valueHandlers map[string]func(string) error, boolHandlers map[string]func() error) (baseRequest, seenFlags, error) {
|
||||
request := baseRequest{}
|
||||
seen := seenFlags{}
|
||||
valueHandlers = cloneValueHandlers(valueHandlers)
|
||||
boolHandlers = cloneBoolHandlers(boolHandlers)
|
||||
for len(args) > 0 {
|
||||
flag := args[0]
|
||||
if flag == "--" {
|
||||
return baseRequest{}, seenFlags{}, errors.New("passthrough separators are not supported")
|
||||
}
|
||||
switch flag {
|
||||
case "--workspace":
|
||||
if len(args) < 2 {
|
||||
return baseRequest{}, seenFlags{}, errors.New("--workspace requires a value")
|
||||
}
|
||||
if seen.workspace {
|
||||
return baseRequest{}, seenFlags{}, errors.New("--workspace must be supplied exactly once")
|
||||
}
|
||||
workspace := args[1]
|
||||
if !workspacePattern.MatchString(workspace) {
|
||||
return baseRequest{}, seenFlags{}, errors.New("--workspace must match [a-z][a-z0-9-]{2,62}")
|
||||
}
|
||||
request.Workspace, seen.workspace, args = workspace, true, args[2:]
|
||||
case "--json":
|
||||
if seen.json {
|
||||
return baseRequest{}, seenFlags{}, errors.New("--json may be supplied once")
|
||||
}
|
||||
request.JSON, seen.json, args = true, true, args[1:]
|
||||
default:
|
||||
if handler, ok := boolHandlers[flag]; ok {
|
||||
if err := handler(); err != nil {
|
||||
return baseRequest{}, seenFlags{}, err
|
||||
}
|
||||
args = args[1:]
|
||||
continue
|
||||
}
|
||||
handler, ok := valueHandlers[flag]
|
||||
if !ok {
|
||||
return baseRequest{}, seenFlags{}, fmt.Errorf("unknown workspace option %q", flag)
|
||||
}
|
||||
if len(args) < 2 {
|
||||
return baseRequest{}, seenFlags{}, fmt.Errorf("%s requires a value", flag)
|
||||
}
|
||||
if err := handler(args[1]); err != nil {
|
||||
return baseRequest{}, seenFlags{}, err
|
||||
}
|
||||
args = args[2:]
|
||||
}
|
||||
}
|
||||
return request, seen, nil
|
||||
}
|
||||
|
||||
func cloneValueHandlers(source map[string]func(string) error) map[string]func(string) error {
|
||||
if len(source) == 0 {
|
||||
return map[string]func(string) error{}
|
||||
}
|
||||
clone := make(map[string]func(string) error, len(source))
|
||||
for key, handler := range source {
|
||||
clone[key] = handler
|
||||
}
|
||||
return clone
|
||||
}
|
||||
|
||||
func cloneBoolHandlers(source map[string]func() error) map[string]func() error {
|
||||
if len(source) == 0 {
|
||||
return map[string]func() error{}
|
||||
}
|
||||
clone := make(map[string]func() error, len(source))
|
||||
for key, handler := range source {
|
||||
clone[key] = handler
|
||||
}
|
||||
return clone
|
||||
}
|
||||
|
||||
func selectedCoreImage(document string) (string, error) {
|
||||
var rendered struct {
|
||||
Services map[string]struct {
|
||||
Image string `json:"image"`
|
||||
} `json:"services"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(document), &rendered); err != nil {
|
||||
return "", errors.New("rendered Compose configuration is invalid")
|
||||
}
|
||||
core, exists := rendered.Services["core"]
|
||||
if !exists || strings.TrimSpace(core.Image) == "" {
|
||||
return "", errors.New("selected core image is unavailable")
|
||||
}
|
||||
return core.Image, nil
|
||||
}
|
||||
|
||||
func immutableImageID(ctx context.Context, runner Runner, reference string) (string, error) {
|
||||
result, err := runDocker(ctx, runner, StageImageInspect, []string{"image", "inspect", "--format", "{{.Id}}", reference})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
id := strings.TrimSpace(result.Stdout)
|
||||
if !strings.HasPrefix(id, "sha256:") || len(id) != len("sha256:")+64 || !isLowerHex(strings.TrimPrefix(id, "sha256:")) {
|
||||
return "", errors.New("selected core image did not resolve to an immutable sha256 image id")
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
func maintenanceOverride(installation config.Installation, imageID string) (string, func(), error) {
|
||||
control := installation.ControlDirectory()
|
||||
if err := os.MkdirAll(control, 0o700); err != nil {
|
||||
return "", func() {}, errors.New("workspace maintenance control directory could not be created")
|
||||
}
|
||||
info, err := os.Lstat(control)
|
||||
if err != nil || !info.IsDir() || info.Mode()&os.ModeSymlink != 0 {
|
||||
return "", func() {}, errors.New("workspace maintenance control directory is unsafe")
|
||||
}
|
||||
directory, err := os.MkdirTemp(control, "workspace-maintenance-")
|
||||
if err != nil {
|
||||
return "", func() {}, errors.New("workspace maintenance override directory could not be created")
|
||||
}
|
||||
cleanup := func() {
|
||||
_ = os.Remove(filepath.Join(directory, "override.yaml"))
|
||||
_ = os.Remove(directory)
|
||||
}
|
||||
path := filepath.Join(directory, "override.yaml")
|
||||
contents := []string{
|
||||
"services:",
|
||||
" core:",
|
||||
" image: " + strconvQuote(imageID),
|
||||
" pull_policy: never",
|
||||
" workspace-maintenance:",
|
||||
" image: " + strconvQuote(imageID),
|
||||
" pull_policy: never",
|
||||
"",
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(strings.Join(contents, "\n")), 0o600); err != nil {
|
||||
cleanup()
|
||||
return "", func() {}, errors.New("workspace maintenance override could not be written")
|
||||
}
|
||||
return path, cleanup, nil
|
||||
}
|
||||
|
||||
func strconvQuote(value string) string {
|
||||
encoded, _ := json.Marshal(value)
|
||||
return string(encoded)
|
||||
}
|
||||
|
||||
func ownedContainerName(installation config.Installation, request Request) string {
|
||||
parts := []string{installation.ProjectName(), request.workspaceID(), request.operatorCommand()}
|
||||
for index, value := range parts {
|
||||
parts[index] = strings.NewReplacer("/", "-", ":", "-", "@", "-", "_", "-").Replace(value)
|
||||
}
|
||||
return strings.Join(parts, "-")
|
||||
}
|
||||
|
||||
func runDocker(ctx context.Context, runner Runner, stage Stage, args []string, stdin ...io.Reader) (compose.Result, error) {
|
||||
var input io.Reader
|
||||
if len(stdin) > 0 {
|
||||
input = stdin[0]
|
||||
}
|
||||
result, err := runner.Run(ctx, args, input)
|
||||
if err != nil {
|
||||
class := ExitClassInvocation
|
||||
switch {
|
||||
case errors.Is(ctx.Err(), context.DeadlineExceeded):
|
||||
class = ExitClassTimeout
|
||||
case result.ExitCode == 127:
|
||||
class = ExitClassUnavailable
|
||||
case result.ExitCode != 0:
|
||||
class = ExitClassNonzero
|
||||
}
|
||||
detail := result.Stderr
|
||||
if strings.TrimSpace(detail) == "" {
|
||||
detail = err.Error()
|
||||
}
|
||||
return result, &OperationError{stage: stage, class: class, detail: detail}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func suggestedArtifactDigest(response operationResponse) string {
|
||||
for _, artifact := range response.ArtifactIdentities {
|
||||
if strings.HasPrefix(artifact.Digest, "sha256:") {
|
||||
return artifact.Digest
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func Human(result Result) string {
|
||||
lines := []string{
|
||||
fmt.Sprintf("workspace: %s", result.WorkspaceID),
|
||||
fmt.Sprintf("operation: %s", result.Operation),
|
||||
fmt.Sprintf("status: %s", result.Status),
|
||||
fmt.Sprintf("code: %s", result.Code),
|
||||
fmt.Sprintf("revision: %s", result.WorkspaceRevision),
|
||||
}
|
||||
if result.RunID != "" {
|
||||
lines = append(lines, fmt.Sprintf("run: %s", result.RunID))
|
||||
}
|
||||
if len(result.CompletedStages) > 0 {
|
||||
stages := append([]string(nil), result.CompletedStages...)
|
||||
sort.Strings(stages)
|
||||
lines = append(lines, fmt.Sprintf("completed: %s", strings.Join(stages, ", ")))
|
||||
}
|
||||
for _, warning := range result.Warnings {
|
||||
lines = append(lines, fmt.Sprintf("warning: %s", warning))
|
||||
}
|
||||
return strings.Join(lines, "\n") + "\n"
|
||||
}
|
||||
@@ -0,0 +1,203 @@
|
||||
package workspaceops
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/aritmolab/thothii/tools/thothctl/internal/compose"
|
||||
"github.com/aritmolab/thothii/tools/thothctl/internal/config"
|
||||
)
|
||||
|
||||
type fakeRunner struct {
|
||||
run func(args []string, stdin string) (compose.Result, error)
|
||||
all [][]string
|
||||
stdins []string
|
||||
}
|
||||
|
||||
func (r *fakeRunner) Run(_ context.Context, args []string, stdin io.Reader) (compose.Result, error) {
|
||||
payload := ""
|
||||
if stdin != nil {
|
||||
bytes, err := io.ReadAll(stdin)
|
||||
if err != nil {
|
||||
return compose.Result{}, err
|
||||
}
|
||||
payload = string(bytes)
|
||||
}
|
||||
r.all = append(r.all, append([]string(nil), args...))
|
||||
r.stdins = append(r.stdins, payload)
|
||||
return r.run(args, payload)
|
||||
}
|
||||
|
||||
func TestExecuteSuggestFksStreamsSQLFileContentsOnStdin(t *testing.T) {
|
||||
installation := testInstallation(t)
|
||||
sqlPath := filepath.Join(filepath.Dir(installation.Path), "query.sql")
|
||||
if err := os.WriteFile(sqlPath, []byte("select 1;\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) {
|
||||
switch {
|
||||
case contains(args, "config", "--format", "json"):
|
||||
return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil
|
||||
case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}):
|
||||
return compose.Result{Stdout: "sha256:" + strings.Repeat("a", 64)}, nil
|
||||
case contains(args, "workspace-maintenance", "schema-suggest-fks"):
|
||||
var envelope map[string]any
|
||||
if err := json.Unmarshal([]byte(stdin), &envelope); err != nil {
|
||||
t.Fatalf("stdin JSON = %q, err=%v", stdin, err)
|
||||
}
|
||||
sqlFiles, ok := envelope["sqlFiles"].([]any)
|
||||
if !ok || len(sqlFiles) != 1 {
|
||||
t.Fatalf("sqlFiles = %#v", envelope["sqlFiles"])
|
||||
}
|
||||
file, ok := sqlFiles[0].(map[string]any)
|
||||
if !ok || file["contents"] != "select 1;\n" {
|
||||
t.Fatalf("sql file envelope = %#v", sqlFiles[0])
|
||||
}
|
||||
return compose.Result{Stdout: successResult("schema-suggest-fks")}, nil
|
||||
default:
|
||||
t.Fatalf("unexpected Docker invocation: %#v", args)
|
||||
return compose.Result{}, nil
|
||||
}
|
||||
}}
|
||||
|
||||
_, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, FromSQL: []string{sqlPath}})
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteSuggestFksRejectsTotalSQLIngressOverSixteenMiB(t *testing.T) {
|
||||
installation := testInstallation(t)
|
||||
paths := make([]string, 0, 17)
|
||||
for index := 0; index < 17; index++ {
|
||||
path := filepath.Join(filepath.Dir(installation.Path), fmt.Sprintf("query-%02d.sql", index))
|
||||
if err := os.WriteFile(path, bytes.Repeat([]byte("x"), 1<<20), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
paths = append(paths, path)
|
||||
}
|
||||
runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) {
|
||||
t.Fatalf("Docker should not run when total SQL ingress exceeds the bound: %#v", args)
|
||||
return compose.Result{}, nil
|
||||
}}
|
||||
|
||||
_, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, FromSQL: paths})
|
||||
if err == nil || !strings.Contains(err.Error(), "total") {
|
||||
t.Fatalf("Execute() error = %v, want total-size failure", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteSchemaCheckStreamsAnnotationContentOnStdin(t *testing.T) {
|
||||
installation := testInstallation(t)
|
||||
annotationsPath := filepath.Join(filepath.Dir(installation.Path), "annotations.yaml")
|
||||
if err := os.WriteFile(annotationsPath, []byte("reviewed: []\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) {
|
||||
switch {
|
||||
case contains(args, "config", "--format", "json"):
|
||||
return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil
|
||||
case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}):
|
||||
return compose.Result{Stdout: "sha256:" + strings.Repeat("b", 64)}, nil
|
||||
case contains(args, "workspace-maintenance", "schema-check"):
|
||||
var envelope map[string]any
|
||||
if err := json.Unmarshal([]byte(stdin), &envelope); err != nil {
|
||||
t.Fatalf("stdin JSON = %q, err=%v", stdin, err)
|
||||
}
|
||||
if envelope["annotations"] != "reviewed: []\n" || envelope["reviewedCandidates"] != "sha256:"+strings.Repeat("c", 64) {
|
||||
t.Fatalf("annotation envelope = %#v", envelope)
|
||||
}
|
||||
if _, exists := envelope["annotationsPath"]; exists {
|
||||
t.Fatalf("annotation path leaked into stdin: %#v", envelope)
|
||||
}
|
||||
return compose.Result{Stdout: successResult("schema-check")}, nil
|
||||
default:
|
||||
t.Fatalf("unexpected Docker invocation: %#v", args)
|
||||
return compose.Result{}, nil
|
||||
}
|
||||
}}
|
||||
|
||||
_, err := Execute(context.Background(), installation, runner, CheckSchemaRequest{baseRequest: baseRequest{Workspace: "abc"}, Annotations: annotationsPath, ReviewedCandidates: "sha256:" + strings.Repeat("c", 64)})
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteSuggestFksWritesTheReturnedCandidateArtifact(t *testing.T) {
|
||||
installation := testInstallation(t)
|
||||
outputPath := filepath.Join(filepath.Dir(installation.Path), "candidates.yaml")
|
||||
runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) {
|
||||
switch {
|
||||
case contains(args, "config", "--format", "json"):
|
||||
return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil
|
||||
case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}):
|
||||
return compose.Result{Stdout: "sha256:" + strings.Repeat("d", 64)}, nil
|
||||
case contains(args, "workspace-maintenance", "schema-suggest-fks"):
|
||||
return compose.Result{Stdout: `{"schemaVersion":1,"status":"blocked","code":"manual_review_required","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:` + strings.Repeat("e", 64) + `","operation":"schema-suggest-fks","completedStages":[],"suggestedFksYaml":"reviewed: []\n"}`}, nil
|
||||
default:
|
||||
t.Fatalf("unexpected Docker invocation: %#v", args)
|
||||
return compose.Result{}, nil
|
||||
}
|
||||
}}
|
||||
|
||||
_, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, Output: outputPath})
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() error = %v", err)
|
||||
}
|
||||
contents, err := os.ReadFile(outputPath)
|
||||
if err != nil {
|
||||
t.Fatalf("output artifact missing: %v", err)
|
||||
}
|
||||
if string(contents) != "reviewed: []\n" {
|
||||
t.Fatalf("output contents = %q", contents)
|
||||
}
|
||||
}
|
||||
|
||||
func testInstallation(t *testing.T) config.Installation {
|
||||
t.Helper()
|
||||
temporaryRoot, err := filepath.EvalSymlinks(os.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root, err := os.MkdirTemp(temporaryRoot, "thothctl-workspaceops-")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(root) })
|
||||
project := filepath.Join(root, "project")
|
||||
if err := os.MkdirAll(filepath.Join(project, "deploy"), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(project, "compose.yaml"), []byte("services: {}\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(project, "deploy", "compose.local.yaml"), []byte("services: {}\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
envFile := filepath.Join(root, "installation.env")
|
||||
if err := os.WriteFile(envFile, []byte("SAFE_VALUE=1\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return config.Installation{Path: filepath.Join(root, "thothii-installation.yaml"), Profile: "local", ProjectDirectory: project, EnvFile: envFile}
|
||||
}
|
||||
|
||||
func contains(values []string, sequence ...string) bool {
|
||||
for start := range values {
|
||||
if start+len(sequence) <= len(values) && reflect.DeepEqual(values[start:start+len(sequence)], sequence) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func successResult(operation string) string {
|
||||
return `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:` + strings.Repeat("f", 64) + `","operation":"` + operation + `","completedStages":[]}`
|
||||
}
|
||||
Reference in New Issue
Block a user