Move backend/ to internal/backend/

This is part of a general effort to move all of Terraform's non-library
package surface under internal in order to reinforce that these are for
internal use within Terraform only.

If you were previously importing packages under this prefix into an
external codebase, you could pin to an earlier release tag as an interim
solution until you've make a plan to achieve the same functionality some
other way.

internal/backend/local/backend.go

@@ -0,0 +1,487 @@
+package local
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/hashicorp/terraform/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+)
+
+const (
+	DefaultWorkspaceDir    = "terraform.tfstate.d"
+	DefaultWorkspaceFile   = "environment"
+	DefaultStateFilename   = "terraform.tfstate"
+	DefaultBackupExtension = ".backup"
+)
+
+// Local is an implementation of EnhancedBackend that performs all operations
+// locally. This is the "default" backend and implements normal Terraform
+// behavior as it is well known.
+type Local struct {
+	// The State* paths are set from the backend config, and may be left blank
+	// to use the defaults. If the actual paths for the local backend state are
+	// needed, use the StatePaths method.
+	//
+	// StatePath is the local path where state is read from.
+	//
+	// StateOutPath is the local path where the state will be written.
+	// If this is empty, it will default to StatePath.
+	//
+	// StateBackupPath is the local path where a backup file will be written.
+	// Set this to "-" to disable state backup.
+	//
+	// StateWorkspaceDir is the path to the folder containing data for
+	// non-default workspaces. This defaults to DefaultWorkspaceDir if not set.
+	StatePath         string
+	StateOutPath      string
+	StateBackupPath   string
+	StateWorkspaceDir string
+
+	// The OverrideState* paths are set based on per-operation CLI arguments
+	// and will override what'd be built from the State* fields if non-empty.
+	// While the interpretation of the State* fields depends on the active
+	// workspace, the OverrideState* fields are always used literally.
+	OverrideStatePath       string
+	OverrideStateOutPath    string
+	OverrideStateBackupPath string
+
+	// We only want to create a single instance of a local state, so store them
+	// here as they're loaded.
+	states map[string]statemgr.Full
+
+	// Terraform context. Many of these will be overridden or merged by
+	// Operation. See Operation for more details.
+	ContextOpts *terraform.ContextOpts
+
+	// OpInput will ask for necessary input prior to performing any operations.
+	//
+	// OpValidation will perform validation prior to running an operation. The
+	// variable naming doesn't match the style of others since we have a func
+	// Validate.
+	OpInput      bool
+	OpValidation bool
+
+	// Backend, if non-nil, will use this backend for non-enhanced behavior.
+	// This allows local behavior with remote state storage. It is a way to
+	// "upgrade" a non-enhanced backend to an enhanced backend with typical
+	// behavior.
+	//
+	// If this is nil, local performs normal state loading and storage.
+	Backend backend.Backend
+
+	// opLock locks operations
+	opLock sync.Mutex
+}
+
+var _ backend.Backend = (*Local)(nil)
+
+// New returns a new initialized local backend.
+func New() *Local {
+	return NewWithBackend(nil)
+}
+
+// NewWithBackend returns a new local backend initialized with a
+// dedicated backend for non-enhanced behavior.
+func NewWithBackend(backend backend.Backend) *Local {
+	return &Local{
+		Backend: backend,
+	}
+}
+
+func (b *Local) ConfigSchema() *configschema.Block {
+	if b.Backend != nil {
+		return b.Backend.ConfigSchema()
+	}
+	return &configschema.Block{
+		Attributes: map[string]*configschema.Attribute{
+			"path": {
+				Type:     cty.String,
+				Optional: true,
+			},
+			"workspace_dir": {
+				Type:     cty.String,
+				Optional: true,
+			},
+		},
+	}
+}
+
+func (b *Local) PrepareConfig(obj cty.Value) (cty.Value, tfdiags.Diagnostics) {
+	if b.Backend != nil {
+		return b.Backend.PrepareConfig(obj)
+	}
+
+	var diags tfdiags.Diagnostics
+
+	if val := obj.GetAttr("path"); !val.IsNull() {
+		p := val.AsString()
+		if p == "" {
+			diags = diags.Append(tfdiags.AttributeValue(
+				tfdiags.Error,
+				"Invalid local state file path",
+				`The "path" attribute value must not be empty.`,
+				cty.Path{cty.GetAttrStep{Name: "path"}},
+			))
+		}
+	}
+
+	if val := obj.GetAttr("workspace_dir"); !val.IsNull() {
+		p := val.AsString()
+		if p == "" {
+			diags = diags.Append(tfdiags.AttributeValue(
+				tfdiags.Error,
+				"Invalid local workspace directory path",
+				`The "workspace_dir" attribute value must not be empty.`,
+				cty.Path{cty.GetAttrStep{Name: "workspace_dir"}},
+			))
+		}
+	}
+
+	return obj, diags
+}
+
+func (b *Local) Configure(obj cty.Value) tfdiags.Diagnostics {
+	if b.Backend != nil {
+		return b.Backend.Configure(obj)
+	}
+
+	var diags tfdiags.Diagnostics
+
+	if val := obj.GetAttr("path"); !val.IsNull() {
+		p := val.AsString()
+		b.StatePath = p
+		b.StateOutPath = p
+	} else {
+		b.StatePath = DefaultStateFilename
+		b.StateOutPath = DefaultStateFilename
+	}
+
+	if val := obj.GetAttr("workspace_dir"); !val.IsNull() {
+		p := val.AsString()
+		b.StateWorkspaceDir = p
+	} else {
+		b.StateWorkspaceDir = DefaultWorkspaceDir
+	}
+
+	return diags
+}
+
+func (b *Local) Workspaces() ([]string, error) {
+	// If we have a backend handling state, defer to that.
+	if b.Backend != nil {
+		return b.Backend.Workspaces()
+	}
+
+	// the listing always start with "default"
+	envs := []string{backend.DefaultStateName}
+
+	entries, err := ioutil.ReadDir(b.stateWorkspaceDir())
+	// no error if there's no envs configured
+	if os.IsNotExist(err) {
+		return envs, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	var listed []string
+	for _, entry := range entries {
+		if entry.IsDir() {
+			listed = append(listed, filepath.Base(entry.Name()))
+		}
+	}
+
+	sort.Strings(listed)
+	envs = append(envs, listed...)
+
+	return envs, nil
+}
+
+// DeleteWorkspace removes a workspace.
+//
+// The "default" workspace cannot be removed.
+func (b *Local) DeleteWorkspace(name string) error {
+	// If we have a backend handling state, defer to that.
+	if b.Backend != nil {
+		return b.Backend.DeleteWorkspace(name)
+	}
+
+	if name == "" {
+		return errors.New("empty state name")
+	}
+
+	if name == backend.DefaultStateName {
+		return errors.New("cannot delete default state")
+	}
+
+	delete(b.states, name)
+	return os.RemoveAll(filepath.Join(b.stateWorkspaceDir(), name))
+}
+
+func (b *Local) StateMgr(name string) (statemgr.Full, error) {
+	// If we have a backend handling state, delegate to that.
+	if b.Backend != nil {
+		return b.Backend.StateMgr(name)
+	}
+
+	if s, ok := b.states[name]; ok {
+		return s, nil
+	}
+
+	if err := b.createState(name); err != nil {
+		return nil, err
+	}
+
+	statePath, stateOutPath, backupPath := b.StatePaths(name)
+	log.Printf("[TRACE] backend/local: state manager for workspace %q will:\n - read initial snapshot from %s\n - write new snapshots to %s\n - create any backup at %s", name, statePath, stateOutPath, backupPath)
+
+	s := statemgr.NewFilesystemBetweenPaths(statePath, stateOutPath)
+	if backupPath != "" {
+		s.SetBackupPath(backupPath)
+	}
+
+	if b.states == nil {
+		b.states = map[string]statemgr.Full{}
+	}
+	b.states[name] = s
+	return s, nil
+}
+
+// Operation implements backend.Enhanced
+//
+// This will initialize an in-memory terraform.Context to perform the
+// operation within this process.
+//
+// The given operation parameter will be merged with the ContextOpts on
+// the structure with the following rules. If a rule isn't specified and the
+// name conflicts, assume that the field is overwritten if set.
+func (b *Local) Operation(ctx context.Context, op *backend.Operation) (*backend.RunningOperation, error) {
+	if op.View == nil {
+		panic("Operation called with nil View")
+	}
+
+	// Determine the function to call for our operation
+	var f func(context.Context, context.Context, *backend.Operation, *backend.RunningOperation)
+	switch op.Type {
+	case backend.OperationTypeRefresh:
+		f = b.opRefresh
+	case backend.OperationTypePlan:
+		f = b.opPlan
+	case backend.OperationTypeApply:
+		f = b.opApply
+	default:
+		return nil, fmt.Errorf(
+			"Unsupported operation type: %s\n\n"+
+				"This is a bug in Terraform and should be reported. The local backend\n"+
+				"is built-in to Terraform and should always support all operations.",
+			op.Type)
+	}
+
+	// Lock
+	b.opLock.Lock()
+
+	// Build our running operation
+	// the runninCtx is only used to block until the operation returns.
+	runningCtx, done := context.WithCancel(context.Background())
+	runningOp := &backend.RunningOperation{
+		Context: runningCtx,
+	}
+
+	// stopCtx wraps the context passed in, and is used to signal a graceful Stop.
+	stopCtx, stop := context.WithCancel(ctx)
+	runningOp.Stop = stop
+
+	// cancelCtx is used to cancel the operation immediately, usually
+	// indicating that the process is exiting.
+	cancelCtx, cancel := context.WithCancel(context.Background())
+	runningOp.Cancel = cancel
+
+	op.StateLocker = op.StateLocker.WithContext(stopCtx)
+
+	// Do it
+	go func() {
+		defer done()
+		defer stop()
+		defer cancel()
+
+		defer b.opLock.Unlock()
+		f(stopCtx, cancelCtx, op, runningOp)
+	}()
+
+	// Return
+	return runningOp, nil
+}
+
+// opWait waits for the operation to complete, and a stop signal or a
+// cancelation signal.
+func (b *Local) opWait(
+	doneCh <-chan struct{},
+	stopCtx context.Context,
+	cancelCtx context.Context,
+	tfCtx *terraform.Context,
+	opStateMgr statemgr.Persister,
+	view views.Operation) (canceled bool) {
+	// Wait for the operation to finish or for us to be interrupted so
+	// we can handle it properly.
+	select {
+	case <-stopCtx.Done():
+		view.Stopping()
+
+		// try to force a PersistState just in case the process is terminated
+		// before we can complete.
+		if err := opStateMgr.PersistState(); err != nil {
+			// We can't error out from here, but warn the user if there was an error.
+			// If this isn't transient, we will catch it again below, and
+			// attempt to save the state another way.
+			var diags tfdiags.Diagnostics
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Error saving current state",
+				fmt.Sprintf(earlyStateWriteErrorFmt, err),
+			))
+			view.Diagnostics(diags)
+		}
+
+		// Stop execution
+		log.Println("[TRACE] backend/local: waiting for the running operation to stop")
+		go tfCtx.Stop()
+
+		select {
+		case <-cancelCtx.Done():
+			log.Println("[WARN] running operation was forcefully canceled")
+			// if the operation was canceled, we need to return immediately
+			canceled = true
+		case <-doneCh:
+			log.Println("[TRACE] backend/local: graceful stop has completed")
+		}
+	case <-cancelCtx.Done():
+		// this should not be called without first attempting to stop the
+		// operation
+		log.Println("[ERROR] running operation canceled without Stop")
+		canceled = true
+	case <-doneCh:
+	}
+	return
+}
+
+// StatePaths returns the StatePath, StateOutPath, and StateBackupPath as
+// configured from the CLI.
+func (b *Local) StatePaths(name string) (stateIn, stateOut, backupOut string) {
+	statePath := b.OverrideStatePath
+	stateOutPath := b.OverrideStateOutPath
+	backupPath := b.OverrideStateBackupPath
+
+	isDefault := name == backend.DefaultStateName || name == ""
+
+	baseDir := ""
+	if !isDefault {
+		baseDir = filepath.Join(b.stateWorkspaceDir(), name)
+	}
+
+	if statePath == "" {
+		if isDefault {
+			statePath = b.StatePath // s.StatePath applies only to the default workspace, since StateWorkspaceDir is used otherwise
+		}
+		if statePath == "" {
+			statePath = filepath.Join(baseDir, DefaultStateFilename)
+		}
+	}
+	if stateOutPath == "" {
+		stateOutPath = statePath
+	}
+	if backupPath == "" {
+		backupPath = b.StateBackupPath
+	}
+	switch backupPath {
+	case "-":
+		backupPath = ""
+	case "":
+		backupPath = stateOutPath + DefaultBackupExtension
+	}
+
+	return statePath, stateOutPath, backupPath
+}
+
+// PathsConflictWith returns true if any state path used by a workspace in
+// the receiver is the same as any state path used by the other given
+// local backend instance.
+//
+// This should be used when "migrating" from one local backend configuration to
+// another in order to avoid deleting the "old" state snapshots if they are
+// in the same files as the "new" state snapshots.
+func (b *Local) PathsConflictWith(other *Local) bool {
+	otherPaths := map[string]struct{}{}
+	otherWorkspaces, err := other.Workspaces()
+	if err != nil {
+		// If we can't enumerate the workspaces then we'll conservatively
+		// assume that paths _do_ overlap, since we can't be certain.
+		return true
+	}
+	for _, name := range otherWorkspaces {
+		p, _, _ := other.StatePaths(name)
+		otherPaths[p] = struct{}{}
+	}
+
+	ourWorkspaces, err := other.Workspaces()
+	if err != nil {
+		// If we can't enumerate the workspaces then we'll conservatively
+		// assume that paths _do_ overlap, since we can't be certain.
+		return true
+	}
+
+	for _, name := range ourWorkspaces {
+		p, _, _ := b.StatePaths(name)
+		if _, exists := otherPaths[p]; exists {
+			return true
+		}
+	}
+	return false
+}
+
+// this only ensures that the named directory exists
+func (b *Local) createState(name string) error {
+	if name == backend.DefaultStateName {
+		return nil
+	}
+
+	stateDir := filepath.Join(b.stateWorkspaceDir(), name)
+	s, err := os.Stat(stateDir)
+	if err == nil && s.IsDir() {
+		// no need to check for os.IsNotExist, since that is covered by os.MkdirAll
+		// which will catch the other possible errors as well.
+		return nil
+	}
+
+	err = os.MkdirAll(stateDir, 0755)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// stateWorkspaceDir returns the directory where state environments are stored.
+func (b *Local) stateWorkspaceDir() string {
+	if b.StateWorkspaceDir != "" {
+		return b.StateWorkspaceDir
+	}
+
+	return DefaultWorkspaceDir
+}
+
+const earlyStateWriteErrorFmt = `Error: %s
+
+Terraform encountered an error attempting to save the state before cancelling the current operation. Once the operation is complete another attempt will be made to save the final state.`

internal/backend/local/backend_apply.go

@@ -0,0 +1,276 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/hashicorp/errwrap"
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/plans"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statefile"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/hashicorp/terraform/tfdiags"
+)
+
+func (b *Local) opApply(
+	stopCtx context.Context,
+	cancelCtx context.Context,
+	op *backend.Operation,
+	runningOp *backend.RunningOperation) {
+	log.Printf("[INFO] backend/local: starting Apply operation")
+
+	var diags tfdiags.Diagnostics
+
+	// If we have a nil module at this point, then set it to an empty tree
+	// to avoid any potential crashes.
+	if op.PlanFile == nil && op.PlanMode != plans.DestroyMode && !op.HasConfig() {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"No configuration files",
+			"Apply requires configuration to be present. Applying without a configuration "+
+				"would mark everything for destruction, which is normally not what is desired. "+
+				"If you would like to destroy everything, run 'terraform destroy' instead.",
+		))
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	stateHook := new(StateHook)
+	op.Hooks = append(op.Hooks, stateHook)
+
+	// Get our context
+	tfCtx, _, opState, contextDiags := b.context(op)
+	diags = diags.Append(contextDiags)
+	if contextDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+	// the state was locked during succesfull context creation; unlock the state
+	// when the operation completes
+	defer func() {
+		diags := op.StateLocker.Unlock()
+		if diags.HasErrors() {
+			op.View.Diagnostics(diags)
+			runningOp.Result = backend.OperationFailure
+		}
+	}()
+
+	runningOp.State = tfCtx.State()
+
+	// If we weren't given a plan, then we refresh/plan
+	if op.PlanFile == nil {
+		// Perform the plan
+		log.Printf("[INFO] backend/local: apply calling Plan")
+		plan, planDiags := tfCtx.Plan()
+		diags = diags.Append(planDiags)
+		if planDiags.HasErrors() {
+			op.ReportResult(runningOp, diags)
+			return
+		}
+
+		trivialPlan := !plan.CanApply()
+		hasUI := op.UIOut != nil && op.UIIn != nil
+		mustConfirm := hasUI && !op.AutoApprove && !trivialPlan
+		op.View.Plan(plan, tfCtx.Schemas())
+
+		if mustConfirm {
+			var desc, query string
+			switch op.PlanMode {
+			case plans.DestroyMode:
+				if op.Workspace != "default" {
+					query = "Do you really want to destroy all resources in workspace \"" + op.Workspace + "\"?"
+				} else {
+					query = "Do you really want to destroy all resources?"
+				}
+				desc = "Terraform will destroy all your managed infrastructure, as shown above.\n" +
+					"There is no undo. Only 'yes' will be accepted to confirm."
+			case plans.RefreshOnlyMode:
+				if op.Workspace != "default" {
+					query = "Would you like to update the Terraform state for \"" + op.Workspace + "\" to reflect these detected changes?"
+				} else {
+					query = "Would you like to update the Terraform state to reflect these detected changes?"
+				}
+				desc = "Terraform will write these changes to the state without modifying any real infrastructure.\n" +
+					"There is no undo. Only 'yes' will be accepted to confirm."
+			default:
+				if op.Workspace != "default" {
+					query = "Do you want to perform these actions in workspace \"" + op.Workspace + "\"?"
+				} else {
+					query = "Do you want to perform these actions?"
+				}
+				desc = "Terraform will perform the actions described above.\n" +
+					"Only 'yes' will be accepted to approve."
+			}
+
+			// We'll show any accumulated warnings before we display the prompt,
+			// so the user can consider them when deciding how to answer.
+			if len(diags) > 0 {
+				op.View.Diagnostics(diags)
+				diags = nil // reset so we won't show the same diagnostics again later
+			}
+
+			v, err := op.UIIn.Input(stopCtx, &terraform.InputOpts{
+				Id:          "approve",
+				Query:       "\n" + query,
+				Description: desc,
+			})
+			if err != nil {
+				diags = diags.Append(errwrap.Wrapf("Error asking for approval: {{err}}", err))
+				op.ReportResult(runningOp, diags)
+				return
+			}
+			if v != "yes" {
+				op.View.Cancelled(op.PlanMode)
+				runningOp.Result = backend.OperationFailure
+				return
+			}
+		}
+	} else {
+		plan, err := op.PlanFile.ReadPlan()
+		if err != nil {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Invalid plan file",
+				fmt.Sprintf("Failed to read plan from plan file: %s.", err),
+			))
+			op.ReportResult(runningOp, diags)
+			return
+		}
+		for _, change := range plan.Changes.Resources {
+			if change.Action != plans.NoOp {
+				op.View.PlannedChange(change)
+			}
+		}
+	}
+
+	// Set up our hook for continuous state updates
+	stateHook.StateMgr = opState
+
+	// Start the apply in a goroutine so that we can be interrupted.
+	var applyState *states.State
+	var applyDiags tfdiags.Diagnostics
+	doneCh := make(chan struct{})
+	go func() {
+		defer close(doneCh)
+		log.Printf("[INFO] backend/local: apply calling Apply")
+		_, applyDiags = tfCtx.Apply()
+		// we always want the state, even if apply failed
+		applyState = tfCtx.State()
+	}()
+
+	if b.opWait(doneCh, stopCtx, cancelCtx, tfCtx, opState, op.View) {
+		return
+	}
+
+	// Store the final state
+	runningOp.State = applyState
+	err := statemgr.WriteAndPersist(opState, applyState)
+	if err != nil {
+		// Export the state file from the state manager and assign the new
+		// state. This is needed to preserve the existing serial and lineage.
+		stateFile := statemgr.Export(opState)
+		if stateFile == nil {
+			stateFile = &statefile.File{}
+		}
+		stateFile.State = applyState
+
+		diags = diags.Append(b.backupStateForError(stateFile, err, op.View))
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	diags = diags.Append(applyDiags)
+	if applyDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	// If we've accumulated any warnings along the way then we'll show them
+	// here just before we show the summary and next steps. If we encountered
+	// errors then we would've returned early at some other point above.
+	op.View.Diagnostics(diags)
+}
+
+// backupStateForError is called in a scenario where we're unable to persist the
+// state for some reason, and will attempt to save a backup copy of the state
+// to local disk to help the user recover. This is a "last ditch effort" sort
+// of thing, so we really don't want to end up in this codepath; we should do
+// everything we possibly can to get the state saved _somewhere_.
+func (b *Local) backupStateForError(stateFile *statefile.File, err error, view views.Operation) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+
+	diags = diags.Append(tfdiags.Sourceless(
+		tfdiags.Error,
+		"Failed to save state",
+		fmt.Sprintf("Error saving state: %s", err),
+	))
+
+	local := statemgr.NewFilesystem("errored.tfstate")
+	writeErr := local.WriteStateForMigration(stateFile, true)
+	if writeErr != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"Failed to create local state file",
+			fmt.Sprintf("Error creating local state file for recovery: %s", writeErr),
+		))
+
+		// To avoid leaving the user with no state at all, our last resort
+		// is to print the JSON state out onto the terminal. This is an awful
+		// UX, so we should definitely avoid doing this if at all possible,
+		// but at least the user has _some_ path to recover if we end up
+		// here for some reason.
+		if dumpErr := view.EmergencyDumpState(stateFile); dumpErr != nil {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Failed to serialize state",
+				fmt.Sprintf(stateWriteFatalErrorFmt, dumpErr),
+			))
+		}
+
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"Failed to persist state to backend",
+			stateWriteConsoleFallbackError,
+		))
+		return diags
+	}
+
+	diags = diags.Append(tfdiags.Sourceless(
+		tfdiags.Error,
+		"Failed to persist state to backend",
+		stateWriteBackedUpError,
+	))
+
+	return diags
+}
+
+const stateWriteBackedUpError = `The error shown above has prevented Terraform from writing the updated state to the configured backend. To allow for recovery, the state has been written to the file "errored.tfstate" in the current working directory.
+
+Running "terraform apply" again at this point will create a forked state, making it harder to recover.
+
+To retry writing this state, use the following command:
+    terraform state push errored.tfstate
+`
+
+const stateWriteConsoleFallbackError = `The errors shown above prevented Terraform from writing the updated state to
+the configured backend and from creating a local backup file. As a fallback,
+the raw state data is printed above as a JSON object.
+
+To retry writing this state, copy the state data (from the first { to the last } inclusive) and save it into a local file called errored.tfstate, then run the following command:
+    terraform state push errored.tfstate
+`
+
+const stateWriteFatalErrorFmt = `Failed to save state after apply.
+
+Error serializing state: %s
+
+A catastrophic error has prevented Terraform from persisting the state file or creating a backup. Unfortunately this means that the record of any resources created during this apply has been lost, and such resources may exist outside of Terraform's management.
+
+For resources that support import, it is possible to recover by manually importing each resource using its id from the target system.
+
+This is a serious bug in Terraform and should be reported.
+`

