mirror of https://github.com/python/cpython
392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""Tests for the packaging.install module."""
|
|
import os
|
|
import logging
|
|
from tempfile import mkstemp
|
|
from sysconfig import is_python_build
|
|
|
|
from packaging import install
|
|
from packaging.pypi.xmlrpc import Client
|
|
from packaging.metadata import Metadata
|
|
from packaging.tests.support import (LoggingCatcher, TempdirManager, unittest,
|
|
fake_dec)
|
|
try:
|
|
import threading
|
|
from packaging.tests.pypi_server import use_xmlrpc_server
|
|
except ImportError:
|
|
threading = None
|
|
use_xmlrpc_server = fake_dec
|
|
|
|
|
|
class InstalledDist:
|
|
"""Distribution object, represent distributions currently installed on the
|
|
system"""
|
|
def __init__(self, name, version, deps):
|
|
self.metadata = Metadata()
|
|
self.name = name
|
|
self.version = version
|
|
self.metadata['Name'] = name
|
|
self.metadata['Version'] = version
|
|
self.metadata['Requires-Dist'] = deps
|
|
|
|
def __repr__(self):
|
|
return '<InstalledDist %r>' % self.metadata['Name']
|
|
|
|
|
|
class ToInstallDist:
|
|
"""Distribution that will be installed"""
|
|
|
|
def __init__(self, files=False):
|
|
self._files = files
|
|
self.install_called = False
|
|
self.install_called_with = {}
|
|
self.uninstall_called = False
|
|
self._real_files = []
|
|
self.name = "fake"
|
|
self.version = "fake"
|
|
if files:
|
|
for f in range(0, 3):
|
|
fp, fn = mkstemp()
|
|
os.close(fp)
|
|
self._real_files.append(fn)
|
|
|
|
def _unlink_installed_files(self):
|
|
if self._files:
|
|
for fn in self._real_files:
|
|
os.unlink(fn)
|
|
|
|
def list_installed_files(self, **args):
|
|
if self._files:
|
|
return self._real_files
|
|
|
|
def get_install(self, **args):
|
|
return self.list_installed_files()
|
|
|
|
|
|
class MagicMock:
|
|
def __init__(self, return_value=None, raise_exception=False):
|
|
self.called = False
|
|
self._times_called = 0
|
|
self._called_with = []
|
|
self._return_value = return_value
|
|
self._raise = raise_exception
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
self.called = True
|
|
self._times_called = self._times_called + 1
|
|
self._called_with.append((args, kwargs))
|
|
iterable = hasattr(self._raise, '__iter__')
|
|
if self._raise:
|
|
if ((not iterable and self._raise)
|
|
or self._raise[self._times_called - 1]):
|
|
raise Exception
|
|
return self._return_value
|
|
|
|
def called_with(self, *args, **kwargs):
|
|
return (args, kwargs) in self._called_with
|
|
|
|
|
|
def get_installed_dists(dists):
|
|
"""Return a list of fake installed dists.
|
|
The list is name, version, deps"""
|
|
objects = []
|
|
for name, version, deps in dists:
|
|
objects.append(InstalledDist(name, version, deps))
|
|
return objects
|
|
|
|
|
|
class TestInstall(LoggingCatcher, TempdirManager, unittest.TestCase):
|
|
def _get_client(self, server, *args, **kwargs):
|
|
return Client(server.full_address, *args, **kwargs)
|
|
|
|
def _get_results(self, output):
|
|
"""return a list of results"""
|
|
installed = [(o.name, str(o.version)) for o in output['install']]
|
|
remove = [(o.name, str(o.version)) for o in output['remove']]
|
|
conflict = [(o.name, str(o.version)) for o in output['conflict']]
|
|
return installed, remove, conflict
|
|
|
|
@unittest.skipIf(threading is None, 'needs threading')
|
|
@use_xmlrpc_server()
|
|
def test_existing_deps(self, server):
|
|
# Test that the installer get the dependencies from the metadatas
|
|
# and ask the index for this dependencies.
|
|
# In this test case, we have choxie that is dependent from towel-stuff
|
|
# 0.1, which is in-turn dependent on bacon <= 0.2:
|
|
# choxie -> towel-stuff -> bacon.
|
|
# Each release metadata is not provided in metadata 1.2.
|
|
client = self._get_client(server)
|
|
archive_path = '%s/distribution.tar.gz' % server.full_address
|
|
server.xmlrpc.set_distributions([
|
|
{'name': 'choxie',
|
|
'version': '2.0.0.9',
|
|
'requires_dist': ['towel-stuff (0.1)'],
|
|
'url': archive_path},
|
|
{'name': 'towel-stuff',
|
|
'version': '0.1',
|
|
'requires_dist': ['bacon (<= 0.2)'],
|
|
'url': archive_path},
|
|
{'name': 'bacon',
|
|
'version': '0.1',
|
|
'requires_dist': [],
|
|
'url': archive_path},
|
|
])
|
|
installed = get_installed_dists([('bacon', '0.1', [])])
|
|
output = install.get_infos("choxie", index=client,
|
|
installed=installed)
|
|
|
|
# we don't have installed bacon as it's already installed system-wide
|
|
self.assertEqual(0, len(output['remove']))
|
|
self.assertEqual(2, len(output['install']))
|
|
readable_output = [(o.name, str(o.version))
|
|
for o in output['install']]
|
|
self.assertIn(('towel-stuff', '0.1'), readable_output)
|
|
self.assertIn(('choxie', '2.0.0.9'), readable_output)
|
|
|
|
@unittest.skipIf(threading is None, 'needs threading')
|
|
@use_xmlrpc_server()
|
|
def test_upgrade_existing_deps(self, server):
|
|
client = self._get_client(server)
|
|
archive_path = '%s/distribution.tar.gz' % server.full_address
|
|
server.xmlrpc.set_distributions([
|
|
{'name': 'choxie',
|
|
'version': '2.0.0.9',
|
|
'requires_dist': ['towel-stuff (0.1)'],
|
|
'url': archive_path},
|
|
{'name': 'towel-stuff',
|
|
'version': '0.1',
|
|
'requires_dist': ['bacon (>= 0.2)'],
|
|
'url': archive_path},
|
|
{'name': 'bacon',
|
|
'version': '0.2',
|
|
'requires_dist': [],
|
|
'url': archive_path},
|
|
])
|
|
|
|
output = install.get_infos("choxie", index=client,
|
|
installed=get_installed_dists([('bacon', '0.1', [])]))
|
|
installed = [(o.name, str(o.version)) for o in output['install']]
|
|
|
|
# we need bacon 0.2, but 0.1 is installed.
|
|
# So we expect to remove 0.1 and to install 0.2 instead.
|
|
remove = [(o.name, str(o.version)) for o in output['remove']]
|
|
self.assertIn(('choxie', '2.0.0.9'), installed)
|
|
self.assertIn(('towel-stuff', '0.1'), installed)
|
|
self.assertIn(('bacon', '0.2'), installed)
|
|
self.assertIn(('bacon', '0.1'), remove)
|
|
self.assertEqual(0, len(output['conflict']))
|
|
|
|
@unittest.skipIf(threading is None, 'needs threading')
|
|
@use_xmlrpc_server()
|
|
def test_conflicts(self, server):
|
|
# Tests that conflicts are detected
|
|
client = self._get_client(server)
|
|
archive_path = '%s/distribution.tar.gz' % server.full_address
|
|
|
|
# choxie depends on towel-stuff, which depends on bacon.
|
|
server.xmlrpc.set_distributions([
|
|
{'name': 'choxie',
|
|
'version': '2.0.0.9',
|
|
'requires_dist': ['towel-stuff (0.1)'],
|
|
'url': archive_path},
|
|
{'name': 'towel-stuff',
|
|
'version': '0.1',
|
|
'requires_dist': ['bacon (>= 0.2)'],
|
|
'url': archive_path},
|
|
{'name': 'bacon',
|
|
'version': '0.2',
|
|
'requires_dist': [],
|
|
'url': archive_path},
|
|
])
|
|
|
|
# name, version, deps.
|
|
already_installed = [('bacon', '0.1', []),
|
|
('chicken', '1.1', ['bacon (0.1)'])]
|
|
output = install.get_infos(
|
|
'choxie', index=client,
|
|
installed=get_installed_dists(already_installed))
|
|
|
|
# we need bacon 0.2, but 0.1 is installed.
|
|
# So we expect to remove 0.1 and to install 0.2 instead.
|
|
installed, remove, conflict = self._get_results(output)
|
|
self.assertIn(('choxie', '2.0.0.9'), installed)
|
|
self.assertIn(('towel-stuff', '0.1'), installed)
|
|
self.assertIn(('bacon', '0.2'), installed)
|
|
self.assertIn(('bacon', '0.1'), remove)
|
|
self.assertIn(('chicken', '1.1'), conflict)
|
|
|
|
@unittest.skipIf(threading is None, 'needs threading')
|
|
@use_xmlrpc_server()
|
|
def test_installation_unexisting_project(self, server):
|
|
# Test that the isntalled raises an exception if the project does not
|
|
# exists.
|
|
client = self._get_client(server)
|
|
self.assertRaises(install.InstallationException,
|
|
install.get_infos,
|
|
'unexisting project', index=client)
|
|
|
|
def test_move_files(self):
|
|
# test that the files are really moved, and that the new path is
|
|
# returned.
|
|
path = self.mkdtemp()
|
|
newpath = self.mkdtemp()
|
|
files = [os.path.join(path, str(x)) for x in range(1, 20)]
|
|
for f in files:
|
|
open(f, 'ab+').close()
|
|
output = [o for o in install._move_files(files, newpath)]
|
|
|
|
# check that output return the list of old/new places
|
|
for file_ in files:
|
|
name = os.path.split(file_)[-1]
|
|
newloc = os.path.join(newpath, name)
|
|
self.assertIn((file_, newloc), output)
|
|
|
|
# remove the files
|
|
for f in [o[1] for o in output]: # o[1] is the new place
|
|
os.remove(f)
|
|
|
|
def test_update_infos(self):
|
|
tests = [[
|
|
{'foo': ['foobar', 'foo', 'baz'], 'baz': ['foo', 'foo']},
|
|
{'foo': ['additional_content', 'yeah'], 'baz': ['test', 'foo']},
|
|
{'foo': ['foobar', 'foo', 'baz', 'additional_content', 'yeah'],
|
|
'baz': ['foo', 'foo', 'test', 'foo']},
|
|
]]
|
|
|
|
for dict1, dict2, expect in tests:
|
|
install._update_infos(dict1, dict2)
|
|
for key in expect:
|
|
self.assertEqual(expect[key], dict1[key])
|
|
|
|
def test_install_dists_rollback(self):
|
|
# if one of the distribution installation fails, call uninstall on all
|
|
# installed distributions.
|
|
|
|
old_install_dist = install._install_dist
|
|
old_uninstall = getattr(install, 'uninstall', None)
|
|
|
|
install._install_dist = MagicMock(return_value=[],
|
|
raise_exception=(False, True))
|
|
install.remove = MagicMock()
|
|
try:
|
|
d1 = ToInstallDist()
|
|
d2 = ToInstallDist()
|
|
path = self.mkdtemp()
|
|
self.assertRaises(Exception, install.install_dists, [d1, d2], path)
|
|
self.assertTrue(install._install_dist.called_with(d1, path))
|
|
self.assertTrue(install.remove.called)
|
|
finally:
|
|
install._install_dist = old_install_dist
|
|
install.remove = old_uninstall
|
|
|
|
def test_install_dists_success(self):
|
|
old_install_dist = install._install_dist
|
|
install._install_dist = MagicMock(return_value=[])
|
|
try:
|
|
# test that the install method is called on each distributions
|
|
d1 = ToInstallDist()
|
|
d2 = ToInstallDist()
|
|
|
|
# should call install
|
|
path = self.mkdtemp()
|
|
install.install_dists([d1, d2], path)
|
|
for dist in (d1, d2):
|
|
self.assertTrue(install._install_dist.called_with(dist, path))
|
|
finally:
|
|
install._install_dist = old_install_dist
|
|
|
|
def test_install_from_infos_conflict(self):
|
|
# assert conflicts raise an exception
|
|
self.assertRaises(install.InstallationConflict,
|
|
install.install_from_infos,
|
|
conflicts=[ToInstallDist()])
|
|
|
|
def test_install_from_infos_remove_success(self):
|
|
old_install_dists = install.install_dists
|
|
install.install_dists = lambda x, y=None: None
|
|
try:
|
|
dists = []
|
|
for i in range(2):
|
|
dists.append(ToInstallDist(files=True))
|
|
install.install_from_infos(remove=dists)
|
|
|
|
# assert that the files have been removed
|
|
for dist in dists:
|
|
for f in dist.list_installed_files():
|
|
self.assertFalse(os.path.exists(f))
|
|
finally:
|
|
install.install_dists = old_install_dists
|
|
|
|
def test_install_from_infos_remove_rollback(self):
|
|
old_install_dist = install._install_dist
|
|
old_uninstall = getattr(install, 'uninstall', None)
|
|
|
|
install._install_dist = MagicMock(return_value=[],
|
|
raise_exception=(False, True))
|
|
install.uninstall = MagicMock()
|
|
try:
|
|
# assert that if an error occurs, the removed files are restored.
|
|
remove = []
|
|
for i in range(2):
|
|
remove.append(ToInstallDist(files=True))
|
|
to_install = [ToInstallDist(), ToInstallDist()]
|
|
temp_dir = self.mkdtemp()
|
|
|
|
self.assertRaises(Exception, install.install_from_infos,
|
|
install_path=temp_dir, install=to_install,
|
|
remove=remove)
|
|
# assert that the files are in the same place
|
|
# assert that the files have been removed
|
|
for dist in remove:
|
|
for f in dist.list_installed_files():
|
|
self.assertTrue(os.path.exists(f))
|
|
dist._unlink_installed_files()
|
|
finally:
|
|
install._install_dist = old_install_dist
|
|
install.uninstall = old_uninstall
|
|
|
|
def test_install_from_infos_install_succes(self):
|
|
old_install_dist = install._install_dist
|
|
install._install_dist = MagicMock([])
|
|
try:
|
|
# assert that the distribution can be installed
|
|
install_path = "my_install_path"
|
|
to_install = [ToInstallDist(), ToInstallDist()]
|
|
|
|
install.install_from_infos(install_path, install=to_install)
|
|
for dist in to_install:
|
|
install._install_dist.called_with(install_path)
|
|
finally:
|
|
install._install_dist = old_install_dist
|
|
|
|
def test_install_permission_denied(self):
|
|
# if we don't have access to the installation path, we should abort
|
|
# immediately
|
|
project = os.path.join(os.path.dirname(__file__), 'package.tgz')
|
|
|
|
# when running from an uninstalled build, a warning is emitted and the
|
|
# installation is not attempted
|
|
if is_python_build():
|
|
self.assertFalse(install.install(project))
|
|
self.assertEqual(1, len(self.get_logs(logging.ERROR)))
|
|
return
|
|
|
|
install_path = self.mkdtemp()
|
|
old_get_path = install.get_path
|
|
install.get_path = lambda path: install_path
|
|
old_mod = os.stat(install_path).st_mode
|
|
os.chmod(install_path, 0)
|
|
try:
|
|
self.assertFalse(install.install(project))
|
|
finally:
|
|
os.chmod(install_path, old_mod)
|
|
install.get_path = old_get_path
|
|
|
|
|
|
def test_suite():
|
|
suite = unittest.TestSuite()
|
|
suite.addTest(unittest.makeSuite(TestInstall))
|
|
return suite
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(defaultTest='test_suite')
|