Add defer to rpc test method

core/dbt/contracts/rpc.py

@@ -79,6 +79,7 @@ class RPCTestParameters(RPCCompileParameters):
     data: bool = False
     schema: bool = False
     state: Optional[str] = None
+    defer: Optional[bool] = None
 
 
 @dataclass

core/dbt/task/rpc/project_commands.py

@@ -129,6 +129,10 @@ def set_args(self, params: RPCTestParameters) -> None:
         self.args.schema = params.schema
         if params.threads is not None:
             self.args.threads = params.threads
+        if params.defer is None:
+            self.args.defer = flags.DEFER_MODE
+        else:
+            self.args.defer = params.defer
 
         self.args.state = state_path(params.state)
         self.set_previous_state()

test/rpc/test_test.py

@@ -105,3 +105,9 @@ def test_rpc_test_state(
             querier.test(state='./state', models=['state:modified']),
         )
         assert len(results['results']) == 0
+
+        # a better test of defer would require multiple targets
+        results = querier.async_wait_for_result(
+            querier.run(state='./state', models=['state:modified'], defer=True)
+        )
+        assert len(results['results']) == 0

test/rpc/util.py

@@ -335,6 +335,7 @@ def test(
         data: bool = None,
         schema: bool = None,
         request_id: int = 1,
+        defer: Optional[bool] = None,
         state: Optional[bool] = None,
     ):
         params = {}
@@ -348,6 +349,8 @@ def test(
             params['schema'] = schema
         if threads is not None:
             params['threads'] = threads
+        if defer is not None:
+            params['defer'] = defer
         if state is not None:
             params['state'] = state
         return self.request(