From bad3cb926c3de500b82e3ef1375bde2ebabacfc3 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Thu, 23 Nov 2023 14:26:09 +0100
Subject: [PATCH 1/7] Correctly overwrite local state if remote state is newer

A bug in the code that pulls the remote state could cause the local state to be
empty instead of a copy of the remote state. This happened only if the local
state was present and stale, compared to the remote version.

We correctly checked for the state serial to see if the local state had to be
replaced but didn't seek back on the remote state before writing it out.
Because the staleness check would read the remote state in full, copying from
the same reader would immediately yield an EOF.

This change includes:
* A fix for this problem.
* Unit tests for state pull and push mutators that rely on a mocked filer.
* An integration test that deploys the same bundle from multiple paths,
  triggering the staleness logic.
---
 bundle/deploy/terraform/filer.go              |  21 +++
 bundle/deploy/terraform/state_pull.go         |  49 ++++--
 bundle/deploy/terraform/state_pull_test.go    | 128 ++++++++++++++++
 bundle/deploy/terraform/state_push.go         |   8 +-
 bundle/deploy/terraform/state_push_test.go    |  63 ++++++++
 bundle/deploy/terraform/state_test.go         |  40 +++++
 bundle/deploy/terraform/util_test.go          |  81 ++--------
 go.mod                                        |   2 +
 go.sum                                        |   2 +
 .../basic/databricks_template_schema.json     |  16 ++
 .../basic/template/databricks.yml.tmpl        |  18 +++
 .../bundles/basic/template/hello_world.py     |   1 +
 internal/bundle/helpers.go                    |  15 +-
 internal/bundle/local_state_staleness_test.go |  70 +++++++++
 internal/mocks/README.md                      |   7 +
 internal/mocks/libs/filer/filer_mock.go       | 139 ++++++++++++++++++
 libs/filer/filer.go                           |   6 +-
 17 files changed, 579 insertions(+), 87 deletions(-)
 create mode 100644 bundle/deploy/terraform/filer.go
 create mode 100644 bundle/deploy/terraform/state_pull_test.go
 create mode 100644 bundle/deploy/terraform/state_push_test.go
 create mode 100644 bundle/deploy/terraform/state_test.go
 create mode 100644 internal/bundle/bundles/basic/databricks_template_schema.json
 create mode 100644 internal/bundle/bundles/basic/template/databricks.yml.tmpl
 create mode 100644 internal/bundle/bundles/basic/template/hello_world.py
 create mode 100644 internal/bundle/local_state_staleness_test.go
 create mode 100644 internal/mocks/README.md
 create mode 100644 internal/mocks/libs/filer/filer_mock.go

diff --git a/bundle/deploy/terraform/filer.go b/bundle/deploy/terraform/filer.go
new file mode 100644
index 0000000000..bae979c647
--- /dev/null
+++ b/bundle/deploy/terraform/filer.go
@@ -0,0 +1,21 @@
+package terraform
+
+import (
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/filer"
+)
+
+// filerFunc is a function that returns a filer.Filer.
+type filerFunc func(b *bundle.Bundle) (filer.Filer, error)
+
+// stateFiler returns a filer.Filer that can be used to read/write state files.
+func stateFiler(b *bundle.Bundle) (filer.Filer, error) {
+	return filer.NewWorkspaceFilesClient(b.WorkspaceClient(), b.Config.Workspace.StatePath)
+}
+
+// identityFiler returns a filerFunc that returns the specified filer.
+func identityFiler(f filer.Filer) filerFunc {
+	return func(_ *bundle.Bundle) (filer.Filer, error) {
+		return f, nil
+	}
+}
diff --git a/bundle/deploy/terraform/state_pull.go b/bundle/deploy/terraform/state_pull.go
index 6dd12ccfc6..1435ca3acc 100644
--- a/bundle/deploy/terraform/state_pull.go
+++ b/bundle/deploy/terraform/state_pull.go
@@ -1,6 +1,7 @@
 package terraform
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
@@ -13,14 +14,38 @@ import (
 	"github.com/databricks/cli/libs/log"
 )
 
-type statePull struct{}
+type statePull struct {
+	filerFunc
+}
 
 func (l *statePull) Name() string {
 	return "terraform:state-pull"
 }
 
+func (l *statePull) remoteState(ctx context.Context, f filer.Filer) (*bytes.Buffer, error) {
+	// Download state file from filer to local cache directory.
+	remote, err := f.Read(ctx, TerraformStateFileName)
+	if err != nil {
+		// On first deploy this state file doesn't yet exist.
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	defer remote.Close()
+
+	var buf bytes.Buffer
+	_, err = io.Copy(&buf, remote)
+	if err != nil {
+		return nil, err
+	}
+
+	return &buf, nil
+}
+
 func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
-	f, err := filer.NewWorkspaceFilesClient(b.WorkspaceClient(), b.Config.Workspace.StatePath)
+	f, err := l.filerFunc(b)
 	if err != nil {
 		return err
 	}
@@ -30,17 +55,15 @@ func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
 		return err
 	}
 
