first pass at adding fixture node

* use unit test name as top level

* remove breakpoints

* finish converting tests

* fix unit test node name

* breakpoints

* fix partial parsing bug

* comment out duplicate test

* fix test and make unique id match other uniqu id patterns

* clean up

* fix incremental test

* Update tests/functional/unit_testing/test_unit_testing.py

.changes/unreleased/Features-20230802-145011.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Initial implementation of unit testing
+time: 2023-08-02T14:50:11.391992-04:00
+custom:
+  Author: gshank
+  Issue: "8287"

.changes/unreleased/Features-20230828-101825.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Unit test manifest artifacts and selection
+time: 2023-08-28T10:18:25.958929-04:00
+custom:
+  Author: gshank
+  Issue: "8295"

.changes/unreleased/Features-20230906-234741.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support config with tags & meta for unit tests
+time: 2023-09-06T23:47:41.059915-04:00
+custom:
+  Author: michelleark
+  Issue: "8294"

.changes/unreleased/Features-20230928-163205.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Enable inline csv fixtures in unit tests
+time: 2023-09-28T16:32:05.573776-04:00
+custom:
+  Author: gshank
+  Issue: "8626"

.changes/unreleased/Features-20231101-101845.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support unit testing incremental models
+time: 2023-11-01T10:18:45.341781-04:00
+custom:
+  Author: michelleark
+  Issue: "8422"

.changes/unreleased/Features-20231106-194752.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support of csv file fixtures to unit testing
+time: 2023-11-06T19:47:52.501495-06:00
+custom:
+  Author: emmyoop
+  Issue: "8290"

.changes/unreleased/Under the Hood-20230912-190506.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add unit testing functional tests
+time: 2023-09-12T19:05:06.023126-04:00
+custom:
+  Author: gshank
+  Issue: "8512"

core/dbt/adapters/relation_configs/config_change.py

@@ -12,7 +12,7 @@ class RelationConfigChangeAction(StrEnum):
     drop = "drop"
 
 
-@dataclass(frozen=True, eq=True, unsafe_hash=True)
+@dataclass(frozen=True, eq=True, unsafe_hash=True)  # type: ignore
 class RelationConfigChange(RelationConfigBase, ABC):
     action: RelationConfigChangeAction
     context: Hashable  # this is usually a RelationConfig, e.g. IndexConfig, but shouldn't be limited

core/dbt/cli/flags.py

@@ -396,6 +396,7 @@ def command_args(command: CliCommand) -> ArgsList:
         CliCommand.SOURCE_FRESHNESS: cli.freshness,
         CliCommand.TEST: cli.test,
         CliCommand.RETRY: cli.retry,
+        CliCommand.UNIT_TEST: cli.unit_test,
     }
     click_cmd: Optional[ClickCommand] = CMD_DICT.get(command, None)
     if click_cmd is None:

core/dbt/cli/main.py

@@ -40,6 +40,7 @@ from dbt.task.serve import ServeTask
 from dbt.task.show import ShowTask
 from dbt.task.snapshot import SnapshotTask
 from dbt.task.test import TestTask
+from dbt.task.unit_test import UnitTestTask
 
 
 @dataclass
@@ -896,6 +897,50 @@ def test(ctx, **kwargs):
     return results, success
 
 
+# dbt unit-test
+@cli.command("unit-test")
+@click.pass_context
+@p.defer
+@p.deprecated_defer
+@p.exclude
+@p.fail_fast
+@p.favor_state
+@p.deprecated_favor_state
+@p.indirect_selection
+@p.show_output_format
+@p.profile
+@p.profiles_dir
+@p.project_dir
+@p.select
+@p.selector
+@p.state
+@p.defer_state
+@p.deprecated_state
+@p.store_failures
+@p.target
+@p.target_path
+@p.threads
+@p.vars
+@p.version_check
+@requires.postflight
+@requires.preflight
+@requires.profile
+@requires.project
+@requires.runtime_config
+@requires.manifest
+def unit_test(ctx, **kwargs):
+    """Runs tests on data in deployed models. Run this after `dbt run`"""
+    task = UnitTestTask(
+        ctx.obj["flags"],
+        ctx.obj["runtime_config"],
+        ctx.obj["manifest"],
+    )
+
+    results = task.run()
+    success = task.interpret_results(results)
+    return results, success
+
+
 # Support running as a module
 if __name__ == "__main__":
     cli()

core/dbt/cli/types.py

@@ -24,6 +24,7 @@ class Command(Enum):
     SOURCE_FRESHNESS = "freshness"
     TEST = "test"
     RETRY = "retry"
+    UNIT_TEST = "unit-test"
 
     @classmethod
     def from_str(cls, s: str) -> "Command":

core/dbt/clients/jinja.py

@@ -330,6 +330,26 @@ class MacroGenerator(BaseMacroGenerator):
             return self.call_macro(*args, **kwargs)
 
 
+class UnitTestMacroGenerator(MacroGenerator):
+    # this makes UnitTestMacroGenerator objects callable like functions
+    def __init__(
+        self,
+        macro_generator: MacroGenerator,
+        call_return_value: Any,
+    ) -> None:
+        super().__init__(
+            macro_generator.macro,
+            macro_generator.context,
+            macro_generator.node,
+            macro_generator.stack,
+        )
+        self.call_return_value = call_return_value
+
+    def __call__(self, *args, **kwargs):
+        with self.track_call():
+            return self.call_return_value
+
+
 class QueryStringGenerator(BaseMacroGenerator):
     def __init__(self, template_str: str, context: Dict[str, Any]) -> None:
         super().__init__(context)

core/dbt/compilation.py

@@ -12,7 +12,10 @@ from dbt.flags import get_flags
 from dbt.adapters.factory import get_adapter
 from dbt.clients import jinja
 from dbt.clients.system import make_directory
