From c2e8b69a4aa0b0d460af773a7f607ff025e1d887 Mon Sep 17 00:00:00 2001 From: Grigory Panov Date: Wed, 8 Jul 2026 11:51:42 +0200 Subject: [PATCH 1/3] Add local-env six-phase pipeline orchestrator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator runs the layers in order (preflight → resolve → fetch → merge → provision → validate), records per-phase status into Result, and returns typed PipelineErrors carrying FailurePhase and DiskMutated. Under --check it computes and reports the plan (with a diff) and performs no writes — skipping the writability probe and package-manager availability, neither of which is needed to compute the plan. Backups are created only when no .bak exists and the original is a genuine os.ErrNotExist-free read; an unreadable or unstattable existing file fails loudly rather than being misread as greenfield and overwritten. Stacked on the package-manager/detection layer; provisions through the PackageManager interface against a fake in tests. Co-authored-by: Isaac --- libs/localenv/pipeline.go | 525 +++++++++++++++++++++++++++++++++ libs/localenv/pipeline_test.go | 484 ++++++++++++++++++++++++++++++ 2 files changed, 1009 insertions(+) create mode 100644 libs/localenv/pipeline.go create mode 100644 libs/localenv/pipeline_test.go diff --git a/libs/localenv/pipeline.go b/libs/localenv/pipeline.go new file mode 100644 index 00000000000..861edb89b29 --- /dev/null +++ b/libs/localenv/pipeline.go @@ -0,0 +1,525 @@ +package localenv + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/databricks/cli/libs/log" + "github.com/hexops/gotextdiff" + "github.com/hexops/gotextdiff/myers" + "github.com/hexops/gotextdiff/span" +) + +// Filenames and directories the pipeline reads and writes. venvDir is also the +// virtualenv directory the package manager provisions. pyprojectFile lives in +// detect.go, the first consumer of it. +const ( + backupFile = "pyproject.toml.bak" + venvDir = ".venv" +) + +// artifactSource values reported in --json resolved.artifactSource (spec §6). +const ( + artifactNetwork = "network" + artifactCache = "cache" +) + +// allPhases is the canonical phase order for the --json "phases" array. The +// pipeline seeds every phase pending from this slice and flips each to ok/error +// as the run progresses. +var allPhases = []PhaseName{ + PhasePreflight, + PhaseResolve, + PhaseFetch, + PhaseMerge, + PhaseProvision, + PhaseValidate, +} + +// Pipeline orchestrates the dbconnect sync phases against a project directory. +type Pipeline struct { + Mode Mode + Check bool + ProjectDir string + ConstraintBaseURL string + CacheDir string + Flags TargetFlags + Bundle BundleTarget + Compute ComputeClient + PM PackageManager + + // res accumulates phase statuses and result fields as the run progresses. + res *Result +} + +// Run executes all pipeline phases in order and returns a fully populated Result. +// On a phase error, Result.Error is set and the same error is also returned. The +// Result always carries the full canonical phase list: phases completed before a +// failure are "ok", the failing phase is "error", and the rest are "pending". +func (p *Pipeline) Run(ctx context.Context) (*Result, error) { + log.Debugf(ctx, "local-env: mode=%s check=%v project=%s cacheDir=%s constraintBaseURL=%s flags=%+v", + p.Mode, + p.Check, + filepath.ToSlash(p.ProjectDir), + filepath.ToSlash(p.CacheDir), + p.ConstraintBaseURL, + p.Flags, + ) + + // NewResult seeds Phases/Warnings to non-nil slices so --json always emits + // [] not null; override Phases with the canonical pending phase list. + p.res = NewResult() + p.res.SchemaVersion = SchemaVersion + p.res.Command = CommandName + p.res.Mode = p.Mode.String() + p.res.DryRun = p.Check + // Phases start as pending and flip to ok/error as the run progresses. + p.res.Phases = initialPhases() + + if err := p.run(ctx); err != nil { + return p.res, err + } + p.res.OK = true + return p.res, nil +} + +// run drives the phases and returns the first phase error. Result bookkeeping +// (phase status, error object) is handled by fail / markOK. +func (p *Pipeline) run(ctx context.Context) error { + // Phase: preflight — manager detection, writability, package-manager availability. + // P0 supports only uv; any other detected manager is a clean, non-blaming exit. + if m := detectManager(p.ProjectDir); m != managerUv { + return p.fail(PhasePreflight, false, NewError(ErrManagerUnsupported, nil, "%s", managerGuidance(m))) + } + // Under --check the pipeline only reads and reports a plan, so it must not + // mutate anything at preflight. Two preflight steps can write: + // - ensureWritable creates and removes a temp file (and would fail a + // read-only project the user only wants to inspect); + // - PackageManager.EnsureAvailable may install the manager (uv) if missing. + // Both exist to fail fast before real writes, which --check never performs, so + // they are skipped in a dry run. Neither result is needed to compute the plan. + if p.Check { + p.markOK(PhasePreflight, "check") + } else { + if err := ensureWritable(p.ProjectDir); err != nil { + return p.fail(PhasePreflight, false, NewError(ErrNotWritable, err, "project directory %s is not writable", filepath.ToSlash(p.ProjectDir))) + } + version, err := p.PM.EnsureAvailable(ctx) + if err != nil { + return p.fail(PhasePreflight, false, asPipelineError(err, ErrUvMissing, "%s unavailable", p.PM.Name())) + } + p.markOK(PhasePreflight, p.PM.Name()+" "+version) + } + + // Phase: resolve — compute target → environment key. + target, err := p.resolve(ctx) + if err != nil { + return err + } + + // Phase: fetch — constraint artifact for the resolved env key. + c, err := p.fetch(ctx, target) + if err != nil { + return err + } + + // Parse the required Python minor from the artifact; a failure here reflects a + // bad artifact, reported at the fetch phase. + pyMinor, err := PythonMinorFromRequires(c.RequiresPython) + if err != nil { + return p.fail(PhaseFetch, false, NewError(ErrFetch, err, "cannot parse python version from constraints %q", c.RequiresPython)) + } + + dbcPin := c.DatabricksConnect + if p.Mode == ModeConstraintsOnly { + // constraints-only omits the databricks-connect dependency entirely. + dbcPin = "" + } + p.res.Resolved = &ResolvedInfo{ + PythonVersion: pyMinor, + DBConnectVersion: versionFromPin(dbcPin), + ArtifactSource: artifactSource(c.FromCache), + } + + // Phase: merge — compute the merged pyproject.toml (in-memory, no writes yet). + mergedBytes, greenfield, err := p.mergePlan(ctx, pyMinor, c, dbcPin) + if err != nil { + return err + } + p.res.Greenfield = greenfield + + // Check mode stops after planning — nothing below mutates disk. + if p.Check { + p.markOK(PhaseMerge, "") + p.markOK(PhaseProvision, "") + p.markOK(PhaseValidate, "") + return nil + } + + // Apply the merged content to disk (backup first for existing projects). + if err := p.applyMerge(ctx, mergedBytes, greenfield); err != nil { + return err + } + p.markOK(PhaseMerge, "") + + // Phase: provision — ensure Python, run uv sync, seed pip. + if err := p.provision(ctx, pyMinor); err != nil { + return err + } + + // Phase: validate — assert the venv matches the target. + return p.validate(ctx, pyMinor, dbcPin) +} + +// resolve runs ResolveTarget and records the resolve phase. +func (p *Pipeline) resolve(ctx context.Context) (*TargetInfo, error) { + target, err := ResolveTarget(ctx, p.Flags, p.Compute, p.Bundle) + if err != nil { + return nil, p.fail(PhaseResolve, false, asPipelineError(err, ErrResolve, "target resolution failed")) + } + p.res.Target = target + p.markOK(PhaseResolve, fmt.Sprintf("source=%s envKey=%s", target.Source, target.EnvKey)) + return target, nil +} + +// fetch fetches constraints for the resolved target and records the fetch phase. +func (p *Pipeline) fetch(ctx context.Context, target *TargetInfo) (*Constraints, error) { + c, err := FetchConstraints(ctx, p.ConstraintBaseURL, target.EnvKey, p.CacheDir) + if err != nil { + // FetchConstraints classifies the cause: E_ENV_UNSUPPORTED for a missing + // key (404) versus E_FETCH for transport failure with no cache. Both are + // discovered here, so both are reported at the fetch phase (the JSON keeps + // the failing phase and error.failurePhase in agreement). + return nil, p.fail(PhaseFetch, false, asPipelineError(err, ErrFetch, "fetch constraints failed")) + } + p.markOK(PhaseFetch, fmt.Sprintf("source=%s fromCache=%v", c.SourceURL, c.FromCache)) + return c, nil +} + +// pyprojectPath returns the path to pyproject.toml in the project directory. +func (p *Pipeline) pyprojectPath() string { + return filepath.Join(p.ProjectDir, pyprojectFile) +} + +// backupPath returns the path to the pyproject.toml backup file. +func (p *Pipeline) backupPath() string { + return filepath.Join(p.ProjectDir, backupFile) +} + +// mergePlan computes the merged pyproject.toml bytes (without writing to disk), +// decides greenfield vs. existing, and builds the Plan (populated only under +// --check). dbcPin is the databricks-connect pin to inject, or "" in +// constraints-only mode. +func (p *Pipeline) mergePlan(_ context.Context, pyMinor string, c *Constraints, dbcPin string) (merged []byte, greenfield bool, err error) { + pyproject := p.pyprojectPath() + backup := p.backupPath() + + // Determine base bytes for the merge. For an existing project with a backup, + // the backup is the canonical base so the merge starts from the original + // unmanaged state. + var baseBytes []byte + if data, rerr := os.ReadFile(backup); rerr == nil { + baseBytes = data + } else if !errors.Is(rerr, os.ErrNotExist) { + // A backup that exists but can't be read (permissions, I/O) must not be + // silently ignored: falling through would treat the project as greenfield + // and overwrite it. Fail loudly instead. + return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read backup %s failed", filepath.ToSlash(backup))) + } + if baseBytes == nil { + data, rerr := os.ReadFile(pyproject) + switch { + case rerr == nil: + baseBytes = data + case !errors.Is(rerr, os.ErrNotExist): + // Only a genuine not-exist means greenfield. Any other read error on an + // existing pyproject.toml (permission change, transient I/O, delete race + // after detectManager saw it) must not be misread as greenfield — that + // would render a fresh file and overwrite the user's project with no + // backup. Fail instead of destroying unrecoverable state (invariant 2). + return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read pyproject.toml %s failed", filepath.ToSlash(pyproject))) + } + } + greenfield = baseBytes == nil + + // The artifact drives the merge; in constraints-only mode we clear the + // databricks-connect pin so it is neither written nor asserted. + effective := *c + effective.DatabricksConnect = dbcPin + + var changedRegions []string + if greenfield { + // No existing pyproject.toml — render a fresh one. The project name comes + // from the directory name as a reasonable default. + projectName := filepath.Base(p.ProjectDir) + merged = RenderFreshPyproject(projectName, effective) + changedRegions = []string{regionRequiresPython, regionToolUv} + if dbcPin != "" { + changedRegions = append(changedRegions, regionDatabricksConnect) + } + } else { + merged, changedRegions, err = MergeManaged(baseBytes, effective) + if err != nil { + return nil, greenfield, p.fail(PhaseMerge, false, NewError(ErrMerge, err, "merge managed regions failed")) + } + } + + // Under --check, build the plan (with a diff) for reporting. A real run does + // not need the diff. + if p.Check { + oldStr := "" + newStr := string(merged) + oldName := pyprojectFile + newName := pyprojectFile + if !greenfield { + oldStr = string(baseBytes) + newName = pyprojectFile + ".new" + } + edits := myers.ComputeEdits(span.URIFromPath(oldName), oldStr, newStr) + diff := fmt.Sprint(gotextdiff.ToUnified(oldName, newName, oldStr, edits)) + + plan := &Plan{ + WouldWrite: filepath.ToSlash(pyproject), + Diff: diff, + ChangedRegions: changedRegions, + WouldInstallPython: pyMinor, + } + if !greenfield { + plan.WouldBackup = filepath.ToSlash(backup) + } + p.res.Plan = plan + } + return merged, greenfield, nil +} + +// applyMerge writes the merged bytes to disk, backing up an existing +// pyproject.toml first. From this point on, disk has been mutated. +func (p *Pipeline) applyMerge(_ context.Context, mergedBytes []byte, greenfield bool) error { + pyproject := p.pyprojectPath() + backup := p.backupPath() + + if !greenfield { + // Back up before modifying so the user's original is recoverable + // (invariant 2). Only create the backup when one does not already exist: + // on a re-run the existing .bak is the canonical original unmanaged state + // (mergePlan used it as the base), so overwriting it with the already-merged + // pyproject.toml would destroy that baseline. + _, statErr := os.Stat(backup) + switch { + case statErr == nil: + // Backup already exists — keep it as the canonical baseline. + case errors.Is(statErr, os.ErrNotExist): + // copyFile creates/truncates the backup path, so a failure mid-copy may + // leave a partial .bak: report disk as mutated. + if err := copyFile(pyproject, backup); err != nil { + return p.fail(PhaseMerge, true, NewError(ErrMerge, err, "backup pyproject.toml failed")) + } + default: + // An existing-but-unstattable backup must not be overwritten (that would + // destroy the recoverable original); fail before any write instead. + return p.fail(PhaseMerge, false, NewError(ErrMerge, statErr, "cannot stat backup %s", filepath.ToSlash(backup))) + } + p.res.BackupPath = filepath.ToSlash(backup) + } + + if err := os.WriteFile(pyproject, mergedBytes, 0o644); err != nil { + code := ErrMerge + if greenfield { + code = ErrWrite + } + return p.fail(PhaseMerge, true, NewError(code, err, "write pyproject.toml failed")) + } + return nil +} + +// provision ensures the required Python version is installed, runs uv sync, and +// seeds pip. All three are reported under the provision phase. +func (p *Pipeline) provision(ctx context.Context, pyMinor string) error { + if err := p.PM.EnsurePython(ctx, pyMinor); err != nil { + return p.fail(PhaseProvision, true, asPipelineError(err, ErrPythonInstall, "ensure python %s failed", pyMinor)) + } + if err := p.PM.Provision(ctx, p.ProjectDir); err != nil { + return p.fail(PhaseProvision, true, asPipelineError(err, ErrProvision, "provision failed")) + } + if err := p.PM.PostProvision(ctx, p.ProjectDir); err != nil { + return p.fail(PhaseProvision, true, asPipelineError(err, ErrProvision, "post-provision failed")) + } + p.markOK(PhaseProvision, "") + return nil +} + +// validate reads the Python and databricks-connect versions from the venv and +// populates the venv path. dbcPin is "" in constraints-only mode, where the DB +// Connect assertion is skipped. +func (p *Pipeline) validate(ctx context.Context, expectedPyMinor, dbcPin string) error { + pyVer, dbcVer, err := p.PM.Validate(ctx, p.ProjectDir) + if err != nil { + return p.fail(PhaseValidate, true, asPipelineError(err, ErrValidate, "validation failed")) + } + + // Assert the installed Python minor matches the target. + if pyVer != expectedPyMinor { + return p.fail(PhaseValidate, true, NewError(ErrValidate, nil, + "python version mismatch: want %s, got %s", expectedPyMinor, pyVer)) + } + + // In default mode, assert the installed databricks-connect major matches the + // pin's major. dbcPin is e.g. "databricks-connect~=17.2.0"; dbcVer is "17.2.0". + if dbcPin != "" { + pinMajor := dbcMajorFromPin(dbcPin) + if pinMajor == "" { + return p.fail(PhaseValidate, true, NewError(ErrValidate, nil, + "cannot determine databricks-connect major version from pin %q", dbcPin)) + } + installedMajor := majorVersion(dbcVer) + if installedMajor == "" { + return p.fail(PhaseValidate, true, NewError(ErrValidate, nil, + "cannot determine installed databricks-connect major version from %q", dbcVer)) + } + if pinMajor != installedMajor { + return p.fail(PhaseValidate, true, NewError(ErrValidate, nil, + "databricks-connect major version mismatch: want %s.x, got %s", pinMajor, dbcVer)) + } + } + + // Report the installed databricks-connect version only in default mode. In + // constraints-only mode databricks-connect is not a managed dependency, so the + // spec omits dbconnectVersion even if the package is present transitively. + defaultMode := dbcPin != "" + + detail := "python=" + pyVer + if defaultMode && dbcVer != "" { + detail += " databricks-connect=" + dbcVer + } + p.markOK(PhaseValidate, detail) + + p.res.VenvPath = filepath.ToSlash(filepath.Join(p.ProjectDir, venvDir)) + if p.res.Resolved != nil { + if defaultMode { + p.res.Resolved.DBConnectVersion = dbcVer + } else { + p.res.Resolved.DBConnectVersion = "" + } + } + return nil +} + +// initialPhases returns the canonical phase list with every phase pending. +func initialPhases() []PhaseStatus { + phases := make([]PhaseStatus, len(allPhases)) + for i, name := range allPhases { + phases[i] = PhaseStatus{Phase: name, Status: StatusPending} + } + return phases +} + +// markOK marks a phase ok with an optional human-readable detail. +func (p *Pipeline) markOK(name PhaseName, detail string) { + for i := range p.res.Phases { + if p.res.Phases[i].Phase == name { + p.res.Phases[i].Status = StatusOK + p.res.Phases[i].Detail = detail + return + } + } +} + +// fail marks the given phase as errored, attaches the error (with its phase and +// disk-mutation flag) to the Result, and returns it. Phases after the failing +// one remain pending. +func (p *Pipeline) fail(phase PhaseName, diskMutated bool, pe *PipelineError) error { + pe.FailurePhase = phase + pe.DiskMutated = diskMutated + for i := range p.res.Phases { + if p.res.Phases[i].Phase == phase { + p.res.Phases[i].Status = StatusError + p.res.Phases[i].Detail = pe.Error() + break + } + } + p.res.Error = pe + return pe +} + +// asPipelineError returns err as a *PipelineError if it already is one, otherwise +// wraps it with the fallback code and message. +func asPipelineError(err error, fallback ErrorCode, format string, args ...any) *PipelineError { + if pe, ok := errors.AsType[*PipelineError](err); ok { + return pe + } + return NewError(fallback, err, format, args...) +} + +// artifactSource maps the from-cache flag to the JSON artifactSource value. +func artifactSource(fromCache bool) string { + if fromCache { + return artifactCache + } + return artifactNetwork +} + +// versionFromPin extracts the bare version from a dependency pin such as +// "databricks-connect~=17.2.0" → "17.2.0". Returns "" when pin is empty or has +// no version component. +func versionFromPin(pin string) string { + for i, c := range pin { + if c >= '0' && c <= '9' { + return pin[i:] + } + } + return "" +} + +// dbcMajorFromPin extracts the major version number from a databricks-connect +// pin string such as "databricks-connect~=17.2.0". Returns "" if unparseable. +func dbcMajorFromPin(pin string) string { + return majorVersion(versionFromPin(pin)) +} + +// majorVersion returns the major portion of a version string (digits before the +// first dot), e.g. "17" from "17.2.0". A bare integer like "17" returns "17". +// Returns "" for an empty string or a non-numeric major component, so the +// validate phase rejects a malformed version rather than comparing arbitrary +// strings as major versions. +func majorVersion(v string) string { + if v == "" { + return "" + } + before, _, ok := strings.Cut(v, ".") + if !ok { + before = v + } + if before == "" || !isAllDigits(before) { + return "" + } + return before +} + +// isAllDigits reports whether s is non-empty and every rune is an ASCII digit. +func isAllDigits(s string) bool { + if s == "" { + return false + } + for _, c := range s { + if c < '0' || c > '9' { + return false + } + } + return true +} + +// copyFile copies src to dst, creating or overwriting dst. +func copyFile(src, dst string) error { + data, err := os.ReadFile(src) + if err != nil { + return fmt.Errorf("read %s: %w", src, err) + } + if err := os.WriteFile(dst, data, 0o644); err != nil { + return fmt.Errorf("write %s: %w", dst, err) + } + return nil +} diff --git a/libs/localenv/pipeline_test.go b/libs/localenv/pipeline_test.go new file mode 100644 index 00000000000..fdc76fbd248 --- /dev/null +++ b/libs/localenv/pipeline_test.go @@ -0,0 +1,484 @@ +package localenv + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakePM struct{ py, dbc string } + +func (fakePM) Name() string { return "fake" } +func (fakePM) EnsureAvailable(context.Context) (string, error) { return "fake 1.0", nil } +func (fakePM) EnsurePython(context.Context, string) error { return nil } +func (fakePM) Provision(context.Context, string) error { return nil } +func (fakePM) PostProvision(context.Context, string) error { return nil } +func (f fakePM) Validate(context.Context, string) (string, string, error) { + return f.py, f.dbc, nil +} + +// noProvisionPM fails any method that could touch the machine (install the +// manager, install Python, sync, seed pip, validate). It asserts that --check +// never reaches those write-side operations. +type noProvisionPM struct{} + +func (noProvisionPM) Name() string { return "noprov" } +func (noProvisionPM) EnsureAvailable(context.Context) (string, error) { + return "", errors.New("EnsureAvailable must not be called under --check") +} + +func (noProvisionPM) EnsurePython(context.Context, string) error { + return errors.New("EnsurePython must not be called under --check") +} + +func (noProvisionPM) Provision(context.Context, string) error { + return errors.New("Provision must not be called under --check") +} + +func (noProvisionPM) PostProvision(context.Context, string) error { + return errors.New("PostProvision must not be called under --check") +} + +func (noProvisionPM) Validate(context.Context, string) (string, string, error) { + return "", "", errors.New("Validate must not be called under --check") +} + +func writeProject(t *testing.T) string { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte(`[project] +name = "demo" +requires-python = ">=3.10" + +[dependency-groups] +dev = ["databricks-connect~=16.0.0"] +`), 0o644)) + return dir +} + +func newTestServer(t *testing.T) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(sampleToml)) + })) +} + +func TestPipelineCheckMutatesNothing(t *testing.T) { + dir := writeProject(t) + before, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, Check: true, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.DryRun) + assert.True(t, res.OK) + require.NotNil(t, res.Plan) + assert.Contains(t, res.Plan.Diff, "==3.12.*") + after, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + assert.Equal(t, string(before), string(after)) // unchanged +} + +func TestPipelineCheckDoesNotProvision(t *testing.T) { + // --check must not call any PackageManager method that could mutate the + // machine (EnsureAvailable may install uv). noProvisionPM errors on all of + // them; the dry run must still succeed and produce a plan. + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, Check: true, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: noProvisionPM{}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + require.NotNil(t, res.Plan) +} + +func TestPipelineCheckWorksOnReadOnlyDir(t *testing.T) { + if runtime.GOOS == "windows" || os.Getuid() == 0 { + t.Skip("read-only-dir enforcement not available") + } + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + // Make the project dir read-only: --check must still compute the plan without + // a writability probe (which would both mutate disk and fail here). + require.NoError(t, os.Chmod(dir, 0o555)) + t.Cleanup(func() { _ = os.Chmod(dir, 0o755) }) + + p := &Pipeline{ + Mode: ModeDefault, Check: true, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + require.NotNil(t, res.Plan) + // No writecheck temp file was left behind in the project dir. + entries, err := os.ReadDir(dir) + require.NoError(t, err) + for _, e := range entries { + assert.NotContains(t, e.Name(), "writecheck", "dry run must not create temp files") + } +} + +func TestPipelineProvisionsAndValidatesExisting(t *testing.T) { + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + assert.False(t, res.Greenfield) + require.NotNil(t, res.Resolved) + assert.Equal(t, "3.12", res.Resolved.PythonVersion) + assert.Equal(t, "17.2.0", res.Resolved.DBConnectVersion) + assert.Equal(t, ".venv", filepath.Base(res.VenvPath)) + merged, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + assert.Contains(t, string(merged), `"databricks-connect~=17.2.0"`) + assert.FileExists(t, filepath.Join(dir, "pyproject.toml.bak")) +} + +func TestPipelineGreenfieldCreatesNewPyproject(t *testing.T) { + dir := t.TempDir() + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + assert.True(t, res.Greenfield) + data, readErr := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + require.NoError(t, readErr) + assert.Contains(t, string(data), `"databricks-connect~=17.2.0",`) + // No backup created when pyproject.toml did not previously exist. + assert.NoFileExists(t, filepath.Join(dir, "pyproject.toml.bak")) +} + +func TestPipelineExistingBacksUp(t *testing.T) { + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + assert.FileExists(t, filepath.Join(dir, "pyproject.toml.bak")) + assert.Equal(t, "pyproject.toml.bak", filepath.Base(res.BackupPath)) +} + +func TestApplyMergeFailsOnUnstattableBackupWithoutOverwrite(t *testing.T) { + if runtime.GOOS == "windows" || os.Getuid() == 0 { + // chmod-based stat blocking does not apply for root or on Windows. + t.Skip("stat-permission enforcement not available") + } + // Both pyproject.toml and its .bak live in a project dir that is made + // unsearchable, so os.Stat of the backup fails with a permission error rather + // than not-exist — isolating applyMerge's "can't stat" branch. applyMerge is + // called directly, bypassing the writability preflight. + dir := filepath.Join(t.TempDir(), "proj") + require.NoError(t, os.Mkdir(dir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte("[project]\n"), 0o644)) + backup := filepath.Join(dir, "pyproject.toml.bak") + require.NoError(t, os.WriteFile(backup, []byte("ORIGINAL BACKUP\n"), 0o644)) + require.NoError(t, os.Chmod(dir, 0o000)) + t.Cleanup(func() { _ = os.Chmod(dir, 0o755) }) + + p := &Pipeline{ProjectDir: dir, res: &Result{Phases: initialPhases()}} + err := p.applyMerge(t.Context(), []byte("merged"), false) + require.Error(t, err) + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, PhaseMerge, pe.FailurePhase) + assert.False(t, pe.DiskMutated, "no write should have happened before the stat check") + + // The original backup must be intact. + require.NoError(t, os.Chmod(dir, 0o755)) + got, readErr := os.ReadFile(backup) + require.NoError(t, readErr) + assert.Equal(t, "ORIGINAL BACKUP\n", string(got), "the canonical backup must not be overwritten") +} + +func TestPipelineConstraintsOnlyOmitsDBConnect(t *testing.T) { + dir := t.TempDir() + srv := newTestServer(t) + defer srv.Close() + + // Even though databricks-connect is present in the venv (fakePM reports a + // version), constraints-only must not assert it, not write the pin, and not + // report dbconnectVersion (spec §6 omits it in this mode). + p := &Pipeline{ + Mode: ModeConstraintsOnly, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + assert.True(t, res.OK) + assert.Equal(t, "constraints-only", res.Mode) + require.NotNil(t, res.Resolved) + assert.Empty(t, res.Resolved.DBConnectVersion, "constraints-only must omit dbconnectVersion") + data, readErr := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + require.NoError(t, readErr) + assert.NotContains(t, string(data), "databricks-connect") + // The dev group is emitted empty, not with a blank-string entry. + assert.Contains(t, string(data), "dev = []") + assert.NotContains(t, string(data), `""`) +} + +func TestPipelineNoTarget(t *testing.T) { + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{}, + Compute: stubCompute{}, PM: fakePM{}, + } + res, err := p.Run(t.Context()) + require.Error(t, err) + require.NotNil(t, res) + require.NotNil(t, res.Error) + assert.False(t, res.OK) + assert.Equal(t, ErrNoTarget, res.Error.Code) + // No-target is detected during target resolution, so it is reported at the + // resolve phase (the failing phase in phases[] and error.failurePhase agree). + assert.Equal(t, PhaseResolve, res.Error.FailurePhase) + // Preflight succeeded, resolve errored, everything after stays pending. + assert.Equal(t, StatusOK, phaseStatus(res, PhasePreflight)) + assert.Equal(t, StatusError, phaseStatus(res, PhaseResolve)) + assert.Equal(t, StatusPending, phaseStatus(res, PhaseProvision)) +} + +func TestPipelineRestoresBackupBeforeMerge(t *testing.T) { + dir := t.TempDir() + // Write an original pyproject.toml and a pre-existing .bak. + original := []byte(`[project] +name = "demo" +requires-python = ">=3.9" + +[dependency-groups] +dev = ["databricks-connect~=15.0.0"] +`) + require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml.bak"), original, 0o644)) + // Current pyproject.toml has been mutated by a previous run. + mutated := []byte(`[project] +name = "demo" +requires-python = "==3.12.*" + +[dependency-groups] +dev = ["databricks-connect~=17.2.0"] +`) + require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), mutated, 0o644)) + + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + require.NotNil(t, res) + // The bak content (requires-python = ">=3.9") was the base; merged result should + // contain the newly pinned version. + data, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + assert.Contains(t, string(data), `"databricks-connect~=17.2.0"`) + assert.Contains(t, string(data), `requires-python = "==3.12.*"`) +} + +func TestPipelineUnreadableExistingIsNotTreatedAsGreenfield(t *testing.T) { + if runtime.GOOS == "windows" || os.Getuid() == 0 { + // chmod 000 does not block reads for root or on Windows. + t.Skip("read-permission enforcement not available") + } + dir := t.TempDir() + pyproject := filepath.Join(dir, "pyproject.toml") + original := []byte(`[project] +name = "demo" +requires-python = ">=3.10" + +[dependency-groups] +dev = ["databricks-connect~=16.0.0"] +`) + require.NoError(t, os.WriteFile(pyproject, original, 0o644)) + // Make it unreadable so os.ReadFile fails with a non-not-exist error after + // detectManager has already seen the file exists. + require.NoError(t, os.Chmod(pyproject, 0o000)) + t.Cleanup(func() { _ = os.Chmod(pyproject, 0o644) }) + + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + // The run must fail at merge rather than silently overwrite the user's file. + require.Error(t, err) + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, ErrMerge, pe.Code) + assert.Equal(t, PhaseMerge, pe.FailurePhase) + assert.False(t, res.Greenfield, "an unreadable existing file must not be treated as greenfield") + + // The original file must be untouched (still unreadable, not overwritten). + require.NoError(t, os.Chmod(pyproject, 0o644)) + after, readErr := os.ReadFile(pyproject) + require.NoError(t, readErr) + assert.Equal(t, string(original), string(after), "the user's pyproject.toml must not be overwritten") +} + +func TestPipelineResultPopulatesResolved(t *testing.T) { + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, Check: true, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.NoError(t, err) + require.NotNil(t, res.Resolved) + assert.Equal(t, "3.12", res.Resolved.PythonVersion) + assert.Equal(t, "17.2.0", res.Resolved.DBConnectVersion) + assert.Equal(t, "network", res.Resolved.ArtifactSource) +} + +// newServerWithDBC returns a test server that serves a constraints TOML with the +// given databricks-connect pin value in the dev dependency group. +func newServerWithDBC(t *testing.T, dbcPin string) *httptest.Server { + t.Helper() + body := `[project] +requires-python = "==3.12.*" + +[dependency-groups] +dev = ["` + dbcPin + `"] + +[tool.uv] +constraint-dependencies = [] +` + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(body)) + })) +} + +func TestPipelineValidateRejectsUnparseablePin(t *testing.T) { + dir := writeProject(t) + // Serve a TOML whose dev group has a malformed databricks-connect entry + // (no version digits after the package name). + srv := newServerWithDBC(t, "databricks-connect") + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + require.Error(t, err) + require.NotNil(t, res.Error) + assert.Equal(t, ErrValidate, res.Error.Code) + assert.Equal(t, PhaseValidate, res.Error.FailurePhase) +} + +func TestPipelineValidateRejectsUnparseableInstalledVersion(t *testing.T) { + dir := writeProject(t) + // sampleToml has databricks-connect~=17.2.0 as the pin; use an empty installed + // version string to simulate an installed version that can't be parsed. + srv := newTestServer(t) + defer srv.Close() + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: ""}, + } + res, err := p.Run(t.Context()) + require.Error(t, err) + require.NotNil(t, res.Error) + assert.Equal(t, ErrValidate, res.Error.Code) +} + +func TestMajorVersion(t *testing.T) { + cases := []struct { + input string + want string + }{ + {"17.2.0", "17"}, + {"17", "17"}, + {"", ""}, + {"3.12", "3"}, + // Non-numeric major components are rejected (empty) so validation does + // not compare arbitrary strings as versions. + {"bad.version", ""}, + {"v17.2", ""}, + {"abc", ""}, + } + for _, tc := range cases { + assert.Equal(t, tc.want, majorVersion(tc.input), "input=%q", tc.input) + } +} + +// phaseStatus returns the status recorded for the named phase in res. +func phaseStatus(res *Result, name PhaseName) string { + for _, ph := range res.Phases { + if ph.Phase == name { + return ph.Status + } + } + return "" +} From 6f315797059cff0429adc5f254310420ac94cfa3 Mon Sep 17 00:00:00 2001 From: Grigory Panov Date: Thu, 9 Jul 2026 16:05:15 +0200 Subject: [PATCH 2/3] Preserve pyproject.toml mode in backup; test preflight exits Address review on the pipeline PR: - copyFile created the .bak at a hardcoded 0o644, silently widening a locked-down pyproject.toml (e.g. 0o600 carrying a private index URL) to world-readable. Stat the source and create the backup with its permission bits; os.WriteFile applies the mode only when creating the file, which is always the case for the fresh .bak. - Add pipeline-level tests for the two preflight exits that were only covered in isolation before: E_MANAGER_UNSUPPORTED (a non-uv project) and E_UV_MISSING (package manager unavailable), asserting phase attribution, DiskMutated=false, and that later phases stay pending. Add a direct copyFile mode-preservation test. Co-authored-by: Isaac --- libs/localenv/pipeline.go | 12 +++++- libs/localenv/pipeline_test.go | 78 ++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) diff --git a/libs/localenv/pipeline.go b/libs/localenv/pipeline.go index 861edb89b29..c1dab760abd 100644 --- a/libs/localenv/pipeline.go +++ b/libs/localenv/pipeline.go @@ -512,13 +512,21 @@ func isAllDigits(s string) bool { return true } -// copyFile copies src to dst, creating or overwriting dst. +// copyFile copies src to dst, creating or overwriting dst. dst is created with +// src's permission bits: the backup preserves a locked-down pyproject.toml +// (e.g. 0o600 because it carries a private index URL) rather than widening it to +// a hardcoded 0o644. os.WriteFile only applies the mode when it creates the +// file, which is always the case for the freshly-created .bak. func copyFile(src, dst string) error { + info, err := os.Stat(src) + if err != nil { + return fmt.Errorf("stat %s: %w", src, err) + } data, err := os.ReadFile(src) if err != nil { return fmt.Errorf("read %s: %w", src, err) } - if err := os.WriteFile(dst, data, 0o644); err != nil { + if err := os.WriteFile(dst, data, info.Mode().Perm()); err != nil { return fmt.Errorf("write %s: %w", dst, err) } return nil diff --git a/libs/localenv/pipeline_test.go b/libs/localenv/pipeline_test.go index fdc76fbd248..e284e3b50a9 100644 --- a/libs/localenv/pipeline_test.go +++ b/libs/localenv/pipeline_test.go @@ -51,6 +51,15 @@ func (noProvisionPM) Validate(context.Context, string) (string, string, error) { return "", "", errors.New("Validate must not be called under --check") } +// uvMissingPM fails EnsureAvailable, simulating a machine where the package +// manager binary is absent and cannot be installed. The remaining methods +// inherit fakePM but must never be reached, since preflight aborts first. +type uvMissingPM struct{ fakePM } + +func (uvMissingPM) EnsureAvailable(context.Context) (string, error) { + return "", errors.New("uv not found and install failed") +} + func writeProject(t *testing.T) string { dir := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte(`[project] @@ -206,6 +215,75 @@ func TestPipelineExistingBacksUp(t *testing.T) { assert.Equal(t, "pyproject.toml.bak", filepath.Base(res.BackupPath)) } +func TestCopyFilePreservesMode(t *testing.T) { + if runtime.GOOS == "windows" { + // Windows does not honor Unix permission bits. + t.Skip("permission-bit preservation is Unix-only") + } + // A locked-down pyproject.toml (e.g. 0o600 because it carries a private index + // URL) must not be widened when copied to the backup. + dir := t.TempDir() + src := filepath.Join(dir, "pyproject.toml") + require.NoError(t, os.WriteFile(src, []byte("[project]\n"), 0o600)) + dst := filepath.Join(dir, "pyproject.toml.bak") + require.NoError(t, copyFile(src, dst)) + info, err := os.Stat(dst) + require.NoError(t, err) + assert.Equal(t, os.FileMode(0o600), info.Mode().Perm()) +} + +// assertPreflightFailure checks the invariants shared by every preflight exit: +// the run returns the error, preflight is attributed as the failing phase, no +// disk mutation is claimed, and every later phase is left pending. +func assertPreflightFailure(t *testing.T, res *Result, err error, wantCode ErrorCode) { + t.Helper() + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, wantCode, pe.Code) + assert.Equal(t, PhasePreflight, pe.FailurePhase) + assert.False(t, pe.DiskMutated, "preflight fails before any write") + assert.False(t, res.OK) + for _, ph := range res.Phases { + switch ph.Phase { + case PhasePreflight: + assert.Equal(t, StatusError, ph.Status) + default: + assert.Equal(t, StatusPending, ph.Status, "phase %s must stay pending after a preflight failure", ph.Phase) + } + } +} + +func TestPipelineManagerUnsupportedFailsAtPreflight(t *testing.T) { + // A non-uv project (environment.yml, no pyproject.toml) is detected as conda + // and must exit cleanly at preflight with E_MANAGER_UNSUPPORTED. + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "environment.yml"), []byte("name: demo\n"), 0o644)) + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + res, err := p.Run(t.Context()) + assertPreflightFailure(t, res, err, ErrManagerUnsupported) + // Detection is pure-fs; no pyproject.toml was created. + assert.NoFileExists(t, filepath.Join(dir, "pyproject.toml")) +} + +func TestPipelineUvMissingFailsAtPreflight(t *testing.T) { + // When the package manager can't be made available, preflight exits with + // E_UV_MISSING before resolve/fetch/merge run. + dir := writeProject(t) + + p := &Pipeline{ + Mode: ModeDefault, ProjectDir: dir, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: uvMissingPM{}, + } + res, err := p.Run(t.Context()) + assertPreflightFailure(t, res, err, ErrUvMissing) +} + func TestApplyMergeFailsOnUnstattableBackupWithoutOverwrite(t *testing.T) { if runtime.GOOS == "windows" || os.Getuid() == 0 { // chmod-based stat blocking does not apply for root or on Windows. From 2ef43e0b6f1776489b1023f4f4f86d9f66d6a19a Mon Sep 17 00:00:00 2001 From: Grigory Panov Date: Fri, 10 Jul 2026 15:36:39 +0200 Subject: [PATCH 3/3] Merge on the live pyproject.toml, not the backup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on the pipeline PR (concerns #1 and #2). Re-basing the merge on .bak silently discarded edits the user made to pyproject.toml between syncs: mergePlan used .bak as the merge base whenever one existed, so a re-run started from the pristine pre-sync original and any dependency the user added afterward was lost. MergeManaged rewrites only the three managed regions and is idempotent on its own output, so merging onto the live file yields identical managed regions without throwing away edits. - mergePlan now reads the live pyproject.toml as the merge base; .bak stays a one-time safety copy of the pre-sync original. - --check no longer reports a backup on a re-run (applyMerge keeps the existing .bak, so WouldBackup is set only when no .bak exists yet), and the diff is now against the live file, so an idempotent re-run reports no change — matching what a real run would do. - Replace TestPipelineRestoresBackupBeforeMerge (which encoded the old discard-edits behavior) with TestPipelineMergesOnLiveFileNotBackup, and add TestPipelineCheckReRunPlanMatchesRealRun. Also clarify the constraints-only comment: it stops managing the databricks-connect pin (greenfield dev = []; existing pin retained) rather than removing it, matching mergeDatabricksConnect's no-op-on-empty behavior. Co-authored-by: Isaac --- libs/localenv/pipeline.go | 54 ++++++++++++++++++---------------- libs/localenv/pipeline_test.go | 51 +++++++++++++++++++++++++++----- 2 files changed, 73 insertions(+), 32 deletions(-) diff --git a/libs/localenv/pipeline.go b/libs/localenv/pipeline.go index c1dab760abd..e3f152d7304 100644 --- a/libs/localenv/pipeline.go +++ b/libs/localenv/pipeline.go @@ -136,7 +136,11 @@ func (p *Pipeline) run(ctx context.Context) error { dbcPin := c.DatabricksConnect if p.Mode == ModeConstraintsOnly { - // constraints-only omits the databricks-connect dependency entirely. + // constraints-only stops *managing* the databricks-connect pin rather than + // removing it. Clearing dbcPin means the merge neither injects nor asserts a + // pin: greenfield renders dev = [] (no databricks-connect), while an existing + // project that already pins databricks-connect keeps its pin untouched (see + // mergeDatabricksConnect — an empty value is a no-op, not a deletion). dbcPin = "" } p.res.Resolved = &ResolvedInfo{ @@ -218,31 +222,25 @@ func (p *Pipeline) mergePlan(_ context.Context, pyMinor string, c *Constraints, pyproject := p.pyprojectPath() backup := p.backupPath() - // Determine base bytes for the merge. For an existing project with a backup, - // the backup is the canonical base so the merge starts from the original - // unmanaged state. + // The merge base is the live pyproject.toml, not the backup. MergeManaged + // rewrites only the three managed regions and preserves every other byte, and + // it is idempotent on its own output — so merging onto the current file yields + // managed regions identical to merging onto the pristine .bak, but without + // discarding edits the user made between syncs (add a dependency, then + // re-sync). The .bak stays a one-time safety copy of the pre-sync original, not + // the merge base. var baseBytes []byte - if data, rerr := os.ReadFile(backup); rerr == nil { + data, rerr := os.ReadFile(pyproject) + switch { + case rerr == nil: baseBytes = data - } else if !errors.Is(rerr, os.ErrNotExist) { - // A backup that exists but can't be read (permissions, I/O) must not be - // silently ignored: falling through would treat the project as greenfield - // and overwrite it. Fail loudly instead. - return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read backup %s failed", filepath.ToSlash(backup))) - } - if baseBytes == nil { - data, rerr := os.ReadFile(pyproject) - switch { - case rerr == nil: - baseBytes = data - case !errors.Is(rerr, os.ErrNotExist): - // Only a genuine not-exist means greenfield. Any other read error on an - // existing pyproject.toml (permission change, transient I/O, delete race - // after detectManager saw it) must not be misread as greenfield — that - // would render a fresh file and overwrite the user's project with no - // backup. Fail instead of destroying unrecoverable state (invariant 2). - return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read pyproject.toml %s failed", filepath.ToSlash(pyproject))) - } + case !errors.Is(rerr, os.ErrNotExist): + // Only a genuine not-exist means greenfield. Any other read error on an + // existing pyproject.toml (permission change, transient I/O, delete race + // after detectManager saw it) must not be misread as greenfield — that + // would render a fresh file and overwrite the user's project with no + // backup. Fail instead of destroying unrecoverable state (invariant 2). + return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read pyproject.toml %s failed", filepath.ToSlash(pyproject))) } greenfield = baseBytes == nil @@ -288,8 +286,14 @@ func (p *Pipeline) mergePlan(_ context.Context, pyMinor string, c *Constraints, ChangedRegions: changedRegions, WouldInstallPython: pyMinor, } + // Report a backup only when a real run would actually write one: for an + // existing project with no .bak yet. On a re-run the .bak already exists and + // applyMerge keeps it (does not re-write), so claiming a backup here would + // describe a write that won't happen. if !greenfield { - plan.WouldBackup = filepath.ToSlash(backup) + if _, statErr := os.Stat(backup); errors.Is(statErr, os.ErrNotExist) { + plan.WouldBackup = filepath.ToSlash(backup) + } } p.res.Plan = plan } diff --git a/libs/localenv/pipeline_test.go b/libs/localenv/pipeline_test.go index e284e3b50a9..1a318cd1872 100644 --- a/libs/localenv/pipeline_test.go +++ b/libs/localenv/pipeline_test.go @@ -100,6 +100,36 @@ func TestPipelineCheckMutatesNothing(t *testing.T) { assert.Equal(t, string(before), string(after)) // unchanged } +func TestPipelineCheckReRunPlanMatchesRealRun(t *testing.T) { + // On a re-run where the .bak already exists and the live file already equals + // the merged output, --check must report a plan a real run would perform: no + // backup (the .bak is kept, not rewritten) and an empty diff (nothing changes). + dir := writeProject(t) + srv := newTestServer(t) + defer srv.Close() + + newPipe := func(check bool) *Pipeline { + return &Pipeline{ + Mode: ModeDefault, Check: check, ProjectDir: dir, + ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(), + Flags: TargetFlags{Serverless: "v4"}, + Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"}, + } + } + + // First a real sync: writes the merged file and creates the .bak. + _, err := newPipe(false).Run(t.Context()) + require.NoError(t, err) + require.FileExists(t, filepath.Join(dir, "pyproject.toml.bak")) + + // Now --check on the already-synced project. + res, err := newPipe(true).Run(t.Context()) + require.NoError(t, err) + require.NotNil(t, res.Plan) + assert.Empty(t, res.Plan.WouldBackup, "a re-run keeps the existing .bak, so --check must not claim a backup") + assert.Empty(t, res.Plan.Diff, "the live file already equals the merged output; the diff must be empty") +} + func TestPipelineCheckDoesNotProvision(t *testing.T) { // --check must not call any PackageManager method that could mutate the // machine (EnsureAvailable may install uv). noProvisionPM errors on all of @@ -370,9 +400,9 @@ func TestPipelineNoTarget(t *testing.T) { assert.Equal(t, StatusPending, phaseStatus(res, PhaseProvision)) } -func TestPipelineRestoresBackupBeforeMerge(t *testing.T) { +func TestPipelineMergesOnLiveFileNotBackup(t *testing.T) { dir := t.TempDir() - // Write an original pyproject.toml and a pre-existing .bak. + // A pre-existing .bak holds the pristine pre-sync original (no user edit). original := []byte(`[project] name = "demo" requires-python = ">=3.9" @@ -381,15 +411,17 @@ requires-python = ">=3.9" dev = ["databricks-connect~=15.0.0"] `) require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml.bak"), original, 0o644)) - // Current pyproject.toml has been mutated by a previous run. - mutated := []byte(`[project] + // The live pyproject.toml is a prior sync's output plus an edit the developer + // made afterwards: a non-managed dependency the .bak never saw. + live := []byte(`[project] name = "demo" requires-python = "==3.12.*" +dependencies = ["rich"] [dependency-groups] dev = ["databricks-connect~=17.2.0"] `) - require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), mutated, 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), live, 0o644)) srv := newTestServer(t) defer srv.Close() @@ -403,11 +435,16 @@ dev = ["databricks-connect~=17.2.0"] res, err := p.Run(t.Context()) require.NoError(t, err) require.NotNil(t, res) - // The bak content (requires-python = ">=3.9") was the base; merged result should - // contain the newly pinned version. data, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml")) + // The between-sync user edit survives: the merge based on the live file, not + // the stale .bak, which would have discarded it. + assert.Contains(t, string(data), `dependencies = ["rich"]`) + // Managed regions are still applied. assert.Contains(t, string(data), `"databricks-connect~=17.2.0"`) assert.Contains(t, string(data), `requires-python = "==3.12.*"`) + // The .bak is left as the one-time original safety copy, not overwritten. + bak, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml.bak")) + assert.Equal(t, string(original), string(bak)) } func TestPipelineUnreadableExistingIsNotTreatedAsGreenfield(t *testing.T) {