Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_postgres_malformed_schema_nonstrict_will_break_run(self):
with self.assertRaises(CompilationException):
self.run_dbt_with_vars(['seed'], strict=False)
def test_seed(self):
try:
_, success = self.run_dbt_and_check(['seed'])
self.assertTrue(False)
except dbt.exceptions.CompilationException as e:
pass
def test_postgres_dbt_run_skips_seeds(self):
# run does not try to parse the seed files
self.assertEqual(len(self.run_dbt(['run'])), 1)
# make sure 'dbt seed' fails, otherwise our test is invalid!
with self.assertRaises(CompilationException):
self.run_dbt(['seed'])
def _get_dicts_for(
self, yaml: YamlBlock, key: str
) -> Iterable[Dict[str, Any]]:
data = yaml.data.get(key, [])
if not isinstance(data, list):
raise CompilationException(
'{} must be a list, got {} instead: ({})'
.format(key, type(data), _trimmed(str(data)))
)
path = yaml.path.original_file_path
for entry in data:
str_keys = (
isinstance(entry, dict) and
all(isinstance(k, str) for k in entry)
)
if str_keys:
yield entry
else:
msg = error_context(
path, key, data, 'expected a dict with string keys'
)
def render_update(
self, node: IntermediateNode, config: SourceConfig
) -> None:
try:
self.render_with_context(node, config)
self.update_parsed_node(node, config)
except ValidationError as exc:
# we got a ValidationError - probably bad types in config()
msg = validator_error_message(exc)
raise CompilationException(msg, node=node) from exc
def extract_blocks(self, source_file: FileBlock) -> Iterable[BlockTag]:
try:
blocks = extract_toplevel_blocks(
source_file.contents,
allowed_blocks=self.allowed_blocks,
collect_raw_data=False
)
# this makes mypy happy, and this is an invariant we really need
for block in blocks:
assert isinstance(block, BlockTag)
yield block
except CompilationException as exc:
if exc.node is None:
# TODO(jeb): attach info about resource type/file path here
exc.node = NotImplemented
raise
def _yaml_from_file(
self, source_file: SourceFile
) -> Optional[Dict[str, Any]]:
"""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:
reason = validator_error_message(e)
raise CompilationException(
'Error reading {}: {} - {}'
.format(self.project.project_name, path, reason)
)
return None
def read_yaml_sources(
self, yaml: YamlBlock
) -> Iterable[SourceTarget]:
path = yaml.path.original_file_path
yaml_key = 'sources'
for data in self._get_dicts_for(yaml, yaml_key):
try:
data = self._renderer.render_schema_source(data)
source = UnparsedSourceDefinition.from_dict(data)
except (ValidationError, JSONValidationException) as exc:
msg = error_context(path, yaml_key, data, exc)
raise CompilationException(msg) from exc
else:
for table in source.tables:
yield SourceTarget(source, table)
block = SchemaTestBlock.from_target_block(
src=target_block,
test=test,
column_name=column_name
)
try:
self.parse_node(block)
except CompilationException as exc:
context = _trimmed(str(block.target))
msg = (
'Invalid test config given in {}:'
'\n\t{}\n\t@: {}'
.format(block.path.original_file_path, exc.msg, context)
)
raise CompilationException(msg) from exc
context = {}
# change these to actual kwargs
base_node = UnparsedMacro(
path=macro_file_path,
original_file_path=macro_file_path,
package_name=package_name,
raw_sql=macro_file_contents,
root_path=root_path,
)
try:
template = dbt.clients.jinja.get_template(
macro_file_contents, context, node=base_node)
except dbt.exceptions.CompilationException as e:
e.node = base_node
raise e
for key, item in template.module.__dict__.items():
if type(item) != jinja2.runtime.Macro:
continue
node_type = None
if key.startswith(dbt.utils.MACRO_PREFIX):
node_type = NodeType.Macro
name = key.replace(dbt.utils.MACRO_PREFIX, '')
elif key.startswith(dbt.utils.OPERATION_PREFIX):
node_type = NodeType.Operation
name = key.replace(dbt.utils.OPERATION_PREFIX, '')