-from dbt.context.providers import generate_runtime_model_context
+from dbt.context.providers import (
+    generate_runtime_model_context,
+    generate_runtime_unit_test_context,
+)
 from dbt.contracts.graph.manifest import Manifest, UniqueID
 from dbt.contracts.graph.nodes import (
     ManifestNode,
@@ -21,6 +24,8 @@ from dbt.contracts.graph.nodes import (
     GraphMemberNode,
     InjectedCTE,
     SeedNode,
+    FixtureNode,
+    UnitTestNode,
 )
 from dbt.exceptions import (
     GraphDependencyNotFoundError,
@@ -44,6 +49,8 @@ def print_compile_stats(stats):
     names = {
         NodeType.Model: "model",
         NodeType.Test: "test",
+        NodeType.Fixture: "fixture",
+        NodeType.Unit: "unit test",
         NodeType.Snapshot: "snapshot",
         NodeType.Analysis: "analysis",
         NodeType.Macro: "macro",
@@ -91,6 +98,7 @@ def _generate_stats(manifest: Manifest):
     stats[NodeType.Macro] += len(manifest.macros)
     stats[NodeType.Group] += len(manifest.groups)
     stats[NodeType.SemanticModel] += len(manifest.semantic_models)
+    stats[NodeType.Unit] += len(manifest.unit_tests)
 
     # TODO: should we be counting dimensions + entities?
 
@@ -191,6 +199,8 @@ class Linker:
             self.link_node(exposure, manifest)
         for metric in manifest.metrics.values():
             self.link_node(metric, manifest)
+        for unit_test in manifest.unit_tests.values():
+            self.link_node(unit_test, manifest)
         for saved_query in manifest.saved_queries.values():
             self.link_node(saved_query, manifest)
 
@@ -291,8 +301,10 @@ class Compiler:
         manifest: Manifest,
         extra_context: Dict[str, Any],
     ) -> Dict[str, Any]:
-
-        context = generate_runtime_model_context(node, self.config, manifest)
+        if isinstance(node, UnitTestNode):
+            context = generate_runtime_unit_test_context(node, self.config, manifest)
+        else:
+            context = generate_runtime_model_context(node, self.config, manifest)
         context.update(extra_context)
 
         if isinstance(node, GenericTestNode):
@@ -332,8 +344,8 @@ class Compiler:
 
         # Just to make it plain that nothing is actually injected for this case
         if len(model.extra_ctes) == 0:
-            # SeedNodes don't have compilation attributes
-            if not isinstance(model, SeedNode):
+            # SeedNodes and FixtureNodes don't have compilation attributes
+            if not isinstance(model, (SeedNode, FixtureNode)):
                 model.extra_ctes_injected = True
             return (model, [])
 
@@ -353,7 +365,7 @@ class Compiler:
                     f"could not be resolved: {cte.id}"
                 )
             cte_model = manifest.nodes[cte.id]
-            assert not isinstance(cte_model, SeedNode)
+            assert not isinstance(cte_model, (SeedNode, FixtureNode))
 
             if not cte_model.is_ephemeral_model:
                 raise DbtInternalError(f"{cte.id} is not ephemeral")
@@ -507,6 +519,7 @@ class Compiler:
         if not node.extra_ctes_injected or node.resource_type in (
             NodeType.Snapshot,
             NodeType.Seed,
+            NodeType.Fixture,
         ):
             return node
         fire_event(WritingInjectedSQLForNode(node_info=get_node_info()))

core/dbt/config/project.py

@@ -432,6 +432,7 @@ class PartialProject(RenderComponents):
         snapshots: Dict[str, Any]
         sources: Dict[str, Any]
         tests: Dict[str, Any]
+        unit_tests: Dict[str, Any]
         metrics: Dict[str, Any]
         semantic_models: Dict[str, Any]
         saved_queries: Dict[str, Any]
@@ -445,6 +446,7 @@ class PartialProject(RenderComponents):
         snapshots = cfg.snapshots
         sources = cfg.sources
         tests = cfg.tests
+        unit_tests = cfg.unit_tests
         metrics = cfg.metrics
         semantic_models = cfg.semantic_models
         saved_queries = cfg.saved_queries
@@ -505,6 +507,7 @@ class PartialProject(RenderComponents):
             query_comment=query_comment,
             sources=sources,
             tests=tests,
+            unit_tests=unit_tests,
             metrics=metrics,
             semantic_models=semantic_models,
             saved_queries=saved_queries,
@@ -615,6 +618,7 @@ class Project:
     snapshots: Dict[str, Any]
     sources: Dict[str, Any]
     tests: Dict[str, Any]
+    unit_tests: Dict[str, Any]
     metrics: Dict[str, Any]
     semantic_models: Dict[str, Any]
     saved_queries: Dict[str, Any]
@@ -648,6 +652,13 @@ class Project:
             generic_test_paths.append(os.path.join(test_path, "generic"))
         return generic_test_paths
 
+    @property
+    def fixture_paths(self):
+        fixture_paths = []
+        for test_path in self.test_paths:
+            fixture_paths.append(os.path.join(test_path, "fixtures"))
+        return fixture_paths
+
     def __str__(self):
         cfg = self.to_project_config(with_packages=True)
         return str(cfg)
@@ -693,6 +704,7 @@ class Project:
                 "snapshots": self.snapshots,
                 "sources": self.sources,
                 "tests": self.tests,
+                "unit-tests": self.unit_tests,
                 "metrics": self.metrics,
                 "semantic-models": self.semantic_models,
                 "saved-queries": self.saved_queries,

core/dbt/config/runtime.py

@@ -166,6 +166,7 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
             query_comment=project.query_comment,
             sources=project.sources,
             tests=project.tests,
+            unit_tests=project.unit_tests,
             metrics=project.metrics,
             semantic_models=project.semantic_models,
             saved_queries=project.saved_queries,
@@ -324,6 +325,7 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
             "snapshots": self._get_config_paths(self.snapshots),
             "sources": self._get_config_paths(self.sources),
             "tests": self._get_config_paths(self.tests),
+            "unit_tests": self._get_config_paths(self.unit_tests),
             "metrics": self._get_config_paths(self.metrics),
             "semantic_models": self._get_config_paths(self.semantic_models),
             "saved_queries": self._get_config_paths(self.saved_queries),

core/dbt/constants.py

@@ -16,4 +16,5 @@ PACKAGE_LOCK_FILE_NAME = "package-lock.yml"
 MANIFEST_FILE_NAME = "manifest.json"
 SEMANTIC_MANIFEST_FILE_NAME = "semantic_manifest.json"
 PARTIAL_PARSE_FILE_NAME = "partial_parse.msgpack"
+UNIT_TEST_MANIFEST_FILE_NAME = "unit_test_manifest.json"
 PACKAGE_LOCK_HASH_KEY = "sha1_hash"

core/dbt/context/context_config.py

@@ -43,6 +43,8 @@ class UnrenderedConfig(ConfigSource):
             model_configs = unrendered.get("sources")
         elif resource_type == NodeType.Test:
             model_configs = unrendered.get("tests")
+        elif resource_type == NodeType.Fixture:
+            model_configs = unrendered.get("fixtures")
         elif resource_type == NodeType.Metric:
             model_configs = unrendered.get("metrics")
         elif resource_type == NodeType.SemanticModel:
@@ -51,6 +53,8 @@ class UnrenderedConfig(ConfigSource):
             model_configs = unrendered.get("saved_queries")
         elif resource_type == NodeType.Exposure:
             model_configs = unrendered.get("exposures")
+        elif resource_type == NodeType.Unit:
+            model_configs = unrendered.get("unit_tests")
         else:
             model_configs = unrendered.get("models")
         if model_configs is None:
@@ -72,6 +76,8 @@ class RenderedConfig(ConfigSource):
             model_configs = self.project.sources
         elif resource_type == NodeType.Test:
             model_configs = self.project.tests
+        elif resource_type == NodeType.Fixture:
+            model_configs = self.project.fixtures
         elif resource_type == NodeType.Metric:
             model_configs = self.project.metrics
         elif resource_type == NodeType.SemanticModel:
@@ -80,6 +86,8 @@ class RenderedConfig(ConfigSource):
             model_configs = self.project.saved_queries
         elif resource_type == NodeType.Exposure:
             model_configs = self.project.exposures
+        elif resource_type == NodeType.Unit:
+            model_configs = self.project.unit_tests
         else:
             model_configs = self.project.models
         return model_configs

core/dbt/context/providers.py

@@ -1,4 +1,5 @@
 import abc
+from copy import deepcopy
 import os
 from typing import (
     Callable,
@@ -17,7 +18,7 @@ from typing_extensions import Protocol
 from dbt.adapters.base.column import Column
 from dbt.adapters.factory import get_adapter, get_adapter_package_names, get_adapter_type_names
 from dbt.clients import agate_helper
-from dbt.clients.jinja import get_rendered, MacroGenerator, MacroStack
+from dbt.clients.jinja import get_rendered, MacroGenerator, MacroStack, UnitTestMacroGenerator
 from dbt.config import RuntimeConfig, Project
 from dbt.constants import SECRET_ENV_PREFIX, DEFAULT_ENV_PLACEHOLDER
 from dbt.context.base import contextmember, contextproperty, Var
@@ -39,6 +40,7 @@ from dbt.contracts.graph.nodes import (
     RefArgs,
     AccessType,
     SemanticModel,
+    UnitTestNode,
 )
 from dbt.contracts.graph.metrics import MetricReference, ResolvedMetricReference
 from dbt.contracts.graph.unparsed import NodeVersion
@@ -566,6 +568,17 @@ class OperationRefResolver(RuntimeRefResolver):
             return super().create_relation(target_model)
 
 
+class RuntimeUnitTestRefResolver(RuntimeRefResolver):
+    def resolve(
+        self,
+        target_name: str,
+        target_package: Optional[str] = None,
+        target_version: Optional[NodeVersion] = None,
+    ) -> RelationProxy:
+        target_name = f"{self.model.name}__{target_name}"
+        return super().resolve(target_name, target_package, target_version)
+
+
 # `source` implementations
 class ParseSourceResolver(BaseSourceResolver):
     def resolve(self, source_name: str, table_name: str):
@@ -670,6 +683,22 @@ class RuntimeVar(ModelConfiguredVar):
     pass
 
 
+class UnitTestVar(RuntimeVar):
+    def __init__(
+        self,
+        context: Dict[str, Any],
+        config: RuntimeConfig,
+        node: Resource,
+    ) -> None:
+        config_copy = None
+        assert isinstance(node, UnitTestNode)
+        if node.overrides and node.overrides.vars:
+            config_copy = deepcopy(config)
+            config_copy.cli_vars.update(node.overrides.vars)
+
+        super().__init__(context, config_copy or config, node=node)
+
+
 # Providers
 class Provider(Protocol):
     execute: bool
@@ -711,6 +740,16 @@ class RuntimeProvider(Provider):
     metric = RuntimeMetricResolver
 
 
+class RuntimeUnitTestProvider(Provider):
+    execute = True
+    Config = RuntimeConfigObject
+    DatabaseWrapper = RuntimeDatabaseWrapper
+    Var = UnitTestVar
+    ref = RuntimeUnitTestRefResolver
+    source = RuntimeSourceResolver  # TODO: RuntimeUnitTestSourceResolver
+    metric = RuntimeMetricResolver
+
+
 class OperationProvider(RuntimeProvider):
     ref = OperationRefResolver
 
@@ -1382,7 +1421,12 @@ class ModelContext(ProviderContext):
 
     @contextproperty()
     def pre_hooks(self) -> List[Dict[str, Any]]:
-        if self.model.resource_type in [NodeType.Source, NodeType.Test]:
+        if self.model.resource_type in [
+            NodeType.Source,
+            NodeType.Test,
+            NodeType.Unit,
+            NodeType.Fixture,
+        ]:
             return []
         # TODO CT-211
         return [
@@ -1391,7 +1435,12 @@ class ModelContext(ProviderContext):
 
     @contextproperty()
     def post_hooks(self) -> List[Dict[str, Any]]:
-        if self.model.resource_type in [NodeType.Source, NodeType.Test]:
+        if self.model.resource_type in [
+            NodeType.Source,
+            NodeType.Test,
+            NodeType.Unit,
+            NodeType.Fixture,
+        ]:
             return []
         # TODO CT-211
         return [
@@ -1484,6 +1533,33 @@ class ModelContext(ProviderContext):
             return None
 
 
+class UnitTestContext(ModelContext):
+    model: UnitTestNode
+
+    @contextmember()
+    def env_var(self, var: str, default: Optional[str] = None) -> str:
+        """The env_var() function. Return the overriden unit test environment variable named 'var'.
+
+        If there is no unit test override, return the environment variable named 'var'.
+
+        If there is no such environment variable set, return the default.
+
+        If the default is None, raise an exception for an undefined variable.
+        """
+        if self.model.overrides and var in self.model.overrides.env_vars:
+            return self.model.overrides.env_vars[var]
+        else:
+            return super().env_var(var, default)
+
+    @contextproperty()
+    def this(self) -> Optional[str]:
+        if self.model.this_input_node_unique_id:
+            this_node = self.manifest.expect(self.model.this_input_node_unique_id)
+            self.model.set_cte(this_node.unique_id, None)  # type: ignore
+            return self.adapter.Relation.add_ephemeral_prefix(this_node.name)
+        return None
+
+
 # This is called by '_context_for', used in 'render_with_context'
 def generate_parser_model_context(
     model: ManifestNode,
@@ -1528,6 +1604,24 @@ def generate_runtime_macro_context(
     return ctx.to_dict()
 
 
+def generate_runtime_unit_test_context(
+    unit_test: UnitTestNode,
+    config: RuntimeConfig,
+    manifest: Manifest,
+) -> Dict[str, Any]:
+    ctx = UnitTestContext(unit_test, config, manifest, RuntimeUnitTestProvider(), None)
+    ctx_dict = ctx.to_dict()
+
+    if unit_test.overrides and unit_test.overrides.macros:
+        for macro_name, macro_value in unit_test.overrides.macros.items():
+            context_value = ctx_dict.get(macro_name)
+            if isinstance(context_value, MacroGenerator):
+                ctx_dict[macro_name] = UnitTestMacroGenerator(context_value, macro_value)
+            else:
+                ctx_dict[macro_name] = macro_value
+    return ctx_dict
+
+
 class ExposureRefResolver(BaseResolver):
     def __call__(self, *args, **kwargs) -> str:
         package = None

core/dbt/contracts/files.py

@@ -18,6 +18,7 @@ class ParseFileType(StrEnum):
     Analysis = "analysis"
     SingularTest = "singular_test"
     GenericTest = "generic_test"
+    Fixture = "fixture"
     Seed = "seed"
     Documentation = "docs"
     Schema = "schema"
@@ -31,6 +32,7 @@ parse_file_type_to_parser = {
     ParseFileType.Analysis: "AnalysisParser",
     ParseFileType.SingularTest: "SingularTestParser",
     ParseFileType.GenericTest: "GenericTestParser",
+    ParseFileType.Fixture: "FixtureParser",
     ParseFileType.Seed: "SeedParser",
     ParseFileType.Documentation: "DocumentationParser",
     ParseFileType.Schema: "SchemaParser",
@@ -231,6 +233,7 @@ class SchemaSourceFile(BaseSourceFile):
     # node patches contain models, seeds, snapshots, analyses
     ndp: List[str] = field(default_factory=list)
     semantic_models: List[str] = field(default_factory=list)
+    unit_tests: List[str] = field(default_factory=list)
     saved_queries: List[str] = field(default_factory=list)
     # any macro patches in this file by macro unique_id.
     mcp: Dict[str, str] = field(default_factory=dict)

core/dbt/contracts/graph/manifest.py

@@ -41,6 +41,7 @@ from dbt.contracts.graph.nodes import (
     SemanticModel,
     SourceDefinition,
     UnpatchedSourceDefinition,
+    UnitTestDefinition,
 )
 from dbt.contracts.graph.unparsed import SourcePatch, NodeVersion, UnparsedVersion
 from dbt.contracts.graph.manifest_upgrade import upgrade_manifest_json
@@ -792,6 +793,7 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
     disabled: MutableMapping[str, List[GraphMemberNode]] = field(default_factory=dict)
     env_vars: MutableMapping[str, str] = field(default_factory=dict)
     semantic_models: MutableMapping[str, SemanticModel] = field(default_factory=dict)
+    unit_tests: MutableMapping[str, UnitTestDefinition] = field(default_factory=dict)
     saved_queries: MutableMapping[str, SavedQuery] = field(default_factory=dict)
 
     _doc_lookup: Optional[DocLookup] = field(
@@ -953,6 +955,7 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             files={k: _deepcopy(v) for k, v in self.files.items()},
             state_check=_deepcopy(self.state_check),
             semantic_models={k: _deepcopy(v) for k, v in self.semantic_models.items()},
+            unit_tests={k: _deepcopy(v) for k, v in self.unit_tests.items()},
             saved_queries={k: _deepcopy(v) for k, v in self.saved_queries.items()},
         )
         copy.build_flat_graph()
@@ -1023,6 +1026,7 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             parent_map=self.parent_map,
             group_map=self.group_map,
             semantic_models=self.semantic_models,
+            unit_tests=self.unit_tests,
             saved_queries=self.saved_queries,
         )
 
@@ -1042,6 +1046,8 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             return self.metrics[unique_id]
         elif unique_id in self.semantic_models:
             return self.semantic_models[unique_id]
+        elif unique_id in self.unit_tests:
+            return self.unit_tests[unique_id]
         elif unique_id in self.saved_queries:
             return self.saved_queries[unique_id]
         else:
@@ -1486,6 +1492,12 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
         self.semantic_models[semantic_model.unique_id] = semantic_model
         source_file.semantic_models.append(semantic_model.unique_id)
 
+    def add_unit_test(self, source_file: SchemaSourceFile, unit_test: UnitTestDefinition):
+        if unit_test.unique_id in self.unit_tests:
+            raise DuplicateResourceNameError(unit_test, self.unit_tests[unit_test.unique_id])
+        self.unit_tests[unit_test.unique_id] = unit_test
+        source_file.unit_tests.append(unit_test.unique_id)
+
     def add_saved_query(self, source_file: SchemaSourceFile, saved_query: SavedQuery) -> None:
         _check_duplicates(saved_query, self.saved_queries)
         self.saved_queries[saved_query.unique_id] = saved_query
@@ -1518,6 +1530,7 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             self.disabled,
             self.env_vars,
             self.semantic_models,
+            self.unit_tests,
             self.saved_queries,
             self._doc_lookup,
             self._source_lookup,
@@ -1600,6 +1613,11 @@ class WritableManifest(ArtifactMixin):
             description="Metadata about the manifest",
         )
     )
+    unit_tests: Mapping[UniqueID, UnitTestDefinition] = field(
+        metadata=dict(
+            description="The unit tests defined in the project",
+        )
+    )
 
     @classmethod
     def compatible_previous_versions(self):

core/dbt/contracts/graph/manifest_upgrade.py

@@ -145,6 +145,9 @@ def upgrade_manifest_json(manifest: dict, manifest_schema_version: int) -> dict:
         manifest["groups"] = {}
     if "group_map" not in manifest:
         manifest["group_map"] = {}
+    # add unit_tests key
+    if "unit_tests" not in manifest:
+        manifest["unit_tests"] = {}
     for metric_content in manifest.get("metrics", {}).values():
         # handle attr renames + value translation ("expression" -> "derived")
         metric_content = upgrade_ref_content(metric_content)

core/dbt/contracts/graph/model_config.py

@@ -551,6 +551,24 @@ class ModelConfig(NodeConfig):
     )
 
 
+@dataclass
+class UnitTestNodeConfig(NodeConfig):
+    expected_rows: List[Dict[str, Any]] = field(default_factory=list)
+
+
+@dataclass
+class FixtureConfig(NodeConfig):
+    materialized: str = "fixture"  # TODO: ?? does this get materialized?
+    delimiter: str = ","
+    quote_columns: Optional[bool] = None
+
+    @classmethod
+    def validate(cls, data):
+        super().validate(data)
+        if data.get("materialized") and data.get("materialized") != "fixture":
+            raise ValidationError("A fixture must have a materialized value of 'seed'")
+
+
 @dataclass
 class SeedConfig(NodeConfig):
     materialized: str = "seed"
@@ -723,6 +741,18 @@ class SnapshotConfig(EmptySnapshotConfig):
         return self.from_dict(data)
 
 
+@dataclass
+class UnitTestConfig(BaseConfig):
+    tags: Union[str, List[str]] = field(
+        default_factory=list_str,
+        metadata=metas(ShowBehavior.Hide, MergeBehavior.Append, CompareBehavior.Exclude),
+    )
+    meta: Dict[str, Any] = field(
+        default_factory=dict,
+        metadata=MergeBehavior.Update.meta(),
+    )
+
+
 RESOURCE_TYPES: Dict[NodeType, Type[BaseConfig]] = {
     NodeType.Metric: MetricConfig,
     NodeType.SemanticModel: SemanticModelConfig,
@@ -731,8 +761,10 @@ RESOURCE_TYPES: Dict[NodeType, Type[BaseConfig]] = {
     NodeType.Source: SourceConfig,
     NodeType.Seed: SeedConfig,
     NodeType.Test: TestConfig,
+    NodeType.Fixture: FixtureConfig,
     NodeType.Model: NodeConfig,
     NodeType.Snapshot: SnapshotConfig,
+    NodeType.Unit: UnitTestConfig,
 }
 
 

core/dbt/contracts/graph/nodes.py

@@ -35,12 +35,18 @@ from dbt.contracts.graph.unparsed import (
     UnparsedSourceDefinition,
     UnparsedSourceTableDefinition,
     UnparsedColumn,
+    UnitTestOverrides,
+    UnitTestInputFixture,
+    UnitTestOutputFixture,
 )
 from dbt.contracts.graph.node_args import ModelNodeArgs
 from dbt.contracts.graph.semantic_layer_common import WhereFilterIntersection
 from dbt.contracts.util import Replaceable, AdditionalPropertiesMixin
 from dbt.events.functions import warn_or_error
-from dbt.exceptions import ParsingError, ContractBreakingChangeError
+from dbt.exceptions import (
+    ParsingError,
+    ContractBreakingChangeError,
+)
 from dbt.events.types import (
     SeedIncreased,
     SeedExceedsLimitSamePath,
@@ -66,12 +72,15 @@ from .model_config import (
     ModelConfig,
     SeedConfig,
     TestConfig,
+    FixtureConfig,
     SourceConfig,
     MetricConfig,
     ExposureConfig,
     EmptySnapshotConfig,
     SnapshotConfig,
     SemanticModelConfig,
+    UnitTestConfig,
+    UnitTestNodeConfig,
     SavedQueryConfig,
 )
 
@@ -83,8 +92,8 @@ from .model_config import (
 # manifest contains "macros", "sources", "metrics", "exposures", "docs",
 # and "disabled" dictionaries.
 #
-# The SeedNode is a ManifestNode, but can't be compiled because it has
-# no SQL.
+# The SeedNode & FixtureNode is a ManifestNode, but can't be compiled
+# because they have no SQL.
 #
 # All objects defined in this file should have BaseNode as a parent
 # class.
@@ -397,10 +406,10 @@ class ParsedNode(NodeInfoMixin, ParsedNodeMandatory, SerializableType):
                 return GenericTestNode.from_dict(dct)
             else:
                 return SingularTestNode.from_dict(dct)
+        elif resource_type == "fixture":
+            return FixtureNode.from_dict(dct)
         elif resource_type == "operation":
             return HookNode.from_dict(dct)
-        elif resource_type == "seed":
-            return SeedNode.from_dict(dct)
         elif resource_type == "snapshot":
             return SnapshotNode.from_dict(dct)
         else:
@@ -1054,6 +1063,156 @@ class GenericTestNode(TestShouldStoreFailures, CompiledNode, HasTestMetadata):
         return "generic"
 
 
+# ====================================
+# Fixture node
+# ====================================
+
+
+@dataclass
+class FixtureNode(ParsedNode):  # No SQLDefaults!
+    resource_type: Literal[NodeType.Fixture]
+    config: FixtureConfig = field(default_factory=FixtureConfig)
+    # fixtures need the root_path because the contents are not loaded initially
+    # and we need the root_path to load the fixture later
+    root_path: Optional[str] = None
+    depends_on: MacroDependsOn = field(default_factory=MacroDependsOn)
+    defer_relation: Optional[DeferRelation] = None
+
+    def same_fixtures(self, other: "FixtureNode") -> bool:
+        # for fixtures, we check the hashes. If the hashes are different types,
+        # no match. If the hashes are both the same 'path', log a warning and
+        # assume they are the same
+        # if the current checksum is a path, we want to log a warning.
+        result = self.checksum == other.checksum
+
+        # TODO: implement this for fixtures
+        # if self.checksum.name == "path":
+        #     msg: str
+        #     if other.checksum.name != "path":
+        #         warn_or_error(
+        #             SeedIncreased(package_name=self.package_name, name=self.name), node=self
+        #         )
+        #     elif result:
+        #         warn_or_error(
+        #             SeedExceedsLimitSamePath(package_name=self.package_name, name=self.name),
+        #             node=self,
+        #         )
+        #     elif not result:
+        #         warn_or_error(
+        #             SeedExceedsLimitAndPathChanged(package_name=self.package_name, name=self.name),
+        #             node=self,
+        #         )
+        #     else:
+        #         warn_or_error(
+        #             SeedExceedsLimitChecksumChanged(
+        #                 package_name=self.package_name,
+        #                 name=self.name,
+        #                 checksum_name=other.checksum.name,
+        #             ),
+        #             node=self,
+        #         )
+
+        return result
+
+    # TODO: should this be true?  probably because they at least have columns names
+    @property
+    def empty(self):
+        """Fixtures are never empty"""
+        return False
+
+    # TODO: I think we need to do something like this for fixtures
+    def _disallow_implicit_dependencies(self):
+        """Disallow seeds to take implicit upstream dependencies via pre/post hooks"""
+        # Seeds are root nodes in the DAG. They cannot depend on other nodes.
+        # However, it's possible to define pre- and post-hooks on seeds, and for those
+        # hooks to include {{ ref(...) }}. This worked in previous versions, but it
+        # was never officially documented or supported behavior. Let's raise an explicit error,
+        # which will surface during parsing if the user has written code such that we attempt
+        # to capture & record a ref/source/metric call on the SeedNode.
+        # For more details: https://github.com/dbt-labs/dbt-core/issues/6806
+        hooks = [f'- pre_hook: "{hook.sql}"' for hook in self.config.pre_hook] + [
+            f'- post_hook: "{hook.sql}"' for hook in self.config.post_hook
+        ]
+        hook_list = "\n".join(hooks)
+        message = f"""
+Seeds cannot depend on other nodes. dbt detected a seed with a pre- or post-hook
+that calls 'ref', 'source', or 'metric', either directly or indirectly via other macros.
+
+Error raised for '{self.unique_id}', which has these hooks defined: \n{hook_list}
+        """
+        raise ParsingError(message)
+
+    @property
+    def refs(self):
+        self._disallow_implicit_dependencies()
+
+    @property
+    def sources(self):
+        self._disallow_implicit_dependencies()
+
+    @property
+    def metrics(self):
+        self._disallow_implicit_dependencies()
+
+    def same_body(self, other) -> bool:
+        return self.same_fixtures(other)
+
+    @property
+    def depends_on_nodes(self):
+        return []
+
+    @property
+    def depends_on_macros(self) -> List[str]:
+        return self.depends_on.macros
+
+    @property
+    def extra_ctes(self):
+        return []
+
+    @property
+    def extra_ctes_injected(self):
+        return False
+
+    # TODO: ??
+    @property
+    def language(self):
+        return "sql"
+
+
+# ====================================
+# Unit Test node
+# ====================================
+
+
+@dataclass
+class UnitTestNode(CompiledNode):
+    resource_type: NodeType = field(metadata={"restrict": [NodeType.Unit]})
+    tested_node_unique_id: Optional[str] = None
+    this_input_node_unique_id: Optional[str] = None
+    overrides: Optional[UnitTestOverrides] = None
+    config: UnitTestNodeConfig = field(default_factory=UnitTestNodeConfig)
+
+
+@dataclass
+class UnitTestDefinition(GraphNode):
+    model: str
+    given: Sequence[UnitTestInputFixture]
+    expect: UnitTestOutputFixture
+    description: str = ""
+    overrides: Optional[UnitTestOverrides] = None
+    depends_on: DependsOn = field(default_factory=DependsOn)
+    config: UnitTestConfig = field(default_factory=UnitTestConfig)
+
+    @property
+    def depends_on_nodes(self):
+        return self.depends_on.nodes
+
+    @property
+    def tags(self) -> List[str]:
+        tags = self.config.tags
+        return [tags] if isinstance(tags, str) else tags
+
+
 # ====================================
 # Snapshot node
 # ====================================
@@ -1310,6 +1469,10 @@ class SourceDefinition(NodeInfoMixin, ParsedSourceMandatory):
     def search_name(self):
         return f"{self.source_name}.{self.name}"
 
+    @property
+    def group(self):
+        return None
+
 
 # ====================================
 # Exposure node
@@ -1849,13 +2012,11 @@ ManifestSQLNode = Union[
     SqlNode,
     GenericTestNode,
     SnapshotNode,
+    UnitTestNode,
 ]
 
-# All SQL nodes plus SeedNode (csv files)
-ManifestNode = Union[
-    ManifestSQLNode,
-    SeedNode,
-]
+# All SQL nodes plus SeedNode/FixtureNode (csv files)
+ManifestNode = Union[ManifestSQLNode, SeedNode, FixtureNode]
 
 ResultNode = Union[
     ManifestNode,
@@ -1869,6 +2030,7 @@ GraphMemberNode = Union[
     Metric,
     SavedQuery,
     SemanticModel,
+    UnitTestDefinition,
 ]
 
 # All "nodes" (or node-like objects) in this file
@@ -1879,7 +2041,4 @@ Resource = Union[
     Group,
 ]
 
-TestNode = Union[
-    SingularTestNode,
-    GenericTestNode,
-]
+TestNode = Union[SingularTestNode, GenericTestNode]

core/dbt/contracts/graph/unparsed.py

@@ -1,5 +1,7 @@
 import datetime
 import re
+import csv
+from io import StringIO
 
 from dbt import deprecations
 from dbt.node_types import NodeType
@@ -769,3 +771,92 @@ def normalize_date(d: Optional[datetime.date]) -> Optional[datetime.datetime]:
         dt = dt.astimezone()
 
     return dt
+
+
+class UnitTestFormat(StrEnum):
+    CSV = "csv"
+    Dict = "dict"
+
+
+class UnitTestFixture:
+    @property
+    def format(self) -> UnitTestFormat:
+        return UnitTestFormat.Dict
+
+    @property
+    def rows(self) -> Optional[Union[str, List[Dict[str, Any]]]]:
+        return None
+
+    @property
+    def fixture(self) -> Optional[str]:  # TODO: typing
+        return None
+
+    def get_rows(self) -> List[Dict[str, Any]]:
+        if self.format == UnitTestFormat.Dict:
+            assert isinstance(self.rows, List)
+            return self.rows
+        elif self.format == UnitTestFormat.CSV:
+            rows: List[Dict[str, Any]] = []
+            if self.fixture is not None:
+                # resolve name to file Path
+                # read file into row dict
+                # breakpoint()
+                # assert isinstance(self.rows, str)
+                # dummy_file = StringIO(self.rows)
+                # reader = {csv.DictReader(dummy_file)}
+                # for row in reader:
+                #     rows.append(row)
+                pass
+            else:  # using inline csv
+                assert isinstance(self.rows, str)
+                dummy_file = StringIO(self.rows)
+                reader = csv.DictReader(dummy_file)
+                for row in reader:
+                    rows.append(row)
+            return rows
+
+    def validate_fixture(self, fixture_type, test_name) -> None:
+        if self.format == UnitTestFormat.Dict and not isinstance(self.rows, list):
+            raise ParsingError(
+                f"Unit test {test_name} has {fixture_type} rows which do not match format {self.format}"
+            )
+        if self.format == UnitTestFormat.CSV and not (
+            isinstance(self.rows, str) or isinstance(self.fixture, str)
+        ):
+            # TODO: update this message
+            raise ParsingError(
+                f"Unit test {test_name} has {fixture_type} rows which do not match format {self.format}"
+            )
+
+
+@dataclass
+class UnitTestInputFixture(dbtClassMixin, UnitTestFixture):
+    input: str
+    rows: Optional[Union[str, List[Dict[str, Any]]]] = None
+    format: UnitTestFormat = UnitTestFormat.Dict
+    fixture: Optional[str] = None
+
+
+@dataclass
+class UnitTestOutputFixture(dbtClassMixin, UnitTestFixture):
+    rows: Optional[Union[str, List[Dict[str, Any]]]] = None
+    format: UnitTestFormat = UnitTestFormat.Dict
+    fixture: Optional[str] = None
+
+
+@dataclass
+class UnitTestOverrides(dbtClassMixin):
+    macros: Dict[str, Any] = field(default_factory=dict)
+    vars: Dict[str, Any] = field(default_factory=dict)
+    env_vars: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class UnparsedUnitTest(dbtClassMixin):
+    name: str
+    model: str  # name of the model being unit tested
+    given: Sequence[UnitTestInputFixture]
+    expect: UnitTestOutputFixture
+    description: str = ""
+    overrides: Optional[UnitTestOverrides] = None
+    config: Dict[str, Any] = field(default_factory=dict)

core/dbt/contracts/project.py

@@ -213,6 +213,7 @@ class Project(dbtClassMixin, Replaceable):
     analyses: Dict[str, Any] = field(default_factory=dict)
     sources: Dict[str, Any] = field(default_factory=dict)
     tests: Dict[str, Any] = field(default_factory=dict)
+    unit_tests: Dict[str, Any] = field(default_factory=dict)
     metrics: Dict[str, Any] = field(default_factory=dict)
     semantic_models: Dict[str, Any] = field(default_factory=dict)
     saved_queries: Dict[str, Any] = field(default_factory=dict)
@@ -255,6 +256,7 @@ class Project(dbtClassMixin, Replaceable):
             "semantic_models": "semantic-models",
             "saved_queries": "saved-queries",
             "dbt_cloud": "dbt-cloud",
+            "unit_tests": "unit-tests",
         }
 
     @classmethod

core/dbt/events/types.py

@@ -2216,7 +2216,7 @@ class SQLCompiledPath(InfoLevel):
         return "Z026"
 
     def message(self) -> str:
-        return f"  compiled Code at {self.path}"
+        return f"  compiled code at {self.path}"
 
 
 class CheckNodeTestFailure(InfoLevel):

core/dbt/exceptions.py

@@ -192,7 +192,7 @@ class DbtDatabaseError(DbtRuntimeError):
         lines = []
 
         if hasattr(self.node, "build_path") and self.node.build_path:
-            lines.append(f"compiled Code at {self.node.build_path}")
+            lines.append(f"compiled code at {self.node.build_path}")
 
         return lines + DbtRuntimeError.process_stack(self)
 
@@ -1220,6 +1220,12 @@ class InvalidAccessTypeError(ParsingError):
         super().__init__(msg=msg)
 
 
+class InvalidUnitTestGivenInput(ParsingError):
+    def __init__(self, input: str) -> None:
+        msg = f"Unit test given inputs must be either a 'ref', 'source' or 'this' call. Got: '{input}'."
+        super().__init__(msg=msg)
+
+
 class SameKeyNestedError(CompilationError):
     def __init__(self) -> None:
         msg = "Test cannot have the same key at the top-level and in config"

core/dbt/graph/selector.py

@@ -31,6 +31,8 @@ def can_select_indirectly(node):
     """
     if node.resource_type == NodeType.Test:
         return True
+    elif node.resource_type == NodeType.Unit:
+        return True
     else:
         return False
 
@@ -171,9 +173,12 @@ class NodeSelector(MethodManager):
         elif unique_id in self.manifest.semantic_models:
             semantic_model = self.manifest.semantic_models[unique_id]
             return semantic_model.config.enabled
+        elif unique_id in self.manifest.unit_tests:
+            return True
         elif unique_id in self.manifest.saved_queries:
             saved_query = self.manifest.saved_queries[unique_id]
             return saved_query.config.enabled
+
         node = self.manifest.nodes[unique_id]
 
         if self.include_empty_nodes:
@@ -199,6 +204,8 @@ class NodeSelector(MethodManager):
             node = self.manifest.metrics[unique_id]
         elif unique_id in self.manifest.semantic_models:
             node = self.manifest.semantic_models[unique_id]
+        elif unique_id in self.manifest.unit_tests:
+            node = self.manifest.unit_tests[unique_id]
         elif unique_id in self.manifest.saved_queries:
             node = self.manifest.saved_queries[unique_id]
         else:
@@ -246,8 +253,11 @@ class NodeSelector(MethodManager):
             )
 
         for unique_id in self.graph.select_successors(selected):
-            if unique_id in self.manifest.nodes:
-                node = self.manifest.nodes[unique_id]
+            if unique_id in self.manifest.nodes or unique_id in self.manifest.unit_tests:
+                if unique_id in self.manifest.nodes:
+                    node = self.manifest.nodes[unique_id]
+                elif unique_id in self.manifest.unit_tests:
+                    node = self.manifest.unit_tests[unique_id]  # type: ignore
                 if can_select_indirectly(node):
                     # should we add it in directly?
                     if indirect_selection == IndirectSelection.Eager or set(

core/dbt/graph/selector_methods.py

@@ -18,6 +18,7 @@ from dbt.contracts.graph.nodes import (
     ResultNode,
     ManifestNode,
     ModelNode,
+    UnitTestDefinition,
     SavedQuery,
     SemanticModel,
 )
@@ -148,6 +149,21 @@ class SelectorMethod(metaclass=abc.ABCMeta):
                 continue
             yield unique_id, metric
 
+    def unit_tests(
+        self, included_nodes: Set[UniqueId]
+    ) -> Iterator[Tuple[UniqueId, UnitTestDefinition]]:
+        for unique_id, unit_test in self.manifest.unit_tests.items():
+            unique_id = UniqueId(unique_id)
+            if unique_id not in included_nodes:
+                continue
+            yield unique_id, unit_test
+
+    def parsed_and_unit_nodes(self, included_nodes: Set[UniqueId]):
+        yield from chain(
+            self.parsed_nodes(included_nodes),
+            self.unit_tests(included_nodes),
+        )
+
     def semantic_model_nodes(
         self, included_nodes: Set[UniqueId]
     ) -> Iterator[Tuple[UniqueId, SemanticModel]]:
@@ -176,6 +192,7 @@ class SelectorMethod(metaclass=abc.ABCMeta):
             self.source_nodes(included_nodes),
             self.exposure_nodes(included_nodes),
             self.metric_nodes(included_nodes),
+            self.unit_tests(included_nodes),
             self.semantic_model_nodes(included_nodes),
         )
 
@@ -192,6 +209,7 @@ class SelectorMethod(metaclass=abc.ABCMeta):
             self.parsed_nodes(included_nodes),
             self.exposure_nodes(included_nodes),
             self.metric_nodes(included_nodes),
+            self.unit_tests(included_nodes),
             self.semantic_model_nodes(included_nodes),
             self.saved_query_nodes(included_nodes),
         )
@@ -519,10 +537,13 @@ class TestNameSelectorMethod(SelectorMethod):
     __test__ = False
 
     def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
-        for node, real_node in self.parsed_nodes(included_nodes):
-            if real_node.resource_type == NodeType.Test and hasattr(real_node, "test_metadata"):
-                if fnmatch(real_node.test_metadata.name, selector):  # type: ignore[union-attr]
-                    yield node
+        for unique_id, node in self.parsed_and_unit_nodes(included_nodes):
+            if node.resource_type == NodeType.Test and hasattr(node, "test_metadata"):
+                if fnmatch(node.test_metadata.name, selector):  # type: ignore[union-attr]
+                    yield unique_id
+            elif node.resource_type == NodeType.Unit:
+                if fnmatch(node.name, selector):
+                    yield unique_id
 
 
 class TestTypeSelectorMethod(SelectorMethod):

core/dbt/graph/selector_spec.py

@@ -99,6 +99,7 @@ class SelectionCriteria:
         except ValueError as exc:
             raise InvalidSelectorError(f"'{method_parts[0]}' is not a valid method name") from exc
 
+        # Following is for cases like config.severity and config.materialized
         method_arguments: List[str] = method_parts[1:]
 
         return method_name, method_arguments

core/dbt/include/global_project/macros/materializations/tests/helpers.sql

@@ -12,3 +12,31 @@
       {{ "limit " ~ limit if limit != none }}
     ) dbt_internal_test
 {%- endmacro %}
+
+
+{% macro get_unit_test_sql(main_sql, expected_fixture_sql, expected_column_names) -%}
+  {{ adapter.dispatch('get_unit_test_sql', 'dbt')(main_sql, expected_fixture_sql, expected_column_names) }}
+{%- endmacro %}
+
+{% macro default__get_unit_test_sql(main_sql, expected_fixture_sql, expected_column_names) -%}
+-- Build actual result given inputs
+with dbt_internal_unit_test_actual AS (
+  select
+    {% for expected_column_name in expected_column_names %}{{expected_column_name}}{% if not loop.last -%},{% endif %}{%- endfor -%}, {{ dbt.string_literal("actual") }} as actual_or_expected
+  from (
+    {{ main_sql }}
+  ) _dbt_internal_unit_test_actual
+),
+-- Build expected result
+dbt_internal_unit_test_expected AS (
+  select
+    {% for expected_column_name in expected_column_names %}{{expected_column_name}}{% if not loop.last -%}, {% endif %}{%- endfor -%}, {{ dbt.string_literal("expected") }} as actual_or_expected
+  from (
+    {{ expected_fixture_sql }}
+  ) _dbt_internal_unit_test_expected
+)
+-- Union actual and expected results
+select * from dbt_internal_unit_test_actual
+union all
+select * from dbt_internal_unit_test_expected
+{%- endmacro %}

core/dbt/include/global_project/macros/materializations/tests/unit.sql

@@ -0,0 +1,29 @@
+{%- materialization unit, default -%}
+
+  {% set relations = [] %}
+
+  {% set expected_rows = config.get('expected_rows') %}
+  {% set tested_expected_column_names = expected_rows[0].keys() if (expected_rows | length ) > 0 else get_columns_in_query(sql) %} %}
+
+  {%- set target_relation = this.incorporate(type='table') -%}
+  {%- set temp_relation = make_temp_relation(target_relation)-%}
+  {% do run_query(get_create_table_as_sql(True, temp_relation, get_empty_subquery_sql(sql))) %}
+  {%- set columns_in_relation = adapter.get_columns_in_relation(temp_relation) -%}
+  {%- set column_name_to_data_types = {} -%}
+  {%- for column in columns_in_relation -%}
+  {%- do column_name_to_data_types.update({column.name: column.dtype}) -%}
+  {%- endfor -%}
+
+  {% set unit_test_sql = get_unit_test_sql(sql, get_expected_sql(expected_rows, column_name_to_data_types), tested_expected_column_names) %}
+
+  {% call statement('main', fetch_result=True) -%}
+
+    {{ unit_test_sql }}
+
+  {%- endcall %}
+
+  {% do adapter.drop_relation(temp_relation) %}
+
+  {{ return({'relations': relations}) }}
+
+{%- endmaterialization -%}

core/dbt/include/global_project/macros/unit_test_sql/get_fixture_sql.sql

@@ -0,0 +1,77 @@
+{% macro get_fixture_sql(rows, column_name_to_data_types) %}
+-- Fixture for {{ model.name }}
+{% set default_row = {} %}
+
+{%- if not column_name_to_data_types -%}
+{%- set columns_in_relation = adapter.get_columns_in_relation(this) -%}
+{%- set column_name_to_data_types = {} -%}
+{%- for column in columns_in_relation -%}
+{%- do column_name_to_data_types.update({column.name: column.dtype}) -%}
+{%- endfor -%}
+{%- endif -%}
+
+{%- if not column_name_to_data_types -%}
+    {{ exceptions.raise_compiler_error("columns not available for" ~ model.name) }}
+{%- endif -%}
+
+{%- for column_name, column_type in column_name_to_data_types.items() -%}
+    {%- do default_row.update({column_name: (safe_cast("null", column_type) | trim )}) -%}
+{%- endfor -%}
+
+{%- for row in rows -%}
+{%- do format_row(row, column_name_to_data_types) -%}
+
+{%- set default_row_copy = default_row.copy() -%}
+{%- do default_row_copy.update(row) -%}
+select
+{%- for column_name, column_value in default_row_copy.items() %} {{ column_value }} AS {{ column_name }}{% if not loop.last -%}, {%- endif %}
+{%- endfor %}
+{%- if not loop.last %}
+union all
+{% endif %}
+{%- endfor -%}
+
+{%- if (rows | length) == 0 -%}
+    select
+    {%- for column_name, column_value in default_row.items() %} {{ column_value }} AS {{ column_name }}{% if not loop.last -%},{%- endif %}
+    {%- endfor %}
+    limit 0
+{%- endif -%}
+{% endmacro %}
+
+
+{% macro get_expected_sql(rows, column_name_to_data_types) %}
+
+{%- if (rows | length) == 0 -%}
+    select * FROM dbt_internal_unit_test_actual
+    limit 0
+{%- else -%}
+{%- for row in rows -%}
+{%- do format_row(row, column_name_to_data_types) -%}
+select
+{%- for column_name, column_value in row.items() %} {{ column_value }} AS {{ column_name }}{% if not loop.last -%}, {%- endif %}
+{%- endfor %}
+{%- if not loop.last %}
+union all
+{% endif %}
+{%- endfor -%}
+{%- endif -%}
+
+{% endmacro %}
+
+{%- macro format_row(row, column_name_to_data_types) -%}
+
+{#-- wrap yaml strings in quotes, apply cast --#}
+{%- for column_name, column_value in row.items() -%}
+{% set row_update = {column_name: column_value} %}
+{%- if column_value is string -%}
+{%- set row_update = {column_name: safe_cast(dbt.string_literal(column_value), column_name_to_data_types[column_name]) } -%}
+{%- elif column_value is none -%}
+{%- set row_update = {column_name: safe_cast('null', column_name_to_data_types[column_name]) } -%}
+{%- else -%}
+{%- set row_update = {column_name: safe_cast(column_value, column_name_to_data_types[column_name]) } -%}
+{%- endif -%}
+{%- do row.update(row_update) -%}
+{%- endfor -%}
+
+{%- endmacro -%}

core/dbt/node_types.py

@@ -21,6 +21,7 @@ class NodeType(StrEnum):
     Model = "model"
     Analysis = "analysis"
     Test = "test"
+    Fixture = "fixture"
     Snapshot = "snapshot"
     Operation = "operation"
     Seed = "seed"
@@ -35,6 +36,7 @@ class NodeType(StrEnum):
     Group = "group"
     SavedQuery = "saved_query"
     SemanticModel = "semantic_model"
+    Unit = "unit_test"
 
     @classmethod
     def executable(cls) -> List["NodeType"]:

core/dbt/parser/fixtures.py

@@ -0,0 +1,28 @@
+from dbt.context.context_config import ContextConfig
+from dbt.contracts.graph.nodes import FixtureNode
+from dbt.node_types import NodeType
+from dbt.parser.base import SimpleSQLParser
+from dbt.parser.search import FileBlock
+
+
+class FixtureParser(SimpleSQLParser[FixtureNode]):
+    def parse_from_dict(self, dct, validate=True) -> FixtureNode:
+        # fixtures need the root_path because the contents are not loaded
+        dct["root_path"] = self.project.project_root
+        if "language" in dct:  # TODO: ?? this was there for seeds
+            del dct["language"]
+        # raw_code is not currently used, but it might be in the future
+        if validate:
+            FixtureNode.validate(dct)
+        return FixtureNode.from_dict(dct)
+
+    @property
+    def resource_type(self) -> NodeType:
+        return NodeType.Fixture  # TODO: ??
+
+    @classmethod
+    def get_compiled_path(cls, block: FileBlock):
+        return block.path.relative_path
+
+    def render_with_context(self, parsed_node: FixtureNode, config: ContextConfig) -> None:
+        """Fixtures don't need to do any rendering."""

core/dbt/parser/manifest.py

@@ -39,6 +39,7 @@ from dbt.constants import (
     MANIFEST_FILE_NAME,
     PARTIAL_PARSE_FILE_NAME,
     SEMANTIC_MANIFEST_FILE_NAME,
+    UNIT_TEST_MANIFEST_FILE_NAME,
 )
 from dbt.helper_types import PathSet
 from dbt.events.functions import fire_event, get_invocation_id, warn_or_error
@@ -1765,8 +1766,13 @@ def write_semantic_manifest(manifest: Manifest, target_path: str) -> None:
     semantic_manifest.write_json_to_file(path)
 
 
-def write_manifest(manifest: Manifest, target_path: str):
-    path = os.path.join(target_path, MANIFEST_FILE_NAME)
+def write_manifest(manifest: Manifest, target_path: str, which: Optional[str] = None):
+    if which and which == "unit-test":
+        file_name = UNIT_TEST_MANIFEST_FILE_NAME
+    else:
+        file_name = MANIFEST_FILE_NAME
+
+    path = os.path.join(target_path, file_name)
     manifest.write(path)
 
     write_semantic_manifest(manifest=manifest, target_path=target_path)

core/dbt/parser/partial.py

@@ -608,7 +608,7 @@ class PartialParsing:
         self.saved_manifest.files.pop(file_id)
 
     # For each key in a schema file dictionary, process the changed, deleted, and added
-    # elemnts for the key lists
+    # elements for the key lists
     def handle_schema_file_changes(self, schema_file, saved_yaml_dict, new_yaml_dict):
         # loop through comparing previous dict_from_yaml with current dict_from_yaml
         # Need to do the deleted/added/changed thing, just like the files lists
@@ -681,6 +681,7 @@ class PartialParsing:
         handle_change("metrics", self.delete_schema_metric)
         handle_change("groups", self.delete_schema_group)
         handle_change("semantic_models", self.delete_schema_semantic_model)
+        handle_change("unit_tests", self.delete_schema_unit_test)
         handle_change("saved_queries", self.delete_schema_saved_query)
 
     def _handle_element_change(
@@ -938,6 +939,17 @@ class PartialParsing:
             elif unique_id in self.saved_manifest.disabled:
                 self.delete_disabled(unique_id, schema_file.file_id)
 
+    def delete_schema_unit_test(self, schema_file, unit_test_dict):
+        unit_test_name = unit_test_dict["name"]
+        unit_tests = schema_file.unit_tests.copy()
+        for unique_id in unit_tests:
+            if unique_id in self.saved_manifest.unit_tests:
+                unit_test = self.saved_manifest.unit_tests[unique_id]
+                if unit_test.name == unit_test_name:
+                    self.saved_manifest.unit_tests.pop(unique_id)
+                    schema_file.unit_tests.remove(unique_id)
+            # No disabled unit tests yet
+
     def get_schema_element(self, elem_list, elem_name):
         for element in elem_list:
             if "name" in element and element["name"] == elem_name:

core/dbt/parser/read_files.py

@@ -407,6 +407,11 @@ def get_file_types_for_project(project):
             "extensions": [".sql"],
             "parser": "GenericTestParser",
         },
+        ParseFileType.Fixture: {
+            "paths": project.fixture_paths,
+            "extensions": [".csv"],
+            "parser": "FixtureParser",
+        },
         ParseFileType.Seed: {
             "paths": project.seed_paths,
             "extensions": [".csv"],

core/dbt/parser/schemas.py

@@ -139,6 +139,11 @@ class SchemaParser(SimpleParser[YamlBlock, ModelNode]):
             self.root_project, self.project.project_name, self.schema_yaml_vars
         )
 
+    # This is unnecessary, but mypy was requiring it. Clean up parser code so
+    # we don't have to do this.
+    def parse_from_dict(self, dct):
+        pass
+
     @classmethod
     def get_compiled_path(cls, block: FileBlock) -> str:
         # should this raise an error?
@@ -226,6 +231,12 @@ class SchemaParser(SimpleParser[YamlBlock, ModelNode]):
                 semantic_model_parser = SemanticModelParser(self, yaml_block)
                 semantic_model_parser.parse()
 
+            if "unit_tests" in dct:
+                from dbt.parser.unit_tests import UnitTestParser
+
+                unit_test_parser = UnitTestParser(self, yaml_block)
+                unit_test_parser.parse()
+
             if "saved_queries" in dct:
                 from dbt.parser.schema_yaml_readers import SavedQueryParser
 
@@ -251,12 +262,13 @@ class ParseResult:
 
 
 # abstract base class (ABCMeta)
-# Four subclasses: MetricParser, ExposureParser, GroupParser, SourceParser, PatchParser
+# Many subclasses: MetricParser, ExposureParser, GroupParser, SourceParser,
+# PatchParser, SemanticModelParser, SavedQueryParser, UnitTestParser
 class YamlReader(metaclass=ABCMeta):
     def __init__(self, schema_parser: SchemaParser, yaml: YamlBlock, key: str) -> None:
         self.schema_parser = schema_parser
         # key: models, seeds, snapshots, sources, macros,
-        # analyses, exposures
+        # analyses, exposures, unit_tests
         self.key = key
         self.yaml = yaml
         self.schema_yaml_vars = SchemaYamlVars()
@@ -304,7 +316,7 @@ class YamlReader(metaclass=ABCMeta):
             if coerce_dict_str(entry) is None:
                 raise YamlParseListError(path, self.key, data, "expected a dict with string keys")
 
-            if "name" not in entry:
+            if "name" not in entry and "model" not in entry:
                 raise ParsingError("Entry did not contain a name")
 
             # Render the data (except for tests and descriptions).

core/dbt/parser/unit_tests.py

@@ -0,0 +1,271 @@
+from typing import List, Set, Dict, Any
+
+from dbt.config import RuntimeConfig
+from dbt.context.context_config import ContextConfig
+from dbt.context.providers import generate_parse_exposure, get_rendered
+from dbt.contracts.files import FileHash
+from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.graph.model_config import UnitTestNodeConfig, ModelConfig
+from dbt.contracts.graph.nodes import (
+    ModelNode,
+    UnitTestNode,
+    UnitTestDefinition,
+    DependsOn,
+    UnitTestConfig,
+)
+from dbt.contracts.graph.unparsed import UnparsedUnitTest
+from dbt.exceptions import ParsingError, InvalidUnitTestGivenInput
+from dbt.graph import UniqueId
+from dbt.node_types import NodeType
+from dbt.parser.schemas import (
+    SchemaParser,
+    YamlBlock,
+    ValidationError,
+    JSONValidationError,
+    YamlParseDictError,
+    YamlReader,
+    ParseResult,
+)
+from dbt.utils import get_pseudo_test_path
+from dbt_extractor import py_extract_from_source, ExtractionError  # type: ignore
+
+
+class UnitTestManifestLoader:
+    def __init__(self, manifest, root_project, selected) -> None:
+        self.manifest: Manifest = manifest
+        self.root_project: RuntimeConfig = root_project
+        # selected comes from the initial selection against a "regular" manifest
+        self.selected: Set[UniqueId] = selected
+        self.unit_test_manifest = Manifest(macros=manifest.macros)
+
+    def load(self) -> Manifest:
+        for unique_id in self.selected:
+            unit_test_case = self.manifest.unit_tests[unique_id]
+            self.parse_unit_test_case(unit_test_case)
+
+        return self.unit_test_manifest
+
+    def parse_unit_test_case(self, test_case: UnitTestDefinition):
+        package_name = self.root_project.project_name
+
+        # Create unit test node based on the node being tested
+        tested_node = self.manifest.ref_lookup.perform_lookup(
+            f"model.{package_name}.{test_case.model}", self.manifest
+        )
+        assert isinstance(tested_node, ModelNode)
+
+        # Create UnitTestNode based on model being tested. Since selection has
+        # already been done, we don't have to care about fields that are necessary
+        # for selection.
+        # Note: no depends_on, that's added later using input nodes
+        name = f"{test_case.model}__{test_case.name}"
+        unit_test_node = UnitTestNode(
+            name=name,
+            resource_type=NodeType.Unit,
+            package_name=package_name,
+            path=get_pseudo_test_path(name, test_case.original_file_path),
+            original_file_path=test_case.original_file_path,
+            unique_id=test_case.unique_id,
+            config=UnitTestNodeConfig(
+                materialized="unit", expected_rows=test_case.expect.get_rows()
+            ),
+            raw_code=tested_node.raw_code,
+            database=tested_node.database,
+            schema=tested_node.schema,
+            alias=name,
+            fqn=test_case.unique_id.split("."),
+            checksum=FileHash.empty(),
+            tested_node_unique_id=tested_node.unique_id,
+            overrides=test_case.overrides,
+        )
+
+        # TODO: generalize this method
+        ctx = generate_parse_exposure(
+            unit_test_node,  # type: ignore
+            self.root_project,
+            self.manifest,
+            package_name,
+        )
+        get_rendered(unit_test_node.raw_code, ctx, unit_test_node, capture_macros=True)
+        # unit_test_node now has a populated refs/sources
+
+        self.unit_test_manifest.nodes[unit_test_node.unique_id] = unit_test_node
+
+        # Now create input_nodes for the test inputs
+        """
+        given:
+          - input: ref('my_model_a')
+            rows: []
+          - input: ref('my_model_b')
+            rows:
+              - {id: 1, b: 2}
+              - {id: 2, b: 2}
+        """
+        # Add the model "input" nodes, consisting of all referenced models in the unit test.
+        # This creates a model for every input in every test, so there may be multiple
+        # input models substituting for the same input ref'd model.
+        for given in test_case.given:
+            # extract the original_input_node from the ref in the "input" key of the given list
+            original_input_node = self._get_original_input_node(given.input, tested_node)
+
+            original_input_node_columns = None
+            if (
+                original_input_node.resource_type == NodeType.Model
+                and original_input_node.config.contract.enforced
+            ):
+                original_input_node_columns = {
+                    column.name: column.data_type for column in original_input_node.columns
+                }
+
+            # TODO: include package_name?
+            input_name = f"{unit_test_node.name}__{original_input_node.name}"
+            input_unique_id = f"model.{package_name}.{input_name}"
+            input_node = ModelNode(
+                raw_code=self._build_fixture_raw_code(
+                    given.get_rows(), original_input_node_columns
+                ),
+                resource_type=NodeType.Model,
+                package_name=package_name,
+                path=original_input_node.path,
+                original_file_path=original_input_node.original_file_path,
+                unique_id=input_unique_id,
+                name=input_name,
+                config=ModelConfig(materialized="ephemeral"),
+                database=original_input_node.database,
+                schema=original_input_node.schema,
+                alias=original_input_node.alias,
+                fqn=input_unique_id.split("."),
+                checksum=FileHash.empty(),
+            )
+            self.unit_test_manifest.nodes[input_node.unique_id] = input_node
+
+            # Populate this_input_node_unique_id if input fixture represents node being tested
+            if original_input_node == tested_node:
+                unit_test_node.this_input_node_unique_id = input_node.unique_id
+
+            # Add unique ids of input_nodes to depends_on
+            unit_test_node.depends_on.nodes.append(input_node.unique_id)
+
+    def _build_fixture_raw_code(self, rows, column_name_to_data_types) -> str:
+        return ("{{{{ get_fixture_sql({rows}, {column_name_to_data_types}) }}}}").format(
+            rows=rows, column_name_to_data_types=column_name_to_data_types
+        )
+
+    def _get_original_input_node(self, input: str, tested_node: ModelNode):
+        """
+        Returns the original input node as defined in the project given an input reference
+        and the node being tested.
+
+        input: str representing how input node is referenced in tested model sql
+          * examples:
+            - "ref('my_model_a')"
+            - "source('my_source_schema', 'my_source_name')"
+            - "this"
+        tested_node: ModelNode of representing node being tested
+        """
+        if input.strip() == "this":
+            original_input_node = tested_node
+        else:
+            try:
+                statically_parsed = py_extract_from_source(f"{{{{ {input} }}}}")
+            except ExtractionError:
+                raise InvalidUnitTestGivenInput(input=input)
+
+            if statically_parsed["refs"]:
+                for ref in statically_parsed["refs"]:
+                    name = ref.get("name")
+                    package = ref.get("package")
+                    version = ref.get("version")
+                    # TODO: disabled lookup, versioned lookup, public models
+                    original_input_node = self.manifest.ref_lookup.find(
+                        name, package, version, self.manifest
+                    )
+            elif statically_parsed["sources"]:
+                input_package_name, input_source_name = statically_parsed["sources"][0]
+                original_input_node = self.manifest.source_lookup.find(
+                    input_source_name, input_package_name, self.manifest
+                )
+            else:
+                raise InvalidUnitTestGivenInput(input=input)
+
+        return original_input_node
+
+
+class UnitTestParser(YamlReader):
+    def __init__(self, schema_parser: SchemaParser, yaml: YamlBlock) -> None:
+        super().__init__(schema_parser, yaml, "unit_tests")
+        self.schema_parser = schema_parser
+        self.yaml = yaml
+
+    def parse(self) -> ParseResult:
+        for data in self.get_key_dicts():
+            unit_test = self._get_unit_test(data)
+            model_name_split = unit_test.model.split()
+            tested_model_node = self._find_tested_model_node(unit_test)
+            unit_test_case_unique_id = (
+                f"{NodeType.Unit}.{self.project.project_name}.{unit_test.model}.{unit_test.name}"
+            )
+            unit_test_fqn = [self.project.project_name] + model_name_split + [unit_test.name]
+            unit_test_config = self._build_unit_test_config(unit_test_fqn, unit_test.config)
+
+            # Check that format and type of rows matches for each given input
+            for input in unit_test.given:
+                input.validate_fixture("input", unit_test.name)
+            unit_test.expect.validate_fixture("expected", unit_test.name)
+
+            unit_test_definition = UnitTestDefinition(
+                name=unit_test.name,
+                model=unit_test.model,
+                resource_type=NodeType.Unit,
+                package_name=self.project.project_name,
+                path=self.yaml.path.relative_path,
+                original_file_path=self.yaml.path.original_file_path,
+                unique_id=unit_test_case_unique_id,
+                given=unit_test.given,
+                expect=unit_test.expect,
+                description=unit_test.description,
+                overrides=unit_test.overrides,
+                depends_on=DependsOn(nodes=[tested_model_node.unique_id]),
+                fqn=unit_test_fqn,
+                config=unit_test_config,
+            )
+            self.manifest.add_unit_test(self.yaml.file, unit_test_definition)
+
+        return ParseResult()
+
+    def _get_unit_test(self, data: Dict[str, Any]) -> UnparsedUnitTest:
+        try:
+            UnparsedUnitTest.validate(data)
+            return UnparsedUnitTest.from_dict(data)
+        except (ValidationError, JSONValidationError) as exc:
+            raise YamlParseDictError(self.yaml.path, self.key, data, exc)
+
+    def _find_tested_model_node(self, unit_test: UnparsedUnitTest) -> ModelNode:
+        package_name = self.project.project_name
+        model_name_split = unit_test.model.split()
+        model_name = model_name_split[0]
+        model_version = model_name_split[1] if len(model_name_split) == 2 else None
+
+        tested_node = self.manifest.ref_lookup.find(
+            model_name, package_name, model_version, self.manifest
+        )
+        if not tested_node:
+            raise ParsingError(
+                f"Unable to find model '{package_name}.{unit_test.model}' for unit tests in {self.yaml.path.original_file_path}"
+            )
+
+        return tested_node
+
+    def _build_unit_test_config(
+        self, unit_test_fqn: List[str], config_dict: Dict[str, Any]
+    ) -> UnitTestConfig:
+        config = ContextConfig(
+            self.schema_parser.root_project,
+            unit_test_fqn,
+            NodeType.Unit,
+            self.schema_parser.project.project_name,
+        )
+        unit_test_config_dict = config.build_config_dict(patch_config_dict=config_dict)
+        unit_test_config_dict = self.render_entry(unit_test_config_dict)
+
+        return UnitTestConfig.from_dict(unit_test_config_dict)

core/dbt/task/retry.py

@@ -17,6 +17,7 @@ from dbt.task.run_operation import RunOperationTask
 from dbt.task.seed import SeedTask
 from dbt.task.snapshot import SnapshotTask
 from dbt.task.test import TestTask
+from dbt.task.unit_test import UnitTestTask
 
 RETRYABLE_STATUSES = {NodeStatus.Error, NodeStatus.Fail, NodeStatus.Skipped, NodeStatus.RuntimeErr}
 OVERRIDE_PARENT_FLAGS = {
@@ -40,6 +41,7 @@ TASK_DICT = {
     "test": TestTask,
     "run": RunTask,
     "run-operation": RunOperationTask,
+    "unit-test": UnitTestTask,
 }
 
 CMD_DICT = {
@@ -52,6 +54,7 @@ CMD_DICT = {
     "test": CliCommand.TEST,
     "run": CliCommand.RUN,
     "run-operation": CliCommand.RUN_OPERATION,
+    "unit-test": CliCommand.UNIT_TEST,
 }
 
 

core/dbt/task/runnable.py

@@ -122,6 +122,7 @@ class GraphRunnableTask(ConfiguredTask):
             fire_event(DefaultSelector(name=default_selector_name))
             spec = self.config.get_selector(default_selector_name)
         else:
+            # This is what's used with no default selector and no selection
             # use --select and --exclude args
             spec = parse_difference(self.selection_arg, self.exclusion_arg, indirect_selection)
         return spec
@@ -136,9 +137,14 @@ class GraphRunnableTask(ConfiguredTask):
 
     def get_graph_queue(self) -> GraphQueue:
         selector = self.get_node_selector()
+        # Following uses self.selection_arg and self.exclusion_arg
         spec = self.get_selection_spec()
         return selector.get_graph_queue(spec)
 
+    # A callback for unit testing
+    def reset_job_queue_and_manifest(self):
+        pass
+
     def _runtime_initialize(self):
         self.compile_manifest()
         if self.manifest is None or self.graph is None:
@@ -146,6 +152,9 @@ class GraphRunnableTask(ConfiguredTask):
 
         self.job_queue = self.get_graph_queue()
 
+        # for unit testing
+        self.reset_job_queue_and_manifest()
+
         # we use this a couple of times. order does not matter.
         self._flattened_nodes = []
         for uid in self.job_queue.get_selected_nodes():
@@ -486,7 +495,8 @@ class GraphRunnableTask(ConfiguredTask):
             )
 
         if self.args.write_json:
-            write_manifest(self.manifest, self.config.project_target_path)
+            # args.which used to determine file name for unit test manifest
+            write_manifest(self.manifest, self.config.project_target_path, self.args.which)
             if hasattr(result, "write"):
                 result.write(self.result_path())
 

core/dbt/task/unit_test.py

@@ -0,0 +1,227 @@
+from dataclasses import dataclass
+from dbt.dataclass_schema import dbtClassMixin
+import threading
+from typing import Dict, Any, Optional
+import io
+
+from .compile import CompileRunner
+from .run import RunTask
+
+from dbt.contracts.graph.nodes import UnitTestNode
+from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.results import TestStatus, RunResult
+from dbt.context.providers import generate_runtime_model_context
+from dbt.clients.jinja import MacroGenerator
+from dbt.events.functions import fire_event
+from dbt.events.types import (
+    LogTestResult,
+    LogStartLine,
+)
+from dbt.graph import ResourceTypeSelector
+from dbt.exceptions import (
+    DbtInternalError,
+    MissingMaterializationError,
+)
+from dbt.node_types import NodeType
+from dbt.parser.unit_tests import UnitTestManifestLoader
+
+
+@dataclass
+class UnitTestResultData(dbtClassMixin):
+    should_error: bool
+    adapter_response: Dict[str, Any]
+    diff: Optional[str] = None
+
+
+class UnitTestRunner(CompileRunner):
+    def describe_node(self):
+        return f"{self.node.resource_type} {self.node.name}"
+
+    def print_result_line(self, result):
+        model = result.node
+
+        fire_event(
+            LogTestResult(
+                name=model.name,
+                status=str(result.status),
+                index=self.node_index,
+                num_models=self.num_nodes,
+                execution_time=result.execution_time,
+                node_info=model.node_info,
+                num_failures=result.failures,
+            ),
+            level=LogTestResult.status_to_level(str(result.status)),
+        )
+
+    def print_start_line(self):
+        fire_event(
+            LogStartLine(
+                description=self.describe_node(),
+                index=self.node_index,
+                total=self.num_nodes,
+                node_info=self.node.node_info,
+            )
+        )
+
+    def before_execute(self):
+        self.print_start_line()
+
+    def execute_unit_test(self, node: UnitTestNode, manifest: Manifest) -> UnitTestResultData:
+        # generate_runtime_unit_test_context not strictly needed - this is to run the 'unit'
+        # materialization, not compile the node.compiled_code
+        context = generate_runtime_model_context(node, self.config, manifest)
+
+        materialization_macro = manifest.find_materialization_macro_by_name(
+            self.config.project_name, node.get_materialization(), self.adapter.type()
+        )
+
+        if materialization_macro is None:
+            raise MissingMaterializationError(
+                materialization=node.get_materialization(), adapter_type=self.adapter.type()
+            )
+
+        if "config" not in context:
+            raise DbtInternalError(
+                "Invalid materialization context generated, missing config: {}".format(context)
+            )
+
+        # generate materialization macro
+        macro_func = MacroGenerator(materialization_macro, context)
+        # execute materialization macro
+        macro_func()
+        # load results from context
+        # could eventually be returned directly by materialization
+        result = context["load_result"]("main")
+        adapter_response = result["response"].to_dict(omit_none=True)
+        table = result["table"]
+        actual = self._get_unit_test_table(table, "actual")
+        expected = self._get_unit_test_table(table, "expected")
+        should_error = actual.rows != expected.rows
+        diff = None
+        if should_error:
+            actual_output = self._agate_table_to_str(actual)
+            expected_output = self._agate_table_to_str(expected)
+
+            diff = f"\n\nActual:\n{actual_output}\n\nExpected:\n{expected_output}\n"
+        return UnitTestResultData(
+            diff=diff,
+            should_error=should_error,
+            adapter_response=adapter_response,
+        )
+
+    def execute(self, node: UnitTestNode, manifest: Manifest):
+        result = self.execute_unit_test(node, manifest)
+        thread_id = threading.current_thread().name
+
+        status = TestStatus.Pass
+        message = None
+        failures = 0
+        if result.should_error:
+            status = TestStatus.Fail
+            message = result.diff
+            failures = 1
+
+        return RunResult(
+            node=node,
+            status=status,
+            timing=[],
+            thread_id=thread_id,
+            execution_time=0,
+            message=message,
+            adapter_response=result.adapter_response,
+            failures=failures,
+        )
+
+    def after_execute(self, result):
+        self.print_result_line(result)
+
+    def _get_unit_test_table(self, result_table, actual_or_expected: str):
+        unit_test_table = result_table.where(
+            lambda row: row["actual_or_expected"] == actual_or_expected
+        )
+        columns = list(unit_test_table.columns.keys())
+        columns.remove("actual_or_expected")
+        return unit_test_table.select(columns)
+
+    def _agate_table_to_str(self, table) -> str:
+        # Hack to get Agate table output as string
+        output = io.StringIO()
+        if self.config.args.output == "json":
+            table.to_json(path=output)
+        else:
+            table.print_table(output=output, max_rows=None)
+        return output.getvalue().strip()
+
+
+class UnitTestSelector(ResourceTypeSelector):
+    # This is what filters out nodes except Unit Tests, in filter_selection
+    def __init__(self, graph, manifest, previous_state):
+        super().__init__(
+            graph=graph,
+            manifest=manifest,
+            previous_state=previous_state,
+            resource_types=[NodeType.Unit],
+        )
+
+
+class UnitTestTask(RunTask):
+    """
+    Unit testing:
+        Read schema files + custom data tests and validate that
+        constraints are satisfied.
+    """
+
+    def __init__(self, args, config, manifest):
+        # This will initialize the RunTask with the regular manifest
+        super().__init__(args, config, manifest)
+        # TODO: We might not need this, but leaving here for now.
+        self.original_manifest = manifest
+        self.using_unit_test_manifest = False
+
+    __test__ = False
+
+    def raise_on_first_error(self):
+        return False
+
+    @property
+    def selection_arg(self):
+        if self.using_unit_test_manifest is False:
+            return self.args.select
+        else:
+            # Everything in the unit test should be selected, since we
+            # created in from a selection list.
+            return ()
+
+    @property
+    def exclusion_arg(self):
+        if self.using_unit_test_manifest is False:
+            return self.args.exclude
+        else:
+            # Everything in the unit test should be selected, since we
+            # created in from a selection list.
+            return ()
+
+    def build_unit_test_manifest(self):
+        loader = UnitTestManifestLoader(self.manifest, self.config, self.job_queue._selected)
+        return loader.load()
+
+    def reset_job_queue_and_manifest(self):
+        # We have the selected models from the "regular" manifest, now we switch
+        # to using the unit_test_manifest to run the unit tests.
+        self.using_unit_test_manifest = True
+        self.manifest = self.build_unit_test_manifest()
+        self.compile_manifest()  # create the networkx graph
+        self.job_queue = self.get_graph_queue()
+
+    def get_node_selector(self) -> ResourceTypeSelector:
+        if self.manifest is None or self.graph is None:
+            raise DbtInternalError("manifest and graph must be set to get perform node selection")
+        # Filter out everything except unit tests
+        return UnitTestSelector(
+            graph=self.graph,
+            manifest=self.manifest,
+            previous_state=self.previous_state,
+        )
+
+    def get_runner_type(self, _):
+        return UnitTestRunner

schemas/dbt/manifest/v11.json

@@ -3401,6 +3401,517 @@
         "config"
       ]
     },
+    "UnitTestNodeConfig": {
+      "type": "object",
+      "title": "UnitTestNodeConfig",
+      "properties": {
+        "_extra": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "enabled": {
+          "type": "boolean",
+          "default": true
+        },
+        "alias": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "schema": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "database": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "tags": {
+          "anyOf": [
+            {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            },
+            {
+              "type": "string"
+            }
+          ]
+        },
+        "meta": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "group": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "materialized": {
+          "type": "string",
+          "default": "view"
+        },
+        "incremental_strategy": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "persist_docs": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "post-hook": {
+          "type": "array",
+          "items": {
+            "$ref": "#/$defs/Hook"
+          }
+        },
+        "pre-hook": {
+          "type": "array",
+          "items": {
+            "$ref": "#/$defs/Hook"
+          }
+        },
+        "quoting": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "column_types": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "full_refresh": {
+          "anyOf": [
+            {
+              "type": "boolean"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "unique_key": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "on_schema_change": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": "ignore"
+        },
+        "on_configuration_change": {
+          "enum": [
+            "apply",
+            "continue",
+            "fail"
+          ]
+        },
+        "grants": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "packages": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "docs": {
+          "$ref": "#/$defs/Docs"
+        },
+        "contract": {
+          "$ref": "#/$defs/ContractConfig"
+        },
+        "expected_rows": {
+          "type": "array",
+          "items": {
+            "type": "object",
+            "propertyNames": {
+              "type": "string"
+            }
+          }
+        }
+      },
+      "additionalProperties": true
+    },
+    "UnitTestOverrides": {
+      "type": "object",
+      "title": "UnitTestOverrides",
+      "properties": {
+        "macros": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "vars": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "env_vars": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        }
+      },
+      "additionalProperties": false
+    },
+    "UnitTestNode": {
+      "type": "object",
+      "title": "UnitTestNode",
+      "properties": {
+        "database": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "schema": {
+          "type": "string"
+        },
+        "name": {
+          "type": "string"
+        },
+        "resource_type": {
+          "enum": [
+            "model",
+            "analysis",
+            "test",
+            "snapshot",
+            "operation",
+            "seed",
+            "rpc",
+            "sql_operation",
+            "doc",
+            "source",
+            "macro",
+            "exposure",
+            "metric",
+            "group",
+            "saved_query",
+            "semantic_model",
+            "unit_test"
+          ]
+        },
+        "package_name": {
+          "type": "string"
+        },
+        "path": {
+          "type": "string"
+        },
+        "original_file_path": {
+          "type": "string"
+        },
+        "unique_id": {
+          "type": "string"
+        },
+        "fqn": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "alias": {
+          "type": "string"
+        },
+        "checksum": {
+          "$ref": "#/$defs/FileHash"
+        },
+        "config": {
+          "$ref": "#/$defs/UnitTestNodeConfig"
+        },
+        "_event_status": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "tags": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "description": {
+          "type": "string",
+          "default": ""
+        },
+        "columns": {
+          "type": "object",
+          "additionalProperties": {
+            "$ref": "#/$defs/ColumnInfo"
+          },
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "meta": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "group": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "docs": {
+          "$ref": "#/$defs/Docs"
+        },
+        "patch_path": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "build_path": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "deferred": {
+          "type": "boolean",
+          "default": false
+        },
+        "unrendered_config": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "created_at": {
+          "type": "number"
+        },
+        "config_call_dict": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "relation_name": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "raw_code": {
+          "type": "string",
+          "default": ""
+        },
+        "language": {
+          "type": "string",
+          "default": "sql"
+        },
+        "refs": {
+          "type": "array",
+          "items": {
+            "$ref": "#/$defs/RefArgs"
+          }
+        },
+        "sources": {
+          "type": "array",
+          "items": {
+            "type": "array",
+            "items": {
+              "type": "string"
+            }
+          }
+        },
+        "metrics": {
+          "type": "array",
+          "items": {
+            "type": "array",
+            "items": {
+              "type": "string"
+            }
+          }
+        },
+        "depends_on": {
+          "$ref": "#/$defs/DependsOn"
+        },
+        "compiled_path": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "compiled": {
+          "type": "boolean",
+          "default": false
+        },
+        "compiled_code": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "extra_ctes_injected": {
+          "type": "boolean",
+          "default": false
+        },
+        "extra_ctes": {
+          "type": "array",
+          "items": {
+            "$ref": "#/$defs/InjectedCTE"
+          }
+        },
+        "_pre_injected_sql": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "contract": {
+          "$ref": "#/$defs/Contract"
+        },
+        "attached_node": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "overrides": {
+          "anyOf": [
+            {
+              "$ref": "#/$defs/UnitTestOverrides"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        }
+      },
+      "additionalProperties": false,
+      "required": [
+        "database",
+        "schema",
+        "name",
+        "resource_type",
+        "package_name",
+        "path",
+        "original_file_path",
+        "unique_id",
+        "fqn",
+        "alias",
+        "checksum"
+      ]
+    },
     "SeedConfig": {
       "type": "object",
       "title": "SeedConfig",
@@ -5251,7 +5762,8 @@
             "metric",
             "group",
             "saved_query",
-            "semantic_model"
+            "semantic_model",
+            "unit_test"
           ]
         },
         "package_name": {
@@ -5822,7 +6334,8 @@
             "metric",
             "group",
             "saved_query",
-            "semantic_model"
+            "semantic_model",
+            "unit_test"
           ]
         },
         "package_name": {
@@ -5975,6 +6488,200 @@
         "node_relation"
       ]
     },
+    "UnitTestInputFixture": {
+      "type": "object",
+      "title": "UnitTestInputFixture",
+      "properties": {
+        "input": {
+          "type": "string"
+        },
+        "rows": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "array",
+              "items": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              }
+            }
+          ],
+          "default": ""
+        },
+        "format": {
+          "enum": [
+            "csv",
+            "dict"
+          ],
+          "default": "dict"
+        }
+      },
+      "additionalProperties": false,
+      "required": [
+        "input"
+      ]
+    },
+    "UnitTestOutputFixture": {
+      "type": "object",
+      "title": "UnitTestOutputFixture",
+      "properties": {
+        "rows": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "array",
+              "items": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              }
+            }
+          ],
+          "default": ""
+        },
+        "format": {
+          "enum": [
+            "csv",
+            "dict"
+          ],
+          "default": "dict"
+        }
+      },
+      "additionalProperties": false
+    },
+    "UnitTestConfig": {
+      "type": "object",
+      "title": "UnitTestConfig",
+      "properties": {
+        "_extra": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        },
+        "tags": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            }
+          ]
+        },
+        "meta": {
+          "type": "object",
+          "propertyNames": {
+            "type": "string"
+          }
+        }
+      },
+      "additionalProperties": true
+    },
+    "UnitTestDefinition": {
+      "type": "object",
+      "title": "UnitTestDefinition",
+      "properties": {
+        "name": {
+          "type": "string"
+        },
+        "resource_type": {
+          "enum": [
+            "model",
+            "analysis",
+            "test",
+            "snapshot",
+            "operation",
+            "seed",
+            "rpc",
+            "sql_operation",
+            "doc",
+            "source",
+            "macro",
+            "exposure",
+            "metric",
+            "group",
+            "saved_query",
+            "semantic_model",
+            "unit_test"
+          ]
+        },
+        "package_name": {
+          "type": "string"
+        },
+        "path": {
+          "type": "string"
+        },
+        "original_file_path": {
+          "type": "string"
+        },
+        "unique_id": {
+          "type": "string"
+        },
+        "fqn": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
+        "model": {
+          "type": "string"
+        },
+        "given": {
+          "type": "array",
+          "items": {
+            "$ref": "#/$defs/UnitTestInputFixture"
+          }
+        },
+        "expect": {
+          "$ref": "#/$defs/UnitTestOutputFixture"
+        },
+        "description": {
+          "type": "string",
+          "default": ""
+        },
+        "overrides": {
+          "anyOf": [
+            {
+              "$ref": "#/$defs/UnitTestOverrides"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null
+        },
+        "depends_on": {
+          "$ref": "#/$defs/DependsOn"
+        },
+        "config": {
+          "$ref": "#/$defs/UnitTestConfig"
+        }
+      },
+      "additionalProperties": false,
+      "required": [
+        "name",
+        "resource_type",
+        "package_name",
+        "path",
+        "original_file_path",
+        "unique_id",
+        "fqn",
+        "model",
+        "given",
+        "expect"
+      ]
+    },
     "WritableManifest": {
       "type": "object",
       "title": "WritableManifest",
@@ -6012,6 +6719,9 @@
               {
                 "$ref": "#/$defs/SnapshotNode"
               },
+              {
+                "$ref": "#/$defs/UnitTestNode"
+              },
               {
                 "$ref": "#/$defs/SeedNode"
               }
@@ -6121,6 +6831,9 @@
                     {
                       "$ref": "#/$defs/SnapshotNode"
                     },
+                    {
+                      "$ref": "#/$defs/UnitTestNode"
+                    },
                     {
                       "$ref": "#/$defs/SeedNode"
                     },
@@ -6138,6 +6851,9 @@
                     },
                     {
                       "$ref": "#/$defs/SemanticModel"
+                    },
+                    {
+                      "$ref": "#/$defs/UnitTestDefinition"
                     }
                   ]
                 }
