[Python-checkins] distutils2: Update installer tests.

Sun Jan 30 10:43:57 CET 2011

tarek.ziade pushed 30c453712740 to distutils2:
http://hg.python.org/distutils2/rev/30c453712740
changeset: 926:30c453712740
user: Alexis Metaireau <alexis at notmyidea.org>
date: Sat Jan 29 16:37:11 2011 +0100
summary:
 Update installer tests.
files:
 distutils2/index/xmlrpc.py
 distutils2/install.py
 distutils2/tests/pypi_server.py
 distutils2/tests/test_install.py

diff --git a/distutils2/index/xmlrpc.py b/distutils2/index/xmlrpc.py
--- a/distutils2/index/xmlrpc.py
+++ b/distutils2/index/xmlrpc.py
@@ -131,7 +131,7 @@
 
 Return a ReleaseInfo object, with metadata informations filled in.
 """
- # to be case-insensitive
+ # to be case-insensitive, get the informations from the XMLRPC API
 projects = [d['name'] for d in
 self.proxy.search({'name': project_name})
 if d['name'].lower() == project_name]
diff --git a/distutils2/install.py b/distutils2/install.py
--- a/distutils2/install.py
+++ b/distutils2/install.py
@@ -139,7 +139,7 @@
 
 # reverting
 for d in installed_dists:
- _uninstall(d)
+ uninstall(d)
 raise e
 return installed_files
 
@@ -226,8 +226,6 @@
 Conflict contains all the conflicting distributions, if there is a
 conflict.
 """
- from ipdb import set_trace
- set_trace()
 if not installed:
 logger.info('Reading installed distributions')
 installed = get_distributions(use_egg_info=True)
@@ -237,16 +235,20 @@
 predicate = get_version_predicate(requirements)
 found = False
 installed = list(installed)
+
+ # check that the project isnt already installed
 for installed_project in installed:
 # is it a compatible project ?
 if predicate.name.lower() != installed_project.name.lower():
 continue
 found = True
 logger.info('Found %s %s' % (installed_project.name,
- installed_project.metadata.version))
- if predicate.match(installed_project.metadata.version):
+ installed_project.version))
+
+ # if we already have something installed, check it matches the
+ # requirements
+ if predicate.match(installed_project.version):
 return infos
-
 break
 
 if not found:
@@ -260,7 +262,7 @@
 releases = index.get_releases(requirements)
 except (ReleaseNotFound, ProjectNotFound), e:
 raise InstallationException('Release not found: "%s"' % requirements)
-
+ 
 # Pick up a release, and try to get the dependency tree
 release = releases.get_last(requirements, prefer_final=prefer_final)
 
@@ -284,7 +286,7 @@
 # Get what the missing deps are
 dists = depgraph.missing[release]
 if dists:
- logger.info("missing dependencies found, installing them")
+ logger.info("missing dependencies found, retrieving metadata")
 # we have missing deps
 for dist in dists:
 _update_infos(infos, get_infos(dist, index, installed))
@@ -333,7 +335,7 @@
 install_path=install_path)
 
 except InstallationConflict, e:
- projects = ['%s %s' % (p.name, p.metadata.version) for p in e.args[0]]
+ projects = ['%s %s' % (p.name, p.version) for p in e.args[0]]
 logger.info('"%s" conflicts with "%s"' % (project, ','.join(projects)))
 
 
diff --git a/distutils2/tests/pypi_server.py b/distutils2/tests/pypi_server.py
--- a/distutils2/tests/pypi_server.py
+++ b/distutils2/tests/pypi_server.py
@@ -375,6 +375,7 @@
 
 def __init__(self, dists=[]):
 self._dists = dists
+ self._search_result = []
 
 def add_distributions(self, dists):
 for dist in dists:
diff --git a/distutils2/tests/test_install.py b/distutils2/tests/test_install.py
--- a/distutils2/tests/test_install.py
+++ b/distutils2/tests/test_install.py
@@ -29,27 +29,16 @@
 class ToInstallDist(object):
 """Distribution that will be installed"""
 
- def __init__(self, raise_error=False, files=False):
- self._raise_error = raise_error
+ 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):
 self._real_files.append(mkstemp())
 
- def install(self, *args):
- self.install_called = True
- self.install_called_with = args
- if self._raise_error:
- raise Exception('Oops !')
- return ['/path/to/foo', '/path/to/bar']
-
- def uninstall(self, **args):
- self.uninstall_called = True
-
 def get_installed_files(self, **args):
 if self._files:
 return [f[1] for f in self._real_files]