-	// Download state file from filer to local cache directory.
-	log.Infof(ctx, "Opening remote state file")
-	remote, err := f.Read(ctx, TerraformStateFileName)
+	remote, err := l.remoteState(ctx, f)
 	if err != nil {
-		// On first deploy this state file doesn't yet exist.
-		if errors.Is(err, fs.ErrNotExist) {
-			log.Infof(ctx, "Remote state file does not exist")
-			return nil
-		}
+		log.Infof(ctx, "Unable to open remote state file: %s", err)
 		return err
 	}
+	if remote == nil {
+		log.Infof(ctx, "Remote state file does not exist")
+		return nil
+	}
 
 	// Expect the state file to live under dir.
 	local, err := os.OpenFile(filepath.Join(dir, TerraformStateFileName), os.O_CREATE|os.O_RDWR, 0600)
@@ -49,7 +72,7 @@ func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
 	}
 	defer local.Close()
 
-	if !IsLocalStateStale(local, remote) {
+	if !IsLocalStateStale(local, bytes.NewReader(remote.Bytes())) {
 		log.Infof(ctx, "Local state is the same or newer, ignoring remote state")
 		return nil
 	}
@@ -60,7 +83,7 @@ func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
 
 	// Write file to disk.
 	log.Infof(ctx, "Writing remote state file to local cache directory")
-	_, err = io.Copy(local, remote)
+	_, err = io.Copy(local, bytes.NewReader(remote.Bytes()))
 	if err != nil {
 		return err
 	}
@@ -69,5 +92,5 @@ func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
 }
 
 func StatePull() bundle.Mutator {
-	return &statePull{}
+	return &statePull{stateFiler}
 }
diff --git a/bundle/deploy/terraform/state_pull_test.go b/bundle/deploy/terraform/state_pull_test.go
new file mode 100644
index 0000000000..71bb3852b6
--- /dev/null
+++ b/bundle/deploy/terraform/state_pull_test.go
@@ -0,0 +1,128 @@
+package terraform
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"io/fs"
+	"os"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	mock "github.com/databricks/cli/internal/mocks/libs/filer"
+	"github.com/databricks/cli/libs/filer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+)
+
+func mockStateFilerForPull(t *testing.T, contents map[string]int, merr error) filer.Filer {
+	buf, err := json.Marshal(contents)
+	require.NoError(t, err)
+
+	ctrl := gomock.NewController(t)
+	filer := mock.NewMockFiler(ctrl)
+	filer.
+		EXPECT().
+		Read(gomock.Any(), gomock.Eq(TerraformStateFileName)).
+		Return(io.NopCloser(bytes.NewReader(buf)), merr).
+		Times(1)
+	return filer
+}
+
+func statePullTestBundle(t *testing.T) *bundle.Bundle {
+	return &bundle.Bundle{
+		Config: config.Root{
+			Bundle: config.Bundle{
+				Target: "default",
+			},
+			Path: t.TempDir(),
+		},
+	}
+}
+
+func TestStatePullLocalMissingRemoteMissing(t *testing.T) {
+	m := &statePull{
+		identityFiler(mockStateFilerForPull(t, nil, os.ErrNotExist)),
+	}
+
+	ctx := context.Background()
+	b := statePullTestBundle(t)
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+
+	// Confirm that no local state file has been written.
+	_, err = os.Stat(localStateFile(t, ctx, b))
+	assert.ErrorIs(t, err, fs.ErrNotExist)
+}
+
+func TestStatePullLocalMissingRemotePresent(t *testing.T) {
+	m := &statePull{
+		identityFiler(mockStateFilerForPull(t, map[string]int{"serial": 5}, nil)),
+	}
+
+	ctx := context.Background()
+	b := statePullTestBundle(t)
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+
+	// Confirm that the local state file has been updated.
+	localState := readLocalState(t, ctx, b)
+	assert.Equal(t, map[string]int{"serial": 5}, localState)
+}
+
+func TestStatePullLocalStale(t *testing.T) {
+	m := &statePull{
+		identityFiler(mockStateFilerForPull(t, map[string]int{"serial": 5}, nil)),
+	}
+
+	ctx := context.Background()
+	b := statePullTestBundle(t)
+
+	// Write a stale local state file.
+	writeLocalState(t, ctx, b, map[string]int{"serial": 4})
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+
+	// Confirm that the local state file has been updated.
+	localState := readLocalState(t, ctx, b)
+	assert.Equal(t, map[string]int{"serial": 5}, localState)
+}
+
+func TestStatePullLocalEqual(t *testing.T) {
+	m := &statePull{
+		identityFiler(mockStateFilerForPull(t, map[string]int{"serial": 5, "some_other_key": 123}, nil)),
+	}
+
+	ctx := context.Background()
+	b := statePullTestBundle(t)
+
+	// Write a local state file with the same serial as the remote.
+	writeLocalState(t, ctx, b, map[string]int{"serial": 5})
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+
+	// Confirm that the local state file has not been updated.
+	localState := readLocalState(t, ctx, b)
+	assert.Equal(t, map[string]int{"serial": 5}, localState)
+}
+
+func TestStatePullLocalNewer(t *testing.T) {
+	m := &statePull{
+		identityFiler(mockStateFilerForPull(t, map[string]int{"serial": 5, "some_other_key": 123}, nil)),
+	}
+
+	ctx := context.Background()
+	b := statePullTestBundle(t)
+
+	// Write a local state file with a newer serial as the remote.
+	writeLocalState(t, ctx, b, map[string]int{"serial": 6})
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+
+	// Confirm that the local state file has not been updated.
+	localState := readLocalState(t, ctx, b)
+	assert.Equal(t, map[string]int{"serial": 6}, localState)
+}
diff --git a/bundle/deploy/terraform/state_push.go b/bundle/deploy/terraform/state_push.go
index ae1d8b8b3e..30a4359620 100644
--- a/bundle/deploy/terraform/state_push.go
+++ b/bundle/deploy/terraform/state_push.go
@@ -10,14 +10,16 @@ import (
 	"github.com/databricks/cli/libs/log"
 )
 