@@ -6230,6 +6946,16 @@
           "propertyNames": {
             "type": "string"
           }
+        },
+        "unit_tests": {
+          "type": "object",
+          "description": "The unit tests defined in the project",
+          "additionalProperties": {
+            "$ref": "#/$defs/UnitTestDefinition"
+          },
+          "propertyNames": {
+            "type": "string"
+          }
         }
       },
       "additionalProperties": false,
@@ -6248,7 +6974,8 @@
         "child_map",
         "group_map",
         "saved_queries",
-        "semantic_models"
+        "semantic_models",
+        "unit_tests"
       ]
     }
   },

tests/functional/artifacts/expected_manifest.py

@@ -890,6 +890,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
         },
         "disabled": {},
         "semantic_models": {},
+        "unit_tests": {},
         "saved_queries": {},
     }
 
@@ -1450,6 +1451,7 @@ def expected_references_manifest(project):
             }
         },
         "semantic_models": {},
+        "unit_tests": {},
         "saved_queries": {},
     }
 
@@ -1930,5 +1932,6 @@ def expected_versions_manifest(project):
         "disabled": {},
         "macros": {},
         "semantic_models": {},
+        "unit_tests": {},
         "saved_queries": {},
     }

tests/functional/artifacts/test_artifacts.py

@@ -469,6 +469,7 @@ def verify_manifest(project, expected_manifest, start_time, manifest_schema_path
         "exposures",
         "selectors",
         "semantic_models",
+        "unit_tests",
         "saved_queries",
     }
 

