Parse model hooks for refs

Add a parse pass for model hooks, parse them like models
Attach the hook refs as if they were on the model
Add a test exercising refs

core/dbt/parser/base.py

@@ -1,4 +1,5 @@
 import abc
+import itertools
 import os
 from typing import (
     List, Dict, Any, Callable, Iterable, Optional, Generic, TypeVar
@@ -288,6 +289,13 @@ def _create_parsetime_node(
             )
             raise CompilationException(msg, node=node)
 
+    def _context_for(
+        self, parsed_node: IntermediateNode, config: SourceConfig
+    ) -> Dict[str, Any]:
+        return dbt.context.parser.generate(
+            parsed_node, self.root_project, self.macro_manifest, config
+        )
+
     def render_with_context(
         self, parsed_node: IntermediateNode, config: SourceConfig
     ) -> None:
@@ -296,9 +304,7 @@ def render_with_context(
 
         Note: this mutates the config object when config() calls are rendered.
         """
-        context = dbt.context.parser.generate(
-            parsed_node, self.root_project, self.macro_manifest, config
-        )
+        context = self._context_for(parsed_node, config)
 
         get_rendered(parsed_node.raw_sql, context, parsed_node,
                      capture_macros=True)
@@ -363,6 +369,19 @@ def update_parsed_node(
         self.update_parsed_node_schema(parsed_node, config_dict)
         self.update_parsed_node_alias(parsed_node, config_dict)
 
+        # at this point, we've collected our hooks. Use the node context to
+        # render each hook and collect refs/sources
+        hooks = list(itertools.chain(parsed_node.config.pre_hook,
+                                     parsed_node.config.post_hook))
+        # skip context rebuilding if there aren't any hooks
+        if not hooks:
+            return
+        # we could cache the original context from parsing this node. Is that
+        # worth the cost in memory/complexity?
+        context = self._context_for(parsed_node, config)
+        for hook in hooks:
+            get_rendered(hook.sql, context, parsed_node, capture_macros=True)
+
     def initial_config(self, fqn: List[str]) -> SourceConfig:
         return SourceConfig(self.root_project, self.project, fqn,
                             self.resource_type)

test/integration/014_hook_tests/ref-hook-models/hooked.sql

@@ -0,0 +1,21 @@
+{{
+    config({
+        "pre_hook": "\
+            insert into {{this.schema}}.on_model_hook select
+                state,
+                '{{ target.dbname }}' as \"target.dbname\",\
+                '{{ target.host }}' as \"target.host\",\
+                '{{ target.name }}' as \"target.name\",\
+                '{{ target.schema }}' as \"target.schema\",\
+                '{{ target.type }}' as \"target.type\",\
+                '{{ target.user }}' as \"target.user\",\
+                '{{ target.get(\"pass\", \"\") }}' as \"target.pass\",\
+                {{ target.port }} as \"target.port\",\
+                {{ target.threads }} as \"target.threads\",\
+                '{{ run_started_at }}' as \"run_started_at\",\
+                '{{ invocation_id }}' as \"invocation_id\"\
+                from {{ ref('pre') }}\
+        "
+    })
+}}
+select 1 as id

test/integration/014_hook_tests/ref-hook-models/post.sql

@@ -0,0 +1 @@
+select 'end' as state

test/integration/014_hook_tests/ref-hook-models/pre.sql

@@ -0,0 +1 @@
+select 'start' as state

test/integration/014_hook_tests/test_model_hooks.py

@@ -62,6 +62,7 @@
    )
 """
 
+
 class BaseTestPrePost(DBTIntegrationTest):
     def setUp(self):
         DBTIntegrationTest.setUp(self)
@@ -153,6 +154,45 @@ def test_postgres_pre_and_post_model_hooks(self):
         self.check_hooks('end')
 
 
+class TestHookRefs(BaseTestPrePost):
+    @property
+    def project_config(self):
+        return {
+            'models': {
+                'test': {
+                    'hooked': {
+                        'post-hook': ['''
+                        insert into {{this.schema}}.on_model_hook select
+                        state,
+                        '{{ target.dbname }}' as "target.dbname",
+                        '{{ target.host }}' as "target.host",
+                        '{{ target.name }}' as "target.name",
+                        '{{ target.schema }}' as "target.schema",
+                        '{{ target.type }}' as "target.type",
+                        '{{ target.user }}' as "target.user",
+                        '{{ target.get("pass", "") }}' as "target.pass",
+                        {{ target.port }} as "target.port",
+                        {{ target.threads }} as "target.threads",
+                        '{{ run_started_at }}' as "run_started_at",
+                        '{{ invocation_id }}' as "invocation_id"
+                    from {{ ref('post') }}'''.strip()],
+                    }
+                },
+            }
+        }
+
+    @property
+    def models(self):
+        return 'ref-hook-models'
+
+    @use_profile('postgres')
+    def test_postgres_pre_post_model_hooks_refed(self):
+        self.run_dbt(['run'])
+
+        self.check_hooks('start', count=1)
+        self.check_hooks('end', count=1)
+
+
 class TestPrePostModelHooksOnSeeds(DBTIntegrationTest):
     @property
     def schema(self):