-type statePush struct{}
+type statePush struct {
+	filerFunc
+}
 
 func (l *statePush) Name() string {
 	return "terraform:state-push"
 }
 
 func (l *statePush) Apply(ctx context.Context, b *bundle.Bundle) error {
-	f, err := filer.NewWorkspaceFilesClient(b.WorkspaceClient(), b.Config.Workspace.StatePath)
+	f, err := l.filerFunc(b)
 	if err != nil {
 		return err
 	}
@@ -45,5 +47,5 @@ func (l *statePush) Apply(ctx context.Context, b *bundle.Bundle) error {
 }
 
 func StatePush() bundle.Mutator {
-	return &statePush{}
+	return &statePush{stateFiler}
 }
diff --git a/bundle/deploy/terraform/state_push_test.go b/bundle/deploy/terraform/state_push_test.go
new file mode 100644
index 0000000000..4167b3cb98
--- /dev/null
+++ b/bundle/deploy/terraform/state_push_test.go
@@ -0,0 +1,63 @@
+package terraform
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	mock "github.com/databricks/cli/internal/mocks/libs/filer"
+	"github.com/databricks/cli/libs/filer"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/mock/gomock"
+)
+
+func mockStateFilerForPush(t *testing.T, fn func(body io.Reader)) filer.Filer {
+	ctrl := gomock.NewController(t)
+	mock := mock.NewMockFiler(ctrl)
+	mock.
+		EXPECT().
+		Write(gomock.Any(), gomock.Any(), gomock.Any(), filer.CreateParentDirectories, filer.OverwriteIfExists).
+		Do(func(ctx context.Context, path string, reader io.Reader, mode ...filer.WriteMode) error {
+			fn(reader)
+			return nil
+		}).
+		Return(nil).
+		Times(1)
+	return mock
+}
+
+func statePushTestBundle(t *testing.T) *bundle.Bundle {
+	return &bundle.Bundle{
+		Config: config.Root{
+			Bundle: config.Bundle{
+				Target: "default",
+			},
+			Path: t.TempDir(),
+		},
+	}
+}
+
+func TestStatePush(t *testing.T) {
+	mock := mockStateFilerForPush(t, func(body io.Reader) {
+		dec := json.NewDecoder(body)
+		var contents map[string]int
+		err := dec.Decode(&contents)
+		assert.NoError(t, err)
+		assert.Equal(t, map[string]int{"serial": 4}, contents)
+	})
+
+	m := &statePush{
+		identityFiler(mock),
+	}
+
+	ctx := context.Background()
+	b := statePushTestBundle(t)
+
+	// Write a stale local state file.
+	writeLocalState(t, ctx, b, map[string]int{"serial": 4})
+	err := bundle.Apply(ctx, b, m)
+	assert.NoError(t, err)
+}
diff --git a/bundle/deploy/terraform/state_test.go b/bundle/deploy/terraform/state_test.go
new file mode 100644
index 0000000000..c24776f4bf
--- /dev/null
+++ b/bundle/deploy/terraform/state_test.go
@@ -0,0 +1,40 @@
+package terraform
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/stretchr/testify/require"
+)
+
+func localStateFile(t *testing.T, ctx context.Context, b *bundle.Bundle) string {
+	dir, err := Dir(ctx, b)
+	require.NoError(t, err)
+	return filepath.Join(dir, TerraformStateFileName)
+}
+
+func readLocalState(t *testing.T, ctx context.Context, b *bundle.Bundle) map[string]int {
+	f, err := os.Open(localStateFile(t, ctx, b))
+	require.NoError(t, err)
+	defer f.Close()
+
+	var contents map[string]int
+	dec := json.NewDecoder(f)
+	err = dec.Decode(&contents)
+	require.NoError(t, err)
+	return contents
+}
+
+func writeLocalState(t *testing.T, ctx context.Context, b *bundle.Bundle, contents map[string]int) {
+	f, err := os.Create(localStateFile(t, ctx, b))
+	require.NoError(t, err)
+	defer f.Close()
+
+	enc := json.NewEncoder(f)
+	err = enc.Encode(contents)
+	require.NoError(t, err)
+}
diff --git a/bundle/deploy/terraform/util_test.go b/bundle/deploy/terraform/util_test.go
index 1ddfbab351..4f2cf29187 100644
--- a/bundle/deploy/terraform/util_test.go
+++ b/bundle/deploy/terraform/util_test.go
@@ -2,92 +2,39 @@ package terraform
 
 import (
 	"fmt"
-	"io"
+	"strings"
 	"testing"
 	"testing/iotest"
 
 	"github.com/stretchr/testify/assert"
 )
 