tests/functional/graph_selection/test_graph_selection.py

@@ -157,7 +157,6 @@ class TestGraphSelection(SelectionFixtures):
             ]
             #           ["list", "--project-dir", str(project.project_root), "--select", "models/test/subdir*"]
         )
-        print(f"--- results: {results}")
         assert len(results) == 1
 
     def test_locally_qualified_name_model_with_dots(self, project):

tests/functional/unit_testing/fixtures.py

@@ -0,0 +1,483 @@
+my_model_sql = """
+SELECT
+a+b as c,
+concat(string_a, string_b) as string_c,
+not_testing, date_a,
+{{ dbt.string_literal(type_numeric()) }} as macro_call,
+{{ dbt.string_literal(var('my_test')) }} as var_call,
+{{ dbt.string_literal(env_var('MY_TEST', 'default')) }} as env_var_call,
+{{ dbt.string_literal(invocation_id) }} as invocation_id
+FROM {{ ref('my_model_a')}} my_model_a
+JOIN {{ ref('my_model_b' )}} my_model_b
+ON my_model_a.id = my_model_b.id
+"""
+
+my_model_a_sql = """
+SELECT
+1 as a,
+1 as id,
+2 as not_testing,
+'a' as string_a,
+DATE '2020-01-02' as date_a
+"""
+
+my_model_b_sql = """
+SELECT
+2 as b,
+1 as id,
+2 as c,
+'b' as string_b
+"""
+
+test_my_model_yml = """
+unit_tests:
+  - name: test_my_model
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows:
+          - {id: 1, a: 1}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1, b: 2}
+          - {id: 2, b: 2}
+    expect:
+      rows:
+        - {c: 2}
+
+  - name: test_my_model_empty
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows: []
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1, b: 2}
+          - {id: 2, b: 2}
+    expect:
+      rows: []
+
+  - name: test_my_model_overrides
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows:
+          - {id: 1, a: 1}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1, b: 2}
+          - {id: 2, b: 2}
+    overrides:
+      macros:
+        type_numeric: override
+        invocation_id: 123
+      vars:
+        my_test: var_override
+      env_vars:
+        MY_TEST: env_var_override
+    expect:
+      rows:
+        - {macro_call: override, var_call: var_override, env_var_call: env_var_override, invocation_id: 123}
+
+  - name: test_my_model_string_concat
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows:
+          - {id: 1, string_a: a}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1, string_b: b}
+    expect:
+      rows:
+        - {string_c: ab}
+    config:
+        tags: test_this
+"""
+
+datetime_test = """
+  - name: test_my_model_datetime
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows:
+          - {id: 1, date_a: "2020-01-01"}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1}
+    expect:
+      rows:
+        - {date_a: "2020-01-01"}
+"""
+
+event_sql = """
+select DATE '2020-01-01' as event_time, 1 as event
+union all
+select DATE '2020-01-02' as event_time, 2 as event
+union all
+select DATE '2020-01-03' as event_time, 3 as event
+"""
+
+datetime_test_invalid_format_key = """
+  - name: test_my_model_datetime
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: xxxx
+        rows:
+          - {id: 1, date_a: "2020-01-01"}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1}
+    expect:
+      rows:
+        - {date_a: "2020-01-01"}
+"""
+
+datetime_test_invalid_csv_values = """
+  - name: test_my_model_datetime
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows:
+          - {id: 1, date_a: "2020-01-01"}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1}
+    expect:
+      rows:
+        - {date_a: "2020-01-01"}
+"""
+
+datetime_test_invalid_csv_file_values = """
+  - name: test_my_model_datetime
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows:
+          - {id: 1, date_a: "2020-01-01"}
+      - input: ref('my_model_b')
+        rows:
+          - {id: 1}
+    expect:
+      rows:
+        - {date_a: "2020-01-01"}
+"""
+
+event_sql = """
+select DATE '2020-01-01' as event_time, 1 as event
+union all
+select DATE '2020-01-02' as event_time, 2 as event
+union all
+select DATE '2020-01-03' as event_time, 3 as event
+"""
+
+my_incremental_model_sql = """
+{{
+    config(
+        materialized='incremental'
+    )
+}}
+
+select * from {{ ref('events') }}
+{% if is_incremental() %}
+where event_time > (select max(event_time) from {{ this }})
+{% endif %}
+"""
+
+test_my_model_incremental_yml = """
+unit_tests:
+  - name: incremental_false
+    model: my_incremental_model
+    overrides:
+      macros:
+        is_incremental: false
+    given:
+      - input: ref('events')
+        rows:
+          - {event_time: "2020-01-01", event: 1}
+    expect:
+      rows:
+        - {event_time: "2020-01-01", event: 1}
+  - name: incremental_true
+    model: my_incremental_model
+    overrides:
+      macros:
+        is_incremental: true
+    given:
+      - input: ref('events')
+        rows:
+          - {event_time: "2020-01-01", event: 1}
+          - {event_time: "2020-01-02", event: 2}
+          - {event_time: "2020-01-03", event: 3}
+      - input: this
+        rows:
+          - {event_time: "2020-01-01", event: 1}
+    expect:
+      rows:
+        - {event_time: "2020-01-02", event: 2}
+        - {event_time: "2020-01-03", event: 3}
+"""
+
+# -- inline csv tests
+
+test_my_model_csv_yml = """
+unit_tests:
+  - name: test_my_model
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows: |
+          id,a
+          1,1
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,b
+          1,2
+          2,2
+    expect:
+      format: csv
+      rows: |
+        c
+        2
+
+  - name: test_my_model_empty
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        rows: []
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,b
+          1,2
+          2,2
+    expect:
+      rows: []
+  - name: test_my_model_overrides
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows: |
+          id,a
+          1,1
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,b
+          1,2
+          2,2
+    overrides:
+      macros:
+        type_numeric: override
+        invocation_id: 123
+      vars:
+        my_test: var_override
+      env_vars:
+        MY_TEST: env_var_override
+    expect:
+      rows:
+        - {macro_call: override, var_call: var_override, env_var_call: env_var_override, invocation_id: 123}
+  - name: test_my_model_string_concat
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows: |
+          id,string_a
+          1,a
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,string_b
+          1,b
+    expect:
+      format: csv
+      rows: |
+        string_c
+        ab
+    config:
+        tags: test_this
+"""
+
+# -- csv file tests
+test_my_model_file_csv_yml = """
+unit_tests:
+  - name: test_my_model
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_numeric_fixture_csv
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_fixture
+    expect:
+      format: csv
+      fixture: test_my_model_basic_fixture_csv
+
+  - name: test_my_model_empty
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_empty_fixture_csv
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_fixture
+    expect:
+      format: csv
+      fixture: test_my_model_a_empty_fixture_csv
+
+  - name: test_my_model_overrides
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_numeric_fixture_csv
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_fixture
+    overrides:
+      macros:
+        type_numeric: override
+        invocation_id: 123
+      vars:
+        my_test: var_override
+      env_vars:
+        MY_TEST: env_var_override
+    expect:
+      rows:
+        - {macro_call: override, var_call: var_override, env_var_call: env_var_override, invocation_id: 123}
+
+  - name: test_my_model_string_concat
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_fixture
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_b_fixture
+    expect:
+      format: csv
+      fixture: test_my_model_concat_fixture_csv
+    config:
+      tags: test_this
+"""
+
+test_my_model_fixture_csv = """
+id,b
+1,2
+2,2
+"""
+
+test_my_model_a_fixture_csv = """
+id,string_a
+1,a
+"""
+
+test_my_model_a_empty_fixture_csv = """
+"""
+
+test_my_model_a_numeric_fixture_csv = """
+id,a
+1,1
+"""
+
+test_my_model_b_fixture_csv = """
+id,string_b
+1,b
+"""
+test_my_model_basic_fixture_csv = """
+c
+2
+"""
+
+test_my_model_concat_fixture_csv = """
+string_c
+ab
+"""
+
+# -- mixed inline and file csv
+test_my_model_mixed_csv_yml = """
+unit_tests:
+  - name: test_my_model
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows: |
+          id,a
+          1,1
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,b
+          1,2
+          2,2
+    expect:
+      format: csv
+      fixture: test_my_model_basic_fixture_csv
+
+  - name: test_my_model_empty
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_empty_fixture_csv
+      - input: ref('my_model_b')
+        format: csv
+        rows: |
+          id,b
+          1,2
+          2,2
+    expect:
+      format: csv
+      fixture: test_my_model_a_empty_fixture_csv
+
+  - name: test_my_model_overrides
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        rows: |
+          id,a
+          1,1
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_fixture
+    overrides:
+      macros:
+        type_numeric: override
+        invocation_id: 123
+      vars:
+        my_test: var_override
+      env_vars:
+        MY_TEST: env_var_override
+    expect:
+      rows:
+        - {macro_call: override, var_call: var_override, env_var_call: env_var_override, invocation_id: 123}
+
+  - name: test_my_model_string_concat
+    model: my_model
+    given:
+      - input: ref('my_model_a')
+        format: csv
+        fixture: test_my_model_a_fixture
+      - input: ref('my_model_b')
+        format: csv
+        fixture: test_my_model_b_fixture
+    expect:
+      format: csv
+      rows: |
+        string_c
+        ab
+    config:
+      tags: test_this
+"""

