Fail on parse errors (#1493)

core/dbt/loader.py

@@ -1,6 +1,7 @@
 import itertools
 import os
 import pickle
+from contextlib import contextmanager
 from datetime import datetime
 from typing import Dict, Optional, Mapping
 
@@ -81,6 +82,19 @@ def make_parse_result(
     )
 
 
+@contextmanager
+def log_exceptions():
+    try:
+        yield
+    except dbt.exceptions.Exception as exc:
+        logger.error('Parsing failed: {}'.format(exc))
+        raise
+    except Exception as exc:
+        logger.error('Parsing failed with an uncaught exception: {}'
+                     .format(exc))
+        raise
+
+
 class GraphLoader:
     def __init__(
         self, root_project: RuntimeConfig, all_projects: Mapping[str, Project]
@@ -284,19 +298,21 @@ def load_all(
         root_config: RuntimeConfig,
         internal_manifest: Optional[Manifest] = None
     ) -> Manifest:
-        projects = load_all_projects(root_config)
-        loader = cls(root_config, projects)
-        loader.load(internal_manifest=internal_manifest)
-        loader.write_parse_results()
-        manifest = loader.create_manifest()
-        _check_manifest(manifest, root_config)
-        return manifest
+        with log_exceptions():
+            projects = load_all_projects(root_config)
+            loader = cls(root_config, projects)
+            loader.load(internal_manifest=internal_manifest)
+            loader.write_parse_results()
+            manifest = loader.create_manifest()
+            _check_manifest(manifest, root_config)
+            return manifest
 
     @classmethod
     def load_internal(cls, root_config: RuntimeConfig) -> Manifest:
-        projects = load_internal_projects(root_config)
-        loader = cls(root_config, projects)
-        return loader.load_only_macros()
+        with log_exceptions():
+            projects = load_internal_projects(root_config)
+            loader = cls(root_config, projects)
+            return loader.load_only_macros()
 
 
 def _check_resource_uniqueness(manifest):

core/dbt/parser/schemas.py

@@ -18,10 +18,9 @@
 )
 from dbt.context.parser import docs
 from dbt.exceptions import (
-    warn_or_error, validator_error_message, JSONValidationException,
+    validator_error_message, JSONValidationException,
     raise_invalid_schema_yml_version, ValidationException, CompilationException
 )
-from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.node_types import NodeType, SourceType
 from dbt.parser.base import SimpleParser
 from dbt.parser.search import FileBlock, FilesystemSearcher
@@ -37,19 +36,24 @@
 TestDef = Union[str, Dict[str, Any]]
 
 
-def warn_invalid(filepath, key, value, explain):
-    msg = (
-        "Invalid test config given in {} @ {}: {} {}"
-    ).format(filepath, key, value, explain)
-    warn_or_error(msg, value, log_fmt='Compilation warning: {}\n')
-
-
-def warn_validation_error(filepath, key, value, exc):
-    if isinstance(exc, ValidationError):
-        msg = validator_error_message(exc)
+def error_context(
+    path: str,
+    key: str,
+    data: Any,
+    cause: Union[str, ValidationException, JSONValidationException]
+) -> str:
+    """Provide contextual information about an error while parsing
+    """
+    if isinstance(cause, str):
+        reason = cause
+    elif isinstance(cause, ValidationError):
+        reason = validator_error_message(cause)
     else:
-        msg = exc.msg
-    warn_invalid(filepath, key, value, '- ' + msg)
+        reason = cause.msg
+    return (
+        'Invalid {key} config given in {path} @ {key}: {data} - {reason}'
+        .format(key=key, path=path, data=data, reason=reason)
+    )
 
 
 class ParserRef:
@@ -154,7 +158,10 @@ def _get_dicts_for(
             if str_keys:
                 yield entry
             else:
-                warn_invalid(path, key, entry, '(expected a Dict[str])')
+                msg = error_context(
+                    path, key, data, 'expected a dict with string keys'
+                )
+                raise CompilationException(msg)
 
     def read_yaml_models(
         self, yaml: YamlBlock
@@ -165,10 +172,9 @@ def read_yaml_models(
         for data in self._get_dicts_for(yaml, yaml_key):
             try:
                 model = UnparsedNodeUpdate.from_dict(data)
-                # we don't want to fail the full run, but we do want to fail
-                # parsing this block
             except (ValidationError, JSONValidationException) as exc:
-                warn_validation_error(path, yaml_key, data, exc)
+                msg = error_context(path, yaml_key, data, exc)
+                raise CompilationException(msg) from exc
             else:
                 yield model
 
@@ -183,23 +189,26 @@ def read_yaml_sources(
                 data = self._renderer.render_schema_source(data)
                 source = UnparsedSourceDefinition.from_dict(data)
             except (ValidationError, JSONValidationException) as exc:
-                warn_validation_error(path, yaml_key, data, exc)
+                msg = error_context(path, yaml_key, data, exc)
+                raise CompilationException(msg) from exc
             else:
                 for table in source.tables:
                     yield SourceTarget(source, table)
 
     def _yaml_from_file(
         self, source_file: SourceFile
     ) -> Optional[Dict[str, Any]]:
-        """If loading the yaml fails, the file will be skipped with an INFO
-        message. TODO(jeb): should this be a warning?
+        """If loading the yaml fails, raise an exception.
         """
         path: str = source_file.path.relative_path
         try:
             return load_yaml_text(source_file.contents)
         except ValidationException as e:
-            logger.info("Error reading {}:{} - Skipping\n{}".format(
-                        self.project.project_name, path, e))
+            reason = validator_error_message(e)
+            raise CompilationException(
+                'Error reading {}: {} - {}'
+                .format(self.project.project_name, path, reason)
+            )
         return None
 
     def parse_column(
@@ -277,11 +286,11 @@ def parse_test(
         except CompilationException as exc:
             context = _trimmed(str(block.target))
             msg = (
-                'Compilation warning: Invalid test config given in {}:'
+                'Invalid test config given in {}:'
                 '\n\t{}\n\t@: {}'
                 .format(block.path.original_file_path, exc.msg, context)
             )
-            warn_or_error(msg, None)
+            raise CompilationException(msg)
 
     def _calculate_freshness(
         self,

test/integration/008_schema_tests_test/test_schema_v2_tests.py

@@ -76,21 +76,13 @@ def run_schema_validations(self):
         test_task = TestTask(args, self.config)
         return test_task.run()
 
-    @use_profile('postgres')
-    def test_malformed_schema_test_wont_brick_run(self):
-        # dbt run should work (Despite broken schema test)
-        results = self.run_dbt(strict=False)
-        self.assertEqual(len(results), 2)
-
-        # in v2, we skip the entire model
-        ran_tests = self.run_schema_validations()
-        self.assertEqual(len(ran_tests), 5)
-        self.assertEqual(sum(x.status for x in ran_tests), 0)
-
     @use_profile('postgres')
     def test_malformed_schema_strict_will_break_run(self):
         with self.assertRaises(CompilationException):
             self.run_dbt(strict=True)
+        # even if strict = False!
+        with self.assertRaises(CompilationException):
+            self.run_dbt(strict=False)
 
 
 class TestHooksInTests(DBTIntegrationTest):

test/integration/042_sources_test/test_sources.py

@@ -36,7 +36,6 @@ def project_config(self):
     def setUp(self):
         super().setUp()
         os.environ['DBT_TEST_SCHEMA_NAME_VARIABLE'] = 'test_run_schema'
-        self.run_dbt_with_vars(['seed'], strict=False)
 
     def tearDown(self):
         del os.environ['DBT_TEST_SCHEMA_NAME_VARIABLE']
@@ -48,7 +47,13 @@ def run_dbt_with_vars(self, cmd, *args, **kwargs):
         return self.run_dbt(cmd, *args, **kwargs)
 
 
-class TestSources(BaseSourcesTest):
+class SuccessfulSourcesTest(BaseSourcesTest):
+    def setUp(self):
+        super().setUp()
+        self.run_dbt_with_vars(['seed'], strict=False)
+
+
+class TestSources(SuccessfulSourcesTest):
     @property
     def project_config(self):
         cfg = super().project_config
@@ -165,7 +170,7 @@ def test_postgres_run_operation_source(self):
         ])
 
 
-class TestSourceFreshness(BaseSourcesTest):
+class TestSourceFreshness(SuccessfulSourcesTest):
     def setUp(self):
         super().setUp()
         self.maxDiff = None
@@ -278,7 +283,7 @@ def test_bigquery_source_freshness(self):
         self._run_source_freshness()
 
 
-class TestSourceFreshnessErrors(BaseSourcesTest):
+class TestSourceFreshnessErrors(SuccessfulSourcesTest):
     @property
     def models(self):
         return "error_models"
@@ -296,15 +301,17 @@ def test_postgres_error(self):
 
 
 class TestMalformedSources(BaseSourcesTest):
+    # even seeds should fail, because parsing is what's raising
     @property
     def models(self):
         return "malformed_models"
 
     @use_profile('postgres')
-    def test_postgres_malformed_schema_nonstrict_will_not_break_run(self):
-        self.run_dbt_with_vars(['run'], strict=False)
+    def test_postgres_malformed_schema_nonstrict_will_break_run(self):
+        with self.assertRaises(CompilationException):
+            self.run_dbt_with_vars(['seed'], strict=False)
 
     @use_profile('postgres')
     def test_postgres_malformed_schema_strict_will_break_run(self):
         with self.assertRaises(CompilationException):
-            self.run_dbt_with_vars(['run'], strict=True)
+            self.run_dbt_with_vars(['seed'], strict=True)

test/integration/048_rpc_test/test_rpc.py

@@ -958,7 +958,7 @@ def test_postgres_status_error(self):
             'RPC server failed to compile project, call the "status" method for compile status',
             None)
         self.assertIn('message', data)
-        self.assertIn('Compilation warning: Invalid test config', str(data['message']))
+        self.assertIn('Invalid test config', str(data['message']))
 
     def tearDown(self):
         # prevent an OperationalError where the server closes on us in the