@@ -58,7 +47,49 @@
 return self.get_installed_files()
 
 
+class MagicMock(object):
+ 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 patch(parent, to_patch):
+ """monkey match a module"""
+ def wrapper(func):
+ print func
+ print dir(func)
+ old_func = getattr(parent, to_patch)
+ def wrapped(*args, **kwargs):
+ parent.__dict__[to_patch] = MagicMock()
+ try:
+ out = func(*args, **kwargs)
+ finally:
+ setattr(parent, to_patch, old_func)
+ return out
+ return wrapped
+ return wrapper
+
+
 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))
@@ -69,6 +100,12 @@
 def _get_client(self, server, *args, **kwargs):
 return Client(server.full_address, *args, **kwargs)
 
+ def _patch_run_install(self):
+ """Patch run install"""
+
+ def _unpatch_run_install(self):
+ """Unpatch run install for d2 and d1"""
+
 def _get_results(self, output):
 """return a list of results"""
 installed = [(o.name, '%s' % o.version) for o in output['install']]
@@ -150,6 +187,8 @@
 # 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',
@@ -164,7 +203,9 @@
 'requires_dist': [],
 'url': archive_path},
 ])
- already_installed = [('bacon', '0.1', []),
+
+ # 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))
@@ -221,23 +262,39 @@
 # if one of the distribution installation fails, call uninstall on all
 # installed distributions.
 
- d1 = ToInstallDist()
- d2 = ToInstallDist(raise_error=True)
- self.assertRaises(Exception, install.install_dists, [d1, d2])
- for dist in (d1, d2):
- self.assertTrue(dist.install_called)
- self.assertTrue(d1.uninstall_called)
- self.assertFalse(d2.uninstall_called)
+ 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:
+ 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.uninstall.called)
+ finally:
+ install._install_dist = old_install_dist
+ install.uninstall = old_uninstall
+
 
 def test_install_dists_success(self):
- # test that the install method is called on each of the distributions.
- d1 = ToInstallDist()
- d2 = ToInstallDist()
- install.install_dists([d1, d2])
- for dist in (d1, d2):
- self.assertTrue(dist.install_called)
- self.assertFalse(d1.uninstall_called)
- self.assertFalse(d2.uninstall_called)
+ old_install_dist = install._install_dist
+ install._install_dist = MagicMock(return_value=[])
+ try:
+ # test that the install method is called on each of the 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
@@ -262,29 +319,46 @@
 install.install_dists = old_install_dists
 
 def test_install_from_infos_remove_rollback(self):
- # assert that if an error occurs, the removed files are restored.
- remove = []
- for i in range(0,2):
- remove.append(ToInstallDist(files=True, raise_error=True))
- to_install = [ToInstallDist(raise_error=True),
- ToInstallDist()]
+ old_install_dist = install._install_dist
+ old_uninstall = getattr(install, 'uninstall', None)
 
- install.install_from_infos(remove=remove, install=to_install)
- # assert that the files are in the same place
- # assert that the files have been removed
- for dist in remove:
- for f in dist.get_installed_files():
- self.assertTrue(os.path.exists(f))
+ 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(0,2):
+ remove.append(ToInstallDist(files=True))
+ to_install = [ToInstallDist(), ToInstallDist()]
+
+ self.assertRaises(Exception, install.install_from_infos, 
+ remove=remove, install=to_install)
+ # assert that the files are in the same place
+ # assert that the files have been removed
+ for dist in remove:
+ for f in dist.get_installed_files():
+ self.assertTrue(os.path.exists(f))
+ finally:
+ install.install_dist = old_install_dist
+ install.uninstall = old_uninstall
+
 
 def test_install_from_infos_install_succes(self):
- # assert that the distribution can be installed
- install_path = "my_install_path"
- to_install = [ToInstallDist(), ToInstallDist()]
+ 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=to_install,
- install_path=install_path)
- for dist in to_install:
- self.assertEqual(dist.install_called_with, (install_path,))
+ install.install_from_infos(install=to_install,
+ install_path=install_path)
+ for dist in to_install:
+ install._install_dist.called_with(install_path)
+ finally:
+ install._install_dist = old_install_dist
+
 
 def test_suite():
 suite = unittest.TestSuite()
--
Repository URL: http://hg.python.org/distutils2