tests/functional/unit_testing/test_csv_fixtures.py

@@ -0,0 +1,165 @@
+import pytest
+from dbt.exceptions import ParsingError, YamlParseDictError
+from dbt.tests.util import run_dbt, write_file
+from fixtures import (
+    my_model_sql,
+    my_model_a_sql,
+    my_model_b_sql,
+    test_my_model_csv_yml,
+    datetime_test,
+    datetime_test_invalid_format_key,
+    datetime_test_invalid_csv_values,
+    test_my_model_file_csv_yml,
+    test_my_model_fixture_csv,
+    test_my_model_a_fixture_csv,
+    test_my_model_b_fixture_csv,
+    test_my_model_basic_fixture_csv,
+    test_my_model_a_numeric_fixture_csv,
+    test_my_model_a_empty_fixture_csv,
+    test_my_model_concat_fixture_csv,
+    datetime_test_invalid_csv_file_values,
+    test_my_model_mixed_csv_yml,
+)
+
+
+class TestUnitTestsWithInlineCSV:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "my_model_a.sql": my_model_a_sql,
+            "my_model_b.sql": my_model_b_sql,
+            "test_my_model.yml": test_my_model_csv_yml + datetime_test,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"vars": {"my_test": "my_test_var"}}
+
+    def test_csv_inline(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 3
+
+        # Select by model name
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 5
+
+        # Check error with invalid format key
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_format_key,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(YamlParseDictError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+
+        # Check error with csv format defined but dict on rows
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_csv_values,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(ParsingError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+
+
+class TestUnitTestsWithFileCSV:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "my_model_a.sql": my_model_a_sql,
+            "my_model_b.sql": my_model_b_sql,
+            "test_my_model.yml": test_my_model_file_csv_yml + datetime_test,
+            "test_my_model_fixture.csv": test_my_model_fixture_csv,
+            "test_my_model_a_fixture.csv": test_my_model_a_fixture_csv,
+            "test_my_model_b_fixture.csv": test_my_model_b_fixture_csv,
+            "test_my_model_basic_fixture.csv": test_my_model_basic_fixture_csv,
+            "test_my_model_a_numeric_fixture.csv": test_my_model_a_numeric_fixture_csv,
+            "test_my_model_a_empty_fixture.csv": test_my_model_a_empty_fixture_csv,
+            "test_my_model_concat_fixture.csv": test_my_model_concat_fixture_csv,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"vars": {"my_test": "my_test_var"}}
+
+    def test_csv_file(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 3
+
+        # Select by model name
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 5
+
+        # Check error with invalid format key
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_format_key,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(YamlParseDictError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+
+        # Check error with csv format defined but dict on rows
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_csv_file_values,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(ParsingError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+
+
+class TestUnitTestsWithMixedCSV:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "my_model_a.sql": my_model_a_sql,
+            "my_model_b.sql": my_model_b_sql,
+            "test_my_model.yml": test_my_model_mixed_csv_yml + datetime_test,
+            "test_my_model_fixture.csv": test_my_model_fixture_csv,
+            "test_my_model_a_fixture.csv": test_my_model_a_fixture_csv,
+            "test_my_model_b_fixture.csv": test_my_model_b_fixture_csv,
+            "test_my_model_basic_fixture.csv": test_my_model_basic_fixture_csv,
+            "test_my_model_a_numeric_fixture.csv": test_my_model_a_numeric_fixture_csv,
+            "test_my_model_a_empty_fixture.csv": test_my_model_a_empty_fixture_csv,
+            "test_my_model_concat_fixture.csv": test_my_model_concat_fixture_csv,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"vars": {"my_test": "my_test_var"}}
+
+    def test_mixed(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 3
+
+        # Select by model name
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 5
+
+        # Check error with invalid format key
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_format_key,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(YamlParseDictError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+
+        # Check error with csv format defined but dict on rows
+        write_file(
+            test_my_model_csv_yml + datetime_test_invalid_csv_file_values,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(ParsingError):
+            results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)

tests/functional/unit_testing/test_unit_testing.py

@@ -0,0 +1,102 @@
+import pytest
+from dbt.tests.util import run_dbt, write_file, get_manifest, get_artifact
+from dbt.exceptions import DuplicateResourceNameError
+from fixtures import (
+    my_model_sql,
+    my_model_a_sql,
+    my_model_b_sql,
+    test_my_model_yml,
+    datetime_test,
+    my_incremental_model_sql,
+    event_sql,
+    test_my_model_incremental_yml,
+)
+
+
+class TestUnitTests:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "my_model_a.sql": my_model_a_sql,
+            "my_model_b.sql": my_model_b_sql,
+            "test_my_model.yml": test_my_model_yml + datetime_test,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"vars": {"my_test": "my_test_var"}}
+
+    def test_basic(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 3
+
+        # Select by model name
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 5
+
+        # Test select by test name
+        results = run_dbt(["unit-test", "--select", "test_name:test_my_model_string_concat"])
+        assert len(results) == 1
+
+        # Select, method not specified
+        results = run_dbt(["unit-test", "--select", "test_my_model_overrides"])
+        assert len(results) == 1
+
+        # Select using tag
+        results = run_dbt(["unit-test", "--select", "tag:test_this"])
+        assert len(results) == 1
+
+        # Partial parsing... remove test
+        write_file(test_my_model_yml, project.project_root, "models", "test_my_model.yml")
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 4
+
+        # Partial parsing... put back removed test
+        write_file(
+            test_my_model_yml + datetime_test, project.project_root, "models", "test_my_model.yml"
+        )
+        results = run_dbt(["unit-test", "--select", "my_model"], expect_pass=False)
+        assert len(results) == 5
+
+        manifest = get_manifest(project.project_root)
+        assert len(manifest.unit_tests) == 5
+        # Every unit test has a depends_on to the model it tests
+        for unit_test_definition in manifest.unit_tests.values():
+            assert unit_test_definition.depends_on.nodes[0] == "model.test.my_model"
+
+        # We should have a UnitTestNode for every test, plus two input models for each test
+        unit_test_manifest = get_artifact(
+            project.project_root, "target", "unit_test_manifest.json"
+        )
+        assert len(unit_test_manifest["nodes"]) == 15
+
+        # Check for duplicate unit test name
+        # this doesn't currently pass with partial parsing because of the root problem
+        # described in https://github.com/dbt-labs/dbt-core/issues/8982
+        write_file(
+            test_my_model_yml + datetime_test + datetime_test,
+            project.project_root,
+            "models",
+            "test_my_model.yml",
+        )
+        with pytest.raises(DuplicateResourceNameError):
+            run_dbt(["run", "--no-partial-parse", "--select", "my_model"])
+
+
+class TestUnitTestIncrementalModel:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_incremental_model.sql": my_incremental_model_sql,
+            "events.sql": event_sql,
+            "test_my_incremental_model.yml": test_my_model_incremental_yml,
+        }
+
+    def test_basic(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+        # Select by model name
+        results = run_dbt(["unit-test", "--select", "my_incremental_model"], expect_pass=True)
+        assert len(results) == 2

tests/unit/test_manifest.py

@@ -398,6 +398,7 @@ class ManifestTest(unittest.TestCase):
                 "docs": {},
                 "disabled": {},
                 "semantic_models": {},
+                "unit_tests": {},
                 "saved_queries": {},
             },
         )
@@ -582,6 +583,7 @@ class ManifestTest(unittest.TestCase):
                 },
                 "disabled": {},
                 "semantic_models": {},
+                "unit_tests": {},
                 "saved_queries": {},
             },
         )
@@ -921,6 +923,7 @@ class MixedManifestTest(unittest.TestCase):
                 "docs": {},
                 "disabled": {},
                 "semantic_models": {},
+                "unit_tests": {},
                 "saved_queries": {},
             },
         )