-type mockedReader struct {
-	content string
-}
-
-func (r *mockedReader) Read(p []byte) (n int, err error) {
-	content := []byte(r.content)
-	n = copy(p, content)
-	return n, io.EOF
-}
-
 func TestLocalStateIsNewer(t *testing.T) {
-	local := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
-	remote := &mockedReader{content: `
-{
-	"serial": 4
-}
-`}
-
-	stale := IsLocalStateStale(local, remote)
-
-	assert.False(t, stale)
+	local := strings.NewReader(`{"serial": 5}`)
+	remote := strings.NewReader(`{"serial": 4}`)
+	assert.False(t, IsLocalStateStale(local, remote))
 }
 
 func TestLocalStateIsOlder(t *testing.T) {
-	local := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
-	remote := &mockedReader{content: `
-{
-	"serial": 6
-}
-`}
-
-	stale := IsLocalStateStale(local, remote)
-	assert.True(t, stale)
+	local := strings.NewReader(`{"serial": 5}`)
+	remote := strings.NewReader(`{"serial": 6}`)
+	assert.True(t, IsLocalStateStale(local, remote))
 }
 
 func TestLocalStateIsTheSame(t *testing.T) {
-	local := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
-	remote := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
-
-	stale := IsLocalStateStale(local, remote)
-	assert.False(t, stale)
+	local := strings.NewReader(`{"serial": 5}`)
+	remote := strings.NewReader(`{"serial": 5}`)
+	assert.False(t, IsLocalStateStale(local, remote))
 }
 
 func TestLocalStateMarkStaleWhenFailsToLoad(t *testing.T) {
 	local := iotest.ErrReader(fmt.Errorf("Random error"))
-	remote := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
-
-	stale := IsLocalStateStale(local, remote)
-	assert.True(t, stale)
+	remote := strings.NewReader(`{"serial": 5}`)
+	assert.True(t, IsLocalStateStale(local, remote))
 }
 
 func TestLocalStateMarkNonStaleWhenRemoteFailsToLoad(t *testing.T) {
-	local := &mockedReader{content: `
-{
-	"serial": 5
-}
-`}
+	local := strings.NewReader(`{"serial": 5}`)
 	remote := iotest.ErrReader(fmt.Errorf("Random error"))
-
-	stale := IsLocalStateStale(local, remote)
-	assert.False(t, stale)
+	assert.False(t, IsLocalStateStale(local, remote))
 }
diff --git a/go.mod b/go.mod
index 7cef4cd4e0..7359b407ad 100644
--- a/go.mod
+++ b/go.mod
@@ -52,10 +52,12 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/zclconf/go-cty v1.14.1 // indirect
 	go.opencensus.io v0.24.0 // indirect
+	go.uber.org/mock v0.3.0 // indirect
 	golang.org/x/crypto v0.15.0 // indirect
 	golang.org/x/net v0.18.0 // indirect
 	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/time v0.4.0 // indirect
+	golang.org/x/tools v0.14.0 // indirect
 	google.golang.org/api v0.150.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231030173426-d783a09b4405 // indirect
diff --git a/go.sum b/go.sum
index 25409bd6f9..cff48c8fc3 100644
--- a/go.sum
+++ b/go.sum
@@ -158,6 +158,8 @@ github.com/zclconf/go-cty v1.14.1 h1:t9fyA35fwjjUMcmL5hLER+e/rEPqrbCK1/OSE4SI9KA
 github.com/zclconf/go-cty v1.14.1/go.mod h1:VvMs5i0vgZdhYawQNq5kePSpLAoz8u1xvZgrPIxfnZE=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
