Python tempfile 模块，mkdtemp() 实例源码
我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用tempfile.mkdtemp()。
def src_proc_dispatcher(pkg_name, src_tbl_name, src_loc):
tobj = tempfile.mkdtemp(dir='/var/cache/acbs/build/', prefix='acbs.')
src_tbl_loc = os.path.join(src_loc, src_tbl_name)
shadow_ark_loc = os.path.join(tobj, src_tbl_name)
if os.path.isdir(src_tbl_loc):
print('[I] Making a copy of the source directory...', end='')
try:
shutil.copytree(src=src_tbl_loc, dst=shadow_ark_loc)
except:
print('Failed!')
return False
print('Done!')
return True, tobj
else:
os.symlink(src_tbl_loc, shadow_ark_loc)
# print('[D] Source location: {}, Shadow link: {}'.format(src_tbl_loc, shadow_ark_loc))
return decomp_file(shadow_ark_loc, tobj), tobj
def glustermount(hostname, volname):
"""
Context manager for Mounting Gluster Volume
Use as
with glustermount(HOSTNAME, VOLNAME) as MNT:
# Do your stuff
Automatically unmounts it in case of Exceptions/out of context
"""
mnt = tempfile.mkdtemp(prefix="georepsetup_")
execute(["glusterfs",
"--xlator-option=\"*dht.lookup-unhashed=off\"",
"--volfile-server", hostname,
"--volfile-id", volname,
"-l", SESSION_MOUNT_LOG_FILE,
"--client-pid=-1",
mnt],
failure_msg="Unable to Mount Gluster Volume "
"{0}:{1}".format(hostname, volname))
if os.path.ismount(mnt):
yield mnt
else:
output_notok("Unable to Mount Gluster Volume "
"{0}:{1}".format(hostname, volname))
cleanup(hostname, volname, mnt)
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with get_zip_class()(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def generate2():
"""
Call an external Python 2 program to retrieve the AST symbols of that
language version
:return:
"""
import subprocess as sp
import tempfile, shutil, sys, traceback
tempdir = tempfile.mkdtemp()
tempfile = os.path.join(tempdir, "py2_ast_code.py")
py2_proc_out = ""
try:
with open(tempfile, 'w') as py2code:
py2code.write(generate_str + WRITESYMS_CODE)
py2_proc_out = sp.check_output(["python2", tempfile]).decode()
finally:
try:
shutil.rmtree(tempdir)
except:
print("Warning: error trying to delete the temporal directory:", file=sys.stderr)
print(traceback.format_exc(), file=sys.stderr)
return set(py2_proc_out.splitlines())
def __init__(self, name=None, delete=None):
# If we were not given an explicit directory, and we were not given an
# explicit delete option, then we'll default to deleting.
if name is None and delete is None:
delete = True
if name is None:
# We realpath here because some systems have their default tmpdir
# symlinked to another directory. This tends to confuse build
# scripts, so we canonicalize the path by traversing potential
# symlinks here.
name = os.path.realpath(tempfile.mkdtemp(prefix="pip-build-"))
# If we were not given an explicit directory, and we were not given
# an explicit delete option, then we'll default to deleting.
if delete is None:
delete = True
self.name = name
self.delete = delete
def _build_one(self, req, output_dir, python_tag=None):
"""Build one wheel.
:return: The filename of the built wheel, or None if the build failed.
"""
tempd = tempfile.mkdtemp('pip-wheel-')
try:
if self.__build_one(req, tempd, python_tag=python_tag):
try:
wheel_name = os.listdir(tempd)[0]
wheel_path = os.path.join(output_dir, wheel_name)
shutil.move(os.path.join(tempd, wheel_name), wheel_path)
logger.info('Stored in directory: %s', output_dir)
return wheel_path
except:
pass
# Ignore return, we can't do anything else useful.
self._clean_one(req)
return None
finally:
rmtree(tempd)
def test_install():
tempdir = mkdtemp()
def get_supported():
return list(wheel.pep425tags.get_supported()) + [('py3', 'none', 'win32')]
whl = WheelFile(TESTWHEEL, context=get_supported)
assert whl.supports_current_python(get_supported)
try:
locs = {}
for key in ('purelib', 'platlib', 'scripts', 'headers', 'data'):
locs[key] = os.path.join(tempdir, key)
os.mkdir(locs[key])
whl.install(overrides=locs)
assert len(os.listdir(locs['purelib'])) == 0
assert check(locs['platlib'], 'hello.pyd')
assert check(locs['platlib'], 'hello', 'hello.py')
assert check(locs['platlib'], 'hello', '__init__.py')
assert check(locs['data'], 'hello.dat')
assert check(locs['headers'], 'hello.dat')
assert check(locs['scripts'], 'hello.sh')
assert check(locs['platlib'], 'test-1.0.dist-info', 'RECORD')
finally:
shutil.rmtree(tempdir)
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with get_zip_class()(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with get_zip_class()(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def upload_to_fileshare_test(self): #pylint: disable=no-self-use
"""Upload copies files to non-native store correctly with no
progress"""
import shutil
import tempfile
temp_file = tempfile.NamedTemporaryFile(dir=tempfile.mkdtemp())
temp_src_dir = os.path.dirname(temp_file.name)
temp_dst_dir = tempfile.mkdtemp()
shutil_mock = MagicMock()
shutil_mock.copyfile.return_value = None
with patch('sfctl.custom_app.shutil', new=shutil_mock):
sf_c.upload_to_fileshare(temp_src_dir, temp_dst_dir, False)
shutil_mock.copyfile.assert_called_once()
temp_file.close()
shutil.rmtree(os.path.dirname(temp_file.name))
shutil.rmtree(temp_dst_dir)
def __init__(self, name=None, delete=None):
# If we were not given an explicit directory, and we were not given an
# explicit delete option, then we'll default to deleting.
if name is None and delete is None:
delete = True
if name is None:
# We realpath here because some systems have their default tmpdir
# symlinked to another directory. This tends to confuse build
# scripts, so we canonicalize the path by traversing potential
# symlinks here.
name = os.path.realpath(tempfile.mkdtemp(prefix="pip-build-"))
# If we were not given an explicit directory, and we were not given
# an explicit delete option, then we'll default to deleting.
if delete is None:
delete = True
self.name = name
self.delete = delete
def _build_one(self, req, output_dir, python_tag=None):
"""Build one wheel.
:return: The filename of the built wheel, or None if the build failed.
"""
tempd = tempfile.mkdtemp('pip-wheel-')
try:
if self.__build_one(req, tempd, python_tag=python_tag):
try:
wheel_name = os.listdir(tempd)[0]
wheel_path = os.path.join(output_dir, wheel_name)
shutil.move(os.path.join(tempd, wheel_name), wheel_path)
logger.info('Stored in directory: %s', output_dir)
return wheel_path
except:
pass
# Ignore return, we can't do anything else useful.
self._clean_one(req)
return None
finally:
rmtree(tempd)
def build_and_install(self, setup_script, setup_base):
args = ['bdist_egg', '--dist-dir']
dist_dir = tempfile.mkdtemp(
prefix='egg-dist-tmp-', dir=os.path.dirname(setup_script)
)
try:
self._set_fetcher_options(os.path.dirname(setup_script))
args.append(dist_dir)
self.run_setup(setup_script, setup_base, args)
all_eggs = Environment([dist_dir])
eggs = []
for key in all_eggs:
for dist in all_eggs[key]:
eggs.append(self.install_egg(dist.location, setup_base))
if not eggs and not self.dry_run:
log.warn("No eggs found in %s (setup script problem?)",
dist_dir)
return eggs
finally:
rmtree(dist_dir)
log.set_verbosity(self.verbose) # restore our log verbosity
def archive_context(filename):
"""
Unzip filename to a temporary directory, set to the cwd.
The unzipped target is cleaned up after.
"""
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with ContextualZipFile(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def reformat(self, sourcefile, destfile, configfile):
# type: (str, str, str) -> None
"""Reformats sourcefile according to configfile and writes it to destfile.
This method is only used for testing.
"""
tmpdir = tempfile.mkdtemp(prefix='whatstyle_')
cfg = os.path.join(tmpdir, self.configfilename)
copyfile(configfile, cfg)
tmpfilename = os.path.join(tmpdir, os.path.basename(sourcefile))
copyfile(sourcefile, tmpfilename)
cmdargs = [tmpfilename]
exeresult = run_executable(self.exe, cmdargs)
writebinary(destfile, exeresult.stdout)
os.remove(tmpfilename)
os.remove(cfg)
os.rmdir(tmpdir)
def from_file(proto_file):
'''
Take a filename |protoc_file|, compile it via the Protobuf
compiler, and import the module.
Return the module if successfully compiled, otherwise raise either
a ProtocNotFound or BadProtobuf exception.
'''
if not proto_file.endswith('.proto'):
raise BadProtobuf()
dest = tempfile.mkdtemp()
full_path = os.path.abspath(proto_file)
_compile_proto(full_path, dest)
filename = os.path.split(full_path)[-1]
name = re.search(r'^(.*)\.proto$', filename).group(1)
target = os.path.join(dest, name+'_pb2.py')
return _load_module(target)
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with get_zip_class()(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with ContextualZipFile(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def make_virtualenv(args):
with cleanup_dir(tempfile.mkdtemp()) as tempdir:
venv, python, pip = dirs(tempdir)
print_call(
sys.executable, '-m', 'virtualenv', venv,
'-p', args.python, '--never-download',
)
def pip_install(*argv):
print_call(pip, 'install', '-i', args.index_url, *argv)
# Latest pip installs python3.5 wheels
pip_install('pip', 'setuptools', '--upgrade')
pip_install('-r', 'requirements-minimal.txt')
pip_install('-r', 'requirements-dev-minimal.txt')
reexec(
python, __file__.rstrip('c'),
'--tempdir', tempdir,
# Pass along existing args
'--index-url', args.index_url,
'--exec-count', str(args.exec_count),
'--exec-limit', str(args.exec_limit),
reason='to use the virtualenv python',
)
def neo4j_test_ws_dir(datafiles):
return datafiles
# @pytest.fixture(scope="session")
# def workspace(request, data_directory):
# wsconf_file = data_directory.join("workspace.yaml")
# temp_root = tempfile.mkdtemp()
# ws = Workspace("saapy-test-ws",
# temp_root,
# "saapy-test-ws",
# configuration_text=wsconf_file.read_text("utf-8"))
#
# def fin():
# shutil.rmtree(temp_root)
#
# request.addfinalizer(fin)
# return ws # provide the fixture value
def archive_context(filename):
# extracting the archive
tmpdir = tempfile.mkdtemp()
log.warn('Extracting in %s', tmpdir)
old_wd = os.getcwd()
try:
os.chdir(tmpdir)
with get_zip_class()(filename) as archive:
archive.extractall()
# going in the directory
subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
os.chdir(subdir)
log.warn('Now working in %s', subdir)
yield
finally:
os.chdir(old_wd)
shutil.rmtree(tmpdir)
def get_sign_command(self, filename, signer, sign_password):
"""
Return a suitable command for signing a file.
:param filename: The pathname to the file to be signed.
:param signer: The identifier of the signer of the file.
:param sign_password: The passphrase for the signer's
private key used for signing.
:return: The signing command as a list suitable to be
passed to :class:`subprocess.Popen`.
"""
cmd = [self.gpg, '--status-fd', '2', '--no-tty']
if self.gpg_home:
cmd.extend(['--homedir', self.gpg_home])
if sign_password is not None:
cmd.extend(['--batch', '--passphrase-fd', '0'])
td = tempfile.mkdtemp()
sf = os.path.join(td, os.path.basename(filename) + '.asc')
cmd.extend(['--detach-sign', '--armor', '--local-user',
signer, '--output', sf, filename])
logger.debug('invoking: %s', ' '.join(cmd))
return cmd, sf
def build_and_install(self, setup_script, setup_base):
args = ['bdist_egg', '--dist-dir']
dist_dir = tempfile.mkdtemp(
prefix='egg-dist-tmp-', dir=os.path.dirname(setup_script)
)
try:
self._set_fetcher_options(os.path.dirname(setup_script))
args.append(dist_dir)
self.run_setup(setup_script, setup_base, args)
all_eggs = Environment([dist_dir])
eggs = []
for key in all_eggs:
for dist in all_eggs[key]:
eggs.append(self.install_egg(dist.location, setup_base))
if not eggs and not self.dry_run:
log.warn("No eggs found in %s (setup script problem?)",
dist_dir)
return eggs
finally:
rmtree(dist_dir)
log.set_verbosity(self.verbose) # restore our log verbosity
def setUp(self):
if self.datafile is None or self.dataname is None:
return
if not os.path.isfile(self.datafile):
self.old_cwd = None
return
self.old_cwd = os.getcwd()
self.temp_dir = tempfile.mkdtemp()
zip_file, source, target = [None, None, None]
try:
zip_file = zipfile.ZipFile(self.datafile)
for files in zip_file.namelist():
_extract(zip_file, files, self.temp_dir)
finally:
if zip_file:
zip_file.close()
del zip_file
os.chdir(os.path.join(self.temp_dir, self.dataname))
def setUp(self):
self.dir = tempfile.mkdtemp()
setup = os.path.join(self.dir, 'setup.py')
f = open(setup, 'w')
f.write(SETUP_PY)
f.close()
self.old_cwd = os.getcwd()
os.chdir(self.dir)
self.upload_dir = os.path.join(self.dir, 'build')
os.mkdir(self.upload_dir)
# A test document.
f = open(os.path.join(self.upload_dir, 'index.html'), 'w')
f.write("Hello world.")
f.close()
# An empty folder.
os.mkdir(os.path.join(self.upload_dir, 'empty'))
if sys.version >= "2.6":
self.old_base = site.USER_BASE
site.USER_BASE = upload_docs.USER_BASE = tempfile.mkdtemp()
self.old_site = site.USER_SITE
site.USER_SITE = upload_docs.USER_SITE = tempfile.mkdtemp()
def test_create_zipfile(self):
# Test to make sure zipfile creation handles common cases.
# This explicitly includes a folder containing an empty folder.
dist = Distribution()
cmd = upload_docs(dist)
cmd.upload_dir = self.upload_dir
cmd.target_dir = self.upload_dir
tmp_dir = tempfile.mkdtemp()
tmp_file = os.path.join(tmp_dir, 'foo.zip')
try:
zip_file = cmd.create_zipfile(tmp_file)
assert zipfile.is_zipfile(tmp_file)
zip_file = zipfile.ZipFile(tmp_file) # woh...
assert zip_file.namelist() == ['index.html']
zip_file.close()
finally:
shutil.rmtree(tmp_dir)
def setUp(self):
self.temp_dir = tempfile.mkdtemp()
f = open(os.path.join(self.temp_dir, 'setup.py'), 'w')
f.write(SETUP_PY)
f.close()
# Set up the rest of the test package
test_pkg = os.path.join(self.temp_dir, 'sdist_test')
os.mkdir(test_pkg)
# *.rst was not included in package_data, so c.rst should not be
# automatically added to the manifest when not under version control
for fname in ['__init__.py', 'a.txt', 'b.txt', 'c.rst']:
# Just touch the files; their contents are irrelevant
open(os.path.join(test_pkg, fname), 'w').close()
self.old_cwd = os.getcwd()
os.chdir(self.temp_dir)
def test_no_find_links(self):
# new option '--no-find-links', that blocks find-links added at
# the project level
dist = Distribution()
cmd = easy_install(dist)
cmd.check_pth_processing = lambda: True
cmd.no_find_links = True
cmd.find_links = ['link1', 'link2']
cmd.install_dir = os.path.join(tempfile.mkdtemp(), 'ok')
cmd.args = ['ok']
cmd.ensure_finalized()
self.assertEqual(cmd.package_index.scanned_urls, {})
# let's try without it (default behavior)
cmd = easy_install(dist)
cmd.check_pth_processing = lambda: True
cmd.find_links = ['link1', 'link2']
cmd.install_dir = os.path.join(tempfile.mkdtemp(), 'ok')
cmd.args = ['ok']
cmd.ensure_finalized()
keys = sorted(cmd.package_index.scanned_urls.keys())
self.assertEqual(keys, ['link1', 'link2'])
def test_updates_package(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['-r', requirements]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_updates_package_in_nested_requirements(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements-with-nested-reqfile.txt')
requirements_nested = os.path.join(tempdir, 'requirements-nested.txt')
shutil.copy('tests/samples/requirements-with-nested-reqfile.txt', requirements)
shutil.copy('tests/samples/requirements-nested.txt', requirements_nested)
args = ['-r', requirements]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'readtime'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated readtime: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_in_nested_requirements').read()
self.assertEquals(open(requirements).read(), expected_requirements)
expected_requirements = open('tests/samples/results/test_updates_package_in_nested_requirements_nested').read()
self.assertEquals(open(requirements_nested).read(), expected_requirements)
def test_requirements_long_option_accepted(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['--requirement', requirements]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_updates_package_to_output_file(self):
tempdir = tempfile.mkdtemp()
output = os.path.join(tempdir, 'output.txt')
requirements = open('tests/samples/requirements.txt').read()
args = ['-r', 'tests/samples/requirements.txt', '--output', output]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
self.assertEquals(open('tests/samples/requirements.txt').read(), requirements)
def test_exit_code_from_some_updates(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['-r', requirements, '--nonzero-exit-code']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertEqual(result.exception.code, 11)
expected_output = "Updated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 11)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_exit_code_from_nested_requirements_file(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements-with-nested-reqfile.txt')
requirements_nested = os.path.join(tempdir, 'requirements-nested.txt')
shutil.copy('tests/samples/requirements-with-nested-reqfile.txt', requirements)
shutil.copy('tests/samples/requirements-nested.txt', requirements_nested)
args = ['-r', requirements, '--nonzero-exit-code']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'readtime'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertEqual(result.exception.code, 11)
expected_output = "Updated readtime: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 11)
expected_requirements = open('tests/samples/results/test_updates_package_in_nested_requirements').read()
self.assertEquals(open(requirements).read(), expected_requirements)
expected_requirements = open('tests/samples/results/test_updates_package_in_nested_requirements_nested').read()
self.assertEquals(open(requirements_nested).read(), expected_requirements)
def test_no_recursive_option(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements-with-nested-reqfile.txt')
requirements_nested = os.path.join(tempdir, 'requirements-nested.txt')
shutil.copy('tests/samples/requirements-with-nested-reqfile.txt', requirements)
shutil.copy('tests/samples/requirements-nested.txt', requirements_nested)
args = ['-r', requirements, '-n']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'readtime'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/requirements-with-nested-reqfile.txt').read()
self.assertEquals(open(requirements).read(), expected_requirements)
expected_requirements = open('tests/samples/requirements-nested.txt').read()
self.assertEquals(open(requirements_nested).read(), expected_requirements)
def test_skip_package(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements-multiple.txt', requirements)
args = ['-r', requirements, '-s', 'flask']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated Alembic: 0.9 -> 0.10.1\nUpdated sqlalchemy: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_skip_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_skip_multiple_packages(self):
requirements = 'tests/samples/requirements-multiple.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile, '-s', 'flask, alembic , SQLAlchemy']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open(requirements).read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_only_multiple_packages(self):
requirements = 'tests/samples/requirements-multiple.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile, '--only', 'flask, sqlalchemy']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nUpdated sqlalchemy: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_only_multiple_packages').read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_updates_package_with_no_version_specified(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['-r', requirements, '-f']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nUpdated flask: Unknown -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_no_version_specified').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_invalid_package(self):
requirements = 'tests/samples/requirements.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
mock_find_all_candidates.return_value = []
result = self.runner.invoke(pur, args)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
self.assertEquals(open(tmpfile).read(), open(requirements).read())
def test_no_arguments(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = []
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
with self.cd(tempdir):
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_no_arguments_and_no_requirements_file(self):
tempdir = tempfile.mkdtemp()
args = []
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
with self.cd(tempdir):
result = self.runner.invoke(pur, args)
self.assertEqual(result.exception.code, 1)
expected_output = "Error: Could not open requirements file: [Errno 2] No such file or directory: 'requirements.txt'\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 1)
def test_updates_package_with_extras(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements-with-extras.txt', requirements)
args = ['-r', requirements]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'firstpackage'
version = '2.0'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
expected_output = "Updated firstpackage1: 1 -> 2.0\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_extras').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_updates_package_with_max_version_spec(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements-with-max-version-spec.txt', requirements)
args = ['-r', requirements]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'afakepackage'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
expected_output = "Updated afakepackage: 0.9 -> 0.10.1\nUpdated afakepackage: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_max_version_spec').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_max_version_spec_prevents_updating_package(self):
requirements = 'tests/samples/requirements-with-max-version-spec.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'afakepackage'
version = '2.0'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open(tmpfile).read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_notequal_version_spec_prevents_updating_package(self):
requirements = 'tests/samples/requirements-multiline.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'afakepackage'
version = '0.9.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open(tmpfile).read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_updates_package_with_multiline_spec(self):
requirements = 'tests/samples/requirements-multiline.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'afakepackage'
version = '1.0'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Updated afakepackage: 0.9 -> 1.0\nUpdated afakepackage: 0.9 -> 1.0\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_multiline_spec').read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_updates_package_with_min_version_spec(self):
requirements = 'tests/samples/requirements-with-min-version-spec.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'django'
version = '1.8.13'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
expected_output = "Updated django: 1.8.6 -> 1.8.13\nNew version for django found (1.8.13), but current spec prohibits updating: django > 1.8.6, < 1.9\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_min_version_spec').read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_updates_package_with_wildcard_spec(self):
requirements = 'tests/samples/requirements-with-wildcard-spec.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'django'
version = '1.0'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
expected_output = "Updated flask: 0.9 -> 1.0\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_wildcard_spec').read()
self.assertEquals(open(tmpfile).read(), expected_requirements)