internal/backend/local/backend_apply_test.go

@@ -0,0 +1,342 @@
+package local
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/zclconf/go-cty/cty"
+
+	"github.com/hashicorp/terraform/command/arguments"
+	"github.com/hashicorp/terraform/command/clistate"
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/initwd"
+	"github.com/hashicorp/terraform/internal/terminal"
+	"github.com/hashicorp/terraform/plans"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/hashicorp/terraform/tfdiags"
+)
+
+func TestLocal_applyBasic(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", applyFixtureSchema())
+	p.ApplyResourceChangeResponse = &providers.ApplyResourceChangeResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id":  cty.StringVal("yes"),
+		"ami": cty.StringVal("bar"),
+	})}
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/apply")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatal("operation failed")
+	}
+
+	if p.ReadResourceCalled {
+		t.Fatal("ReadResource should not be called")
+	}
+
+	if !p.PlanResourceChangeCalled {
+		t.Fatal("diff should be called")
+	}
+
+	if !p.ApplyResourceChangeCalled {
+		t.Fatal("apply should be called")
+	}
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+  ami = bar
+`)
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func TestLocal_applyEmptyDir(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", &terraform.ProviderSchema{})
+	p.ApplyResourceChangeResponse = &providers.ApplyResourceChangeResponse{NewState: cty.ObjectVal(map[string]cty.Value{"id": cty.StringVal("yes")})}
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/empty")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result == backend.OperationSuccess {
+		t.Fatal("operation succeeded; want error")
+	}
+
+	if p.ApplyResourceChangeCalled {
+		t.Fatal("apply should not be called")
+	}
+
+	if _, err := os.Stat(b.StateOutPath); err == nil {
+		t.Fatal("should not exist")
+	}
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+
+	if got, want := done(t).Stderr(), "Error: No configuration files"; !strings.Contains(got, want) {
+		t.Fatalf("unexpected error output:\n%s\nwant: %s", got, want)
+	}
+}
+
+func TestLocal_applyEmptyDirDestroy(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", &terraform.ProviderSchema{})
+	p.ApplyResourceChangeResponse = &providers.ApplyResourceChangeResponse{}
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/empty")
+	defer configCleanup()
+	op.PlanMode = plans.DestroyMode
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("apply operation failed")
+	}
+
+	if p.ApplyResourceChangeCalled {
+		t.Fatal("apply should not be called")
+	}
+
+	checkState(t, b.StateOutPath, `<no state>`)
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func TestLocal_applyError(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	schema := &terraform.ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"ami": {Type: cty.String, Optional: true},
+					"id":  {Type: cty.String, Computed: true},
+				},
+			},
+		},
+	}
+	p := TestLocalProvider(t, b, "test", schema)
+
+	var lock sync.Mutex
+	errored := false
+	p.ApplyResourceChangeFn = func(
+		r providers.ApplyResourceChangeRequest) providers.ApplyResourceChangeResponse {
+
+		lock.Lock()
+		defer lock.Unlock()
+		var diags tfdiags.Diagnostics
+
+		ami := r.Config.GetAttr("ami").AsString()
+		if !errored && ami == "error" {
+			errored = true
+			diags = diags.Append(errors.New("ami error"))
+			return providers.ApplyResourceChangeResponse{
+				Diagnostics: diags,
+			}
+		}
+		return providers.ApplyResourceChangeResponse{
+			Diagnostics: diags,
+			NewState: cty.ObjectVal(map[string]cty.Value{
+				"id":  cty.StringVal("foo"),
+				"ami": cty.StringVal("bar"),
+			}),
+		}
+	}
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/apply-error")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result == backend.OperationSuccess {
+		t.Fatal("operation succeeded; want failure")
+	}
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = foo
+  provider = provider["registry.terraform.io/hashicorp/test"]
+  ami = bar
+	`)
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+
+	if got, want := done(t).Stderr(), "Error: ami error"; !strings.Contains(got, want) {
+		t.Fatalf("unexpected error output:\n%s\nwant: %s", got, want)
+	}
+}
+
+func TestLocal_applyBackendFail(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", applyFixtureSchema())
+	p.ApplyResourceChangeResponse = &providers.ApplyResourceChangeResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id":  cty.StringVal("yes"),
+		"ami": cty.StringVal("bar"),
+	})}
+
+	wd, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("failed to get current working directory")
+	}
+	err = os.Chdir(filepath.Dir(b.StatePath))
+	if err != nil {
+		t.Fatalf("failed to set temporary working directory")
+	}
+	defer os.Chdir(wd)
+
+	op, configCleanup, done := testOperationApply(t, wd+"/testdata/apply")
+	defer configCleanup()
+
+	b.Backend = &backendWithFailingState{}
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	output := done(t)
+
+	if run.Result == backend.OperationSuccess {
+		t.Fatalf("apply succeeded; want error")
+	}
+
+	diagErr := output.Stderr()
+	if !strings.Contains(diagErr, "Error saving state: fake failure") {
+		t.Fatalf("missing \"fake failure\" message in diags:\n%s", diagErr)
+	}
+
+	// The fallback behavior should've created a file errored.tfstate in the
+	// current working directory.
+	checkState(t, "errored.tfstate", `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+  ami = bar
+	`)
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+}
+
+func TestLocal_applyRefreshFalse(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState())
+
+	op, configCleanup, done := testOperationApply(t, "./testdata/plan")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if p.ReadResourceCalled {
+		t.Fatal("ReadResource should not be called")
+	}
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+type backendWithFailingState struct {
+	Local
+}
+
+func (b *backendWithFailingState) StateMgr(name string) (statemgr.Full, error) {
+	return &failingState{
+		statemgr.NewFilesystem("failing-state.tfstate"),
+	}, nil
+}
+
+type failingState struct {
+	*statemgr.Filesystem
+}
+
+func (s failingState) WriteState(state *states.State) error {
+	return errors.New("fake failure")
+}
+
+func testOperationApply(t *testing.T, configDir string) (*backend.Operation, func(), func(*testing.T) *terminal.TestOutput) {
+	t.Helper()
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+
+	streams, done := terminal.StreamsForTesting(t)
+	view := views.NewOperation(arguments.ViewHuman, false, views.NewView(streams))
+
+	return &backend.Operation{
+		Type:         backend.OperationTypeApply,
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		StateLocker:  clistate.NewNoopLocker(),
+		View:         view,
+	}, configCleanup, done
+}
+
+// applyFixtureSchema returns a schema suitable for processing the
+// configuration in testdata/apply . This schema should be
+// assigned to a mock provider named "test".
+func applyFixtureSchema() *terraform.ProviderSchema {
+	return &terraform.ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"ami": {Type: cty.String, Optional: true},
+					"id":  {Type: cty.String, Computed: true},
+				},
+			},
+		},
+	}
+}

