[Python-checkins] distutils2: implemented run for the simplest case

tarek.ziade python-checkins at python.org
Sun Jul 4 11:48:38 CEST 2010


tarek.ziade pushed 4c79b30a4bc4 to distutils2:

http://hg.python.org/distutils2/rev/4c79b30a4bc4
changeset:   272:4c79b30a4bc4
user:        Konrad Delong <konryd at gmail.com>
date:        Fri May 14 00:42:49 2010 +0200
summary:     implemented run for the simplest case
files:       src/distutils2/command/upload_docs.py, src/distutils2/tests/test_upload_docs.py

diff --git a/src/distutils2/command/upload_docs.py b/src/distutils2/command/upload_docs.py
--- a/src/distutils2/command/upload_docs.py
+++ b/src/distutils2/command/upload_docs.py
@@ -1,10 +1,11 @@
-import os.path, tempfile, zipfile
+import base64, httplib, os.path, tempfile, urlparse, zipfile
+from cStringIO import StringIO
+from distutils2 import log
 from distutils2.core import Command
 
-def zip_dir_into(directory, destination):
-    """Compresses recursively contents of directory into a zipfile located
-    under given destination.
-    """
+def zip_dir(directory):
+    """Compresses recursively contents of directory into a StringIO object"""
+    destination = StringIO()
     zip_file = zipfile.ZipFile(destination, "w")
     for root, dirs, files in os.walk(directory):
         for name in files:
@@ -13,14 +14,14 @@
             dest = os.path.join(relative, name)
             zip_file.write(full, dest)
     zip_file.close()
+    return destination
 
 # grabbed from
 #    http://code.activestate.com/recipes/146306-http-client-to-post-using-multipartform-data/
 def encode_multipart(fields, files, boundary=None):
     """
     fields is a sequence of (name, value) elements for regular form fields.
-    files is a sequence of (name, filename, content_type, value)
-                            elements for data to be uploaded as files
+    files is a sequence of (name, filename, value) elements for data to be uploaded as files
     Return (content_type, body) ready for httplib.HTTP instance
     """
     if boundary is None:
@@ -33,11 +34,10 @@
             'Content-Disposition: form-data; name="%s"' % key,
             '',
             value])
-    for (key, filename, content_type, value) in files:
+    for (key, filename, value) in files:
         l.extend([
             '--' + boundary,
             'Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename),
-            'Content-Type: %s' % content_type,
             '',
             value])
     l.append('--' + boundary + '--')
@@ -61,4 +61,34 @@
             self.upload_dir = os.path.join(build.build_base, "docs")
 
     def run(self):
-        pass
+        tmp_dir = tempfile.mkdtemp()
+        name = self.distribution.metadata['Name']
+        zip_file = zip_dir(self.upload_dir)
+
+        fields = {':action': 'doc_upload', 'name': name}.items()
+        files = [('content', name, zip_file.getvalue())]
+        content_type, body = encode_multipart(fields, files)
+
+        credentials = self.username + ':' + self.password
+        auth = "Basic " + base64.encodestring(credentials).strip()
+
+        self.announce("Submitting documentation to %s" % (self.repository),
+                      log.INFO)
+
+        schema, netloc, url, params, query, fragments = \
+            urlparse.urlparse(self.repository)
+        conn = httplib.HTTPConnection(netloc)
+
+        try:
+            conn.connect()
+            conn.putrequest("POST", url)
+            conn.putheader('Content-type', content_type)
+            conn.putheader('Content-length', str(len(body)))
+            conn.putheader('Authorization', auth)
+            conn.endheaders()
+            conn.send(body)
+        except socket.error, e:
+            self.announce(str(e), log.ERROR)
+            return
+
+        r = conn.getresponse()
diff --git a/src/distutils2/tests/test_upload_docs.py b/src/distutils2/tests/test_upload_docs.py
--- a/src/distutils2/tests/test_upload_docs.py
+++ b/src/distutils2/tests/test_upload_docs.py
@@ -1,8 +1,9 @@
 """Tests for distutils.command.upload_docs."""
 # -*- encoding: utf8 -*-
 import os, os.path, tempfile, unittest2, zipfile
+from cStringIO import StringIO
 
