Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def find_project_file(start_dir, basename):
'''Walk up the directory tree until we find a file of the given name.'''
prefix = os.path.abspath(start_dir)
while True:
candidate = os.path.join(prefix, basename)
if os.path.isfile(candidate):
return candidate
if os.path.exists(candidate):
raise PrintableError(
"Found {}, but it's not a file.".format(candidate))
if os.path.dirname(prefix) == prefix:
# We've walked all the way to the top. Bail.
raise PrintableError("Can't find " + basename)
# Not found at this level. We must go...shallower.
prefix = os.path.dirname(prefix)
def _error(logging_target_name, text, *text_format_args):
text = text.format(*text_format_args)
raise PrintableError('Error in target {}: {}'.format(
logging_target_name, text))
pass
class PluginCommandCandidateError(PrintableError):
pass
class PluginModuleFieldError(PrintableError):
pass
class PluginMetadataMissingError(PrintableError):
pass
class PluginPermissionsError(PrintableError):
pass
class PluginRuntimeError(PrintableError):
def __init__(self, type, fields, errorcode, output):
# Don't depend on plugins using terminating newlines.
stripped_output = output.strip('\n')
super().__init__(stripped_output)
def _get_override_tree(runtime, module, rules):
override_path = runtime.get_override(module.name)
if not os.path.exists(override_path):
raise PrintableError(
"override path for module '{}' does not exist: {}".format(
module.name, override_path))
if not os.path.isdir(override_path):
raise PrintableError(
"override path for module '{}' is not a directory: {}".format(
module.name, override_path))
override_module = module.get_local_override(override_path)
tree = yield from override_module.get_tree(runtime, rules)
return tree
pass
class PluginModuleFieldError(PrintableError):
pass
class PluginMetadataMissingError(PrintableError):
pass
class PluginPermissionsError(PrintableError):
pass
class PluginRuntimeError(PrintableError):
def __init__(self, type, fields, errorcode, output):
# Don't depend on plugins using terminating newlines.
stripped_output = output.strip('\n')
super().__init__(stripped_output)
@contextlib.contextmanager
def debug_parallel_count_context():
global DEBUG_PARALLEL_COUNT, DEBUG_PARALLEL_MAX
DEBUG_PARALLEL_COUNT += 1
DEBUG_PARALLEL_MAX = max(DEBUG_PARALLEL_COUNT, DEBUG_PARALLEL_MAX)
try:
yield
finally:
DEBUG_PARALLEL_COUNT -= 1
def tmp_dir(context):
return tempfile.TemporaryDirectory(dir=context.tmp_root)
class PluginCandidateError(PrintableError):
pass
class PluginCommandCandidateError(PrintableError):
pass
class PluginModuleFieldError(PrintableError):
pass
class PluginMetadataMissingError(PrintableError):
pass
class PluginPermissionsError(PrintableError):
def _get_module_checked(self, name):
if name not in self.modules:
raise PrintableError('Module "{}" doesn\'t exist.', name)
return self.modules[name]
def _get_override_tree(self, runtime, path):
if not os.path.exists(path):
raise PrintableError(
"override path for module '{}' does not exist: {}".format(
self.name, path))
if not os.path.isdir(path):
raise PrintableError(
"override path for module '{}' is not a directory: {}".format(
self.name, path))
return runtime.cache.import_tree(path)
def get_rules(runtime, rule_names):
rules = []
for name in rule_names:
if name not in runtime.scope:
raise PrintableError('rule "{}" does not exist'.format(name))
rule = runtime.scope[name]
# Avoid a circular import.
if type(rule).__name__ != 'Rule':
raise PrintableError('"{}" is not a rule'.format(name))
rules.append(rule)
return rules