internal/backend/local/backend_local.go

@@ -0,0 +1,427 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sort"
+
+	"github.com/hashicorp/errwrap"
+	"github.com/hashicorp/terraform/configs"
+	"github.com/hashicorp/terraform/configs/configload"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/plans/planfile"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/hashicorp/terraform/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+)
+
+// backend.Local implementation.
+func (b *Local) Context(op *backend.Operation) (*terraform.Context, statemgr.Full, tfdiags.Diagnostics) {
+	// Make sure the type is invalid. We use this as a way to know not
+	// to ask for input/validate.
+	op.Type = backend.OperationTypeInvalid
+
+	op.StateLocker = op.StateLocker.WithContext(context.Background())
+
+	ctx, _, stateMgr, diags := b.context(op)
+	return ctx, stateMgr, diags
+}
+
+func (b *Local) context(op *backend.Operation) (*terraform.Context, *configload.Snapshot, statemgr.Full, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+
+	// Get the latest state.
+	log.Printf("[TRACE] backend/local: requesting state manager for workspace %q", op.Workspace)
+	s, err := b.StateMgr(op.Workspace)
+	if err != nil {
+		diags = diags.Append(errwrap.Wrapf("Error loading state: {{err}}", err))
+		return nil, nil, nil, diags
+	}
+	log.Printf("[TRACE] backend/local: requesting state lock for workspace %q", op.Workspace)
+	if diags := op.StateLocker.Lock(s, op.Type.String()); diags.HasErrors() {
+		return nil, nil, nil, diags
+	}
+
+	defer func() {
+		// If we're returning with errors, and thus not producing a valid
+		// context, we'll want to avoid leaving the workspace locked.
+		if diags.HasErrors() {
+			diags = diags.Append(op.StateLocker.Unlock())
+		}
+	}()
+
+	log.Printf("[TRACE] backend/local: reading remote state for workspace %q", op.Workspace)
+	if err := s.RefreshState(); err != nil {
+		diags = diags.Append(errwrap.Wrapf("Error loading state: {{err}}", err))
+		return nil, nil, nil, diags
+	}
+
+	// Initialize our context options
+	var opts terraform.ContextOpts
+	if v := b.ContextOpts; v != nil {
+		opts = *v
+	}
+
+	// Copy set options from the operation
+	opts.PlanMode = op.PlanMode
+	opts.Targets = op.Targets
+	opts.ForceReplace = op.ForceReplace
+	opts.UIInput = op.UIIn
+	opts.Hooks = op.Hooks
+
+	opts.SkipRefresh = op.Type != backend.OperationTypeRefresh && !op.PlanRefresh
+	if opts.SkipRefresh {
+		log.Printf("[DEBUG] backend/local: skipping refresh of managed resources")
+	}
+
+	// Load the latest state. If we enter contextFromPlanFile below then the
+	// state snapshot in the plan file must match this, or else it'll return
+	// error diagnostics.
+	log.Printf("[TRACE] backend/local: retrieving local state snapshot for workspace %q", op.Workspace)
+	opts.State = s.State()
+
+	var tfCtx *terraform.Context
+	var ctxDiags tfdiags.Diagnostics
+	var configSnap *configload.Snapshot
+	if op.PlanFile != nil {
+		var stateMeta *statemgr.SnapshotMeta
+		// If the statemgr implements our optional PersistentMeta interface then we'll
+		// additionally verify that the state snapshot in the plan file has
+		// consistent metadata, as an additional safety check.
+		if sm, ok := s.(statemgr.PersistentMeta); ok {
+			m := sm.StateSnapshotMeta()
+			stateMeta = &m
+		}
+		log.Printf("[TRACE] backend/local: building context from plan file")
+		tfCtx, configSnap, ctxDiags = b.contextFromPlanFile(op.PlanFile, opts, stateMeta)
+		if ctxDiags.HasErrors() {
+			return nil, nil, nil, ctxDiags
+		}
+
+		// Write sources into the cache of the main loader so that they are
+		// available if we need to generate diagnostic message snippets.
+		op.ConfigLoader.ImportSourcesFromSnapshot(configSnap)
+	} else {
+		log.Printf("[TRACE] backend/local: building context for current working directory")
+		tfCtx, configSnap, ctxDiags = b.contextDirect(op, opts)
+	}
+	diags = diags.Append(ctxDiags)
+	if diags.HasErrors() {
+		return nil, nil, nil, diags
+	}
+	log.Printf("[TRACE] backend/local: finished building terraform.Context")
+
+	// If we have an operation, then we automatically do the input/validate
+	// here since every option requires this.
+	if op.Type != backend.OperationTypeInvalid {
+		// If input asking is enabled, then do that
+		if op.PlanFile == nil && b.OpInput {
+			mode := terraform.InputModeProvider
+
+			log.Printf("[TRACE] backend/local: requesting interactive input, if necessary")
+			inputDiags := tfCtx.Input(mode)
+			diags = diags.Append(inputDiags)
+			if inputDiags.HasErrors() {
+				return nil, nil, nil, diags
+			}
+		}
+
+		// If validation is enabled, validate
+		if b.OpValidation {
+			log.Printf("[TRACE] backend/local: running validation operation")
+			validateDiags := tfCtx.Validate()
+			diags = diags.Append(validateDiags)
+		}
+	}
+
+	return tfCtx, configSnap, s, diags
+}
+
+func (b *Local) contextDirect(op *backend.Operation, opts terraform.ContextOpts) (*terraform.Context, *configload.Snapshot, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+
+	// Load the configuration using the caller-provided configuration loader.
+	config, configSnap, configDiags := op.ConfigLoader.LoadConfigWithSnapshot(op.ConfigDir)
+	diags = diags.Append(configDiags)
+	if configDiags.HasErrors() {
+		return nil, nil, diags
+	}
+	opts.Config = config
+
+	var rawVariables map[string]backend.UnparsedVariableValue
+	if op.AllowUnsetVariables {
+		// Rather than prompting for input, we'll just stub out the required
+		// but unset variables with unknown values to represent that they are
+		// placeholders for values the user would need to provide for other
+		// operations.
+		rawVariables = b.stubUnsetRequiredVariables(op.Variables, config.Module.Variables)
+	} else {
+		// If interactive input is enabled, we might gather some more variable
+		// values through interactive prompts.
+		// TODO: Need to route the operation context through into here, so that
+		// the interactive prompts can be sensitive to its timeouts/etc.
+		rawVariables = b.interactiveCollectVariables(context.TODO(), op.Variables, config.Module.Variables, opts.UIInput)
+	}
+
+	variables, varDiags := backend.ParseVariableValues(rawVariables, config.Module.Variables)
+	diags = diags.Append(varDiags)
+	if diags.HasErrors() {
+		return nil, nil, diags
+	}
+	opts.Variables = variables
+
+	tfCtx, ctxDiags := terraform.NewContext(&opts)
+	diags = diags.Append(ctxDiags)
+	return tfCtx, configSnap, diags
+}
+
+func (b *Local) contextFromPlanFile(pf *planfile.Reader, opts terraform.ContextOpts, currentStateMeta *statemgr.SnapshotMeta) (*terraform.Context, *configload.Snapshot, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+
+	const errSummary = "Invalid plan file"
+
+	// A plan file has a snapshot of configuration embedded inside it, which
+	// is used instead of whatever configuration might be already present
+	// in the filesystem.
+	snap, err := pf.ReadConfigSnapshot()
+	if err != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			errSummary,
+			fmt.Sprintf("Failed to read configuration snapshot from plan file: %s.", err),
+		))
+		return nil, snap, diags
+	}
+	loader := configload.NewLoaderFromSnapshot(snap)
+	config, configDiags := loader.LoadConfig(snap.Modules[""].Dir)
+	diags = diags.Append(configDiags)
+	if configDiags.HasErrors() {
+		return nil, snap, diags
+	}
+	opts.Config = config
+
+	// A plan file also contains a snapshot of the prior state the changes
+	// are intended to apply to.
+	priorStateFile, err := pf.ReadStateFile()
+	if err != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			errSummary,
+			fmt.Sprintf("Failed to read prior state snapshot from plan file: %s.", err),
+		))
+		return nil, snap, diags
+	}
+	if currentStateMeta != nil {
+		// If the caller sets this, we require that the stored prior state
+		// has the same metadata, which is an extra safety check that nothing
+		// has changed since the plan was created. (All of the "real-world"
+		// state manager implementations support this, but simpler test backends
+		// may not.)
+		if currentStateMeta.Lineage != "" && priorStateFile.Lineage != "" {
+			if priorStateFile.Serial != currentStateMeta.Serial || priorStateFile.Lineage != currentStateMeta.Lineage {
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Saved plan is stale",
+					"The given plan file can no longer be applied because the state was changed by another operation after the plan was created.",
+				))
+			}
+		}
+	}
+	// The caller already wrote the "current state" here, but we're overriding
+	// it here with the prior state. These two should actually be identical in
+	// normal use, particularly if we validated the state meta above, but
+	// we do this here anyway to ensure consistent behavior.
+	opts.State = priorStateFile.State
+
+	plan, err := pf.ReadPlan()
+	if err != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			errSummary,
+			fmt.Sprintf("Failed to read plan from plan file: %s.", err),
+		))
+		return nil, snap, diags
+	}
+
+	variables := terraform.InputValues{}
+	for name, dyVal := range plan.VariableValues {
+		val, err := dyVal.Decode(cty.DynamicPseudoType)
+		if err != nil {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				errSummary,
+				fmt.Sprintf("Invalid value for variable %q recorded in plan file: %s.", name, err),
+			))
+			continue
+		}
+
+		variables[name] = &terraform.InputValue{
+			Value:      val,
+			SourceType: terraform.ValueFromPlan,
+		}
+	}
+	opts.Variables = variables
+	opts.Changes = plan.Changes
+	opts.Targets = plan.TargetAddrs
+	opts.ForceReplace = plan.ForceReplaceAddrs
+	opts.ProviderSHA256s = plan.ProviderSHA256s
+
+	tfCtx, ctxDiags := terraform.NewContext(&opts)
+	diags = diags.Append(ctxDiags)
+	return tfCtx, snap, diags
+}
+
+// interactiveCollectVariables attempts to complete the given existing
+// map of variables by interactively prompting for any variables that are
+// declared as required but not yet present.
+//
+// If interactive input is disabled for this backend instance then this is
+// a no-op. If input is enabled but fails for some reason, the resulting
+// map will be incomplete. For these reasons, the caller must still validate
+// that the result is complete and valid.
+//
+// This function does not modify the map given in "existing", but may return
+// it unchanged if no modifications are required. If modifications are required,
+// the result is a new map with all of the elements from "existing" plus
+// additional elements as appropriate.
+//
+// Interactive prompting is a "best effort" thing for first-time user UX and
+// not something we expect folks to be relying on for routine use. Terraform
+// is primarily a non-interactive tool and so we prefer to report in error
+// messages that variables are not set rather than reporting that input failed:
+// the primary resolution to missing variables is to provide them by some other
+// means.
+func (b *Local) interactiveCollectVariables(ctx context.Context, existing map[string]backend.UnparsedVariableValue, vcs map[string]*configs.Variable, uiInput terraform.UIInput) map[string]backend.UnparsedVariableValue {
+	var needed []string
+	if b.OpInput && uiInput != nil {
+		for name, vc := range vcs {
+			if !vc.Required() {
+				continue // We only prompt for required variables
+			}
+			if _, exists := existing[name]; !exists {
+				needed = append(needed, name)
+			}
+		}
+	} else {
+		log.Print("[DEBUG] backend/local: Skipping interactive prompts for variables because input is disabled")
+	}
+	if len(needed) == 0 {
+		return existing
+	}
+
+	log.Printf("[DEBUG] backend/local: will prompt for input of unset required variables %s", needed)
+
+	// If we get here then we're planning to prompt for at least one additional
+	// variable's value.
+	sort.Strings(needed) // prompt in lexical order
+	ret := make(map[string]backend.UnparsedVariableValue, len(vcs))
+	for k, v := range existing {
+		ret[k] = v
+	}
+	for _, name := range needed {
+		vc := vcs[name]
+		rawValue, err := uiInput.Input(ctx, &terraform.InputOpts{
+			Id:          fmt.Sprintf("var.%s", name),
+			Query:       fmt.Sprintf("var.%s", name),
+			Description: vc.Description,
+		})
+		if err != nil {
+			// Since interactive prompts are best-effort, we'll just continue
+			// here and let subsequent validation report this as a variable
+			// not specified.
+			log.Printf("[WARN] backend/local: Failed to request user input for variable %q: %s", name, err)
+			continue
+		}
+		ret[name] = unparsedInteractiveVariableValue{Name: name, RawValue: rawValue}
+	}
+	return ret
+}
+
+// stubUnsetVariables ensures that all required variables defined in the
+// configuration exist in the resulting map, by adding new elements as necessary.
+//
+// The stubbed value of any additions will be an unknown variable conforming
+// to the variable's configured type constraint, meaning that no particular
+// value is known and that one must be provided by the user in order to get
+// a complete result.
+//
+// Unset optional attributes (those with default values) will not be populated
+// by this function, under the assumption that a later step will handle those.
+// In this sense, stubUnsetRequiredVariables is essentially a non-interactive,
+// non-error-producing variant of interactiveCollectVariables that creates
+// placeholders for values the user would be prompted for interactively on
+// other operations.
+//
+// This function should be used only in situations where variables values
+// will not be directly used and the variables map is being constructed only
+// to produce a complete Terraform context for some ancillary functionality
+// like "terraform console", "terraform state ...", etc.
+//
+// This function is guaranteed not to modify the given map, but it may return
+// the given map unchanged if no additions are required. If additions are
+// required then the result will be a new map containing everything in the
+// given map plus additional elements.
+func (b *Local) stubUnsetRequiredVariables(existing map[string]backend.UnparsedVariableValue, vcs map[string]*configs.Variable) map[string]backend.UnparsedVariableValue {
+	var missing bool // Do we need to add anything?
+	for name, vc := range vcs {
+		if !vc.Required() {
+			continue // We only stub required variables
+		}
+		if _, exists := existing[name]; !exists {
+			missing = true
+		}
+	}
+	if !missing {
+		return existing
+	}
+
+	// If we get down here then there's at least one variable value to add.
+	ret := make(map[string]backend.UnparsedVariableValue, len(vcs))
+	for k, v := range existing {
+		ret[k] = v
+	}
+	for name, vc := range vcs {
+		if !vc.Required() {
+			continue
+		}
+		if _, exists := existing[name]; !exists {
+			ret[name] = unparsedUnknownVariableValue{Name: name, WantType: vc.Type}
+		}
+	}
+	return ret
+}
+
+type unparsedInteractiveVariableValue struct {
+	Name, RawValue string
+}
+
+var _ backend.UnparsedVariableValue = unparsedInteractiveVariableValue{}
+
+func (v unparsedInteractiveVariableValue) ParseVariableValue(mode configs.VariableParsingMode) (*terraform.InputValue, tfdiags.Diagnostics) {
+	var diags tfdiags.Diagnostics
+	val, valDiags := mode.Parse(v.Name, v.RawValue)
+	diags = diags.Append(valDiags)
+	if diags.HasErrors() {
+		return nil, diags
+	}
+	return &terraform.InputValue{
+		Value:      val,
+		SourceType: terraform.ValueFromInput,
+	}, diags
+}
+
+type unparsedUnknownVariableValue struct {
+	Name     string
+	WantType cty.Type
+}
+
+var _ backend.UnparsedVariableValue = unparsedUnknownVariableValue{}
+
+func (v unparsedUnknownVariableValue) ParseVariableValue(mode configs.VariableParsingMode) (*terraform.InputValue, tfdiags.Diagnostics) {
+	return &terraform.InputValue{
+		Value:      cty.UnknownVal(v.WantType),
+		SourceType: terraform.ValueFromInput,
+	}, nil
+}

