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__executable_does_not_exist(self):
with self.assertRaises(ExecutableError) as exc:
dbt.clients.system.run_cmd(self.run_dir, [self.does_not_exist])
msg = str(exc.exception).lower()
self.assertIn('path', msg)
self.assertIn('could not find', msg)
self.assertIn(self.does_not_exist.lower(), msg)
def test_git(self):
try:
dbt.clients.system.run_cmd(os.getcwd(), ['git', '--help'])
except dbt.exceptions.ExecutableError as exc:
self.messages.append('Error from git --help: {!s}'.format(exc))
return red('ERROR')
return green('OK found')
if tags is None:
tags = []
if dbt.flags.STRICT_MODE:
dbt.contracts.project.ProjectList(**self.all_projects)
file_matches = dbt.clients.system.find_matching(
root_dir,
relative_dirs,
extension)
result = []
for file_match in file_matches:
file_contents = dbt.clients.system.load_file_contents(
file_match.get('absolute_path'))
parts = dbt.utils.split_path(file_match.get('relative_path', ''))
name, _ = os.path.splitext(parts[-1])
path = self.get_compiled_path(name,
file_match.get('relative_path'))
original_file_path = os.path.join(
file_match.get('searched_path'),
path)
result.append({
'name': name,
'root_path': root_dir,
'resource_type': resource_type,
def install(self, project):
dest_path = self.get_installation_path(project)
if os.path.exists(dest_path):
if dbt.clients.system.path_is_symlink(dest_path):
dbt.clients.system.remove_file(dest_path)
else:
dbt.clients.system.rmdir(dest_path)
dbt.clients.system.move(self._checkout(project), dest_path)
def install(self, project):
version_string = self.version_name()
metadata = self.fetch_metadata(project)
tar_name = '{}.{}.tar.gz'.format(self.package, version_string)
tar_path = os.path.realpath(os.path.join(DOWNLOADS_PATH, tar_name))
dbt.clients.system.make_directory(os.path.dirname(tar_path))
download_url = metadata['downloads']['tarball']
dbt.clients.system.download(download_url, tar_path)
deps_path = project.modules_path
package_name = self.get_project_name(project)
dbt.clients.system.untar_package(tar_path, deps_path, package_name)
def read_profile(profiles_dir):
path = os.path.join(profiles_dir, 'profiles.yml')
contents = None
if os.path.isfile(path):
try:
contents = dbt.clients.system.load_file_contents(path, strip=False)
return dbt.clients.yaml_helper.load_yaml_text(contents)
except dbt.exceptions.ValidationException as e:
msg = INVALID_PROFILE_MESSAGE.format(error_string=e)
raise dbt.exceptions.ValidationException(msg)
return {}
def get_addendum(self, project_name, profiles_path):
open_cmd = dbt.clients.system.open_dir_cmd()
return ON_COMPLETE_MESSAGE.format(
open_cmd=open_cmd,
project_name=project_name,
profiles_path=profiles_path,
docs_url=DOCS_URL
)
def make_log_dir_if_missing(log_dir):
import dbt.clients.system
dbt.clients.system.make_directory(log_dir)