Merge pull request #1346 from fishtown-analytics/fix/on-run-hooks-in-…

…tests

[Stephen Girard] fix for on-run- hooks running in tests

core/dbt/task/test.py

@@ -7,17 +7,19 @@
 class TestTask(RunTask):
  """
  Testing:
- 1) Create tmp views w/ 0 rows to ensure all tables, schemas, and SQL
- statements are valid
- 2) Read schema files and validate that constraints are satisfied
- a) not null
- b) uniquenss
- c) referential integrity
- d) accepted value
+ Read schema files + custom data tests and validate that
+ constraints are satisfied.
  """
  def raise_on_first_error(self):
  return False
 
+ def before_run(self, adapter, selected_uids):
+ # Don't execute on-run-* hooks for tests
+ self.populate_adapter_cache(adapter)
+
+ def after_run(self, adapter, results):
+ pass
+
  def build_query(self):
  query = {
  "include": self.args.models,

test/integration/008_schema_tests_test/ephemeral/ephemeral.sql

@@ -0,0 +1,4 @@
+
+{{ config(materialized='ephemeral') }}
+
+select 1 as id

test/integration/008_schema_tests_test/ephemeral/schema.yml

@@ -0,0 +1,8 @@
+
+version: 2
+models:
+ - name: ephemeral
+ columns:
+ - name: id
+ tests:
+ - unique

test/integration/008_schema_tests_test/test_schema_v2_tests.py

@@ -94,6 +94,38 @@ def test_malformed_schema_strict_will_break_run(self):
  self.run_dbt(strict=True)
 
 
+class TestHooksInTests(DBTIntegrationTest):
+
+ @property
+ def schema(self):
+ return "schema_tests_008"
+
+ @property
+ def models(self):
+ # test ephemeral models so we don't need to do a run (which would fail)
+ return "test/integration/008_schema_tests_test/ephemeral"
+
+ @property
+ def project_config(self):
+ return {
+ "on-run-start": ["{{ exceptions.raise_compiler_error('hooks called in tests -- error') if execute }}"],
+ "on-run-end": ["{{ exceptions.raise_compiler_error('hooks called in tests -- error') if execute }}"],
+ }
+
+ @attr(type='postgres')
+ def test_hooks_dont_run_for_tests(self):
+ # This would fail if the hooks ran
+ results = self.run_dbt(['test', '--model', 'ephemeral'])
+ self.assertEqual(len(results), 1)
+ for result in results:
+ self.assertIsNone(result.error)
+ self.assertFalse(result.skipped)
+ # status = # of failing rows
+ self.assertEqual(
+ result.status, 0,
+ 'test {} failed'.format(result.node.get('name'))
+ )
+
 class TestCustomSchemaTests(DBTIntegrationTest):
 
  def setUp(self):