Python code coverage for Lib/packaging/tests/test_install.py

#	count	content
1	n/a	"""Tests for the packaging.install module."""
2	n/a	import os
3	n/a	import logging
4	n/a	from tempfile import mkstemp
5	n/a	from sysconfig import is_python_build
6	n/a
7	n/a	from packaging import install
8	n/a	from packaging.pypi.xmlrpc import Client
9	n/a	from packaging.metadata import Metadata
10	n/a	from packaging.tests.support import (LoggingCatcher, TempdirManager, unittest,
11	n/a	fake_dec)
12	n/a	try:
13	n/a	import threading
14	n/a	from packaging.tests.pypi_server import use_xmlrpc_server
15	n/a	except ImportError:
16	n/a	threading = None
17	n/a	use_xmlrpc_server = fake_dec
18	n/a
19	n/a
20	n/a	class InstalledDist:
21	n/a	"""Distribution object, represent distributions currently installed on the
22	n/a	system"""
23	n/a	def __init__(self, name, version, deps):
24	n/a	self.metadata = Metadata()
25	n/a	self.name = name
26	n/a	self.version = version
27	n/a	self.metadata['Name'] = name
28	n/a	self.metadata['Version'] = version
29	n/a	self.metadata['Requires-Dist'] = deps
30	n/a
31	n/a	def __repr__(self):
32	n/a	return '<InstalledDist %r>' % self.metadata['Name']
33	n/a
34	n/a
35	n/a	class ToInstallDist:
36	n/a	"""Distribution that will be installed"""
37	n/a
38	n/a	def __init__(self, files=False):
39	n/a	self._files = files
40	n/a	self.install_called = False
41	n/a	self.install_called_with = {}
42	n/a	self.uninstall_called = False
43	n/a	self._real_files = []
44	n/a	self.name = "fake"
45	n/a	self.version = "fake"
46	n/a	if files:
47	n/a	for f in range(0, 3):
48	n/a	fp, fn = mkstemp()
49	n/a	os.close(fp)
50	n/a	self._real_files.append(fn)
51	n/a
52	n/a	def _unlink_installed_files(self):
53	n/a	if self._files:
54	n/a	for fn in self._real_files:
55	n/a	os.unlink(fn)
56	n/a
57	n/a	def list_installed_files(self, **args):
58	n/a	if self._files:
59	n/a	return self._real_files
60	n/a
61	n/a	def get_install(self, **args):
62	n/a	return self.list_installed_files()
63	n/a
64	n/a
65	n/a	class MagicMock:
66	n/a	def __init__(self, return_value=None, raise_exception=False):
67	n/a	self.called = False
68	n/a	self._times_called = 0
69	n/a	self._called_with = []
70	n/a	self._return_value = return_value
71	n/a	self._raise = raise_exception
72	n/a
73	n/a	def __call__(self, args, *kwargs):
74	n/a	self.called = True
75	n/a	self._times_called = self._times_called + 1
76	n/a	self._called_with.append((args, kwargs))
77	n/a	iterable = hasattr(self._raise, '__iter__')
78	n/a	if self._raise:
79	n/a	if ((not iterable and self._raise)
80	n/a	or self._raise[self._times_called - 1]):
81	n/a	raise Exception
82	n/a	return self._return_value
83	n/a
84	n/a	def called_with(self, args, *kwargs):
85	n/a	return (args, kwargs) in self._called_with
86	n/a
87	n/a
88	n/a	def get_installed_dists(dists):
89	n/a	"""Return a list of fake installed dists.
90	n/a	The list is name, version, deps"""
91	n/a	objects = []
92	n/a	for name, version, deps in dists:
93	n/a	objects.append(InstalledDist(name, version, deps))
94	n/a	return objects
95	n/a
96	n/a
97	n/a	class TestInstall(LoggingCatcher, TempdirManager, unittest.TestCase):
98	n/a	def _get_client(self, server, args, *kwargs):
99	n/a	return Client(server.full_address, args, *kwargs)
100	n/a
101	n/a	def _get_results(self, output):
102	n/a	"""return a list of results"""
103	n/a	installed = [(o.name, str(o.version)) for o in output['install']]
104	n/a	remove = [(o.name, str(o.version)) for o in output['remove']]
105	n/a	conflict = [(o.name, str(o.version)) for o in output['conflict']]
106	n/a	return installed, remove, conflict
107	n/a
108	n/a	@unittest.skipIf(threading is None, 'needs threading')
109	n/a	@use_xmlrpc_server()
110	n/a	def test_existing_deps(self, server):
111	n/a	# Test that the installer get the dependencies from the metadatas
112	n/a	# and ask the index for this dependencies.
113	n/a	# In this test case, we have choxie that is dependent from towel-stuff
114	n/a	# 0.1, which is in-turn dependent on bacon <= 0.2:
115	n/a	# choxie -> towel-stuff -> bacon.
116	n/a	# Each release metadata is not provided in metadata 1.2.
117	n/a	client = self._get_client(server)
118	n/a	archive_path = '%s/distribution.tar.gz' % server.full_address
119	n/a	server.xmlrpc.set_distributions([
120	n/a	{'name': 'choxie',
121	n/a	'version': '2.0.0.9',
122	n/a	'requires_dist': ['towel-stuff (0.1)'],
123	n/a	'url': archive_path},
124	n/a	{'name': 'towel-stuff',
125	n/a	'version': '0.1',
126	n/a	'requires_dist': ['bacon (<= 0.2)'],
127	n/a	'url': archive_path},
128	n/a	{'name': 'bacon',
129	n/a	'version': '0.1',
130	n/a	'requires_dist': [],
131	n/a	'url': archive_path},
132	n/a	])
133	n/a	installed = get_installed_dists([('bacon', '0.1', [])])
134	n/a	output = install.get_infos("choxie", index=client,
135	n/a	installed=installed)
136	n/a
137	n/a	# we don't have installed bacon as it's already installed system-wide
138	n/a	self.assertEqual(0, len(output['remove']))
139	n/a	self.assertEqual(2, len(output['install']))
140	n/a	readable_output = [(o.name, str(o.version))
141	n/a	for o in output['install']]
142	n/a	self.assertIn(('towel-stuff', '0.1'), readable_output)
143	n/a	self.assertIn(('choxie', '2.0.0.9'), readable_output)
144	n/a
145	n/a	@unittest.skipIf(threading is None, 'needs threading')
146	n/a	@use_xmlrpc_server()
147	n/a	def test_upgrade_existing_deps(self, server):
148	n/a	client = self._get_client(server)
149	n/a	archive_path = '%s/distribution.tar.gz' % server.full_address
150	n/a	server.xmlrpc.set_distributions([
151	n/a	{'name': 'choxie',
152	n/a	'version': '2.0.0.9',
153	n/a	'requires_dist': ['towel-stuff (0.1)'],
154	n/a	'url': archive_path},
155	n/a	{'name': 'towel-stuff',
156	n/a	'version': '0.1',
157	n/a	'requires_dist': ['bacon (>= 0.2)'],
158	n/a	'url': archive_path},
159	n/a	{'name': 'bacon',
160	n/a	'version': '0.2',
161	n/a	'requires_dist': [],
162	n/a	'url': archive_path},
163	n/a	])
164	n/a
165	n/a	output = install.get_infos("choxie", index=client,
166	n/a	installed=get_installed_dists([('bacon', '0.1', [])]))
167	n/a	installed = [(o.name, str(o.version)) for o in output['install']]
168	n/a
169	n/a	# we need bacon 0.2, but 0.1 is installed.
170	n/a	# So we expect to remove 0.1 and to install 0.2 instead.
171	n/a	remove = [(o.name, str(o.version)) for o in output['remove']]
172	n/a	self.assertIn(('choxie', '2.0.0.9'), installed)
173	n/a	self.assertIn(('towel-stuff', '0.1'), installed)
174	n/a	self.assertIn(('bacon', '0.2'), installed)
175	n/a	self.assertIn(('bacon', '0.1'), remove)
176	n/a	self.assertEqual(0, len(output['conflict']))
177	n/a
178	n/a	@unittest.skipIf(threading is None, 'needs threading')
179	n/a	@use_xmlrpc_server()
180	n/a	def test_conflicts(self, server):
181	n/a	# Tests that conflicts are detected
182	n/a	client = self._get_client(server)
183	n/a	archive_path = '%s/distribution.tar.gz' % server.full_address
184	n/a
185	n/a	# choxie depends on towel-stuff, which depends on bacon.
186	n/a	server.xmlrpc.set_distributions([
187	n/a	{'name': 'choxie',
188	n/a	'version': '2.0.0.9',
189	n/a	'requires_dist': ['towel-stuff (0.1)'],
190	n/a	'url': archive_path},
191	n/a	{'name': 'towel-stuff',
192	n/a	'version': '0.1',
193	n/a	'requires_dist': ['bacon (>= 0.2)'],
194	n/a	'url': archive_path},
195	n/a	{'name': 'bacon',
196	n/a	'version': '0.2',
197	n/a	'requires_dist': [],
198	n/a	'url': archive_path},
199	n/a	])
200	n/a
201	n/a	# name, version, deps.
202	n/a	already_installed = [('bacon', '0.1', []),
203	n/a	('chicken', '1.1', ['bacon (0.1)'])]
204	n/a	output = install.get_infos(
205	n/a	'choxie', index=client,
206	n/a	installed=get_installed_dists(already_installed))
207	n/a
208	n/a	# we need bacon 0.2, but 0.1 is installed.
209	n/a	# So we expect to remove 0.1 and to install 0.2 instead.
210	n/a	installed, remove, conflict = self._get_results(output)
211	n/a	self.assertIn(('choxie', '2.0.0.9'), installed)
212	n/a	self.assertIn(('towel-stuff', '0.1'), installed)
213	n/a	self.assertIn(('bacon', '0.2'), installed)
214	n/a	self.assertIn(('bacon', '0.1'), remove)
215	n/a	self.assertIn(('chicken', '1.1'), conflict)
216	n/a
217	n/a	@unittest.skipIf(threading is None, 'needs threading')
218	n/a	@use_xmlrpc_server()
219	n/a	def test_installation_unexisting_project(self, server):
220	n/a	# Test that the isntalled raises an exception if the project does not
221	n/a	# exists.
222	n/a	client = self._get_client(server)
223	n/a	self.assertRaises(install.InstallationException,
224	n/a	install.get_infos,
225	n/a	'unexisting project', index=client)
226	n/a
227	n/a	def test_move_files(self):
228	n/a	# test that the files are really moved, and that the new path is
229	n/a	# returned.
230	n/a	path = self.mkdtemp()
231	n/a	newpath = self.mkdtemp()
232	n/a	files = [os.path.join(path, str(x)) for x in range(1, 20)]
233	n/a	for f in files:
234	n/a	open(f, 'ab+').close()
235	n/a	output = [o for o in install._move_files(files, newpath)]
236	n/a
237	n/a	# check that output return the list of old/new places
238	n/a	for file_ in files:
239	n/a	name = os.path.split(file_)[-1]
240	n/a	newloc = os.path.join(newpath, name)
241	n/a	self.assertIn((file_, newloc), output)
242	n/a
243	n/a	# remove the files
244	n/a	for f in [o[1] for o in output]: # o[1] is the new place
245	n/a	os.remove(f)
246	n/a
247	n/a	def test_update_infos(self):
248	n/a	tests = [[
249	n/a	{'foo': ['foobar', 'foo', 'baz'], 'baz': ['foo', 'foo']},
250	n/a	{'foo': ['additional_content', 'yeah'], 'baz': ['test', 'foo']},
251	n/a	{'foo': ['foobar', 'foo', 'baz', 'additional_content', 'yeah'],
252	n/a	'baz': ['foo', 'foo', 'test', 'foo']},
253	n/a	]]
254	n/a
255	n/a	for dict1, dict2, expect in tests:
256	n/a	install._update_infos(dict1, dict2)
257	n/a	for key in expect:
258	n/a	self.assertEqual(expect[key], dict1[key])
259	n/a
260	n/a	def test_install_dists_rollback(self):
261	n/a	# if one of the distribution installation fails, call uninstall on all
262	n/a	# installed distributions.
263	n/a
264	n/a	old_install_dist = install._install_dist
265	n/a	old_uninstall = getattr(install, 'uninstall', None)
266	n/a
267	n/a	install._install_dist = MagicMock(return_value=[],
268	n/a	raise_exception=(False, True))
269	n/a	install.remove = MagicMock()
270	n/a	try:
271	n/a	d1 = ToInstallDist()
272	n/a	d2 = ToInstallDist()
273	n/a	path = self.mkdtemp()
274	n/a	self.assertRaises(Exception, install.install_dists, [d1, d2], path)
275	n/a	self.assertTrue(install._install_dist.called_with(d1, path))
276	n/a	self.assertTrue(install.remove.called)
277	n/a	finally:
278	n/a	install._install_dist = old_install_dist
279	n/a	install.remove = old_uninstall
280	n/a
281	n/a	def test_install_dists_success(self):
282	n/a	old_install_dist = install._install_dist
283	n/a	install._install_dist = MagicMock(return_value=[])
284	n/a	try:
285	n/a	# test that the install method is called on each distributions
286	n/a	d1 = ToInstallDist()
287	n/a	d2 = ToInstallDist()
288	n/a
289	n/a	# should call install
290	n/a	path = self.mkdtemp()
291	n/a	install.install_dists([d1, d2], path)
292	n/a	for dist in (d1, d2):
293	n/a	self.assertTrue(install._install_dist.called_with(dist, path))
294	n/a	finally:
295	n/a	install._install_dist = old_install_dist
296	n/a
297	n/a	def test_install_from_infos_conflict(self):
298	n/a	# assert conflicts raise an exception
299	n/a	self.assertRaises(install.InstallationConflict,
300	n/a	install.install_from_infos,
301	n/a	conflicts=[ToInstallDist()])
302	n/a
303	n/a	def test_install_from_infos_remove_success(self):
304	n/a	old_install_dists = install.install_dists
305	n/a	install.install_dists = lambda x, y=None: None
306	n/a	try:
307	n/a	dists = []
308	n/a	for i in range(2):
309	n/a	dists.append(ToInstallDist(files=True))
310	n/a	install.install_from_infos(remove=dists)
311	n/a
312	n/a	# assert that the files have been removed
313	n/a	for dist in dists:
314	n/a	for f in dist.list_installed_files():
315	n/a	self.assertFalse(os.path.exists(f))
316	n/a	finally:
317	n/a	install.install_dists = old_install_dists
318	n/a
319	n/a	def test_install_from_infos_remove_rollback(self):
320	n/a	old_install_dist = install._install_dist
321	n/a	old_uninstall = getattr(install, 'uninstall', None)
322	n/a
323	n/a	install._install_dist = MagicMock(return_value=[],
324	n/a	raise_exception=(False, True))
325	n/a	install.uninstall = MagicMock()
326	n/a	try:
327	n/a	# assert that if an error occurs, the removed files are restored.
328	n/a	remove = []
329	n/a	for i in range(2):
330	n/a	remove.append(ToInstallDist(files=True))
331	n/a	to_install = [ToInstallDist(), ToInstallDist()]
332	n/a	temp_dir = self.mkdtemp()
333	n/a
334	n/a	self.assertRaises(Exception, install.install_from_infos,
335	n/a	install_path=temp_dir, install=to_install,
336	n/a	remove=remove)
337	n/a	# assert that the files are in the same place
338	n/a	# assert that the files have been removed
339	n/a	for dist in remove:
340	n/a	for f in dist.list_installed_files():
341	n/a	self.assertTrue(os.path.exists(f))
342	n/a	dist._unlink_installed_files()
343	n/a	finally:
344	n/a	install._install_dist = old_install_dist
345	n/a	install.uninstall = old_uninstall
346	n/a
347	n/a	def test_install_from_infos_install_succes(self):
348	n/a	old_install_dist = install._install_dist
349	n/a	install._install_dist = MagicMock([])
350	n/a	try:
351	n/a	# assert that the distribution can be installed
352	n/a	install_path = "my_install_path"
353	n/a	to_install = [ToInstallDist(), ToInstallDist()]
354	n/a
355	n/a	install.install_from_infos(install_path, install=to_install)
356	n/a	for dist in to_install:
357	n/a	install._install_dist.called_with(install_path)
358	n/a	finally:
359	n/a	install._install_dist = old_install_dist
360	n/a
361	n/a	def test_install_permission_denied(self):
362	n/a	# if we don't have access to the installation path, we should abort
363	n/a	# immediately
364	n/a	project = os.path.join(os.path.dirname(__file__), 'package.tgz')
365	n/a
366	n/a	# when running from an uninstalled build, a warning is emitted and the
367	n/a	# installation is not attempted
368	n/a	if is_python_build():
369	n/a	self.assertFalse(install.install(project))
370	n/a	self.assertEqual(1, len(self.get_logs(logging.ERROR)))
371	n/a	return
372	n/a
373	n/a	install_path = self.mkdtemp()
374	n/a	old_get_path = install.get_path
375	n/a	install.get_path = lambda path: install_path
376	n/a	old_mod = os.stat(install_path).st_mode
377	n/a	os.chmod(install_path, 0)
378	n/a	try:
379	n/a	self.assertFalse(install.install(project))
380	n/a	finally:
381	n/a	os.chmod(install_path, old_mod)
382	n/a	install.get_path = old_get_path
383	n/a
384	n/a
385	n/a	def test_suite():
386	n/a	suite = unittest.TestSuite()
387	n/a	suite.addTest(unittest.makeSuite(TestInstall))
388	n/a	return suite
389	n/a
390	n/a	if __name__ == '__main__':
391	n/a	unittest.main(defaultTest='test_suite')