diff --git a/helper/resource/testing.go b/helper/resource/testing.go
index be7c01c6c..83d299801 100644
--- a/helper/resource/testing.go
+++ b/helper/resource/testing.go
@@ -560,12 +560,25 @@ type TestStep struct {
 	//---------------------------------------------------------------
 	// ImportState testing
 	//---------------------------------------------------------------
+	// Terraform has two workflows for importing resources: the import CLI
+	// command, which writes directly to state, and the import block in HCL,
+	// which imports to state via the normal plan and apply workflow.
 
 	// ImportState, if true, will test the functionality of ImportState
 	// by importing the resource with ResourceName (must be set) and the
 	// ID of that resource.
+	// By default, the "terraform import" command will be run. To test import
+	// block functionality instead, set ImportBlock to true.
 	ImportState bool
 
+	// ImportBlock, if true, enables a sub-mode of ImportState testing. In this
+	// mode, an import block is added to the config, and plan and apply are run.
+	ImportBlock bool
+
+	// ImportBlockConfig is an optional string with the import block
+	// configuration to use when ImportBlock is true.
+	ImportBlockConfig string
+
 	// ImportStateId is the ID to perform an ImportState operation with.
 	// This is optional. If it isn't set, then the resource ID is automatically
 	// determined by inspecting the state for ResourceName's ID.
diff --git a/helper/resource/testing_new.go b/helper/resource/testing_new.go
index da4785f0a..9e4234911 100644
--- a/helper/resource/testing_new.go
+++ b/helper/resource/testing_new.go
@@ -214,7 +214,12 @@ func runNewTest(ctx context.Context, t testing.T, c TestCase, helper *plugintest
 		if step.ImportState {
 			logging.HelperResourceTrace(ctx, "TestStep is ImportState mode")
 
-			err := testStepNewImportState(ctx, t, helper, wd, step, appliedCfg, providers)
+			var err error
+			if step.ImportBlock {
+				err = testStepNewImportBlock(ctx, t, helper, wd, step, appliedCfg, providers)
+			} else {
+				err = testStepNewImportState(ctx, t, helper, wd, step, appliedCfg, providers)
+			}
 			if step.ExpectError != nil {
 				logging.HelperResourceDebug(ctx, "Checking TestStep ExpectError")
 				if err == nil {
diff --git a/helper/resource/testing_new_import_block.go b/helper/resource/testing_new_import_block.go
new file mode 100644
index 000000000..4663f3c4b
--- /dev/null
+++ b/helper/resource/testing_new_import_block.go
@@ -0,0 +1,305 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package resource
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/mitchellh/go-testing-interface"
+
+	"github.com/hashicorp/terraform-plugin-testing/terraform"
+
+	"github.com/hashicorp/terraform-plugin-testing/internal/logging"
+	"github.com/hashicorp/terraform-plugin-testing/internal/plugintest"
+)
+
+// Generates an import block, then plans and applies the import.
+// Optionally attempts to generate resource configuration during the plan step.
+func testStepNewImportBlock(ctx context.Context, t testing.T, helper *plugintest.Helper, wd *plugintest.WorkingDir, step TestStep, cfg string, providers *providerFactories) error {
+	t.Helper()
+
+	if step.ResourceName == "" {
+		t.Fatal("ResourceName is required for an import state test")
+	}
+
+	// get state from check sequence
+	var state *terraform.State
+	var err error
+	err = runProviderCommand(ctx, t, func() error {
+		state, err = getState(ctx, t, wd)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, wd, providers)
+	if err != nil {
+		t.Fatalf("Error getting state: %s", err)
+	}
+
+	// Determine the ID to import
+	var importId string
+	switch {
+	case step.ImportStateIdFunc != nil:
+		logging.HelperResourceTrace(ctx, "Using TestStep ImportStateIdFunc for import identifier")
+
+		var err error
+
+		logging.HelperResourceDebug(ctx, "Calling TestStep ImportStateIdFunc")
+
+		importId, err = step.ImportStateIdFunc(state)
+
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		logging.HelperResourceDebug(ctx, "Called TestStep ImportStateIdFunc")
+	case step.ImportStateId != "":
+		logging.HelperResourceTrace(ctx, "Using TestStep ImportStateId for import identifier")
+
+		importId = step.ImportStateId
+	default:
+		logging.HelperResourceTrace(ctx, "Using resource identifier for import identifier")
+
+		resource, err := testResource(step, state)
+		if err != nil {
+			t.Fatal(err)
+		}
+		importId = resource.Primary.ID
+	}
+
+	if step.ImportStateIdPrefix != "" {
+		logging.HelperResourceTrace(ctx, "Prepending TestStep ImportStateIdPrefix for import identifier")
+
+		importId = step.ImportStateIdPrefix + importId
+	}
+
+	logging.HelperResourceTrace(ctx, fmt.Sprintf("Using import identifier: %s", importId))
+
+	// Create working directory for import tests
+	if step.Config == "" {
+		logging.HelperResourceTrace(ctx, "Using prior TestStep Config for import")
+
+		step.Config = cfg
+		if step.Config == "" {
+			t.Fatal("Cannot import state with no specified config")
+		}
+	}
+
+	var importWd *plugintest.WorkingDir
+
+	// Use the same working directory to persist the state from import
+	if step.ImportStatePersist {
+		importWd = wd
+	} else {
+		importWd = helper.RequireNewWorkingDir(ctx, t, "")
+		defer importWd.Close()
+	}
+
+	var importBlockConfig string
+
+	if step.ImportBlockConfig != "" {
+		importBlockConfig = step.ImportBlockConfig + "\n"
+	} else {
+		importBlockConfig = fmt.Sprintf(`import {
+  to = %s
+  id = "%s"
+}
+`, step.ResourceName, importId)
+	}
+
+	err = importWd.SetConfig(ctx, importBlockConfig+step.Config)
+	if err != nil {
+		t.Fatalf("Error setting test config: %s", err)
+	}
+
+	logging.HelperResourceDebug(ctx, "Running Terraform CLI init and plan")
+
+	if !step.ImportStatePersist {
+		err = runProviderCommand(ctx, t, func() error {
+			return importWd.Init(ctx)
+		}, importWd, providers)
+		if err != nil {
+			t.Fatalf("Error running init: %s", err)
+		}
+	}
+
+	err = runProviderCommand(ctx, t, func() error {
+		return importWd.CreatePlan(ctx)
+	}, importWd, providers)
+	if err != nil {
+		return err
+	}
+
+	logging.HelperResourceDebug(ctx, "Running Terraform CLI apply")
+
+	err = runProviderCommand(ctx, t, func() error {
+		return importWd.Apply(ctx)
+	}, importWd, providers)
+	if err != nil {
+		return err
+	}
+
+	var importState *terraform.State
+	err = runProviderCommand(ctx, t, func() error {
+		importState, err = getState(ctx, t, importWd)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, importWd, providers)
+	if err != nil {
+		t.Fatalf("Error getting state: %s", err)
+	}
+
+	// Go through the imported state and verify
+	if step.ImportStateCheck != nil {
+		logging.HelperResourceTrace(ctx, "Using TestStep ImportStateCheck")
+
+		var states []*terraform.InstanceState
+		for address, r := range importState.RootModule().Resources {
+			if strings.HasPrefix(address, "data.") {
+				continue
+			}
+
+			if r.Primary == nil {
+				continue
+			}
+
+			is := r.Primary.DeepCopy()
+			is.Ephemeral.Type = r.Type // otherwise the check function cannot see the type
+			states = append(states, is)
+		}
+
+		logging.HelperResourceDebug(ctx, "Calling TestStep ImportStateCheck")
+
+		if err := step.ImportStateCheck(states); err != nil {
+			t.Fatal(err)
+		}
+
+		logging.HelperResourceDebug(ctx, "Called TestStep ImportStateCheck")
+	}
+
+	// Verify that all the states match
+	if step.ImportStateVerify {
+		logging.HelperResourceTrace(ctx, "Using TestStep ImportStateVerify")
+
+		// Ensure that we do not match against data sources as they
+		// cannot be imported and are not what we want to verify.
+		// Mode is not present in ResourceState so we use the
+		// stringified ResourceStateKey for comparison.
+		newResources := make(map[string]*terraform.ResourceState)
+		for k, v := range importState.RootModule().Resources {
+			if !strings.HasPrefix(k, "data.") {
+				newResources[k] = v
+			}
+		}
+		oldResources := make(map[string]*terraform.ResourceState)
+		for k, v := range state.RootModule().Resources {
+			if !strings.HasPrefix(k, "data.") {
+				oldResources[k] = v
+			}
+		}
+
+		for _, r := range newResources {
+			// Find the existing resource
+			var oldR *terraform.ResourceState
+			for _, r2 := range oldResources {
+
+				if r2.Primary != nil && r2.Primary.ID == r.Primary.ID && r2.Type == r.Type && r2.Provider == r.Provider {
+					oldR = r2
+					break
+				}
+			}
+			if oldR == nil || oldR.Primary == nil {
+				t.Fatalf(
+					"Failed state verification, resource with ID %s not found",
+					r.Primary.ID)
+			}
+
+			// don't add empty flatmapped containers, so we can more easily
+			// compare the attributes
+			skipEmpty := func(k, v string) bool {
+				if strings.HasSuffix(k, ".#") || strings.HasSuffix(k, ".%") {
+					if v == "0" {
+						return true
+					}
+				}
+				return false
+			}
+
+			// Compare their attributes
+			actual := make(map[string]string)
+			for k, v := range r.Primary.Attributes {
+				if skipEmpty(k, v) {
+					continue
+				}
+				actual[k] = v
+			}
+
+			expected := make(map[string]string)
+			for k, v := range oldR.Primary.Attributes {
+				if skipEmpty(k, v) {
+					continue
+				}
+				expected[k] = v
+			}
+
+			// Remove fields we're ignoring
+			for _, v := range step.ImportStateVerifyIgnore {
+				for k := range actual {
+					if strings.HasPrefix(k, v) {
+						delete(actual, k)
+					}
+				}
+				for k := range expected {
+					if strings.HasPrefix(k, v) {
+						delete(expected, k)
+					}
+				}
+			}
+
+			// timeouts are only _sometimes_ added to state. To
+			// account for this, just don't compare timeouts at
+			// all.
+			for k := range actual {
+				if strings.HasPrefix(k, "timeouts.") {
+					delete(actual, k)
+				}
+				if k == "timeouts" {
+					delete(actual, k)
+				}
+			}
+			for k := range expected {
+				if strings.HasPrefix(k, "timeouts.") {
+					delete(expected, k)
+				}
+				if k == "timeouts" {
+					delete(expected, k)
+				}
+			}
+
+			if !reflect.DeepEqual(actual, expected) {
+				// Determine only the different attributes
+				// go-cmp tries to show surrounding identical map key/value for
+				// context of differences, which may be confusing.
+				for k, v := range expected {
+					if av, ok := actual[k]; ok && v == av {
+						delete(expected, k)
+						delete(actual, k)
+					}
+				}
+
+				if diff := cmp.Diff(expected, actual); diff != "" {
+					return fmt.Errorf("ImportStateVerify attributes not equivalent. Difference is shown below. The - symbol indicates attributes missing after import.\n\n%s", diff)
+				}
+			}
+		}
+	}
+
+	return nil
+}
diff --git a/helper/resource/testing_new_import_block_test.go b/helper/resource/testing_new_import_block_test.go
new file mode 100644
index 000000000..1ec971c6a
--- /dev/null
+++ b/helper/resource/testing_new_import_block_test.go
@@ -0,0 +1,284 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package resource
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"testing"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+
+	"github.com/hashicorp/terraform-plugin-testing/terraform"
+)
+
+func TestTest_TestStep_ImportBlock_ImportStateCheck_SkipDataSourceState(t *testing.T) {
+	t.Parallel()
+
+	UnitTest(t, TestCase{
+		ProviderFactories: map[string]func() (*schema.Provider, error){
+			"examplecloud": func() (*schema.Provider, error) { //nolint:unparam // required signature
+				return &schema.Provider{
+					DataSourcesMap: map[string]*schema.Resource{
+						"examplecloud_thing": {
+							ReadContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								d.SetId("datasource-test")
+
+								return nil
+							},
+							Schema: map[string]*schema.Schema{
+								"id": {
+									Type:     schema.TypeString,
+									Computed: true,
+								},
+							},
+						},
+					},
+					ResourcesMap: map[string]*schema.Resource{
+						"examplecloud_thing": {
+							CreateContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								d.SetId("resource-test")
+
+								return nil
+							},
+							DeleteContext: func(_ context.Context, _ *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								return nil
+							},
+							ReadContext: func(_ context.Context, _ *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								return nil
+							},
+							Schema: map[string]*schema.Schema{
+								"id": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+							},
+							Importer: &schema.ResourceImporter{
+								StateContext: schema.ImportStatePassthroughContext,
+							},
+						},
+					},
+				}, nil
+			},
+		},
+		Steps: []TestStep{
+			{
+				Config: `
+					data "examplecloud_thing" "test" {}
+					resource "examplecloud_thing" "test" {}
+				`,
+			},
+			{
+				ResourceName: "examplecloud_thing.test",
+				ImportState:  true,
+				ImportBlock:  true,
+				ImportStateCheck: func(is []*terraform.InstanceState) error {
+					if len(is) > 1 {
+						return fmt.Errorf("expected 1 state, got: %d", len(is))
+					}
+
+					return nil
+				},
+			},
+		},
+	})
+}
+
+func TestTest_TestStep_ImportBlock_ImportStateVerify(t *testing.T) {
+	t.Parallel()
+
+	UnitTest(t, TestCase{
+		ProviderFactories: map[string]func() (*schema.Provider, error){
+			"examplecloud": func() (*schema.Provider, error) { //nolint:unparam // required signature
+				return &schema.Provider{
+					ResourcesMap: map[string]*schema.Resource{
+						"examplecloud_thing": {
+							CreateContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								d.SetId("resource-test")
+
+								return nil
+							},
+							DeleteContext: func(_ context.Context, _ *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								return nil
+							},
+							ReadContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								_ = d.Set("other", "testvalue")
+
+								return nil
+							},
+							Schema: map[string]*schema.Schema{
+								"other": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+								"id": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+							},
+							Importer: &schema.ResourceImporter{
+								StateContext: schema.ImportStatePassthroughContext,
+							},
+						},
+					},
+				}, nil
+			},
+		},
+		Steps: []TestStep{
+			{
+				Config: `resource "examplecloud_thing" "test" {}`,
+			},
+			{
+				ResourceName:      "examplecloud_thing.test",
+				ImportState:       true,
+				ImportBlock:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestTest_TestStep_ImportBlock_ImportBlockConfig(t *testing.T) {
+	t.Parallel()
+
+	UnitTest(t, TestCase{
+		ProviderFactories: map[string]func() (*schema.Provider, error){
+			"examplecloud": func() (*schema.Provider, error) { //nolint:unparam // required signature
+				return &schema.Provider{
+					ResourcesMap: map[string]*schema.Resource{
+						"examplecloud_thing": {
+							CreateContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								d.SetId("resource-test")
+
+								return nil
+							},
+							DeleteContext: func(_ context.Context, _ *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								return nil
+							},
+							ReadContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								_ = d.Set("other", "testvalue")
+
+								return nil
+							},
+							Schema: map[string]*schema.Schema{
+								"other": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+								"id": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+							},
+							Importer: &schema.ResourceImporter{
+								StateContext: schema.ImportStatePassthroughContext,
+							},
+						},
+					},
+				}, nil
+			},
+		},
+		Steps: []TestStep{
+			{
+				Config: `resource "examplecloud_thing" "test" {}`,
+			},
+			{
+				ResourceName: "examplecloud_thing.test",
+				ImportState:  true,
+				ImportBlock:  true,
+				ImportBlockConfig: `import {
+  to = examplecloud_thing.test
+  id = "resource-test"
+}`,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestTest_TestStep_ImportBlock_ImportStateVerifyIgnore(t *testing.T) {
+	t.Parallel()
+
+	UnitTest(t, TestCase{
+		ProviderFactories: map[string]func() (*schema.Provider, error){
+			"examplecloud": func() (*schema.Provider, error) { //nolint:unparam // required signature
+				return &schema.Provider{
+					ResourcesMap: map[string]*schema.Resource{
+						"examplecloud_thing": {
+							CreateContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								d.SetId("resource-test")
+
+								_ = d.Set("create_only", "testvalue")
+
+								return nil
+							},
+							DeleteContext: func(_ context.Context, _ *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								return nil
+							},
+							ReadContext: func(_ context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+								_ = d.Set("read_only", "testvalue")
+
+								return nil
+							},
+							Schema: map[string]*schema.Schema{
+								"create_only": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+								"read_only": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+								"id": {
+									Computed: true,
+									Type:     schema.TypeString,
+								},
+							},
+							Importer: &schema.ResourceImporter{
+								StateContext: schema.ImportStatePassthroughContext,
+							},
+						},
+					},
+				}, nil
+			},
+		},
+		Steps: []TestStep{
+			{
+				Config: `resource "examplecloud_thing" "test" {}`,
+			},
+			{
+				ResourceName:            "examplecloud_thing.test",
+				ImportState:             true,
+				ImportBlock:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"create_only"},
+			},
+		},
+	})
+}
+
+func TestTest_TestStep_ExpectError_ImportBlock(t *testing.T) {
+	t.Parallel()
+
+	Test(t, TestCase{
+		ExternalProviders: map[string]ExternalProvider{
+			"random": {
+				Source:            "registry.terraform.io/hashicorp/time",
+				VersionConstraint: "0.9.1",
+			},
+		},
+		Steps: []TestStep{
+			{
+				Config:        `resource "time_static" "one" {}`,
+				ImportStateId: "invalid time string",
+				ResourceName:  "time_static.one",
+				ImportState:   true,
+				ImportBlock:   true,
+				ExpectError:   regexp.MustCompile(`Error: Import time static error`),
+			},
+		},
+	})
+}