+go.uber.org/mock v0.3.0 h1:3mUxI1No2/60yUYax92Pt8eNOEecx2D3lcXZh2NEZJo=
+go.uber.org/mock v0.3.0/go.mod h1:a6FSlNadKUHUa9IP5Vyt1zh4fC7uAwxMutEAscFbkZc=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
diff --git a/internal/bundle/bundles/basic/databricks_template_schema.json b/internal/bundle/bundles/basic/databricks_template_schema.json
new file mode 100644
index 0000000000..c1c5cf12eb
--- /dev/null
+++ b/internal/bundle/bundles/basic/databricks_template_schema.json
@@ -0,0 +1,16 @@
+{
+    "properties": {
+        "unique_id": {
+            "type": "string",
+            "description": "Unique ID for job name"
+        },
+        "spark_version": {
+            "type": "string",
+            "description": "Spark version used for job cluster"
+        },
+        "node_type_id": {
+            "type": "string",
+            "description": "Node type id for job cluster"
+        }
+    }
+}
diff --git a/internal/bundle/bundles/basic/template/databricks.yml.tmpl b/internal/bundle/bundles/basic/template/databricks.yml.tmpl
new file mode 100644
index 0000000000..a88cbd30ee
--- /dev/null
+++ b/internal/bundle/bundles/basic/template/databricks.yml.tmpl
@@ -0,0 +1,18 @@
+bundle:
+  name: basic
+
+workspace:
+  root_path: "~/.bundle/{{.unique_id}}"
+
+resources:
+  jobs:
+    foo:
+      name: test-job-basic-{{.unique_id}}
+      tasks:
+        - task_key: my_notebook_task
+          new_cluster:
+            num_workers: 1
+            spark_version: "{{.spark_version}}"
+            node_type_id: "{{.node_type_id}}"
+          spark_python_task:
+            python_file: ./hello_world.py
diff --git a/internal/bundle/bundles/basic/template/hello_world.py b/internal/bundle/bundles/basic/template/hello_world.py
new file mode 100644
index 0000000000..f301245e24
--- /dev/null
+++ b/internal/bundle/bundles/basic/template/hello_world.py
@@ -0,0 +1 @@
+print("Hello World!")
diff --git a/internal/bundle/helpers.go b/internal/bundle/helpers.go
index 3fd4eabc9b..540f397d25 100644
--- a/internal/bundle/helpers.go
+++ b/internal/bundle/helpers.go
@@ -47,12 +47,25 @@ func writeConfigFile(t *testing.T, config map[string]any) (string, error) {
 }
 
 func deployBundle(t *testing.T, path string) error {
+	return deployBundleWithVariables(t, path, nil)
+}
+
+func deployBundleWithVariables(t *testing.T, path string, vars map[string]string) error {
 	t.Setenv("BUNDLE_ROOT", path)
-	c := internal.NewCobraTestRunner(t, "bundle", "deploy", "--force-lock")
+	args := append([]string{"bundle", "deploy", "--force-lock"}, buildVariableArgs(vars)...)
+	c := internal.NewCobraTestRunner(t, args...)
 	_, _, err := c.Run()
 	return err
 }
 