-from distutils2.command.upload_docs import (upload_docs, zip_dir_into,
+from distutils2.command.upload_docs import (upload_docs, zip_dir,
                                     encode_multipart)
 from distutils2.core import Distribution
 
@@ -23,14 +24,12 @@
 'd',
 '---x',
 'Content-Disposition: form-data; name="e"; filename="f"',
-'Content-Type: g',
 '',
-'h',
+'g',
 '---x',
-'Content-Disposition: form-data; name="i"; filename="j"',
-'Content-Type: k',
+'Content-Disposition: form-data; name="h"; filename="i"',
 '',
-'l',
+'j',
 '---x--',
 '',
 ])
@@ -42,33 +41,68 @@
     def setUp(self):
         super(UploadDocsTestCase, self).setUp()
         self.dist = Distribution()
+        self.dist.metadata['Name'] = "distr-name"
         self.cmd = upload_docs(self.dist)
 
     def test_generates_uploaddir_name_not_given(self):
         self.cmd.ensure_finalized()
         self.assertEqual(self.cmd.upload_dir, os.path.join("build", "docs"))
 
-    def test_zip_dir_into(self):
-        source_dir = tempfile.mkdtemp()
-        os.mkdir(os.path.join(source_dir, "some_dir"))
-        some_file = open(os.path.join(source_dir, "some_dir", "some_file"), "w")
+    def prepare_sample_dir(self):
+        sample_dir = tempfile.mkdtemp()
+        os.mkdir(os.path.join(sample_dir, "some_dir"))
+        some_file = open(os.path.join(sample_dir, "some_dir", "some_file"), "w")
         some_file.write("Ce mortel ennui")
         some_file.close()
-        dest_dir = tempfile.mkdtemp()
-        destination = os.path.join(dest_dir, "archive.zip")
-        zip_dir_into(source_dir, destination)
+        return sample_dir
 
-        self.assertTrue(zipfile.is_zipfile(destination))
-        zip_f = zipfile.ZipFile(destination)
+    def test_zip_dir(self):
+        source_dir = self.prepare_sample_dir()
+        compressed = zip_dir(source_dir)
+
+        zip_f = zipfile.ZipFile(compressed)
         self.assertEqual(zip_f.namelist(), ['some_dir/some_file'])
 
     def test_encode_multipart(self):
         fields = [("a", "b"), ("c", "d")]
-        files = [("e", "f", "g", "h"), ("i", "j", "k", "l")]
-        self.assertEqual(encode_multipart(fields, files, "-x")[0], "multipart/form-data; boundary=-x")
-        self.assertEqual(encode_multipart(fields, files, "-x")[1], EXPECTED_MULTIPART_OUTPUT)
+        files = [("e", "f", "g"), ("h", "i", "j")]
+        content_type, body = encode_multipart(fields, files, "-x")
+        self.assertEqual(content_type, "multipart/form-data; boundary=-x")
+        self.assertEqual(body, EXPECTED_MULTIPART_OUTPUT)
 
     def test_upload(self):
+        self.cmd.ensure_finalized()
+        self.cmd.repository = self.pypi.full_address
+        self.cmd.upload_dir = self.prepare_sample_dir()
+        self.cmd.username = "username"
+        self.cmd.password = "password"
+        self.cmd.run()
+
+        self.assertEqual(len(self.pypi.requests), 1)
+        environ, request_data = self.pypi.requests[-1]
+
+        self.assertIn("content", request_data)
+        self.assertIn("Basic", environ['HTTP_AUTHORIZATION'])
+        self.assertTrue(environ['CONTENT_TYPE'].startswith('multipart/form-data;'))
+
+        action, name, content =\
+            request_data.split("----------------GHSKFJDLGDS7543FJKLFHRE75642756743254")[1:4]
+
+        # check that we picked the right chunks
+        self.assertIn('name=":action"', action)
+        self.assertIn('name="name"', name)
+        self.assertIn('name="content"', content)
+
+        # check their contents
+        self.assertIn("doc_upload", action)
+        self.assertIn("distr-name", name)
+        self.assertIn("some_dir/some_file", content)
+        self.assertIn("Ce mortel ennui", content)
+
+    def test_https_connection(self):
+        pass
+
+    def test_handling_response(self):
         pass
 
     def test_honours_dry_run(self):

--
Repository URL: http://hg.python.org/distutils2


More information about the Python-checkins mailing list