Merged revisions 70886,70888-70892 via svnmerge from

svn+ssh://pythondev@svn.python.org/python/trunk

........
  r70886 | tarek.ziade | 2009-03-31 15:50:59 -0500 (Tue, 31 Mar 2009) | 1 line

  added tests for the clean command
........
  r70888 | tarek.ziade | 2009-03-31 15:53:13 -0500 (Tue, 31 Mar 2009) | 1 line

  more tests for the register command
........
  r70889 | tarek.ziade | 2009-03-31 15:53:55 -0500 (Tue, 31 Mar 2009) | 1 line

  more tests for the upload command
........
  r70890 | tarek.ziade | 2009-03-31 15:54:38 -0500 (Tue, 31 Mar 2009) | 1 line

  added test to the install_data command
........
  r70891 | tarek.ziade | 2009-03-31 15:55:21 -0500 (Tue, 31 Mar 2009) | 1 line

  added tests to the install_headers command
........
  r70892 | tarek.ziade | 2009-03-31 15:56:11 -0500 (Tue, 31 Mar 2009) | 1 line

  making sdist and config test silents
........

cmd.py

@@ -333,7 +333,7 @@ class Command:
     # -- External world manipulation -----------------------------------
 
     def warn(self, msg):
-        sys.stderr.write("warning: %s: %s\n" % (self.get_command_name(), msg))
+        log.warn("warning: %s: %s\n" % (self.get_command_name(), msg))
 
     def execute(self, func, args, msg=None, level=1):
         util.execute(func, args, msg, dry_run=self.dry_run)

command/install_data.py

@@ -31,7 +31,6 @@ class install_data(Command):
         self.outfiles = []
         self.root = None
         self.force = 0
-
         self.data_files = self.distribution.data_files
         self.warn_dir = 1
 

command/install_headers.py

@@ -8,6 +8,7 @@ __revision__ = "$Id$"
 from distutils.core import Command
 
 
+# XXX force is never used
 class install_headers(Command):
 
     description = "install C/C++ header files"

command/register.py

@@ -92,15 +92,14 @@ class register(PyPIRCCommand):
         '''
         url = self.repository+'?:action=list_classifiers'
         response = urllib.request.urlopen(url)
-        print(response.read())
+        log.info(response.read())
 
     def verify_metadata(self):
         ''' Send the metadata to the package index server to be checked.
         '''
         # send the info to the server and report the result
         (code, result) = self.post_to_server(self.build_post_data('verify'))
-        print('Server response (%s): %s'%(code, result))
-
+        log.info('Server response (%s): %s' % (code, result))
 
     def send_metadata(self):
         ''' Send the metadata to the package index server.
@@ -211,17 +210,18 @@ Your selection [default 1]: ''', log.INFO)
                 data['email'] = input('   EMail: ')
             code, result = self.post_to_server(data)
             if code != 200:
-                print('Server response (%s): %s'%(code, result))
+                log.info('Server response (%s): %s' % (code, result))
             else:
-                print('You will receive an email shortly.')
-                print('Follow the instructions in it to complete registration.')
+                log.info('You will receive an email shortly.')
+                log.info(('Follow the instructions in it to '
+                          'complete registration.'))
         elif choice == '3':
             data = {':action': 'password_reset'}
             data['email'] = ''
             while not data['email']:
                 data['email'] = input('Your email address: ')
             code, result = self.post_to_server(data)
-            print('Server response (%s): %s'%(code, result))
+            log.info('Server response (%s): %s' % (code, result))
 
     def build_post_data(self, action):
         # figure the data to send - the metadata plus some additional
@@ -254,8 +254,10 @@ Your selection [default 1]: ''', log.INFO)
     def post_to_server(self, data, auth=None):
         ''' Post a query to the server, and return a string response.
         '''
-        self.announce('Registering %s to %s' % (data['name'],
-                                                self.repository), log.INFO)
+        if 'name' in data:
+            self.announce('Registering %s to %s' % (data['name'],
+                                                    self.repository),
+                                                    log.INFO)
         # Build up the MIME payload for the urllib2 POST data
         boundary = '--------------GHSKFJDLGDS7543FJKLFHRE75642756743254'
         sep_boundary = '\n--' + boundary