internal/backend/local/backend_local_test.go

@@ -0,0 +1,68 @@
+package local
+
+import (
+	"testing"
+
+	"github.com/hashicorp/terraform/command/arguments"
+	"github.com/hashicorp/terraform/command/clistate"
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/initwd"
+	"github.com/hashicorp/terraform/internal/terminal"
+)
+
+func TestLocalContext(t *testing.T) {
+	configDir := "./testdata/empty"
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+	defer configCleanup()
+
+	streams, _ := terminal.StreamsForTesting(t)
+	view := views.NewView(streams)
+	stateLocker := clistate.NewLocker(0, views.NewStateLocker(arguments.ViewHuman, view))
+
+	op := &backend.Operation{
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		Workspace:    backend.DefaultStateName,
+		StateLocker:  stateLocker,
+	}
+
+	_, _, diags := b.Context(op)
+	if diags.HasErrors() {
+		t.Fatalf("unexpected error: %s", diags.Err().Error())
+	}
+
+	// Context() retains a lock on success
+	assertBackendStateLocked(t, b)
+}
+
+func TestLocalContext_error(t *testing.T) {
+	configDir := "./testdata/apply"
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+	defer configCleanup()
+
+	streams, _ := terminal.StreamsForTesting(t)
+	view := views.NewView(streams)
+	stateLocker := clistate.NewLocker(0, views.NewStateLocker(arguments.ViewHuman, view))
+
+	op := &backend.Operation{
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		Workspace:    backend.DefaultStateName,
+		StateLocker:  stateLocker,
+	}
+
+	_, _, diags := b.Context(op)
+	if !diags.HasErrors() {
+		t.Fatal("unexpected success")
+	}
+
+	// Context() unlocks the state on failure
+	assertBackendStateUnlocked(t, b)
+}

internal/backend/local/backend_plan.go

@@ -0,0 +1,157 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/plans"
+	"github.com/hashicorp/terraform/plans/planfile"
+	"github.com/hashicorp/terraform/states/statefile"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/hashicorp/terraform/tfdiags"
+)
+
+func (b *Local) opPlan(
+	stopCtx context.Context,
+	cancelCtx context.Context,
+	op *backend.Operation,
+	runningOp *backend.RunningOperation) {
+
+	log.Printf("[INFO] backend/local: starting Plan operation")
+
+	var diags tfdiags.Diagnostics
+
+	if op.PlanFile != nil {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"Can't re-plan a saved plan",
+			"The plan command was given a saved plan file as its input. This command generates "+
+				"a new plan, and so it requires a configuration directory as its argument.",
+		))
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	// Local planning requires a config, unless we're planning to destroy.
+	if op.PlanMode != plans.DestroyMode && !op.HasConfig() {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Error,
+			"No configuration files",
+			"Plan requires configuration to be present. Planning without a configuration would "+
+				"mark everything for destruction, which is normally not what is desired. If you "+
+				"would like to destroy everything, run plan with the -destroy option. Otherwise, "+
+				"create a Terraform configuration file (.tf file) and try again.",
+		))
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	if b.ContextOpts == nil {
+		b.ContextOpts = new(terraform.ContextOpts)
+	}
+
+	// Get our context
+	tfCtx, configSnap, opState, ctxDiags := b.context(op)
+	diags = diags.Append(ctxDiags)
+	if ctxDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+	// the state was locked during succesfull context creation; unlock the state
+	// when the operation completes
+	defer func() {
+		diags := op.StateLocker.Unlock()
+		if diags.HasErrors() {
+			op.View.Diagnostics(diags)
+			runningOp.Result = backend.OperationFailure
+		}
+	}()
+
+	runningOp.State = tfCtx.State()
+
+	// Perform the plan in a goroutine so we can be interrupted
+	var plan *plans.Plan
+	var planDiags tfdiags.Diagnostics
+	doneCh := make(chan struct{})
+	go func() {
+		defer close(doneCh)
+		log.Printf("[INFO] backend/local: plan calling Plan")
+		plan, planDiags = tfCtx.Plan()
+	}()
+
+	if b.opWait(doneCh, stopCtx, cancelCtx, tfCtx, opState, op.View) {
+		// If we get in here then the operation was cancelled, which is always
+		// considered to be a failure.
+		log.Printf("[INFO] backend/local: plan operation was force-cancelled by interrupt")
+		runningOp.Result = backend.OperationFailure
+		return
+	}
+	log.Printf("[INFO] backend/local: plan operation completed")
+
+	diags = diags.Append(planDiags)
+	if planDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	// Record whether this plan includes any side-effects that could be applied.
+	runningOp.PlanEmpty = !plan.CanApply()
+
+	// Save the plan to disk
+	if path := op.PlanOutPath; path != "" {
+		if op.PlanOutBackend == nil {
+			// This is always a bug in the operation caller; it's not valid
+			// to set PlanOutPath without also setting PlanOutBackend.
+			diags = diags.Append(fmt.Errorf(
+				"PlanOutPath set without also setting PlanOutBackend (this is a bug in Terraform)"),
+			)
+			op.ReportResult(runningOp, diags)
+			return
+		}
+		plan.Backend = *op.PlanOutBackend
+
+		// We may have updated the state in the refresh step above, but we
+		// will freeze that updated state in the plan file for now and
+		// only write it if this plan is subsequently applied.
+		plannedStateFile := statemgr.PlannedStateUpdate(opState, plan.PriorState)
+
+		// We also include a file containing the state as it existed before
+		// we took any action at all, but this one isn't intended to ever
+		// be saved to the backend (an equivalent snapshot should already be
+		// there) and so we just use a stub state file header in this case.
+		// NOTE: This won't be exactly identical to the latest state snapshot
+		// in the backend because it's still been subject to state upgrading
+		// to make it consumable by the current Terraform version, and
+		// intentionally doesn't preserve the header info.
+		prevStateFile := &statefile.File{
+			State: plan.PrevRunState,
+		}
+
+		log.Printf("[INFO] backend/local: writing plan output to: %s", path)
+		err := planfile.Create(path, configSnap, prevStateFile, plannedStateFile, plan)
+		if err != nil {
+			diags = diags.Append(tfdiags.Sourceless(
+				tfdiags.Error,
+				"Failed to write plan file",
+				fmt.Sprintf("The plan file could not be written: %s.", err),
+			))
+			op.ReportResult(runningOp, diags)
+			return
+		}
+	}
+
+	// Render the plan
+	op.View.Plan(plan, tfCtx.Schemas())
+
+	// If we've accumulated any warnings along the way then we'll show them
+	// here just before we show the summary and next steps. If we encountered
+	// errors then we would've returned early at some other point above.
+	op.View.Diagnostics(diags)
+
+	if !runningOp.PlanEmpty {
+		op.View.PlanNextStep(op.PlanOutPath)
+	}
+}

