Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _assert_confirm(stdout):
assert_input(capfd, query, "Y", expected = True, input_ = cli.confirm, stdout = stdout)
assert_input(capfd, query, "y", expected = True, input_ = cli.confirm, stdout = stdout)
assert_input(capfd, query,"\n", expected = True, input_ = cli.confirm, stdout = stdout)
assert_input(capfd, query, "n", expected = False, input_ = cli.confirm, stdout = stdout)
assert_input(capfd, query, "1", expected = False, input_ = cli.confirm, stdout = stdout)
@cli.command
def foobar(*args, **kwargs):
args = get_args()
params = merge_dict(args, override)
for k, v in iteritems(values):
assert params[k] == v
if initial:
for k in iterkeys(initial):
assert initial[k] == args[k]
package.latest_version)
except (TypeError, ValueError):
pass
string = "* %s" % cli_format(package.name, _SEMVER_COLOR_MAP.get(difference, cli.CLEAR))
if getattr(package, "has_dependency_conflict", False):
string += cli_format(" [dependency conflict]", cli.RED)
if package.current_version:
string += " (%s)" % package.current_version
if package.latest_version and package.current_version != package.latest_version:
string += " -> (%s)" % _cli_format_semver(package.latest_version, difference)
string += " " + cli_format("[%s]" % package.home_page, cli.CYAN)
return string
def _format(x):
return cli_format(x, cli.YELLOW)
logger.info("Updating pip executables: %s" % " ".join(pip_path))
with parallel.no_daemon_pool(processes = jobs) as pool:
pool.imap_unordered(
partial(
update_pip, **{ "user": user, "quiet": not verbose,
"file": file_ }
),
pip_path
)
if self:
package = __name__
logger.info("Updating %s..." % package)
cli.echo(cli_format("Updating %s..." % package, cli.YELLOW),
file = file_)
_pip.call("install", package, user = user, quiet = not verbose,
no_cache = True, upgrade = True)
cli.echo("%s upto date." % cli_format(package, cli.CYAN),
file = file_)
else:
if project:
project = sequencify(project)
requirements = requirements or [ ]
pipfile = pipfile or [ ]
logger.info("Detecting projects and its dependencies...")
with parallel.no_daemon_pool(processes = jobs) as pool:
from pipupgrade.model import Registry
from pipupgrade.commands.util import cli_format
from pipupgrade.table import Table
from pipupgrade.tree import Node as TreeNode
from pipupgrade.util.string import pluralize, strip
from pipupgrade.util.system import read, write, popen, which
from pipupgrade.util.array import squash
from pipupgrade import (_pip, cli, semver,
log, parallel
)
from pipupgrade.exception import PopenError
logger = log.get_logger()
_SEMVER_COLOR_MAP = dict(
major = cli.RED,
minor = cli.YELLOW,
patch = cli.GREEN
)
_DEPENDENCY_FORMATS = ("tree", "json", "yaml")
def _cli_format_semver(version, type_):
def _format(x):
return cli_format(x, cli.YELLOW)
try:
semver.parse(version)
if type_ == "major":
version = _format(version)
if type_ == "minor":
index = version.find(".", 1) + 1
def format_yellow(string):
string = cli_format(string, cli.YELLOW)
return string
force = False,
verbose = False
):
if not verbose:
logger.setLevel(log.NOTSET)
logger.info("Environment: %s" % environment())
logger.info("Arguments Passed: %s" % locals())
file_ = output
if file_:
logger.info("Writing to output file %s..." % file_)
touch(file_)
cli.echo(cli_format("Checking...", cli.YELLOW), file = file_)
pip_path = pip_path or [ ]
pip_path = [which(p) for p in pip_path] or _pip._PIP_EXECUTABLES
logger.info("`pip` executables found: %s" % pip_path)
logger.info("Using %s jobs..." % jobs)
registries = [ ]
if pip:
logger.info("Updating pip executables: %s" % " ".join(pip_path))
with parallel.no_daemon_pool(processes = jobs) as pool:
pool.imap_unordered(
partial(