+func buildVariableArgs(vars map[string]string) []string {
+	var args []string
+	for k, v := range vars {
+		args = append(args, "--var", k+"="+v)
+	}
+	return args
+}
+
 func runResource(t *testing.T, path string, key string) (string, error) {
 	ctx := context.Background()
 	ctx = cmdio.NewContext(ctx, cmdio.Default())
diff --git a/internal/bundle/local_state_staleness_test.go b/internal/bundle/local_state_staleness_test.go
new file mode 100644
index 0000000000..06cfe0e0d9
--- /dev/null
+++ b/internal/bundle/local_state_staleness_test.go
@@ -0,0 +1,70 @@
+package bundle
+
+import (
+	"context"
+	"testing"
+
+	"github.com/databricks/cli/internal"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/listing"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAccLocalStateStaleness(t *testing.T) {
+	env := internal.GetEnvOrSkipTest(t, "CLOUD_ENV")
+	t.Log(env)
+
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	// The approach for this test is as follows:
+	// 1) First deploy of bundle instance A
+	// 2) First deploy of bundle instance B
+	// 3) Second deploy of bundle instance A
+	// Because of deploy (2), the locally cached state of bundle instance A should be stale.
+	// Then for deploy (3), it must use the remote state over the stale local state.
+
+	nodeTypeId := internal.GetNodeTypeId(env)
+	uniqueId := uuid.New().String()
+	initialize := func() string {
+		root, err := initTestTemplate(t, "basic", map[string]any{
+			"unique_id":     uniqueId,
+			"node_type_id":  nodeTypeId,
+			"spark_version": "13.2.x-snapshot-scala2.12",
+		})
+		require.NoError(t, err)
+
+		t.Cleanup(func() {
+			err = destroyBundle(t, root)
+			require.NoError(t, err)
+		})
+
+		return root
+	}
+
+	bundleA := initialize()
+	bundleB := initialize()
+
+	// 1) Deploy bundle A
+	err = deployBundle(t, bundleA)
+	require.NoError(t, err)
+
+	// 2) Deploy bundle B
+	err = deployBundle(t, bundleB)
+	require.NoError(t, err)
+
+	// 3) Deploy bundle A again
+	err = deployBundle(t, bundleA)
+	require.NoError(t, err)
+
+	// Assert that there is only a single job in the workspace corresponding to this bundle.
+	iter := w.Jobs.List(context.Background(), jobs.ListJobsRequest{
+		Name: "test-job-basic-" + uniqueId,
+	})
+	jobs, err := listing.ToSlice(context.Background(), iter)
+	require.NoError(t, err)
+	assert.Len(t, jobs, 1)
+}
diff --git a/internal/mocks/README.md b/internal/mocks/README.md
new file mode 100644
index 0000000000..231bbfaa46
--- /dev/null
+++ b/internal/mocks/README.md
@@ -0,0 +1,7 @@
+# Interface mocking
+
+Use this directory to store mocks for interfaces in this repository.
+
+Please use the same package structure for the mocks as the interface it is mocking.
+
+See https://github.com/uber-go/mock for more information on how to generate mocks.
diff --git a/internal/mocks/libs/filer/filer_mock.go b/internal/mocks/libs/filer/filer_mock.go
new file mode 100644
index 0000000000..ef00976a2a
--- /dev/null
+++ b/internal/mocks/libs/filer/filer_mock.go
@@ -0,0 +1,139 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/databricks/cli/libs/filer (interfaces: Filer)
+//
+// Generated by this command:
+//
+//	mockgen -destination filer_mock.go github.com/databricks/cli/libs/filer Filer
+//
+// Package mock_filer is a generated GoMock package.
+package mock_filer
+
+import (
+	context "context"
+	io "io"
+	fs "io/fs"
+	reflect "reflect"
+
+	filer "github.com/databricks/cli/libs/filer"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockFiler is a mock of Filer interface.
+type MockFiler struct {
+	ctrl     *gomock.Controller
+	recorder *MockFilerMockRecorder
+}
+
+// MockFilerMockRecorder is the mock recorder for MockFiler.
+type MockFilerMockRecorder struct {
+	mock *MockFiler
+}
+
+// NewMockFiler creates a new mock instance.
+func NewMockFiler(ctrl *gomock.Controller) *MockFiler {
+	mock := &MockFiler{ctrl: ctrl}
+	mock.recorder = &MockFilerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockFiler) EXPECT() *MockFilerMockRecorder {
+	return m.recorder
+}
+
+// Delete mocks base method.
+func (m *MockFiler) Delete(arg0 context.Context, arg1 string, arg2 ...filer.DeleteMode) error {
+	m.ctrl.T.Helper()
+	varargs := []any{arg0, arg1}
+	for _, a := range arg2 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "Delete", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Delete indicates an expected call of Delete.
+func (mr *MockFilerMockRecorder) Delete(arg0, arg1 any, arg2 ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{arg0, arg1}, arg2...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockFiler)(nil).Delete), varargs...)
+}
+
+// Mkdir mocks base method.
+func (m *MockFiler) Mkdir(arg0 context.Context, arg1 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Mkdir", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Mkdir indicates an expected call of Mkdir.
+func (mr *MockFilerMockRecorder) Mkdir(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Mkdir", reflect.TypeOf((*MockFiler)(nil).Mkdir), arg0, arg1)
+}
+
+// Read mocks base method.
+func (m *MockFiler) Read(arg0 context.Context, arg1 string) (io.ReadCloser, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Read", arg0, arg1)
+	ret0, _ := ret[0].(io.ReadCloser)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Read indicates an expected call of Read.
+func (mr *MockFilerMockRecorder) Read(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Read", reflect.TypeOf((*MockFiler)(nil).Read), arg0, arg1)
+}
+
+// ReadDir mocks base method.
+func (m *MockFiler) ReadDir(arg0 context.Context, arg1 string) ([]fs.DirEntry, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ReadDir", arg0, arg1)
+	ret0, _ := ret[0].([]fs.DirEntry)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReadDir indicates an expected call of ReadDir.
+func (mr *MockFilerMockRecorder) ReadDir(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadDir", reflect.TypeOf((*MockFiler)(nil).ReadDir), arg0, arg1)
+}
+
+// Stat mocks base method.
+func (m *MockFiler) Stat(arg0 context.Context, arg1 string) (fs.FileInfo, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Stat", arg0, arg1)
+	ret0, _ := ret[0].(fs.FileInfo)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Stat indicates an expected call of Stat.
+func (mr *MockFilerMockRecorder) Stat(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stat", reflect.TypeOf((*MockFiler)(nil).Stat), arg0, arg1)
+}
+
+// Write mocks base method.
+func (m *MockFiler) Write(arg0 context.Context, arg1 string, arg2 io.Reader, arg3 ...filer.WriteMode) error {
+	m.ctrl.T.Helper()
+	varargs := []any{arg0, arg1, arg2}
+	for _, a := range arg3 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "Write", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Write indicates an expected call of Write.
+func (mr *MockFilerMockRecorder) Write(arg0, arg1, arg2 any, arg3 ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{arg0, arg1, arg2}, arg3...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Write", reflect.TypeOf((*MockFiler)(nil).Write), varargs...)
+}
diff --git a/libs/filer/filer.go b/libs/filer/filer.go
index 8267dc3431..c1c747c546 100644
--- a/libs/filer/filer.go
+++ b/libs/filer/filer.go
@@ -10,14 +10,14 @@ import (
 type WriteMode int
 
 const (
-	OverwriteIfExists       WriteMode = iota
-	CreateParentDirectories           = iota << 1
+	OverwriteIfExists WriteMode = 1 << iota
+	CreateParentDirectories
 )
 
 type DeleteMode int
 
 const (
-	DeleteRecursively DeleteMode = iota
+	DeleteRecursively DeleteMode = 1 << iota
 )
 
 type FileAlreadyExistsError struct {

From 84ed2ed350fc4e67f47d9e9611016ec04731b33c Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Thu, 23 Nov 2023 20:39:13 +0100
Subject: [PATCH 2/7] filer -> mock

---
 bundle/deploy/terraform/state_pull_test.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/bundle/deploy/terraform/state_pull_test.go b/bundle/deploy/terraform/state_pull_test.go
index 71bb3852b6..60eb5d90c5 100644
--- a/bundle/deploy/terraform/state_pull_test.go
+++ b/bundle/deploy/terraform/state_pull_test.go
@@ -23,13 +23,13 @@ func mockStateFilerForPull(t *testing.T, contents map[string]int, merr error) fi
 	require.NoError(t, err)
 
 	ctrl := gomock.NewController(t)
-	filer := mock.NewMockFiler(ctrl)
-	filer.
+	mock := mock.NewMockFiler(ctrl)
+	mock.
 		EXPECT().
 		Read(gomock.Any(), gomock.Eq(TerraformStateFileName)).
 		Return(io.NopCloser(bytes.NewReader(buf)), merr).
 		Times(1)
-	return filer
+	return mock
 }
 
 func statePullTestBundle(t *testing.T) *bundle.Bundle {

From c336f2ca39b687a376c9001d57bbb18144b46672 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Thu, 23 Nov 2023 20:41:20 +0100
Subject: [PATCH 3/7] Revert unrelated changes

---
 internal/bundle/helpers.go | 15 +--------------
 1 file changed, 1 insertion(+), 14 deletions(-)

diff --git a/internal/bundle/helpers.go b/internal/bundle/helpers.go
index 540f397d25..3fd4eabc9b 100644
--- a/internal/bundle/helpers.go
+++ b/internal/bundle/helpers.go
@@ -47,25 +47,12 @@ func writeConfigFile(t *testing.T, config map[string]any) (string, error) {
 }
 
 func deployBundle(t *testing.T, path string) error {
-	return deployBundleWithVariables(t, path, nil)
-}
-
-func deployBundleWithVariables(t *testing.T, path string, vars map[string]string) error {
 	t.Setenv("BUNDLE_ROOT", path)
-	args := append([]string{"bundle", "deploy", "--force-lock"}, buildVariableArgs(vars)...)
-	c := internal.NewCobraTestRunner(t, args...)
+	c := internal.NewCobraTestRunner(t, "bundle", "deploy", "--force-lock")
 	_, _, err := c.Run()
 	return err
 }
 
-func buildVariableArgs(vars map[string]string) []string {
-	var args []string
-	for k, v := range vars {
-		args = append(args, "--var", k+"="+v)
-	}
-	return args
-}
-
 func runResource(t *testing.T, path string, key string) (string, error) {
 	ctx := context.Background()
 	ctx = cmdio.NewContext(ctx, cmdio.Default())

From 205ebbfd2236d984f6b3b2de630454ebd82c61e4 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Thu, 23 Nov 2023 20:47:45 +0100
Subject: [PATCH 4/7] Add back log line and comment

---
 bundle/deploy/terraform/state_pull.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/bundle/deploy/terraform/state_pull.go b/bundle/deploy/terraform/state_pull.go
index 1435ca3acc..14e8ecf121 100644
--- a/bundle/deploy/terraform/state_pull.go
+++ b/bundle/deploy/terraform/state_pull.go
@@ -55,6 +55,8 @@ func (l *statePull) Apply(ctx context.Context, b *bundle.Bundle) error {
 		return err
 	}
 
+	// Download state file from filer to local cache directory.
+	log.Infof(ctx, "Opening remote state file")
 	remote, err := l.remoteState(ctx, f)
 	if err != nil {
 		log.Infof(ctx, "Unable to open remote state file: %s", err)

From 56278672204b7fb25e14c2a7f8aff9dfaa02a859 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Thu, 23 Nov 2023 20:53:56 +0100
Subject: [PATCH 5/7] Run go mod tidy

---
 go.mod | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/go.mod b/go.mod
index 7359b407ad..f703d8b06d 100644
--- a/go.mod
+++ b/go.mod
@@ -31,7 +31,10 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // Apache 2.0
 )
 
-require gopkg.in/yaml.v3 v3.0.1
+require (
+	go.uber.org/mock v0.3.0
+	gopkg.in/yaml.v3 v3.0.1
+)
 
 require (
 	cloud.google.com/go/compute v1.23.1 // indirect
@@ -52,12 +55,10 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/zclconf/go-cty v1.14.1 // indirect
 	go.opencensus.io v0.24.0 // indirect
-	go.uber.org/mock v0.3.0 // indirect
 	golang.org/x/crypto v0.15.0 // indirect
 	golang.org/x/net v0.18.0 // indirect
 	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/time v0.4.0 // indirect
-	golang.org/x/tools v0.14.0 // indirect
 	google.golang.org/api v0.150.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231030173426-d783a09b4405 // indirect

From 5bd20af48eb740cda058341d63765aad2b07c306 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Fri, 24 Nov 2023 12:09:41 +0100
Subject: [PATCH 6/7] Add new library to NOTICE file

---
 NOTICE | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/NOTICE b/NOTICE
index bce870f16b..7c7eb7db46 100644
--- a/NOTICE
+++ b/NOTICE
@@ -16,6 +16,10 @@ go-ini/ini - https://github.com/go-ini/ini
 Copyright ini authors
 License - https://github.com/go-ini/ini/blob/main/LICENSE
 
+uber-go/mock - https://go.uber.org/mock
+Copyright Google Inc.
+License - https://github.com/uber-go/mock/blob/main/LICENSE
+
 —--
 
 This software contains code from the following open source projects, licensed under the MPL 2.0 license:

From d10441f29c17a73e943d84bf298f63cb1ede1ed5 Mon Sep 17 00:00:00 2001
From: Pieter Noordhuis <pieter.noordhuis@databricks.com>
Date: Fri, 24 Nov 2023 12:11:28 +0100
Subject: [PATCH 7/7] Move test-only function

---
 bundle/deploy/terraform/filer.go      | 7 -------
 bundle/deploy/terraform/state_test.go | 8 ++++++++
 2 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/bundle/deploy/terraform/filer.go b/bundle/deploy/terraform/filer.go
index bae979c647..b1fa5a1bd0 100644
--- a/bundle/deploy/terraform/filer.go
+++ b/bundle/deploy/terraform/filer.go
@@ -12,10 +12,3 @@ type filerFunc func(b *bundle.Bundle) (filer.Filer, error)
 func stateFiler(b *bundle.Bundle) (filer.Filer, error) {
 	return filer.NewWorkspaceFilesClient(b.WorkspaceClient(), b.Config.Workspace.StatePath)
 }
-
-// identityFiler returns a filerFunc that returns the specified filer.
-func identityFiler(f filer.Filer) filerFunc {
-	return func(_ *bundle.Bundle) (filer.Filer, error) {
-		return f, nil
-	}
-}
diff --git a/bundle/deploy/terraform/state_test.go b/bundle/deploy/terraform/state_test.go
index c24776f4bf..ee15b953bf 100644
--- a/bundle/deploy/terraform/state_test.go
+++ b/bundle/deploy/terraform/state_test.go
@@ -8,9 +8,17 @@ import (
 	"testing"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/filer"
 	"github.com/stretchr/testify/require"
 )
 
+// identityFiler returns a filerFunc that returns the specified filer.
+func identityFiler(f filer.Filer) filerFunc {
+	return func(_ *bundle.Bundle) (filer.Filer, error) {
+		return f, nil
+	}
+}
+
 func localStateFile(t *testing.T, ctx context.Context, b *bundle.Bundle) string {
 	dir, err := Dir(ctx, b)
 	require.NoError(t, err)