@@ -302,5 +304,6 @@ Your selection [default 1]: ''', log.INFO)
                 data = result.read()
             result = 200, 'OK'
         if self.show_response:
-            print('-'*75, data, '-'*75)
+            dashes = '-' * 75
+            self.announce('%s%s%s' % (dashes, data, dashes))
         return result

command/upload.py

@@ -194,4 +194,4 @@ class upload(PyPIRCCommand):
             self.announce('Upload failed (%s): %s' % (r.status, r.reason),
                           log.ERROR)
         if self.show_response:
-            print('-'*75, r.read(), '-'*75)
+            self.announce('-'*75, r.read(), '-'*75)

tests/support.py

@@ -4,7 +4,7 @@ import shutil
 import tempfile
 
 from distutils import log
-
+from distutils.core import Distribution
 
 class LoggingSilencer(object):
 
@@ -42,7 +42,7 @@ class TempdirManager(object):
         self.tempdirs.append(d)
         return d
 
-    def write_file(self, path, content):
+    def write_file(self, path, content='xxx'):
         """Writes a file in the given path.
 
 
@@ -56,6 +56,23 @@ class TempdirManager(object):
         finally:
             f.close()
 
+    def create_dist(self, pkg_name='foo', **kw):
+        """Will generate a test environment.
+
+        This function creates:
+         - a Distribution instance using keywords
+         - a temporary directory with a package structure
+
+        It returns the package directory and the distribution
+        instance.
+        """
+        tmp_dir = self.mkdtemp()
+        pkg_dir = os.path.join(tmp_dir, pkg_name)
+        os.mkdir(pkg_dir)
+        dist = Distribution(attrs=kw)
+
+        return pkg_dir, dist
+
 class DummyCommand:
     """Class to store options for retrieval via set_undefined_options()."""
 

tests/test_clean.py

+"""Tests for distutils.command.clean."""
+import sys
+import os
+import unittest
+import getpass
+
+from distutils.command.clean import clean
+from distutils.tests import support
+
+class cleanTestCase(support.TempdirManager,
+                    unittest.TestCase):
+
+    def test_simple_run(self):
+        pkg_dir, dist = self.create_dist()
+        cmd = clean(dist)
+
+        # let's add some elements clean should remove
+        dirs = [(d, os.path.join(pkg_dir, d))
+                for d in ('build_temp', 'build_lib', 'bdist_base',
+                'build_scripts', 'build_base')]
+
+        for name, path in dirs:
+            os.mkdir(path)
+            setattr(cmd, name, path)
+            if name == 'build_base':
+                continue
+            for f in ('one', 'two', 'three'):
+                self.write_file(os.path.join(path, f))
+
+        # let's run the command
+        cmd.all = 1
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # make sure the files where removed
+        for name, path in dirs:
+            self.assert_(not os.path.exists(path),
+                         '%s was not removed' % path)
+
+        # let's run the command again (should spit warnings but suceed)
+        cmd.all = 1
+        cmd.ensure_finalized()
+        cmd.run()
+
+def test_suite():
+    return unittest.makeSuite(cleanTestCase)
+
+if __name__ == "__main__":
+    unittest.main(defaultTest="test_suite")

tests/test_config.py

@@ -46,7 +46,9 @@ password:xxx
 """
 
 
-class PyPIRCCommandTestCase(support.TempdirManager, unittest.TestCase):
+class PyPIRCCommandTestCase(support.TempdirManager,
+                            support.LoggingSilencer,
+                            unittest.TestCase):
 
     def setUp(self):
         """Patches the environment."""

tests/test_install_data.py

+"""Tests for distutils.command.install_data."""
+import sys
+import os
+import unittest
+import getpass
+
+from distutils.command.install_data import install_data
+from distutils.tests import support
+
+class InstallDataTestCase(support.TempdirManager,
+                          support.LoggingSilencer,
+                          unittest.TestCase):
+
+    def test_simple_run(self):
+        pkg_dir, dist = self.create_dist()
+        cmd = install_data(dist)
+        cmd.install_dir = inst = os.path.join(pkg_dir, 'inst')
+
+        # data_files can contain
+        #  - simple files
+        #  - a tuple with a path, and a list of file
+        one = os.path.join(pkg_dir, 'one')
+        self.write_file(one, 'xxx')
+        inst2 = os.path.join(pkg_dir, 'inst2')
+        two = os.path.join(pkg_dir, 'two')
+        self.write_file(two, 'xxx')
+
+        cmd.data_files = [one, (inst2, [two])]
+        self.assertEquals(cmd.get_inputs(), [one, (inst2, [two])])
+
+        # let's run the command
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # let's check the result
+        self.assertEquals(len(cmd.get_outputs()), 2)
+        rtwo = os.path.split(two)[-1]
+        self.assert_(os.path.exists(os.path.join(inst2, rtwo)))
+        rone = os.path.split(one)[-1]
+        self.assert_(os.path.exists(os.path.join(inst, rone)))
+        cmd.outfiles = []
+
+        # let's try with warn_dir one
+        cmd.warn_dir = 1
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # let's check the result
+        self.assertEquals(len(cmd.get_outputs()), 2)
+        self.assert_(os.path.exists(os.path.join(inst2, rtwo)))
+        self.assert_(os.path.exists(os.path.join(inst, rone)))
+        cmd.outfiles = []
+
+        # now using root and empty dir
+        cmd.root = os.path.join(pkg_dir, 'root')
+        inst3 = os.path.join(cmd.install_dir, 'inst3')
+        inst4 = os.path.join(pkg_dir, 'inst4')
+        three = os.path.join(cmd.install_dir, 'three')
+        self.write_file(three, 'xx')
+        cmd.data_files = [one, (inst2, [two]),
+                          ('inst3', [three]),
+                          (inst4, [])]
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # let's check the result
+        self.assertEquals(len(cmd.get_outputs()), 4)
+        self.assert_(os.path.exists(os.path.join(inst2, rtwo)))
+        self.assert_(os.path.exists(os.path.join(inst, rone)))
+
+def test_suite():
+    return unittest.makeSuite(InstallDataTestCase)
+
+if __name__ == "__main__":
+    unittest.main(defaultTest="test_suite")

tests/test_install_headers.py

+"""Tests for distutils.command.install_headers."""
+import sys
+import os
+import unittest
+import getpass
+
+from distutils.command.install_headers import install_headers
+from distutils.tests import support
+
+class InstallHeadersTestCase(support.TempdirManager,
+                             support.LoggingSilencer,
+                             unittest.TestCase):
+
+    def test_simple_run(self):
+        # we have two headers
+        header_list = self.mkdtemp()
+        header1 = os.path.join(header_list, 'header1')
+        header2 = os.path.join(header_list, 'header2')
+        self.write_file(header1)
+        self.write_file(header2)
+        headers = [header1, header2]
+
+        pkg_dir, dist = self.create_dist(headers=headers)
+        cmd = install_headers(dist)
+        self.assertEquals(cmd.get_inputs(), headers)
+
+        # let's run the command
+        cmd.install_dir = os.path.join(pkg_dir, 'inst')
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # let's check the results
+        self.assertEquals(len(cmd.get_outputs()), 2)
+
+def test_suite():
+    return unittest.makeSuite(InstallHeadersTestCase)
+
+if __name__ == "__main__":
+    unittest.main(defaultTest="test_suite")

tests/test_register.py

@@ -3,7 +3,9 @@ import sys
 import os
 import unittest
 import getpass
+import urllib
 
+from distutils.command import register as register_module
 from distutils.command.register import register
 from distutils.core import Distribution
 
@@ -42,18 +44,20 @@ class Inputs(object):
         finally:
             self.index += 1
 
-class FakeServer(object):
+class FakeOpener(object):
     """Fakes a PyPI server"""
     def __init__(self):
-        self.calls = []
+        self.reqs = []
 
     def __call__(self, *args):
-        # we want to compare them, so let's store
-        # something comparable
-        els = list(args[0].items())
-        els.sort()
-        self.calls.append(tuple(els))
-        return 200, 'OK'
+        return self
+
+    def open(self, req):
+        self.reqs.append(req)
+        return self
+
+    def read(self):
+        return 'xxx'
 
 class registerTestCase(PyPIRCCommandTestCase):
 
@@ -64,24 +68,27 @@ class registerTestCase(PyPIRCCommandTestCase):
         def _getpass(prompt):
             return 'password'
         getpass.getpass = _getpass
+        self.old_opener = urllib.request.build_opener
+        self.conn = urllib.request.build_opener = FakeOpener()
 
     def tearDown(self):
         getpass.getpass = self._old_getpass
+        urllib.request.build_opener = self.old_opener
         PyPIRCCommandTestCase.tearDown(self)
 
+    def _get_cmd(self):
+        metadata = {'url': 'xxx', 'author': 'xxx',
+                    'author_email': 'xxx',
+                    'name': 'xxx', 'version': 'xxx'}
+        pkg_info, dist = self.create_dist(**metadata)
+        return register(dist)
+
     def test_create_pypirc(self):
         # this test makes sure a .pypirc file
         # is created when requested.
 
-        # let's create a fake distribution
-        # and a register instance
-        dist = Distribution()
-        dist.metadata.url = 'xxx'
-        dist.metadata.author = 'xxx'
-        dist.metadata.author_email = 'xxx'
-        dist.metadata.name = 'xxx'
-        dist.metadata.version =  'xxx'
-        cmd = register(dist)
+        # let's create a register instance
+        cmd = self._get_cmd()
 
         # we shouldn't have a .pypirc file yet
         self.assert_(not os.path.exists(self.rc))
@@ -95,13 +102,12 @@ class registerTestCase(PyPIRCCommandTestCase):
         # Password : 'password'
         # Save your login (y/N)? : 'y'
         inputs = Inputs('1', 'tarek', 'y')
-        from distutils.command import register as register_module
         register_module.input = inputs.__call__
-
-        cmd.post_to_server = pypi_server = FakeServer()
-
         # let's run the command
-        cmd.run()
+        try:
+            cmd.run()
+        finally:
+            del register_module.input
 
         # we should have a brand new .pypirc file
         self.assert_(os.path.exists(self.rc))
@@ -115,32 +121,68 @@ class registerTestCase(PyPIRCCommandTestCase):
         # if we run the command again
         def _no_way(prompt=''):
             raise AssertionError(prompt)
-        register_module.raw_input = _no_way
+        register_module.input = _no_way
 
+        cmd.show_response = 1
         cmd.run()
 
         # let's see what the server received : we should
         # have 2 similar requests
-        self.assert_(len(pypi_server.calls), 2)
-        self.assert_(pypi_server.calls[0], pypi_server.calls[1])
-
-    def test_password_not_in_file(self):
+        self.assert_(self.conn.reqs, 2)
+        req1 = dict(self.conn.reqs[0].headers)
+        req2 = dict(self.conn.reqs[1].headers)
 
-        f = open(self.rc, 'w')
-        f.write(PYPIRC_NOPASSWORD)
-        f.close()
+        self.assertEquals(req1['Content-length'], '1374')
+        self.assertEquals(req2['Content-length'], '1374')
+        self.assert_((b'xxx') in self.conn.reqs[1].data)
 
-        dist = Distribution()
-        cmd = register(dist)
-        cmd.post_to_server = FakeServer()
+    def test_password_not_in_file(self):
 
+        self.write_file(self.rc, PYPIRC_NOPASSWORD)
+        cmd = self._get_cmd()
         cmd._set_config()
         cmd.finalize_options()
         cmd.send_metadata()
 
         # dist.password should be set
         # therefore used afterwards by other commands
-        self.assertEquals(dist.password, 'password')
+        self.assertEquals(cmd.distribution.password, 'password')
+
+    def test_registering(self):
+        # this test runs choice 2
+        cmd = self._get_cmd()
+        inputs = Inputs('2', 'tarek', 'tarek@ziade.org')
+        register_module.input = inputs.__call__
+        try:
+            # let's run the command
+            cmd.run()
+        finally:
+            del register_module.input
+
+        # we should have send a request
+        self.assert_(self.conn.reqs, 1)
+        req = self.conn.reqs[0]
+        headers = dict(req.headers)
+        self.assertEquals(headers['Content-length'], '608')
+        self.assert_((b'tarek') in req.data)
+
+    def test_password_reset(self):
+        # this test runs choice 3
+        cmd = self._get_cmd()
+        inputs = Inputs('3', 'tarek@ziade.org')
+        register_module.input = inputs.__call__
+        try:
+            # let's run the command
+            cmd.run()
+        finally:
+            del register_module.input
+
+        # we should have send a request
+        self.assert_(self.conn.reqs, 1)
+        req = self.conn.reqs[0]
+        headers = dict(req.headers)
+        self.assertEquals(headers['Content-length'], '290')
+        self.assert_((b'tarek') in req.data)
 
 def test_suite():
     return unittest.makeSuite(registerTestCase)

tests/test_sdist.py

@@ -34,7 +34,7 @@ somecode%(sep)sdoc.dat
 somecode%(sep)sdoc.txt
 """
 
-class sdistTestCase(support.LoggingSilencer, PyPIRCCommandTestCase):
+class sdistTestCase(PyPIRCCommandTestCase):
 
     def setUp(self):
         # PyPIRCCommandTestCase creates a temp dir already

tests/test_upload.py

@@ -2,6 +2,7 @@
 import sys
 import os
 import unittest
+import http.client as httpclient
 
 from distutils.command.upload import upload
 from distutils.core import Distribution
@@ -18,17 +19,52 @@ index-servers =
 [server1]
 username:me
 """
+class Response(object):
+    def __init__(self, status=200, reason='OK'):
+        self.status = status
+        self.reason = reason
 
+class FakeConnection(object):
+
+    def __init__(self):
+        self.requests = []
+        self.headers = []
+        self.body = ''
+
+    def __call__(self, netloc):
+        return self
+
+    def connect(self):
+        pass
+    endheaders = connect
+
+    def putrequest(self, method, url):
+        self.requests.append((method, url))
+
+    def putheader(self, name, value):
+        self.headers.append((name, value))
+
+    def send(self, body):
+        self.body = body
+
+    def getresponse(self):
+        return Response()
 
 class uploadTestCase(PyPIRCCommandTestCase):
 
+    def setUp(self):
+        super(uploadTestCase, self).setUp()
+        self.old_class = httpclient.HTTPConnection
+        self.conn = httpclient.HTTPConnection = FakeConnection()
+
+    def tearDown(self):
+        httpclient.HTTPConnection = self.old_class
+        super(uploadTestCase, self).tearDown()
+
     def test_finalize_options(self):
 
         # new format
-        f = open(self.rc, 'w')
-        f.write(PYPIRC)
-        f.close()
-
+        self.write_file(self.rc, PYPIRC)
         dist = Distribution()
         cmd = upload(dist)
         cmd.finalize_options()
@@ -39,9 +75,7 @@ class uploadTestCase(PyPIRCCommandTestCase):
 
     def test_saved_password(self):
         # file with no password
-        f = open(self.rc, 'w')
-        f.write(PYPIRC_NOPASSWORD)
-        f.close()
+        self.write_file(self.rc, PYPIRC_NOPASSWORD)
 
         # make sure it passes
         dist = Distribution()
@@ -56,6 +90,28 @@ class uploadTestCase(PyPIRCCommandTestCase):
         cmd.finalize_options()
         self.assertEquals(cmd.password, 'xxx')
 
+    def test_upload(self):
+        tmp = self.mkdtemp()
+        path = os.path.join(tmp, 'xxx')
+        self.write_file(path)
+        command, pyversion, filename = 'xxx', '2.6', path
+        dist_files = [(command, pyversion, filename)]
+        self.write_file(self.rc, PYPIRC)
+
+        # lets run it
+        pkg_dir, dist = self.create_dist(dist_files=dist_files)
+        cmd = upload(dist)
+        cmd.ensure_finalized()
+        cmd.run()
+
+        # what did we send ?
+        headers = dict(self.conn.headers)
+        self.assertEquals(headers['Content-length'], '2087')
+        self.assert_(headers['Content-type'].startswith('multipart/form-data'))
+
+        self.assertEquals(self.conn.requests, [('POST', '/pypi')])
+        self.assert_((b'xxx') in self.conn.body)
+
 def test_suite():
     return unittest.makeSuite(uploadTestCase)