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_should_unwrap_single_list(self):
self.assertEquals(["spam", "eggs"], as_list(["spam", "eggs"]))
def explicit_namespaces(self, value):
self._explicit_namespaces = as_list(value)
for name in dir(project_module):
candidate = getattr(project_module, name)
name = normalize_candidate_name(candidate)
description = getattr(candidate, DESCRIPTION_ATTRIBUTE, "")
if getattr(candidate, TASK_ATTRIBUTE, None):
dependencies = getattr(candidate, DEPENDS_ATTRIBUTE, None)
task_dependencies = list()
if dependencies:
dependencies = list(as_list(dependencies))
for d in dependencies:
if isinstance(d, optional):
d = as_list(d())
task_dependencies.extend([TaskDependency(item, True) for item in d])
else:
task_dependencies.append(TaskDependency(d))
# Add injected
if name in injected_task_dependencies:
task_dependencies.extend(injected_task_dependencies[name])
del injected_task_dependencies[name]
self.logger.debug("Found task '%s' with dependencies %s", name, task_dependencies)
self.execution_manager.register_task(
Task(name, candidate, task_dependencies, description))
elif getattr(candidate, ACTION_ATTRIBUTE, None):
before = getattr(candidate, BEFORE_ATTRIBUTE, None)
after = getattr(candidate, AFTER_ATTRIBUTE, None)
description = getattr(candidate, DESCRIPTION_ATTRIBUTE) if hasattr(
candidate, DESCRIPTION_ATTRIBUTE) else ""
if hasattr(candidate, TASK_ATTRIBUTE) and getattr(candidate, TASK_ATTRIBUTE):
dependencies = getattr(candidate, DEPENDS_ATTRIBUTE) if hasattr(candidate, DEPENDS_ATTRIBUTE) else None
required_dependencies = []
optional_dependencies = []
# Add injected
if name in task_dependencies:
injected_dependencies = task_dependencies[name]
required_dependencies.extend(injected_dependencies[0])
optional_dependencies.extend(injected_dependencies[1])
if dependencies:
dependencies = list(as_list(dependencies))
for d in dependencies:
if type(d) is optional:
d = as_list(d())
optional_dependencies += d
else:
required_dependencies.append(d)
self.logger.debug("Found task '%s' with required dependencies %s and optional dependencies %s", name,
required_dependencies, optional_dependencies)
self.execution_manager.register_task(
Task(name, candidate, required_dependencies, description, optional_dependencies))
elif hasattr(candidate, ACTION_ATTRIBUTE) and getattr(candidate, ACTION_ATTRIBUTE):
before = getattr(candidate, BEFORE_ATTRIBUTE) if hasattr(
candidate, BEFORE_ATTRIBUTE) else None
after = getattr(candidate, AFTER_ATTRIBUTE) if hasattr(
candidate, AFTER_ATTRIBUTE) else None
add_task_dependency(d, name, True)
else:
add_task_dependency(d, name, False)
for name in dir(project_module):
candidate = getattr(project_module, name)
name = normalize_candidate_name(candidate)
description = getattr(candidate, DESCRIPTION_ATTRIBUTE, "")
if getattr(candidate, TASK_ATTRIBUTE, None):
dependencies = getattr(candidate, DEPENDS_ATTRIBUTE, None)
task_dependencies = list()
if dependencies:
dependencies = list(as_list(dependencies))
for d in dependencies:
if isinstance(d, optional):
d = as_list(d())
task_dependencies.extend([TaskDependency(item, True) for item in d])
else:
task_dependencies.append(TaskDependency(d))
# Add injected
if name in injected_task_dependencies:
task_dependencies.extend(injected_task_dependencies[name])
del injected_task_dependencies[name]
self.logger.debug("Found task '%s' with dependencies %s", name, task_dependencies)
self.execution_manager.register_task(
Task(name, candidate, task_dependencies, description))
def _prepare_tasks(self, tasks):
if not len(tasks):
if self.project.default_task:
tasks += as_list(self.project.default_task)
else:
raise PyBuilderException("No default task given.")
else:
new_tasks = [task for task in tasks if task[0] not in ("+", "^") or task in ("+", "^")]
append_tasks = [task[1:] for task in tasks if task[0] == "+" and task != "+"]
remove_tasks = [task[1:] for task in tasks if task[0] == "^" and task != "^"]
if len(new_tasks):
del tasks[:]
tasks.extend(new_tasks)
tasks.extend(append_tasks)
for task in remove_tasks:
try:
tasks.remove(task)
except ValueError:
pass
def build_pip_install_options(index_url=None, extra_index_url=None, upgrade=False, insecure_installs=None,
force_reinstall=False, target_dir=None, verbose=False, trusted_host=None,
constraint_file=None, eager_upgrade=False):
options = []
if index_url:
options.append("--index-url")
options.append(index_url)
if extra_index_url:
extra_index_urls = as_list(extra_index_url)
for url in extra_index_urls:
options.append("--extra-index-url")
options.append(url)
if trusted_host:
trusted_hosts = as_list(trusted_host)
for host in trusted_hosts:
options.append("--trusted-host")
options.append(host)
if upgrade:
options.append("--upgrade")
if pip_common.pip_version >= "9.0":
options.append("--upgrade-strategy")
if eager_upgrade:
options.append("eager")
def as_pip_install_target(mixed):
arguments = []
targets = as_list(mixed)
for target in targets:
if isinstance(target, RequirementsFile):
arguments.extend(("-r", target.name))
elif isinstance(target, Dependency):
if target.url:
arguments.append(target.url)
else:
arguments.append("{0}{1}".format(target.name, build_dependency_version_string(target)))
else:
arguments.append(str(target))
return arguments
for name in as_list(names):
if not isinstance(name, basestring):
name = normalize_candidate_name(name)
if name not in task_dependencies:
task_dependencies[name] = (list(), list())
task_dependencies[name][1 if optional else 0].append(depends_on)
for name in dir(project_module):
candidate = getattr(project_module, name)
name = normalize_candidate_name(candidate)
if hasattr(candidate, TASK_ATTRIBUTE) and getattr(candidate, TASK_ATTRIBUTE):
dependents = getattr(candidate, DEPENDENTS_ATTRIBUTE) if hasattr(candidate,
DEPENDENTS_ATTRIBUTE) else None
if dependents:
dependents = list(as_list(dependents))
for d in dependents:
if type(d) is optional:
d = d()
add_task_dependency(d, candidate, True)
else:
add_task_dependency(d, candidate, False)
for name in dir(project_module):
candidate = getattr(project_module, name)
name = normalize_candidate_name(candidate)
description = getattr(candidate, DESCRIPTION_ATTRIBUTE) if hasattr(
candidate, DESCRIPTION_ATTRIBUTE) else ""
if hasattr(candidate, TASK_ATTRIBUTE) and getattr(candidate, TASK_ATTRIBUTE):
dependencies = getattr(candidate, DEPENDS_ATTRIBUTE) if hasattr(candidate, DEPENDS_ATTRIBUTE) else None