internal/backend/local/backend_plan_test.go

@@ -0,0 +1,878 @@
+package local
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/terraform/addrs"
+	"github.com/hashicorp/terraform/command/arguments"
+	"github.com/hashicorp/terraform/command/clistate"
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/initwd"
+	"github.com/hashicorp/terraform/internal/terminal"
+	"github.com/hashicorp/terraform/plans"
+	"github.com/hashicorp/terraform/plans/planfile"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/terraform"
+	"github.com/zclconf/go-cty/cty"
+)
+
+func TestLocal_planBasic(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanRefresh = true
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if !p.PlanResourceChangeCalled {
+		t.Fatal("PlanResourceChange should be called")
+	}
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func TestLocal_planInAutomation(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+
+	const msg = `You didn't use the -out option`
+
+	// When we're "in automation" we omit certain text from the plan output.
+	// However, the responsibility for this omission is in the view, so here we
+	// test for its presence while the "in automation" setting is false, to
+	// validate that we are calling the correct view method.
+	//
+	// Ideally this test would be replaced by a call-logging mock view, but
+	// that's future work.
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanRefresh = true
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if output := done(t).Stdout(); !strings.Contains(output, msg) {
+		t.Fatalf("missing next-steps message when not in automation\nwant: %s\noutput:\n%s", msg, output)
+	}
+}
+
+func TestLocal_planNoConfig(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	TestLocalProvider(t, b, "test", &terraform.ProviderSchema{})
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/empty")
+	defer configCleanup()
+	op.PlanRefresh = true
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	output := done(t)
+
+	if run.Result == backend.OperationSuccess {
+		t.Fatal("plan operation succeeded; want failure")
+	}
+
+	if stderr := output.Stderr(); !strings.Contains(stderr, "No configuration files") {
+		t.Fatalf("bad: %s", stderr)
+	}
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+}
+
+// This test validates the state lacking behavior when the inner call to
+// Context() fails
+func TestLocal_plan_context_error(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanRefresh = true
+
+	// we coerce a failure in Context() by omitting the provider schema
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationFailure {
+		t.Fatalf("plan operation succeeded")
+	}
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+
+	if got, want := done(t).Stderr(), "Error: Could not load plugin"; !strings.Contains(got, want) {
+		t.Fatalf("unexpected error output:\n%s\nwant: %s", got, want)
+	}
+}
+
+func TestLocal_planOutputsChanged(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	testStateFile(t, b.StatePath, states.BuildState(func(ss *states.SyncState) {
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance,
+			OutputValue: addrs.OutputValue{Name: "changed"},
+		}, cty.StringVal("before"), false)
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance,
+			OutputValue: addrs.OutputValue{Name: "sensitive_before"},
+		}, cty.StringVal("before"), true)
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance,
+			OutputValue: addrs.OutputValue{Name: "sensitive_after"},
+		}, cty.StringVal("before"), false)
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance,
+			OutputValue: addrs.OutputValue{Name: "removed"}, // not present in the config fixture
+		}, cty.StringVal("before"), false)
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance,
+			OutputValue: addrs.OutputValue{Name: "unchanged"},
+		}, cty.StringVal("before"), false)
+		// NOTE: This isn't currently testing the situation where the new
+		// value of an output is unknown, because to do that requires there to
+		// be at least one managed resource Create action in the plan and that
+		// would defeat the point of this test, which is to ensure that a
+		// plan containing only output changes is considered "non-empty".
+		// For now we're not too worried about testing the "new value is
+		// unknown" situation because that's already common for printing out
+		// resource changes and we already have many tests for that.
+	}))
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan-outputs-changed")
+	defer configCleanup()
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+	if run.PlanEmpty {
+		t.Error("plan should not be empty")
+	}
+
+	expectedOutput := strings.TrimSpace(`
+Changes to Outputs:
+  + added            = "after"
+  ~ changed          = "before" -> "after"
+  - removed          = "before" -> null
+  ~ sensitive_after  = (sensitive value)
+  ~ sensitive_before = (sensitive value)
+
+You can apply this plan to save these new output values to the Terraform
+state, without changing any real infrastructure.
+`)
+
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Errorf("Unexpected output:\n%s\n\nwant output containing:\n%s", output, expectedOutput)
+	}
+}
+
+// Module outputs should not cause the plan to be rendered
+func TestLocal_planModuleOutputsChanged(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	testStateFile(t, b.StatePath, states.BuildState(func(ss *states.SyncState) {
+		ss.SetOutputValue(addrs.AbsOutputValue{
+			Module:      addrs.RootModuleInstance.Child("mod", addrs.NoKey),
+			OutputValue: addrs.OutputValue{Name: "changed"},
+		}, cty.StringVal("before"), false)
+	}))
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan-module-outputs-changed")
+	defer configCleanup()
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+	if !run.PlanEmpty {
+		t.Fatal("plan should be empty")
+	}
+
+	expectedOutput := strings.TrimSpace(`
+No changes. Your infrastructure matches the configuration.
+`)
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Fatalf("Unexpected output:\n%s\n\nwant output containing:\n%s", output, expectedOutput)
+	}
+}
+
+func TestLocal_planTainted(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState_tainted())
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+	if !p.ReadResourceCalled {
+		t.Fatal("ReadResource should be called")
+	}
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	expectedOutput := `Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+-/+ destroy and then create replacement
+
+Terraform will perform the following actions:
+
+  # test_instance.foo is tainted, so must be replaced
+-/+ resource "test_instance" "foo" {
+        # (1 unchanged attribute hidden)
+
+        # (1 unchanged block hidden)
+    }
+
+Plan: 1 to add, 0 to change, 1 to destroy.`
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Fatalf("Unexpected output\ngot\n%s\n\nwant:\n%s", output, expectedOutput)
+	}
+}
+
+func TestLocal_planDeposedOnly(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, states.BuildState(func(ss *states.SyncState) {
+		ss.SetResourceInstanceDeposed(
+			addrs.Resource{
+				Mode: addrs.ManagedResourceMode,
+				Type: "test_instance",
+				Name: "foo",
+			}.Instance(addrs.NoKey).Absolute(addrs.RootModuleInstance),
+			states.DeposedKey("00000000"),
+			&states.ResourceInstanceObjectSrc{
+				Status: states.ObjectReady,
+				AttrsJSON: []byte(`{
+				"ami": "bar",
+				"network_interface": [{
+					"device_index": 0,
+					"description": "Main network interface"
+				}]
+			}`),
+			},
+			addrs.AbsProviderConfig{
+				Provider: addrs.NewDefaultProvider("test"),
+				Module:   addrs.RootModule,
+			},
+		)
+	}))
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+	if !p.ReadResourceCalled {
+		t.Fatal("ReadResource should've been called to refresh the deposed object")
+	}
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	// The deposed object and the current object are distinct, so our
+	// plan includes separate actions for each of them. This strange situation
+	// is not common: it should arise only if Terraform fails during
+	// a create-before-destroy when the create hasn't completed yet but
+	// in a severe way that prevents the previous object from being restored
+	// as "current".
+	//
+	// However, that situation was more common in some earlier Terraform
+	// versions where deposed objects were not managed properly, so this
+	// can arise when upgrading from an older version with deposed objects
+	// already in the state.
+	//
+	// This is one of the few cases where we expose the idea of "deposed" in
+	// the UI, including the user-unfriendly "deposed key" (00000000 in this
+	// case) just so that users can correlate this with what they might
+	// see in `terraform show` and in the subsequent apply output, because
+	// it's also possible for there to be _multiple_ deposed objects, in the
+	// unlikely event that create_before_destroy _keeps_ crashing across
+	// subsequent runs.
+	expectedOutput := `Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+  + create
+  - destroy
+
+Terraform will perform the following actions:
+
+  # test_instance.foo will be created
+  + resource "test_instance" "foo" {
+      + ami = "bar"
+
+      + network_interface {
+          + description  = "Main network interface"
+          + device_index = 0
+        }
+    }
+
+  # test_instance.foo (deposed object 00000000) will be destroyed
+  # (left over from a partially-failed replacement of this instance)
+  - resource "test_instance" "foo" {
+      - ami = "bar" -> null
+
+      - network_interface {
+          - description  = "Main network interface" -> null
+          - device_index = 0 -> null
+        }
+    }
+
+Plan: 1 to add, 0 to change, 1 to destroy.`
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Fatalf("Unexpected output:\n%s", output)
+	}
+}
+
+func TestLocal_planTainted_createBeforeDestroy(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState_tainted())
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan-cbd")
+	defer configCleanup()
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+	if !p.ReadResourceCalled {
+		t.Fatal("ReadResource should be called")
+	}
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	expectedOutput := `Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
++/- create replacement and then destroy
+
+Terraform will perform the following actions:
+
+  # test_instance.foo is tainted, so must be replaced
++/- resource "test_instance" "foo" {
+        # (1 unchanged attribute hidden)
+
+        # (1 unchanged block hidden)
+    }
+
+Plan: 1 to add, 0 to change, 1 to destroy.`
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Fatalf("Unexpected output:\n%s", output)
+	}
+}
+
+func TestLocal_planRefreshFalse(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState())
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if p.ReadResourceCalled {
+		t.Fatal("ReadResource should not be called")
+	}
+
+	if !run.PlanEmpty {
+		t.Fatal("plan should be empty")
+	}
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func TestLocal_planDestroy(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState())
+
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanMode = plans.DestroyMode
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	plan := testReadPlan(t, planPath)
+	for _, r := range plan.Changes.Resources {
+		if r.Action.String() != "Delete" {
+			t.Fatalf("bad: %#v", r.Action.String())
+		}
+	}
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func TestLocal_planDestroy_withDataSources(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState_withDataSource())
+
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/destroy-with-ds")
+	defer configCleanup()
+	op.PlanMode = plans.DestroyMode
+	op.PlanRefresh = true
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	if run.PlanEmpty {
+		t.Fatal("plan should not be empty")
+	}
+
+	// Data source should still exist in the the plan file
+	plan := testReadPlan(t, planPath)
+	if len(plan.Changes.Resources) != 2 {
+		t.Fatalf("Expected exactly 1 resource for destruction, %d given: %q",
+			len(plan.Changes.Resources), getAddrs(plan.Changes.Resources))
+	}
+
+	// Data source should not be rendered in the output
+	expectedOutput := `Terraform will perform the following actions:
+
+  # test_instance.foo[0] will be destroyed
+  - resource "test_instance" "foo" {
+      - ami = "bar" -> null
+
+      - network_interface {
+          - description  = "Main network interface" -> null
+          - device_index = 0 -> null
+        }
+    }
+
+Plan: 0 to add, 0 to change, 1 to destroy.`
+
+	if output := done(t).Stdout(); !strings.Contains(output, expectedOutput) {
+		t.Fatalf("Unexpected output:\n%s", output)
+	}
+}
+
+func getAddrs(resources []*plans.ResourceInstanceChangeSrc) []string {
+	addrs := make([]string, len(resources))
+	for i, r := range resources {
+		addrs[i] = r.Addr.String()
+	}
+	return addrs
+}
+
+func TestLocal_planOutPathNoChange(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	TestLocalProvider(t, b, "test", planFixtureSchema())
+	testStateFile(t, b.StatePath, testPlanState())
+
+	outDir := testTempDir(t)
+	defer os.RemoveAll(outDir)
+	planPath := filepath.Join(outDir, "plan.tfplan")
+
+	op, configCleanup, done := testOperationPlan(t, "./testdata/plan")
+	defer configCleanup()
+	op.PlanOutPath = planPath
+	cfg := cty.ObjectVal(map[string]cty.Value{
+		"path": cty.StringVal(b.StatePath),
+	})
+	cfgRaw, err := plans.NewDynamicValue(cfg, cfg.Type())
+	if err != nil {
+		t.Fatal(err)
+	}
+	op.PlanOutBackend = &plans.Backend{
+		// Just a placeholder so that we can generate a valid plan file.
+		Type:   "local",
+		Config: cfgRaw,
+	}
+	op.PlanRefresh = true
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result != backend.OperationSuccess {
+		t.Fatalf("plan operation failed")
+	}
+
+	plan := testReadPlan(t, planPath)
+
+	if !plan.Changes.Empty() {
+		t.Fatalf("expected empty plan to be written")
+	}
+
+	if errOutput := done(t).Stderr(); errOutput != "" {
+		t.Fatalf("unexpected error output:\n%s", errOutput)
+	}
+}
+
+func testOperationPlan(t *testing.T, configDir string) (*backend.Operation, func(), func(*testing.T) *terminal.TestOutput) {
+	t.Helper()
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+
+	streams, done := terminal.StreamsForTesting(t)
+	view := views.NewOperation(arguments.ViewHuman, false, views.NewView(streams))
+
+	return &backend.Operation{
+		Type:         backend.OperationTypePlan,
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		StateLocker:  clistate.NewNoopLocker(),
+		View:         view,
+	}, configCleanup, done
+}
+
+// testPlanState is just a common state that we use for testing plan.
+func testPlanState() *states.State {
+	state := states.NewState()
+	rootModule := state.RootModule()
+	rootModule.SetResourceInstanceCurrent(
+		addrs.Resource{
+			Mode: addrs.ManagedResourceMode,
+			Type: "test_instance",
+			Name: "foo",
+		}.Instance(addrs.IntKey(0)),
+		&states.ResourceInstanceObjectSrc{
+			Status: states.ObjectReady,
+			AttrsJSON: []byte(`{
+				"ami": "bar",
+				"network_interface": [{
+					"device_index": 0,
+					"description": "Main network interface"
+				}]
+			}`),
+		},
+		addrs.AbsProviderConfig{
+			Provider: addrs.NewDefaultProvider("test"),
+			Module:   addrs.RootModule,
+		},
+	)
+	return state
+}
+
+func testPlanState_withDataSource() *states.State {
+	state := states.NewState()
+	rootModule := state.RootModule()
+	rootModule.SetResourceInstanceCurrent(
+		addrs.Resource{
+			Mode: addrs.ManagedResourceMode,
+			Type: "test_instance",
+			Name: "foo",
+		}.Instance(addrs.IntKey(0)),
+		&states.ResourceInstanceObjectSrc{
+			Status: states.ObjectReady,
+			AttrsJSON: []byte(`{
+				"ami": "bar",
+				"network_interface": [{
+					"device_index": 0,
+					"description": "Main network interface"
+				}]
+			}`),
+		},
+		addrs.AbsProviderConfig{
+			Provider: addrs.NewDefaultProvider("test"),
+			Module:   addrs.RootModule,
+		},
+	)
+	rootModule.SetResourceInstanceCurrent(
+		addrs.Resource{
+			Mode: addrs.DataResourceMode,
+			Type: "test_ds",
+			Name: "bar",
+		}.Instance(addrs.IntKey(0)),
+		&states.ResourceInstanceObjectSrc{
+			Status: states.ObjectReady,
+			AttrsJSON: []byte(`{
+				"filter": "foo"
+			}`),
+		},
+		addrs.AbsProviderConfig{
+			Provider: addrs.NewDefaultProvider("test"),
+			Module:   addrs.RootModule,
+		},
+	)
+	return state
+}
+
+func testPlanState_tainted() *states.State {
+	state := states.NewState()
+	rootModule := state.RootModule()
+	rootModule.SetResourceInstanceCurrent(
+		addrs.Resource{
+			Mode: addrs.ManagedResourceMode,
+			Type: "test_instance",
+			Name: "foo",
+		}.Instance(addrs.NoKey),
+		&states.ResourceInstanceObjectSrc{
+			Status: states.ObjectTainted,
+			AttrsJSON: []byte(`{
+				"ami": "bar",
+				"network_interface": [{
+					"device_index": 0,
+					"description": "Main network interface"
+				}]
+			}`),
+		},
+		addrs.AbsProviderConfig{
+			Provider: addrs.NewDefaultProvider("test"),
+			Module:   addrs.RootModule,
+		},
+	)
+	return state
+}
+
+func testReadPlan(t *testing.T, path string) *plans.Plan {
+	t.Helper()
+
+	p, err := planfile.Open(path)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	defer p.Close()
+
+	plan, err := p.ReadPlan()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	return plan
+}
+
+// planFixtureSchema returns a schema suitable for processing the
+// configuration in testdata/plan . This schema should be
+// assigned to a mock provider named "test".
+func planFixtureSchema() *terraform.ProviderSchema {
+	return &terraform.ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"ami": {Type: cty.String, Optional: true},
+				},
+				BlockTypes: map[string]*configschema.NestedBlock{
+					"network_interface": {
+						Nesting: configschema.NestingList,
+						Block: configschema.Block{
+							Attributes: map[string]*configschema.Attribute{
+								"device_index": {Type: cty.Number, Optional: true},
+								"description":  {Type: cty.String, Optional: true},
+							},
+						},
+					},
+				},
+			},
+		},
+		DataSources: map[string]*configschema.Block{
+			"test_ds": {
+				Attributes: map[string]*configschema.Attribute{
+					"filter": {Type: cty.String, Required: true},
+				},
+			},
+		},
+	}
+}

