diff --git a/internal/configs/configload/loader_load.go b/internal/configs/configload/loader_load.go
index 1dd4c95f433b..b431e5333966 100644
--- a/internal/configs/configload/loader_load.go
+++ b/internal/configs/configload/loader_load.go
@@ -22,7 +22,16 @@ import (
 // LoadConfig performs the basic syntax and uniqueness validations that are
 // required to process the individual modules
 func (l *Loader) LoadConfig(rootDir string) (*configs.Config, hcl.Diagnostics) {
-	rootMod, diags := l.parser.LoadConfigDir(rootDir)
+	return l.loadConfig(l.parser.LoadConfigDir(rootDir))
+}
+
+// LoadConfigWithTests matches LoadConfig, except the configs.Config contains
+// any relevant .tftest files.
+func (l *Loader) LoadConfigWithTests(rootDir string, testDir string) (*configs.Config, hcl.Diagnostics) {
+	return l.loadConfig(l.parser.LoadConfigDirWithTests(rootDir, testDir))
+}
+
+func (l *Loader) loadConfig(rootMod *configs.Module, diags hcl.Diagnostics) (*configs.Config, hcl.Diagnostics) {
 	if rootMod == nil || diags.HasErrors() {
 		// Ensure we return any parsed modules here so that required_version
 		// constraints can be verified even when encountering errors.
diff --git a/internal/configs/module.go b/internal/configs/module.go
index b281f39d9ef9..995863eff452 100644
--- a/internal/configs/module.go
+++ b/internal/configs/module.go
@@ -53,6 +53,8 @@ type Module struct {
 	Import []*Import
 
 	Checks map[string]*Check
+
+	Tests map[string]*TestFile
 }
 
 // File describes the contents of a single configuration file.
@@ -92,6 +94,16 @@ type File struct {
 	Checks []*Check
 }
 
+// NewModuleWithTests matches NewModule except it will also load in the provided
+// test files.
+func NewModuleWithTests(primaryFiles, overrideFiles []*File, testFiles map[string]*TestFile) (*Module, hcl.Diagnostics) {
+	mod, diags := NewModule(primaryFiles, overrideFiles)
+	if mod != nil {
+		mod.Tests = testFiles
+	}
+	return mod, diags
+}
+
 // NewModule takes a list of primary files and a list of override files and
 // produces a *Module by combining the files together.
 //
@@ -113,6 +125,7 @@ func NewModule(primaryFiles, overrideFiles []*File) (*Module, hcl.Diagnostics) {
 		DataResources:      map[string]*Resource{},
 		Checks:             map[string]*Check{},
 		ProviderMetas:      map[addrs.Provider]*ProviderMeta{},
+		Tests:              map[string]*TestFile{},
 	}
 
 	// Process the required_providers blocks first, to ensure that all
diff --git a/internal/configs/parser_config.go b/internal/configs/parser_config.go
index f1a1c553d635..967197e003c6 100644
--- a/internal/configs/parser_config.go
+++ b/internal/configs/parser_config.go
@@ -32,6 +32,22 @@ func (p *Parser) LoadConfigFileOverride(path string) (*File, hcl.Diagnostics) {
 	return p.loadConfigFile(path, true)
 }
 
+// LoadTestFile reads the file at the given path and parses it as a Terraform
+// test file.
+//
+// It references the same LoadHCLFile as LoadConfigFile, so inherits the same
+// syntax selection behaviours.
+func (p *Parser) LoadTestFile(path string) (*TestFile, hcl.Diagnostics) {
+	body, diags := p.LoadHCLFile(path)
+	if body == nil {
+		return nil, diags
+	}
+
+	test, testDiags := loadTestFile(body)
+	diags = append(diags, testDiags...)
+	return test, diags
+}
+
 func (p *Parser) loadConfigFile(path string, override bool) (*File, hcl.Diagnostics) {
 	body, diags := p.LoadHCLFile(path)
 	if body == nil {
diff --git a/internal/configs/parser_config_dir.go b/internal/configs/parser_config_dir.go
index eebc12da37f8..97ece4f001a2 100644
--- a/internal/configs/parser_config_dir.go
+++ b/internal/configs/parser_config_dir.go
@@ -32,7 +32,7 @@ import (
 // .tf files are parsed using the HCL native syntax while .tf.json files are
 // parsed using the HCL JSON syntax.
 func (p *Parser) LoadConfigDir(path string) (*Module, hcl.Diagnostics) {
-	primaryPaths, overridePaths, diags := p.dirFiles(path)
+	primaryPaths, overridePaths, _, diags := p.dirFiles(path, "")
 	if diags.HasErrors() {
 		return nil, diags
 	}
@@ -50,20 +50,51 @@ func (p *Parser) LoadConfigDir(path string) (*Module, hcl.Diagnostics) {
 	return mod, diags
 }
 
+// LoadConfigDirWithTests matches LoadConfigDir, but the return Module also
+// contains any relevant .tftest files.
+func (p *Parser) LoadConfigDirWithTests(path string, testDirectory string) (*Module, hcl.Diagnostics) {
+	primaryPaths, overridePaths, testPaths, diags := p.dirFiles(path, testDirectory)
+	if diags.HasErrors() {
+		return nil, diags
+	}
+
+	primary, fDiags := p.loadFiles(primaryPaths, false)
+	diags = append(diags, fDiags...)
+	override, fDiags := p.loadFiles(overridePaths, true)
+	diags = append(diags, fDiags...)
+	tests, fDiags := p.loadTestFiles(path, testPaths)
+	diags = append(diags, fDiags...)
+
+	mod, modDiags := NewModuleWithTests(primary, override, tests)
+	diags = append(diags, modDiags...)
+
+	mod.SourceDir = path
+
+	return mod, diags
+}
+
 // ConfigDirFiles returns lists of the primary and override files configuration
 // files in the given directory.
 //
 // If the given directory does not exist or cannot be read, error diagnostics
 // are returned. If errors are returned, the resulting lists may be incomplete.
 func (p Parser) ConfigDirFiles(dir string) (primary, override []string, diags hcl.Diagnostics) {
-	return p.dirFiles(dir)
+	primary, override, _, diags = p.dirFiles(dir, "")
+	return primary, override, diags
+}
+
+// ConfigDirFilesWithTests matches ConfigDirFiles except it also returns the
+// paths to any test files within the module.
+func (p Parser) ConfigDirFilesWithTests(dir string, testDirectory string) (primary, override, tests []string, diags hcl.Diagnostics) {
+	return p.dirFiles(dir, testDirectory)
 }
 
 // IsConfigDir determines whether the given path refers to a directory that
 // exists and contains at least one Terraform config file (with a .tf or
-// .tf.json extension.)
+// .tf.json extension.). Note, we explicitely exclude checking for tests here
+// as tests must live alongside actual .tf config files.
 func (p *Parser) IsConfigDir(path string) bool {
-	primaryPaths, overridePaths, _ := p.dirFiles(path)
+	primaryPaths, overridePaths, _, _ := p.dirFiles(path, "")
 	return (len(primaryPaths) + len(overridePaths)) > 0
 }
 
@@ -88,7 +119,16 @@ func (p *Parser) loadFiles(paths []string, override bool) ([]*File, hcl.Diagnost
 	return files, diags
 }
 
-func (p *Parser) dirFiles(dir string) (primary, override []string, diags hcl.Diagnostics) {
+// dirFiles finds Terraform configuration files within dir, splitting them into
+// primary and override files based on the filename.
+//
+// If testsDir is not empty, dirFiles will also retrieve Terraform testing files
+// both directly within dir and within testsDir as a subdirectory of dir. In
+// this way, testsDir acts both as a direction to retrieve test files within the
+// main direction and as the location for additional test files.
+func (p *Parser) dirFiles(dir string, testsDir string) (primary, override, tests []string, diags hcl.Diagnostics) {
+	includeTests := len(testsDir) > 0
+
 	infos, err := p.fs.ReadDir(dir)
 	if err != nil {
 		diags = append(diags, &hcl.Diagnostic{
@@ -101,7 +141,31 @@ func (p *Parser) dirFiles(dir string) (primary, override []string, diags hcl.Dia
 
 	for _, info := range infos {
 		if info.IsDir() {
-			// We only care about files
+			if includeTests && info.Name() == testsDir {
+				testsDir := filepath.Join(dir, info.Name())
+				testInfos, err := p.fs.ReadDir(testsDir)
+				if err != nil {
+					diags = append(diags, &hcl.Diagnostic{
+						Severity: hcl.DiagError,
+						Summary:  "Failed to read module test directory",
+						Detail:   fmt.Sprintf("Module test directory %s does not exist or cannot be read.", testsDir),
+					})
+					return
+				}
+
+				for _, testInfo := range testInfos {
+					if testInfo.IsDir() || IsIgnoredFile(testInfo.Name()) {
+						continue
+					}
+
+					if strings.HasSuffix(testInfo.Name(), ".tftest") || strings.HasSuffix(testInfo.Name(), ".tftest.json") {
+						tests = append(tests, filepath.Join(testsDir, testInfo.Name()))
+					}
+				}
+			}
+
+			// We only care about the tests directory or terraform configuration
+			// files.
 			continue
 		}
 
@@ -111,6 +175,13 @@ func (p *Parser) dirFiles(dir string) (primary, override []string, diags hcl.Dia
 			continue
 		}
 
+		if ext == ".tftest" || ext == ".tftest.json" {
+			if includeTests {
+				tests = append(tests, filepath.Join(dir, name))
+			}
+			continue
+		}
+
 		baseName := name[:len(name)-len(ext)] // strip extension
 		isOverride := baseName == "override" || strings.HasSuffix(baseName, "_override")
 
@@ -125,6 +196,32 @@ func (p *Parser) dirFiles(dir string) (primary, override []string, diags hcl.Dia
 	return
 }
 
+func (p *Parser) loadTestFiles(basePath string, paths []string) (map[string]*TestFile, hcl.Diagnostics) {
+	var diags hcl.Diagnostics
+
+	tfs := make(map[string]*TestFile)
+	for _, path := range paths {
+		tf, fDiags := p.LoadTestFile(path)
+		diags = append(diags, fDiags...)
+		if tf != nil {
+			// We index test files relative to the module they are testing, so
+			// the key is the relative path between basePath and path.
+			relPath, err := filepath.Rel(basePath, path)
+			if err != nil {
+				diags = append(diags, &hcl.Diagnostic{
+					Severity: hcl.DiagWarning,
+					Summary:  "Failed to calculate relative path",
+					Detail:   fmt.Sprintf("Terraform could not calculate the relative path for test file %s and it has been skipped: %s", path, err),
+				})
+				continue
+			}
+			tfs[relPath] = tf
+		}
+	}
+
+	return tfs, diags
+}
+
 // fileExt returns the Terraform configuration extension of the given
 // path, or a blank string if it is not a recognized extension.
 func fileExt(path string) string {
@@ -132,6 +229,10 @@ func fileExt(path string) string {
 		return ".tf"
 	} else if strings.HasSuffix(path, ".tf.json") {
 		return ".tf.json"
+	} else if strings.HasSuffix(path, ".tftest") {
+		return ".tftest"
+	} else if strings.HasSuffix(path, ".tftest.json") {
+		return ".tftest.json"
 	} else {
 		return ""
 	}
@@ -157,7 +258,7 @@ func IsEmptyDir(path string) (bool, error) {
 	}
 
 	p := NewParser(nil)
-	fs, os, diags := p.dirFiles(path)
+	fs, os, _, diags := p.dirFiles(path, "")
 	if diags.HasErrors() {
 		return false, diags
 	}
diff --git a/internal/configs/parser_config_dir_test.go b/internal/configs/parser_config_dir_test.go
index f1afcb208d82..8d2603c8b3d6 100644
--- a/internal/configs/parser_config_dir_test.go
+++ b/internal/configs/parser_config_dir_test.go
@@ -73,6 +73,12 @@ func TestParserLoadConfigDirSuccess(t *testing.T) {
 			if mod.SourceDir != path {
 				t.Errorf("wrong SourceDir value %q; want %s", mod.SourceDir, path)
 			}
+
+			if len(mod.Tests) > 0 {
+				// We only load tests when requested, and we didn't request this
+				// time.
+				t.Errorf("should not have loaded tests, but found %d", len(mod.Tests))
+			}
 		})
 	}
 
@@ -107,6 +113,31 @@ func TestParserLoadConfigDirSuccess(t *testing.T) {
 
 }
 
+func TestParserLoadConfigDirWithTests(t *testing.T) {
+	directories := []string{
+		"testdata/valid-modules/with-tests",
+		"testdata/valid-modules/with-tests-nested",
+		"testdata/valid-modules/with-tests-json",
+	}
+
+	for _, directory := range directories {
+		t.Run(directory, func(t *testing.T) {
+			parser := NewParser(nil)
+			mod, diags := parser.LoadConfigDirWithTests(directory, "tests")
+			if diags.HasErrors() {
+				t.Errorf("unexpected error diagnostics")
+				for _, diag := range diags {
+					t.Logf("- %s", diag)
+				}
+			}
+
+			if len(mod.Tests) != 2 {
+				t.Errorf("incorrect number of test files found: %d", len(mod.Tests))
+			}
+		})
+	}
+}
+
 // TestParseLoadConfigDirFailure is a simple test that just verifies that
 // a number of test configuration directories (in testdata/invalid-modules)
 // produce diagnostics when parsed.
diff --git a/internal/configs/test_file.go b/internal/configs/test_file.go
new file mode 100644
index 000000000000..9171f1cc8fef
--- /dev/null
+++ b/internal/configs/test_file.go
@@ -0,0 +1,335 @@
+package configs
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/hcl/v2"
+	"github.com/hashicorp/hcl/v2/gohcl"
+)
+
+// TestCommand represents the Terraform a given run block will execute, plan
+// or apply. Defaults to apply.
+type TestCommand rune
+
+// TestMode represents the plan mode that Terraform will use for a given run
+// block, normal or refresh-only. Defaults to normal.
+type TestMode rune
+
+const (
+	// ApplyTestCommand causes the run block to execute a Terraform apply
+	// operation.
+	ApplyTestCommand TestCommand = 0
+
+	// PlanTestCommand causes the run block to execute a Terraform plan
+	// operation.
+	PlanTestCommand TestCommand = 'P'
+
+	// NormalTestMode causes the run block to execute in plans.NormalMode.
+	NormalTestMode TestMode = 0
+
+	// RefreshOnlyTestMode causes the run block to execute in
+	// plans.RefreshOnlyMode.
+	RefreshOnlyTestMode TestMode = 'R'
+)
+
+// TestFile represents a single test file within a `terraform test` execution.
+//
+// A test file is made up of a sequential list of run blocks, each designating
+// a command to execute and a series of validations to check after the command.
+type TestFile struct {
+	// Variables defines a set of global variable definitions that should be set
+	// for every run block within the test file.
+	Variables map[string]hcl.Expression
+
+	// Runs defines the sequential list of run blocks that should be executed in
+	// order.
+	Runs []*TestRun
+
+	VariablesDeclRange hcl.Range
+}
+
+// TestRun represents a single run block within a test file.
+//
+// Each run block represents a single Terraform command to be executed and a set
+// of validations to run after the command.
+type TestRun struct {
+	Name string
+
+	// Command is the Terraform command to execute.
+	//
+	// One of ['apply', 'plan'].
+	Command TestCommand
+
+	// Options contains the embedded plan options that will affect the given
+	// Command. These should map to the options documented here:
+	//   - https://developer.hashicorp.com/terraform/cli/commands/plan#planning-options
+	//
+	// Note, that the Variables are a top level concept and not embedded within
+	// the options despite being listed as plan options in the documentation.
+	Options *TestRunOptions
+
+	// Variables defines a set of variable definitions for this command.
+	//
+	// Any variables specified locally that clash with the global variables will
+	// take precedence over the global definition.
+	Variables map[string]hcl.Expression
+
+	// CheckRules defines the list of assertions/validations that should be
+	// checked by this run block.
+	CheckRules []*CheckRule
+
+	NameDeclRange      hcl.Range
+	VariablesDeclRange hcl.Range
+	DeclRange          hcl.Range
+}
+
+// TestRunOptions contains the plan options for a given run block.
+type TestRunOptions struct {
+	// Mode is the planning mode to run in. One of ['normal', 'refresh-only'].
+	Mode TestMode
+
+	// Refresh is analogous to the -refresh=false Terraform plan option.
+	Refresh bool
+
+	// Replace is analogous to the -refresh=ADDRESS Terraform plan option.
+	Replace []hcl.Traversal
+
+	// Target is analogous to the -target=ADDRESS Terraform plan option.
+	Target []hcl.Traversal
+
+	DeclRange hcl.Range
+}
+
+func loadTestFile(body hcl.Body) (*TestFile, hcl.Diagnostics) {
+	var diags hcl.Diagnostics
+
+	content, contentDiags := body.Content(testFileSchema)
+	diags = append(diags, contentDiags...)
+
+	tf := TestFile{}
+
+	for _, block := range content.Blocks {
+		switch block.Type {
+		case "run":
+			run, runDiags := decodeTestRunBlock(block)
+			diags = append(diags, runDiags...)
+			if !runDiags.HasErrors() {
+				tf.Runs = append(tf.Runs, run)
+			}
+		case "variables":
+			if tf.Variables != nil {
+				diags = append(diags, &hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Multiple \"variables\" blocks",
+					Detail:   fmt.Sprintf("This test file already has a variables block defined at %s.", tf.VariablesDeclRange),
+					Subject:  block.DefRange.Ptr(),
+				})
+				continue
+			}
+
+			tf.Variables = make(map[string]hcl.Expression)
+			tf.VariablesDeclRange = block.DefRange
+
+			vars, varsDiags := block.Body.JustAttributes()
+			diags = append(diags, varsDiags...)
+			for _, v := range vars {
+				tf.Variables[v.Name] = v.Expr
+			}
+		}
+	}
+
+	return &tf, diags
+}
+
+func decodeTestRunBlock(block *hcl.Block) (*TestRun, hcl.Diagnostics) {
+	var diags hcl.Diagnostics
+
+	content, contentDiags := block.Body.Content(testRunBlockSchema)
+	diags = append(diags, contentDiags...)
+
+	r := TestRun{
+		Name:          block.Labels[0],
+		NameDeclRange: block.LabelRanges[0],
+		DeclRange:     block.DefRange,
+	}
+	for _, block := range content.Blocks {
+		switch block.Type {
+		case "assert":
+			cr, crDiags := decodeCheckRuleBlock(block, false)
+			diags = append(diags, crDiags...)
+			if !crDiags.HasErrors() {
+				r.CheckRules = append(r.CheckRules, cr)
+			}
+		case "plan_options":
+			if r.Options != nil {
+				diags = append(diags, &hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Multiple \"plan_options\" blocks",
+					Detail:   fmt.Sprintf("This run block already has a plan_options block defined at %s.", r.Options.DeclRange),
+					Subject:  block.DefRange.Ptr(),
+				})
+				continue
+			}
+
+			opts, optsDiags := decodeTestRunOptionsBlock(block)
+			diags = append(diags, optsDiags...)
+			if !optsDiags.HasErrors() {
+				r.Options = opts
+			}
+		case "variables":
+			if r.Variables != nil {
+				diags = append(diags, &hcl.Diagnostic{
+					Severity: hcl.DiagError,
+					Summary:  "Multiple \"variables\" blocks",
+					Detail:   fmt.Sprintf("This run block already has a variables block defined at %s.", r.VariablesDeclRange),
+					Subject:  block.DefRange.Ptr(),
+				})
+				continue
+			}
+
+			r.Variables = make(map[string]hcl.Expression)
+			r.VariablesDeclRange = block.DefRange
+
+			vars, varsDiags := block.Body.JustAttributes()
+			diags = append(diags, varsDiags...)
+			for _, v := range vars {
+				r.Variables[v.Name] = v.Expr
+			}
+
+		}
+	}
+
+	if r.Variables == nil {
+		// There is no distinction between a nil map of variables or an empty
+		// map, but we can avoid any potential nil pointer exceptions by just
+		// creating an empty map.
+		r.Variables = make(map[string]hcl.Expression)
+	}
+
+	if r.Options == nil {
+		// Create an options with default values if the user didn't specify
+		// anything.
+		r.Options = &TestRunOptions{
+			Mode:    NormalTestMode,
+			Refresh: true,
+		}
+	}
+
+	if attr, exists := content.Attributes["command"]; exists {
+		switch hcl.ExprAsKeyword(attr.Expr) {
+		case "apply":
+			r.Command = ApplyTestCommand
+		case "plan":
+			r.Command = PlanTestCommand
+		default:
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid \"command\" keyword",
+				Detail:   "The \"command\" argument requires one of the following keywords without quotes: apply or plan.",
+				Subject:  attr.Expr.Range().Ptr(),
+			})
+		}
+	} else {
+		r.Command = ApplyTestCommand // Default to apply
+	}
+
+	return &r, diags
+}
+
+func decodeTestRunOptionsBlock(block *hcl.Block) (*TestRunOptions, hcl.Diagnostics) {
+	var diags hcl.Diagnostics
+
+	content, contentDiags := block.Body.Content(testRunOptionsBlockSchema)
+	diags = append(diags, contentDiags...)
+
+	opts := TestRunOptions{
+		DeclRange: block.DefRange,
+	}
+
+	if attr, exists := content.Attributes["mode"]; exists {
+		switch hcl.ExprAsKeyword(attr.Expr) {
+		case "refresh-only":
+			opts.Mode = RefreshOnlyTestMode
+		case "normal":
+			opts.Mode = NormalTestMode
+		default:
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "Invalid \"mode\" keyword",
+				Detail:   "The \"mode\" argument requires one of the following keywords without quotes: normal or refresh-only",
+				Subject:  attr.Expr.Range().Ptr(),
+			})
+		}
+	} else {
+		opts.Mode = NormalTestMode // Default to normal
+	}
+
+	if attr, exists := content.Attributes["refresh"]; exists {
+		diags = append(diags, gohcl.DecodeExpression(attr.Expr, nil, &opts.Refresh)...)
+	} else {
+		// Defaults to true.
+		opts.Refresh = true
+	}
+
+	if attr, exists := content.Attributes["replace"]; exists {
+		reps, repsDiags := decodeDependsOn(attr)
+		diags = append(diags, repsDiags...)
+		opts.Replace = reps
+	}
+
+	if attr, exists := content.Attributes["target"]; exists {
+		tars, tarsDiags := decodeDependsOn(attr)
+		diags = append(diags, tarsDiags...)
+		opts.Target = tars
+	}
+
+	if !opts.Refresh && opts.Mode == RefreshOnlyTestMode {
+		// These options are incompatible.
+		diags = append(diags, &hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Incompatible plan options",
+			Detail:   "The \"refresh\" option cannot be set to false when running a test in \"refresh-only\" mode.",
+			Subject:  content.Attributes["refresh"].Range.Ptr(),
+		})
+	}
+
+	return &opts, diags
+}
+
+var testFileSchema = &hcl.BodySchema{
+	Blocks: []hcl.BlockHeaderSchema{
+		{
+			Type:       "run",
+			LabelNames: []string{"name"},
+		},
+		{
+			Type: "variables",
+		},
+	},
+}
+
+var testRunBlockSchema = &hcl.BodySchema{
+	Attributes: []hcl.AttributeSchema{
+		{Name: "command"},
+	},
+	Blocks: []hcl.BlockHeaderSchema{
+		{
+			Type: "plan_options",
+		},
+		{
+			Type: "assert",
+		},
+		{
+			Type: "variables",
+		},
+	},
+}
+
+var testRunOptionsBlockSchema = &hcl.BodySchema{
+	Attributes: []hcl.AttributeSchema{
+		{Name: "mode"},
+		{Name: "refresh"},
+		{Name: "replace"},
+		{Name: "target"},
+	},
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests-json/main.tf.json b/internal/configs/testdata/valid-modules/with-tests-json/main.tf.json
new file mode 100644
index 000000000000..03624c1463ed
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-json/main.tf.json
@@ -0,0 +1,17 @@
+{
+  "variable": {
+    "input": {
+      "type": "string"
+    }
+  },
+  "resource": {
+    "foo_resource": {
+      "a": {
+        "value": "${var.input}"
+      }
+    },
+    "bar_resource": {
+      "c": {}
+    }
+  }
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests-json/test_case_two.tftest.json b/internal/configs/testdata/valid-modules/with-tests-json/test_case_two.tftest.json
new file mode 100644
index 000000000000..43e677db64bc
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-json/test_case_two.tftest.json
@@ -0,0 +1,45 @@
+{
+  "run": {
+    "test_run_one": {
+      "variables": {
+        "input": "test_run_one"
+      },
+      "assert": [
+        {
+          "condition": "${foo_resource.a.value} == test_run_one",
+          "error_message": "invalid value"
+        }
+      ]
+    },
+    "test_run_two": {
+      "plan_options": {
+        "mode": "refresh-only"
+      },
+      "variables": {
+        "input": "test_run_two"
+      },
+      "assert": [
+        {
+          "condition": "${foo_resource.a.value} == test_run_one",
+          "error_message": "invalid value"
+        }
+      ]
+    },
+    "test_run_three": {
+      "variables": {
+        "input": "test_run_three"
+      },
+      "plan_options": {
+        "replace": [
+          "bar_resource.c"
+        ]
+      },
+      "assert": [
+        {
+          "condition": "${foo_resource.a.value} == test_run_three",
+          "error_message": "invalid value"
+        }
+      ]
+    }
+  }
+}
\ No newline at end of file
diff --git a/internal/configs/testdata/valid-modules/with-tests-json/tests/test_case_one.tftest.json b/internal/configs/testdata/valid-modules/with-tests-json/tests/test_case_one.tftest.json
new file mode 100644
index 000000000000..934173aca38b
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-json/tests/test_case_one.tftest.json
@@ -0,0 +1,32 @@
+{
+  "variables": {
+    "input": "default"
+  },
+  "run": {
+    "test_run_one": {
+      "command": "plan",
+      "plan_options": {
+        "target": [
+          "foo_resource.a"
+        ]
+      },
+      "assert": [
+        {
+          "condition": "${foo_resource.a.value} == default",
+          "error_message": "invalid value"
+        }
+      ]
+    },
+    "test_run_two": {
+      "variables": {
+        "input": "custom"
+      },
+      "assert": [
+        {
+          "condition": "${foo_resource.a.value} == custom",
+          "error_message": "invalid value"
+        }
+      ]
+    }
+  }
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests-nested/main.tf b/internal/configs/testdata/valid-modules/with-tests-nested/main.tf
new file mode 100644
index 000000000000..b84d4f3c416b
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-nested/main.tf
@@ -0,0 +1,11 @@
+
+variable "input" {
+  type = string
+}
+
+
+resource "foo_resource" "a" {
+  value = var.input
+}
+
+resource "bar_resource" "c" {}
diff --git a/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_one.tftest b/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_one.tftest
new file mode 100644
index 000000000000..01ef5dff053e
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_one.tftest
@@ -0,0 +1,31 @@
+variables {
+  input = "default"
+}
+
+# test_run_one runs a partial plan
+run "test_run_one" {
+  command = plan
+
+  plan_options {
+    target = [
+      foo_resource.a
+    ]
+  }
+
+  assert {
+    condition = foo_resource.a.value == "default"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_two does a complete apply operation
+run "test_run_two" {
+  variables {
+    input = "custom"
+  }
+
+  assert {
+    condition = foo_resource.a.value == "custom"
+    error_message = "invalid value"
+  }
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_two.tftest b/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_two.tftest
new file mode 100644
index 000000000000..b2be9172b64b
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests-nested/tests/test_case_two.tftest
@@ -0,0 +1,46 @@
+# test_run_one does a complete apply
+run "test_run_one" {
+  variables {
+    input = "test_run_one"
+  }
+
+  assert {
+    condition = foo_resource.a.value == "test_run_one"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_two does a refresh only apply
+run "test_run_two" {
+  plan_options {
+    mode = refresh-only
+  }
+
+  variables {
+    input = "test_run_two"
+  }
+
+  assert {
+    # value shouldn't change, as we're doing a refresh-only apply.
+    condition = foo_resource.a.value == "test_run_one"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_three does an apply with a replace operation
+run "test_run_three" {
+  variables {
+    input = "test_run_three"
+  }
+
+  plan_options {
+    replace = [
+      bar_resource.c
+    ]
+  }
+
+  assert {
+    condition = foo_resource.a.value == "test_run_three"
+    error_message = "invalid value"
+  }
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests/main.tf b/internal/configs/testdata/valid-modules/with-tests/main.tf
new file mode 100644
index 000000000000..b84d4f3c416b
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests/main.tf
@@ -0,0 +1,11 @@
+
+variable "input" {
+  type = string
+}
+
+
+resource "foo_resource" "a" {
+  value = var.input
+}
+
+resource "bar_resource" "c" {}
diff --git a/internal/configs/testdata/valid-modules/with-tests/test_case_one.tftest b/internal/configs/testdata/valid-modules/with-tests/test_case_one.tftest
new file mode 100644
index 000000000000..01ef5dff053e
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests/test_case_one.tftest
@@ -0,0 +1,31 @@
+variables {
+  input = "default"
+}
+
+# test_run_one runs a partial plan
+run "test_run_one" {
+  command = plan
+
+  plan_options {
+    target = [
+      foo_resource.a
+    ]
+  }
+
+  assert {
+    condition = foo_resource.a.value == "default"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_two does a complete apply operation
+run "test_run_two" {
+  variables {
+    input = "custom"
+  }
+
+  assert {
+    condition = foo_resource.a.value == "custom"
+    error_message = "invalid value"
+  }
+}
diff --git a/internal/configs/testdata/valid-modules/with-tests/test_case_two.tftest b/internal/configs/testdata/valid-modules/with-tests/test_case_two.tftest
new file mode 100644
index 000000000000..b2be9172b64b
--- /dev/null
+++ b/internal/configs/testdata/valid-modules/with-tests/test_case_two.tftest
@@ -0,0 +1,46 @@
+# test_run_one does a complete apply
+run "test_run_one" {
+  variables {
+    input = "test_run_one"
+  }
+
+  assert {
+    condition = foo_resource.a.value == "test_run_one"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_two does a refresh only apply
+run "test_run_two" {
+  plan_options {
+    mode = refresh-only
+  }
+
+  variables {
+    input = "test_run_two"
+  }
+
+  assert {
+    # value shouldn't change, as we're doing a refresh-only apply.
+    condition = foo_resource.a.value == "test_run_one"
+    error_message = "invalid value"
+  }
+}
+
+# test_run_three does an apply with a replace operation
+run "test_run_three" {
+  variables {
+    input = "test_run_three"
+  }
+
+  plan_options {
+    replace = [
+      bar_resource.c
+    ]
+  }
+
+  assert {
+    condition = foo_resource.a.value == "test_run_three"
+    error_message = "invalid value"
+  }
+}
diff --git a/internal/moduletest/file.go b/internal/moduletest/file.go
index 69a22613237f..b59ccea04450 100644
--- a/internal/moduletest/file.go
+++ b/internal/moduletest/file.go
@@ -1,6 +1,10 @@
 package moduletest
 
+import "github.com/hashicorp/terraform/internal/configs"
+
 type File struct {
+	Config *configs.TestFile
+
 	Name   string
 	Status Status
 
diff --git a/internal/moduletest/run.go b/internal/moduletest/run.go
index 41294861a234..33f720791323 100644
--- a/internal/moduletest/run.go
+++ b/internal/moduletest/run.go
@@ -1,10 +1,13 @@
 package moduletest
 
 import (
+	"github.com/hashicorp/terraform/internal/configs"
 	"github.com/hashicorp/terraform/internal/tfdiags"
 )
 
 type Run struct {
+	Config *configs.TestRun
+
 	Name   string
 	Status Status