tests/unit/test_node_types.py

@@ -17,6 +17,7 @@ node_type_pluralizations = {
     NodeType.Metric: "metrics",
     NodeType.Group: "groups",
     NodeType.SemanticModel: "semantic_models",
+    NodeType.Unit: "unit_tests",
     NodeType.SavedQuery: "saved_queries",
 }
 

tests/unit/test_parser.py

@@ -176,13 +176,14 @@ class BaseParserTest(unittest.TestCase):
         return FileBlock(file=source_file)
 
     def assert_has_manifest_lengths(
-        self, manifest, macros=3, nodes=0, sources=0, docs=0, disabled=0
+        self, manifest, macros=3, nodes=0, sources=0, docs=0, disabled=0, unit_tests=0
     ):
         self.assertEqual(len(manifest.macros), macros)
         self.assertEqual(len(manifest.nodes), nodes)
         self.assertEqual(len(manifest.sources), sources)
         self.assertEqual(len(manifest.docs), docs)
         self.assertEqual(len(manifest.disabled), disabled)
+        self.assertEqual(len(manifest.unit_tests), unit_tests)
 
 
 def assertEqualNodes(node_one, node_two):
@@ -371,8 +372,8 @@ class SchemaParserTest(BaseParserTest):
             manifest=self.manifest,
         )
 
