first pass adding --resource-type cli arg

core/dbt/main.py

@@ -399,6 +399,14 @@ def _build_build_subparser(subparsers, base_subparser):
         Stop execution upon a first failure.
         '''
     )
+    resource_values: List[str] = [
+        str(s) for s in ListTask.ALL_RESOURCE_VALUES
+    ] + ['default', 'all']
+    sub.add_argument('--resource-type',
+                     choices=resource_values,
+                     action='append',
+                     default=[],
+                     dest='resource_types')
     return sub
 
 

core/dbt/task/build.py

@@ -5,25 +5,34 @@
 
 from dbt.contracts.results import NodeStatus
 from dbt.graph import ResourceTypeSelector
-from dbt.exceptions import InternalException
+from dbt.exceptions import RuntimeException, InternalException
 from dbt.node_types import NodeType
+from dbt.task.test import TestSelector
 
 
 class BuildTask(RunTask):
-    """The Build task processes all assets of a given process and
-    attempts to 'build' them in an opinionated fashion. Every resource
-    type outlined in RUNNER_MAP will be processed by the mapped runner class.
+    """The Build task processes all assets of a given process and attempts to
+    'build' them in an opinionated fashion.  Every resource type outlined in
+    RUNNER_MAP will be processed by the mapped runner class.
 
     I.E. a resource of type Model is handled by the ModelRunner which is
-    imported as run_model_runner.
-    """
+    imported as run_model_runner. """
 
     RUNNER_MAP = {
         NodeType.Model: run_model_runner,
         NodeType.Snapshot: snapshot_model_runner,
         NodeType.Seed: seed_runner,
         NodeType.Test: test_runner,
     }
+    ALL_RESOURCE_VALUES = frozenset({x for x in RUNNER_MAP.keys()})
+
+    @property
+    def resource_types(self):
+        if not self.args.resource_types:
+            return list(self.ALL_RESOURCE_VALUES)
+
+        values = set(self.args.resource_types)
+        return list(values)
 
     MARK_DEPENDENT_ERRORS_STATUSES = [NodeStatus.Error, NodeStatus.Fail]
 
@@ -32,12 +41,19 @@ def get_node_selector(self) -> ResourceTypeSelector:
             raise InternalException(
                 'manifest and graph must be set to get node selection'
             )
+        resource_types = self.resource_types
 
+        if resource_types == [NodeType.Test]:
+            return TestSelector(
+                graph=self.graph,
+                manifest=self.manifest,
+                previous_state=self.previous_state,
+            )
         return ResourceTypeSelector(
             graph=self.graph,
             manifest=self.manifest,
             previous_state=self.previous_state,
-            resource_types=[x for x in self.RUNNER_MAP.keys()],
+            resource_types=resource_types,
         )
 
     def get_runner_type(self, node):

core/dbt/task/list.py

@@ -1,5 +1,4 @@
 import json
-from typing import Type
 
 from dbt.contracts.graph.parsed import (
     ParsedExposure,
@@ -183,7 +182,6 @@ def get_node_selector(self):
             raise InternalException(
                 'manifest and graph must be set to get perform node selection'
             )
-        cls: Type[ResourceTypeSelector]
         if self.resource_types == [NodeType.Test]:
             return TestSelector(
                 graph=self.graph,