internal/backend/local/backend_refresh.go

@@ -0,0 +1,106 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/hashicorp/errwrap"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/tfdiags"
+)
+
+func (b *Local) opRefresh(
+	stopCtx context.Context,
+	cancelCtx context.Context,
+	op *backend.Operation,
+	runningOp *backend.RunningOperation) {
+
+	var diags tfdiags.Diagnostics
+
+	// Check if our state exists if we're performing a refresh operation. We
+	// only do this if we're managing state with this backend.
+	if b.Backend == nil {
+		if _, err := os.Stat(b.StatePath); err != nil {
+			if os.IsNotExist(err) {
+				err = nil
+			}
+
+			if err != nil {
+				diags = diags.Append(tfdiags.Sourceless(
+					tfdiags.Error,
+					"Cannot read state file",
+					fmt.Sprintf("Failed to read %s: %s", b.StatePath, err),
+				))
+				op.ReportResult(runningOp, diags)
+				return
+			}
+		}
+	}
+
+	// Refresh now happens via a plan, so we need to ensure this is enabled
+	op.PlanRefresh = true
+
+	// Get our context
+	tfCtx, _, opState, contextDiags := b.context(op)
+	diags = diags.Append(contextDiags)
+	if contextDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	// the state was locked during succesfull context creation; unlock the state
+	// when the operation completes
+	defer func() {
+		diags := op.StateLocker.Unlock()
+		if diags.HasErrors() {
+			op.View.Diagnostics(diags)
+			runningOp.Result = backend.OperationFailure
+		}
+	}()
+
+	// Set our state
+	runningOp.State = opState.State()
+	if !runningOp.State.HasResources() {
+		diags = diags.Append(tfdiags.Sourceless(
+			tfdiags.Warning,
+			"Empty or non-existent state",
+			"There are currently no resources tracked in the state, so there is nothing to refresh.",
+		))
+	}
+
+	// Perform the refresh in a goroutine so we can be interrupted
+	var newState *states.State
+	var refreshDiags tfdiags.Diagnostics
+	doneCh := make(chan struct{})
+	go func() {
+		defer close(doneCh)
+		newState, refreshDiags = tfCtx.Refresh()
+		log.Printf("[INFO] backend/local: refresh calling Refresh")
+	}()
+
+	if b.opWait(doneCh, stopCtx, cancelCtx, tfCtx, opState, op.View) {
+		return
+	}
+
+	// Write the resulting state to the running op
+	runningOp.State = newState
+	diags = diags.Append(refreshDiags)
+	if refreshDiags.HasErrors() {
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	err := statemgr.WriteAndPersist(opState, newState)
+	if err != nil {
+		diags = diags.Append(errwrap.Wrapf("Failed to write state: {{err}}", err))
+		op.ReportResult(runningOp, diags)
+		return
+	}
+
+	// Show any remaining warnings before exiting
+	op.ReportResult(runningOp, diags)
+}

internal/backend/local/backend_refresh_test.go

@@ -0,0 +1,307 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/terraform/addrs"
+	"github.com/hashicorp/terraform/command/arguments"
+	"github.com/hashicorp/terraform/command/clistate"
+	"github.com/hashicorp/terraform/command/views"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/internal/initwd"
+	"github.com/hashicorp/terraform/internal/terminal"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/terraform"
+
+	"github.com/zclconf/go-cty/cty"
+)
+
+func TestLocal_refresh(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", refreshFixtureSchema())
+	testStateFile(t, b.StatePath, testRefreshState())
+
+	p.ReadResourceFn = nil
+	p.ReadResourceResponse = &providers.ReadResourceResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id": cty.StringVal("yes"),
+	})}
+
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/refresh")
+	defer configCleanup()
+	defer done(t)
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	if !p.ReadResourceCalled {
+		t.Fatal("ReadResource should be called")
+	}
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+	`)
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+}
+
+func TestLocal_refreshInput(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	schema := &terraform.ProviderSchema{
+		Provider: &configschema.Block{
+			Attributes: map[string]*configschema.Attribute{
+				"value": {Type: cty.String, Optional: true},
+			},
+		},
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"id":  {Type: cty.String, Computed: true},
+					"foo": {Type: cty.String, Optional: true},
+					"ami": {Type: cty.String, Optional: true},
+				},
+			},
+		},
+	}
+
+	p := TestLocalProvider(t, b, "test", schema)
+	testStateFile(t, b.StatePath, testRefreshState())
+
+	p.ReadResourceFn = nil
+	p.ReadResourceResponse = &providers.ReadResourceResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id": cty.StringVal("yes"),
+	})}
+	p.ConfigureProviderFn = func(req providers.ConfigureProviderRequest) (resp providers.ConfigureProviderResponse) {
+		val := req.Config.GetAttr("value")
+		if val.IsNull() || val.AsString() != "bar" {
+			resp.Diagnostics = resp.Diagnostics.Append(fmt.Errorf("incorrect value %#v", val))
+		}
+
+		return
+	}
+
+	// Enable input asking since it is normally disabled by default
+	b.OpInput = true
+	b.ContextOpts.UIInput = &terraform.MockUIInput{InputReturnString: "bar"}
+
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/refresh-var-unset")
+	defer configCleanup()
+	defer done(t)
+	op.UIIn = b.ContextOpts.UIInput
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	if !p.ReadResourceCalled {
+		t.Fatal("ReadResource should be called")
+	}
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+	`)
+}
+
+func TestLocal_refreshValidate(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	p := TestLocalProvider(t, b, "test", refreshFixtureSchema())
+	testStateFile(t, b.StatePath, testRefreshState())
+	p.ReadResourceFn = nil
+	p.ReadResourceResponse = &providers.ReadResourceResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id": cty.StringVal("yes"),
+	})}
+
+	// Enable validation
+	b.OpValidation = true
+
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/refresh")
+	defer configCleanup()
+	defer done(t)
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+	`)
+}
+
+func TestLocal_refreshValidateProviderConfigured(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	schema := &terraform.ProviderSchema{
+		Provider: &configschema.Block{
+			Attributes: map[string]*configschema.Attribute{
+				"value": {Type: cty.String, Optional: true},
+			},
+		},
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"id":  {Type: cty.String, Computed: true},
+					"ami": {Type: cty.String, Optional: true},
+				},
+			},
+		},
+	}
+
+	p := TestLocalProvider(t, b, "test", schema)
+	testStateFile(t, b.StatePath, testRefreshState())
+	p.ReadResourceFn = nil
+	p.ReadResourceResponse = &providers.ReadResourceResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id": cty.StringVal("yes"),
+	})}
+
+	// Enable validation
+	b.OpValidation = true
+
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/refresh-provider-config")
+	defer configCleanup()
+	defer done(t)
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	if !p.ValidateProviderConfigCalled {
+		t.Fatal("Validate provider config should be called")
+	}
+
+	checkState(t, b.StateOutPath, `
+test_instance.foo:
+  ID = yes
+  provider = provider["registry.terraform.io/hashicorp/test"]
+	`)
+}
+
+// This test validates the state lacking behavior when the inner call to
+// Context() fails
+func TestLocal_refresh_context_error(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+	testStateFile(t, b.StatePath, testRefreshState())
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/apply")
+	defer configCleanup()
+	defer done(t)
+
+	// we coerce a failure in Context() by omitting the provider schema
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+	if run.Result == backend.OperationSuccess {
+		t.Fatal("operation succeeded; want failure")
+	}
+	assertBackendStateUnlocked(t, b)
+}
+
+func TestLocal_refreshEmptyState(t *testing.T) {
+	b, cleanup := TestLocal(t)
+	defer cleanup()
+
+	p := TestLocalProvider(t, b, "test", refreshFixtureSchema())
+	testStateFile(t, b.StatePath, states.NewState())
+
+	p.ReadResourceFn = nil
+	p.ReadResourceResponse = &providers.ReadResourceResponse{NewState: cty.ObjectVal(map[string]cty.Value{
+		"id": cty.StringVal("yes"),
+	})}
+
+	op, configCleanup, done := testOperationRefresh(t, "./testdata/refresh")
+	defer configCleanup()
+
+	run, err := b.Operation(context.Background(), op)
+	if err != nil {
+		t.Fatalf("bad: %s", err)
+	}
+	<-run.Done()
+
+	output := done(t)
+
+	if stderr := output.Stderr(); stderr != "" {
+		t.Fatalf("expected only warning diags, got errors: %s", stderr)
+	}
+	if got, want := output.Stdout(), "Warning: Empty or non-existent state"; !strings.Contains(got, want) {
+		t.Errorf("wrong diags\n got: %s\nwant: %s", got, want)
+	}
+
+	// the backend should be unlocked after a run
+	assertBackendStateUnlocked(t, b)
+}
+
+func testOperationRefresh(t *testing.T, configDir string) (*backend.Operation, func(), func(*testing.T) *terminal.TestOutput) {
+	t.Helper()
+
+	_, configLoader, configCleanup := initwd.MustLoadConfigForTests(t, configDir)
+
+	streams, done := terminal.StreamsForTesting(t)
+	view := views.NewOperation(arguments.ViewHuman, false, views.NewView(streams))
+
+	return &backend.Operation{
+		Type:         backend.OperationTypeRefresh,
+		ConfigDir:    configDir,
+		ConfigLoader: configLoader,
+		StateLocker:  clistate.NewNoopLocker(),
+		View:         view,
+	}, configCleanup, done
+}
+
+// testRefreshState is just a common state that we use for testing refresh.
+func testRefreshState() *states.State {
+	state := states.NewState()
+	root := state.EnsureModule(addrs.RootModuleInstance)
+	root.SetResourceInstanceCurrent(
+		mustResourceInstanceAddr("test_instance.foo").Resource,
+		&states.ResourceInstanceObjectSrc{
+			Status:    states.ObjectReady,
+			AttrsJSON: []byte(`{"id":"bar"}`),
+		},
+		mustProviderConfig(`provider["registry.terraform.io/hashicorp/test"]`),
+	)
+	return state
+}
+
+// refreshFixtureSchema returns a schema suitable for processing the
+// configuration in testdata/refresh . This schema should be
+// assigned to a mock provider named "test".
+func refreshFixtureSchema() *terraform.ProviderSchema {
+	return &terraform.ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test_instance": {
+				Attributes: map[string]*configschema.Attribute{
+					"ami": {Type: cty.String, Optional: true},
+					"id":  {Type: cty.String, Computed: true},
+				},
+			},
+		},
+	}
+}

internal/backend/local/backend_test.go

@@ -0,0 +1,250 @@
+package local
+
+import (
+	"errors"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/states/statefile"
+	"github.com/hashicorp/terraform/states/statemgr"
+)
+
+func TestLocal_impl(t *testing.T) {
+	var _ backend.Enhanced = New()
+	var _ backend.Local = New()
+	var _ backend.CLI = New()
+}
+
+func TestLocal_backend(t *testing.T) {
+	defer testTmpDir(t)()
+	b := New()
+	backend.TestBackendStates(t, b)
+	backend.TestBackendStateLocks(t, b, b)
+}
+
+func checkState(t *testing.T, path, expected string) {
+	t.Helper()
+	// Read the state
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	state, err := statefile.Read(f)
+	f.Close()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	actual := state.State.String()
+	expected = strings.TrimSpace(expected)
+	if actual != expected {
+		t.Fatalf("state does not match! actual:\n%s\n\nexpected:\n%s", actual, expected)
+	}
+}
+
+func TestLocal_StatePaths(t *testing.T) {
+	b := New()
+
+	// Test the defaults
+	path, out, back := b.StatePaths("")
+
+	if path != DefaultStateFilename {
+		t.Fatalf("expected %q, got %q", DefaultStateFilename, path)
+	}
+
+	if out != DefaultStateFilename {
+		t.Fatalf("expected %q, got %q", DefaultStateFilename, out)
+	}
+
+	dfltBackup := DefaultStateFilename + DefaultBackupExtension
+	if back != dfltBackup {
+		t.Fatalf("expected %q, got %q", dfltBackup, back)
+	}
+
+	// check with env
+	testEnv := "test_env"
+	path, out, back = b.StatePaths(testEnv)
+
+	expectedPath := filepath.Join(DefaultWorkspaceDir, testEnv, DefaultStateFilename)
+	expectedOut := expectedPath
+	expectedBackup := expectedPath + DefaultBackupExtension
+
+	if path != expectedPath {
+		t.Fatalf("expected %q, got %q", expectedPath, path)
+	}
+
+	if out != expectedOut {
+		t.Fatalf("expected %q, got %q", expectedOut, out)
+	}
+
+	if back != expectedBackup {
+		t.Fatalf("expected %q, got %q", expectedBackup, back)
+	}
+
+}
+
+func TestLocal_addAndRemoveStates(t *testing.T) {
+	defer testTmpDir(t)()
+	dflt := backend.DefaultStateName
+	expectedStates := []string{dflt}
+
+	b := New()
+	states, err := b.Workspaces()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(states, expectedStates) {
+		t.Fatalf("expected []string{%q}, got %q", dflt, states)
+	}
+
+	expectedA := "test_A"
+	if _, err := b.StateMgr(expectedA); err != nil {
+		t.Fatal(err)
+	}
+
+	states, err = b.Workspaces()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedStates = append(expectedStates, expectedA)
+	if !reflect.DeepEqual(states, expectedStates) {
+		t.Fatalf("expected %q, got %q", expectedStates, states)
+	}
+
+	expectedB := "test_B"
+	if _, err := b.StateMgr(expectedB); err != nil {
+		t.Fatal(err)
+	}
+
+	states, err = b.Workspaces()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedStates = append(expectedStates, expectedB)
+	if !reflect.DeepEqual(states, expectedStates) {
+		t.Fatalf("expected %q, got %q", expectedStates, states)
+	}
+
+	if err := b.DeleteWorkspace(expectedA); err != nil {
+		t.Fatal(err)
+	}
+
+	states, err = b.Workspaces()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedStates = []string{dflt, expectedB}
+	if !reflect.DeepEqual(states, expectedStates) {
+		t.Fatalf("expected %q, got %q", expectedStates, states)
+	}
+
+	if err := b.DeleteWorkspace(expectedB); err != nil {
+		t.Fatal(err)
+	}
+
+	states, err = b.Workspaces()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedStates = []string{dflt}
+	if !reflect.DeepEqual(states, expectedStates) {
+		t.Fatalf("expected %q, got %q", expectedStates, states)
+	}
+
+	if err := b.DeleteWorkspace(dflt); err == nil {
+		t.Fatal("expected error deleting default state")
+	}
+}
+
+// a local backend which returns sentinel errors for NamedState methods to
+// verify it's being called.
+type testDelegateBackend struct {
+	*Local
+
+	// return a sentinel error on these calls
+	stateErr  bool
+	statesErr bool
+	deleteErr bool
+}
+
+var errTestDelegateState = errors.New("State called")
+var errTestDelegateStates = errors.New("States called")
+var errTestDelegateDeleteState = errors.New("Delete called")
+
+func (b *testDelegateBackend) StateMgr(name string) (statemgr.Full, error) {
+	if b.stateErr {
+		return nil, errTestDelegateState
+	}
+	s := statemgr.NewFilesystem("terraform.tfstate")
+	return s, nil
+}
+
+func (b *testDelegateBackend) Workspaces() ([]string, error) {
+	if b.statesErr {
+		return nil, errTestDelegateStates
+	}
+	return []string{"default"}, nil
+}
+
+func (b *testDelegateBackend) DeleteWorkspace(name string) error {
+	if b.deleteErr {
+		return errTestDelegateDeleteState
+	}
+	return nil
+}
+
+// verify that the MultiState methods are dispatched to the correct Backend.
+func TestLocal_multiStateBackend(t *testing.T) {
+	// assign a separate backend where we can read the state
+	b := NewWithBackend(&testDelegateBackend{
+		stateErr:  true,
+		statesErr: true,
+		deleteErr: true,
+	})
+
+	if _, err := b.StateMgr("test"); err != errTestDelegateState {
+		t.Fatal("expected errTestDelegateState, got:", err)
+	}
+
+	if _, err := b.Workspaces(); err != errTestDelegateStates {
+		t.Fatal("expected errTestDelegateStates, got:", err)
+	}
+
+	if err := b.DeleteWorkspace("test"); err != errTestDelegateDeleteState {
+		t.Fatal("expected errTestDelegateDeleteState, got:", err)
+	}
+}
+
+// change into a tmp dir and return a deferable func to change back and cleanup
+func testTmpDir(t *testing.T) func() {
+	tmp, err := ioutil.TempDir("", "tf")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	old, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := os.Chdir(tmp); err != nil {
+		t.Fatal(err)
+	}
+
+	return func() {
+		// ignore errors and try to clean up
+		os.Chdir(old)
+		os.RemoveAll(tmp)
+	}
+}

internal/backend/local/cli.go

@@ -0,0 +1,32 @@
+package local
+
+import (
+	"log"
+
+	"github.com/hashicorp/terraform/internal/backend"
+)
+
+// backend.CLI impl.
+func (b *Local) CLIInit(opts *backend.CLIOpts) error {
+	b.ContextOpts = opts.ContextOpts
+	b.OpInput = opts.Input
+	b.OpValidation = opts.Validation
+
+	// configure any new cli options
+	if opts.StatePath != "" {
+		log.Printf("[TRACE] backend/local: CLI option -state is overriding state path to %s", opts.StatePath)
+		b.OverrideStatePath = opts.StatePath
+	}
+
+	if opts.StateOutPath != "" {
+		log.Printf("[TRACE] backend/local: CLI option -state-out is overriding state output path to %s", opts.StateOutPath)
+		b.OverrideStateOutPath = opts.StateOutPath
+	}
+
+	if opts.StateBackupPath != "" {
+		log.Printf("[TRACE] backend/local: CLI option -backup is overriding state backup path to %s", opts.StateBackupPath)
+		b.OverrideStateBackupPath = opts.StateBackupPath
+	}
+
+	return nil
+}

internal/backend/local/hook_state.go

@@ -0,0 +1,33 @@
+package local
+
+import (
+	"sync"
+
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+)
+
+// StateHook is a hook that continuously updates the state by calling
+// WriteState on a statemgr.Full.
+type StateHook struct {
+	terraform.NilHook
+	sync.Mutex
+
+	StateMgr statemgr.Writer
+}
+
+var _ terraform.Hook = (*StateHook)(nil)
+
+func (h *StateHook) PostStateUpdate(new *states.State) (terraform.HookAction, error) {
+	h.Lock()
+	defer h.Unlock()
+
+	if h.StateMgr != nil {
+		if err := h.StateMgr.WriteState(new); err != nil {
+			return terraform.HookActionHalt, err
+		}
+	}
+
+	return terraform.HookActionContinue, nil
+}

internal/backend/local/hook_state_test.go

@@ -0,0 +1,29 @@
+package local
+
+import (
+	"testing"
+
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+)
+
+func TestStateHook_impl(t *testing.T) {
+	var _ terraform.Hook = new(StateHook)
+}
+
+func TestStateHook(t *testing.T) {
+	is := statemgr.NewTransientInMemory(nil)
+	var hook terraform.Hook = &StateHook{StateMgr: is}
+
+	s := statemgr.TestFullInitialState()
+	action, err := hook.PostStateUpdate(s)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if action != terraform.HookActionContinue {
+		t.Fatalf("bad: %v", action)
+	}
+	if !is.State().Equal(s) {
+		t.Fatalf("bad state: %#v", is.State())
+	}
+}

internal/backend/local/local_test.go

@@ -0,0 +1,14 @@
+package local
+
+import (
+	"flag"
+	"os"
+	"testing"
+
+	_ "github.com/hashicorp/terraform/internal/logging"
+)
+
+func TestMain(m *testing.M) {
+	flag.Parse()
+	os.Exit(m.Run())
+}

internal/backend/local/testdata/apply-empty/hello.txt

@@ -0,0 +1 @@
+This is an empty dir

internal/backend/local/testdata/apply-error/main.tf

@@ -0,0 +1,7 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+}
+
+resource "test_instance" "bar" {
+    ami = "error"
+}

internal/backend/local/testdata/apply/main.tf

@@ -0,0 +1,3 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+}

internal/backend/local/testdata/destroy-with-ds/main.tf

@@ -0,0 +1,8 @@
+resource "test_instance" "foo" {
+  count = 1
+  ami = "bar"
+}
+
+data "test_ds" "bar" {
+  filter = "foo"
+}

internal/backend/local/testdata/plan-cbd/main.tf

@@ -0,0 +1,13 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+
+    # This is here because at some point it caused a test failure
+    network_interface {
+      device_index = 0
+      description = "Main network interface"
+    }
+
+    lifecycle {
+      create_before_destroy = true
+    }
+}

internal/backend/local/testdata/plan-module-outputs-changed/main.tf

@@ -0,0 +1,3 @@
+module "mod" {
+  source = "./mod"
+}

internal/backend/local/testdata/plan-module-outputs-changed/mod/main.tf

@@ -0,0 +1,3 @@
+output "changed" {
+  value = "after"
+}

internal/backend/local/testdata/plan-outputs-changed/main.tf

@@ -0,0 +1,28 @@
+module "submodule" {
+  source = "./submodule"
+}
+
+output "changed" {
+  value = "after"
+}
+
+output "sensitive_before" {
+  value = "after"
+  # no sensitive = true here, but the prior state is marked as sensitive in the test code
+}
+
+output "sensitive_after" {
+  value = "after"
+
+  # This one is _not_ sensitive in the prior state, but is transitioning to
+  # being sensitive in our new plan.
+  sensitive = true
+}
+
+output "added" { // not present in the prior state
+  value = "after"
+}
+
+output "unchanged" {
+  value = "before"
+}

internal/backend/local/testdata/plan-outputs-changed/submodule/main.tf

@@ -0,0 +1,3 @@
+output "foo" {
+  value = "bar"
+}

internal/backend/local/testdata/plan/main.tf

@@ -0,0 +1,9 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+
+    # This is here because at some point it caused a test failure
+    network_interface {
+      device_index = 0
+      description = "Main network interface"
+    }
+}

internal/backend/local/testdata/refresh-provider-config/main.tf

@@ -0,0 +1,7 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+}
+
+provider "test" {
+    value = "foo"
+}

internal/backend/local/testdata/refresh-var-unset/main.tf

@@ -0,0 +1,9 @@
+variable "should_ask" {}
+
+provider "test" {
+  value = var.should_ask
+}
+
+resource "test_instance" "foo" {
+  foo = "bar"
+}

internal/backend/local/testdata/refresh/main.tf

@@ -0,0 +1,3 @@
+resource "test_instance" "foo" {
+    ami = "bar"
+}

internal/backend/local/testing.go

@@ -0,0 +1,246 @@
+package local
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/zclconf/go-cty/cty"
+
+	"github.com/hashicorp/terraform/addrs"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/internal/backend"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/states"
+	"github.com/hashicorp/terraform/states/statemgr"
+	"github.com/hashicorp/terraform/terraform"
+)
+
+// TestLocal returns a configured Local struct with temporary paths and
+// in-memory ContextOpts.
+//
+// No operations will be called on the returned value, so you can still set
+// public fields without any locks.
+func TestLocal(t *testing.T) (*Local, func()) {
+	t.Helper()
+	tempDir := testTempDir(t)
+
+	local := New()
+	local.StatePath = filepath.Join(tempDir, "state.tfstate")
+	local.StateOutPath = filepath.Join(tempDir, "state.tfstate")
+	local.StateBackupPath = filepath.Join(tempDir, "state.tfstate.bak")
+	local.StateWorkspaceDir = filepath.Join(tempDir, "state.tfstate.d")
+	local.ContextOpts = &terraform.ContextOpts{}
+
+	cleanup := func() {
+		if err := os.RemoveAll(tempDir); err != nil {
+			t.Fatal("error cleanup up test:", err)
+		}
+	}
+
+	return local, cleanup
+}
+
+// TestLocalProvider modifies the ContextOpts of the *Local parameter to
+// have a provider with the given name.
+func TestLocalProvider(t *testing.T, b *Local, name string, schema *terraform.ProviderSchema) *terraform.MockProvider {
+	// Build a mock resource provider for in-memory operations
+	p := new(terraform.MockProvider)
+
+	if schema == nil {
+		schema = &terraform.ProviderSchema{} // default schema is empty
+	}
+	p.GetProviderSchemaResponse = &providers.GetProviderSchemaResponse{
+		Provider:      providers.Schema{Block: schema.Provider},
+		ProviderMeta:  providers.Schema{Block: schema.ProviderMeta},
+		ResourceTypes: map[string]providers.Schema{},
+		DataSources:   map[string]providers.Schema{},
+	}
+	for name, res := range schema.ResourceTypes {
+		p.GetProviderSchemaResponse.ResourceTypes[name] = providers.Schema{
+			Block:   res,
+			Version: int64(schema.ResourceTypeSchemaVersions[name]),
+		}
+	}
+	for name, dat := range schema.DataSources {
+		p.GetProviderSchemaResponse.DataSources[name] = providers.Schema{Block: dat}
+	}
+
+	p.PlanResourceChangeFn = func(req providers.PlanResourceChangeRequest) providers.PlanResourceChangeResponse {
+		rSchema, _ := schema.SchemaForResourceType(addrs.ManagedResourceMode, req.TypeName)
+		if rSchema == nil {
+			rSchema = &configschema.Block{} // default schema is empty
+		}
+		plannedVals := map[string]cty.Value{}
+		for name, attrS := range rSchema.Attributes {
+			val := req.ProposedNewState.GetAttr(name)
+			if attrS.Computed && val.IsNull() {
+				val = cty.UnknownVal(attrS.Type)
+			}
+			plannedVals[name] = val
+		}
+		for name := range rSchema.BlockTypes {
+			// For simplicity's sake we just copy the block attributes over
+			// verbatim, since this package's mock providers are all relatively
+			// simple -- we're testing the backend, not esoteric provider features.
+			plannedVals[name] = req.ProposedNewState.GetAttr(name)
+		}
+
+		return providers.PlanResourceChangeResponse{
+			PlannedState:   cty.ObjectVal(plannedVals),
+			PlannedPrivate: req.PriorPrivate,
+		}
+	}
+	p.ReadResourceFn = func(req providers.ReadResourceRequest) providers.ReadResourceResponse {
+		return providers.ReadResourceResponse{NewState: req.PriorState}
+	}
+	p.ReadDataSourceFn = func(req providers.ReadDataSourceRequest) providers.ReadDataSourceResponse {
+		return providers.ReadDataSourceResponse{State: req.Config}
+	}
+
+	// Initialize the opts
+	if b.ContextOpts == nil {
+		b.ContextOpts = &terraform.ContextOpts{}
+	}
+
+	// Set up our provider
+	b.ContextOpts.Providers = map[addrs.Provider]providers.Factory{
+		addrs.NewDefaultProvider(name): providers.FactoryFixed(p),
+	}
+
+	return p
+
+}
+
+// TestLocalSingleState is a backend implementation that wraps Local
+// and modifies it to only support single states (returns
+// ErrWorkspacesNotSupported for multi-state operations).
+//
+// This isn't an actual use case, this is exported just to provide a
+// easy way to test that behavior.
+type TestLocalSingleState struct {
+	*Local
+}
+
+// TestNewLocalSingle is a factory for creating a TestLocalSingleState.
+// This function matches the signature required for backend/init.
+func TestNewLocalSingle() backend.Backend {
+	return &TestLocalSingleState{Local: New()}
+}
+
+func (b *TestLocalSingleState) Workspaces() ([]string, error) {
+	return nil, backend.ErrWorkspacesNotSupported
+}
+
+func (b *TestLocalSingleState) DeleteWorkspace(string) error {
+	return backend.ErrWorkspacesNotSupported
+}
+
+func (b *TestLocalSingleState) StateMgr(name string) (statemgr.Full, error) {
+	if name != backend.DefaultStateName {
+		return nil, backend.ErrWorkspacesNotSupported
+	}
+
+	return b.Local.StateMgr(name)
+}
+
+// TestLocalNoDefaultState is a backend implementation that wraps
+// Local and modifies it to support named states, but not the
+// default state. It returns ErrDefaultWorkspaceNotSupported when
+// the DefaultStateName is used.
+type TestLocalNoDefaultState struct {
+	*Local
+}
+
+// TestNewLocalNoDefault is a factory for creating a TestLocalNoDefaultState.
+// This function matches the signature required for backend/init.
+func TestNewLocalNoDefault() backend.Backend {
+	return &TestLocalNoDefaultState{Local: New()}
+}
+
+func (b *TestLocalNoDefaultState) Workspaces() ([]string, error) {
+	workspaces, err := b.Local.Workspaces()
+	if err != nil {
+		return nil, err
+	}
+
+	filtered := workspaces[:0]
+	for _, name := range workspaces {
+		if name != backend.DefaultStateName {
+			filtered = append(filtered, name)
+		}
+	}
+
+	return filtered, nil
+}
+
+func (b *TestLocalNoDefaultState) DeleteWorkspace(name string) error {
+	if name == backend.DefaultStateName {
+		return backend.ErrDefaultWorkspaceNotSupported
+	}
+	return b.Local.DeleteWorkspace(name)
+}
+
+func (b *TestLocalNoDefaultState) StateMgr(name string) (statemgr.Full, error) {
+	if name == backend.DefaultStateName {
+		return nil, backend.ErrDefaultWorkspaceNotSupported
+	}
+	return b.Local.StateMgr(name)
+}
+
+func testTempDir(t *testing.T) string {
+	d, err := ioutil.TempDir("", "tf")
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	return d
+}
+
+func testStateFile(t *testing.T, path string, s *states.State) {
+	stateFile := statemgr.NewFilesystem(path)
+	stateFile.WriteState(s)
+}
+
+func mustProviderConfig(s string) addrs.AbsProviderConfig {
+	p, diags := addrs.ParseAbsProviderConfigStr(s)
+	if diags.HasErrors() {
+		panic(diags.Err())
+	}
+	return p
+}
+
+func mustResourceInstanceAddr(s string) addrs.AbsResourceInstance {
+	addr, diags := addrs.ParseAbsResourceInstanceStr(s)
+	if diags.HasErrors() {
+		panic(diags.Err())
+	}
+	return addr
+}
+
+// assertBackendStateUnlocked attempts to lock the backend state. Failure
+// indicates that the state was indeed locked and therefore this function will
+// return true.
+func assertBackendStateUnlocked(t *testing.T, b *Local) bool {
+	t.Helper()
+	stateMgr, _ := b.StateMgr(backend.DefaultStateName)
+	if _, err := stateMgr.Lock(statemgr.NewLockInfo()); err != nil {
+		t.Errorf("state is already locked: %s", err.Error())
+		return false
+	}
+	return true
+}
+
+// assertBackendStateLocked attempts to lock the backend state. Failure
+// indicates that the state was already locked and therefore this function will
+// return false.
+func assertBackendStateLocked(t *testing.T, b *Local) bool {
+	t.Helper()
+	stateMgr, _ := b.StateMgr(backend.DefaultStateName)
+	if _, err := stateMgr.Lock(statemgr.NewLockInfo()); err != nil {
+		return true
+	}
+	t.Error("unexpected success locking state")
+	return true
+}