-    def file_block_for(self, data, filename):
-        return super().file_block_for(data, filename, "models")
+    def file_block_for(self, data, filename, searched="models"):
+        return super().file_block_for(data, filename, searched)
 
     def yaml_block_for(self, test_yml: str, filename: str):
         file_block = self.file_block_for(data=test_yml, filename=filename)

tests/unit/test_unit_test_parser.py

@@ -0,0 +1,180 @@
+from dbt.contracts.graph.nodes import UnitTestDefinition, UnitTestConfig, DependsOn, NodeType
+from dbt.exceptions import ParsingError
+from dbt.parser import SchemaParser
+from dbt.parser.unit_tests import UnitTestParser
+
+from .utils import MockNode
+from .test_parser import SchemaParserTest, assertEqualNodes
+
+from unittest import mock
+from dbt.contracts.graph.unparsed import UnitTestOutputFixture
+
+
+UNIT_TEST_MODEL_NOT_FOUND_SOURCE = """
+unit_tests:
+    - name: test_my_model_doesnt_exist
+      model: my_model_doesnt_exist
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
+
+UNIT_TEST_SOURCE = """
+unit_tests:
+    - name: test_my_model
+      model: my_model
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
+
+UNIT_TEST_VERSIONED_MODEL_SOURCE = """
+unit_tests:
+    - name: test_my_model_versioned
+      model: my_model_versioned.v1
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
+
+UNIT_TEST_CONFIG_SOURCE = """
+unit_tests:
+    - name: test_my_model
+      model: my_model
+      config:
+        tags: "schema_tag"
+        meta:
+          meta_key: meta_value
+          meta_jinja_key: '{{ 1 + 1 }}'
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
+
+UNIT_TEST_MULTIPLE_SOURCE = """
+unit_tests:
+    - name: test_my_model
+      model: my_model
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+    - name: test_my_model2
+      model: my_model
+      description: "unit test description"
+      given: []
+      expect:
+        rows:
+          - {a: 1}
+"""
+
+
+class UnitTestParserTest(SchemaParserTest):
+    def setUp(self):
+        super().setUp()
+        my_model_node = MockNode(
+            package="snowplow",
+            name="my_model",
+            config=mock.MagicMock(enabled=True),
+            refs=[],
+            sources=[],
+            patch_path=None,
+        )
+        self.manifest.nodes = {my_model_node.unique_id: my_model_node}
+        self.parser = SchemaParser(
+            project=self.snowplow_project_config,
+            manifest=self.manifest,
+            root_project=self.root_project_config,
+        )
+
+    def file_block_for(self, data, filename):
+        return super().file_block_for(data, filename, "unit_tests")
+
+    def test_basic_model_not_found(self):
+        block = self.yaml_block_for(UNIT_TEST_MODEL_NOT_FOUND_SOURCE, "test_my_model.yml")
+
+        with self.assertRaises(ParsingError):
+            UnitTestParser(self.parser, block).parse()
+
+    def test_basic(self):
+        block = self.yaml_block_for(UNIT_TEST_SOURCE, "test_my_model.yml")
+
+        UnitTestParser(self.parser, block).parse()
+
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=1, unit_tests=1)
+        unit_test = list(self.parser.manifest.unit_tests.values())[0]
+        expected = UnitTestDefinition(
+            name="test_my_model",
+            model="my_model",
+            resource_type=NodeType.Unit,
+            package_name="snowplow",
+            path=block.path.relative_path,
+            original_file_path=block.path.original_file_path,
+            unique_id="unit_test.snowplow.my_model.test_my_model",
+            given=[],
+            expect=UnitTestOutputFixture(rows=[{"a": 1}]),
+            description="unit test description",
+            overrides=None,
+            depends_on=DependsOn(nodes=["model.snowplow.my_model"]),
+            fqn=["snowplow", "my_model", "test_my_model"],
+            config=UnitTestConfig(),
+        )
+        assertEqualNodes(unit_test, expected)
+
+    def test_unit_test_config(self):
+        block = self.yaml_block_for(UNIT_TEST_CONFIG_SOURCE, "test_my_model.yml")
+        self.root_project_config.unit_tests = {
+            "snowplow": {"my_model": {"+tags": ["project_tag"]}}
+        }
+
+        UnitTestParser(self.parser, block).parse()
+
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=1, unit_tests=1)
+        unit_test = self.parser.manifest.unit_tests["unit_test.snowplow.my_model.test_my_model"]
+        self.assertEqual(sorted(unit_test.config.tags), sorted(["schema_tag", "project_tag"]))
+        self.assertEqual(unit_test.config.meta, {"meta_key": "meta_value", "meta_jinja_key": "2"})
+
+    def test_unit_test_versioned_model(self):
+        block = self.yaml_block_for(UNIT_TEST_VERSIONED_MODEL_SOURCE, "test_my_model.yml")
+        my_model_versioned_node = MockNode(
+            package="snowplow",
+            name="my_model_versioned",
+            config=mock.MagicMock(enabled=True),
+            refs=[],
+            sources=[],
+            patch_path=None,
+            version=1,
+        )
+        self.manifest.nodes[my_model_versioned_node.unique_id] = my_model_versioned_node
+
+        UnitTestParser(self.parser, block).parse()
+
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=2, unit_tests=1)
+        unit_test = self.parser.manifest.unit_tests[
+            "unit_test.snowplow.my_model_versioned.v1.test_my_model_versioned"
+        ]
+        self.assertEqual(len(unit_test.depends_on.nodes), 1)
+        self.assertEqual(unit_test.depends_on.nodes[0], "model.snowplow.my_model_versioned.v1")
+
+    def test_multiple_unit_tests(self):
+        block = self.yaml_block_for(UNIT_TEST_MULTIPLE_SOURCE, "test_my_model.yml")
+
+        UnitTestParser(self.parser, block).parse()
+
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=1, unit_tests=2)
+        for unit_test in self.parser.manifest.unit_tests.values():
+            self.assertEqual(len(unit_test.depends_on.nodes), 1)
+            self.assertEqual(unit_test.depends_on.nodes[0], "model.snowplow.my_model")

tests/unit/utils.py

@@ -336,7 +336,7 @@ def MockNode(package, name, resource_type=None, **kwargs):
 
     version = kwargs.get("version")
     search_name = name if version is None else f"{name}.v{version}"
-    unique_id = f"{str(resource_type)}.{package}.{name}"
+    unique_id = f"{str(resource_type)}.{package}.{search_name}"
     node = mock.MagicMock(
         __class__=cls,
         resource_type=resource_type,