import socket
import platform
import httplib
-import base64
+from base64 import standard_b64encode
import urlparse
import cStringIO as StringIO
from ConfigParser import ConfigParser
open(filename+".asc").read())
# set up the authentication
- auth = "Basic " + base64.encodestring(self.username + ":" + self.password).strip()
+ auth = "Basic " + standard_b64encode(self.username + ":" +
+ self.password)
# Build up the MIME payload for the POST data
boundary = '--------------GHSKFJDLGDS7543FJKLFHRE75642756743254'
"""Support code for distutils test cases."""
-
+import os
import shutil
import tempfile
from distutils import log
-
+from distutils.dist import Distribution
class LoggingSilencer(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()."""
import sys
import os
import unittest
+import httplib
from distutils.command.upload import upload
from distutils.core import Distribution
from distutils.tests import support
from distutils.tests.test_config import PYPIRC, PyPIRCCommandTestCase
+PYPIRC_LONG_PASSWORD = """\
+[distutils]
+
+index-servers =
+ server1
+ server2
+
+[server1]
+username:me
+password:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
+
+[server2]
+username:meagain
+password: secret
+realm:acme
+repository:http://another.pypi/
+"""
+
+class _Resp(object):
+ def __init__(self, status):
+ self.status = status
+ self.reason = 'OK'
+
+_CONNECTIONS = []
+
+class _FakeHTTPConnection(object):
+ def __init__(self, netloc):
+ self.requests = []
+ self.headers = {}
+ self.body = None
+ self.netloc = netloc
+ _CONNECTIONS.append(self)
+
+ def connect(self):
+ pass
+ endheaders = connect
+
+ def send(self, body):
+ self.body = body
+
+ def putrequest(self, type_, data):
+ self.requests.append((type_, data))
+
+ def putheader(self, name, value):
+ self.headers[name] = value
+
+ def getresponse(self):
+ return _Resp(200)
+
class uploadTestCase(PyPIRCCommandTestCase):
+ def setUp(self):
+ super(uploadTestCase, self).setUp()
+ self.old_klass = httplib.HTTPConnection
+ httplib.HTTPConnection = _FakeHTTPConnection
+
+ def tearDown(self):
+ httplib.HTTPConnection = self.old_klass
+ super(uploadTestCase, self).tearDown()
+
def test_finalize_options(self):
# new format
self.assertEquals(getattr(cmd, attr), waited)
+ 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_LONG_PASSWORD)
+
+ # 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 ?
+ res = _CONNECTIONS[-1]
+
+ headers = res.headers
+ self.assertEquals(headers['Content-length'], '2086')
+ self.assertTrue(headers['Content-type'].startswith('multipart/form-data'))
+
+ method, request = res.requests[-1]
+ self.assertEquals(method, 'POST')
+ self.assertEquals(res.netloc, 'pypi.python.org')
+ self.assertTrue('xxx' in res.body)
+ self.assertFalse('\n' in headers['Authorization'])
+
def test_suite():
return unittest.makeSuite(uploadTestCase)