changes

Author: allanlei

.gitignore

@@ -34,3 +34,5 @@ nosetests.xml
 .mr.developer.cfg
 .project
 .pydevproject
+
+*.zip

.travis.yml

@@ -0,0 +1,10 @@
+language: python
+python:
+  - "2.6"
+  - "2.7"
+  - "3.2"
+  - "3.3"
+  - "pypy"
+install:
+  - "pip install ."
+script: nosetests

README.markdown

@@ -1,10 +1,12 @@
 
-# ZipStream
+# python-zipstream
 
-zipstream.py is a zip archive generator based on zipfile.py. It was created to
-generate a zip file on-the-fly for download in a web.py (http://webpy.org/)
-application. This is beneficial for when you want to provide a downloadable
-archive of a large collection of regular files, which would be infeasible to
+[![Build Status](https://travis-ci.org/allanlei/python-zipstream.png?branch=master)](https://travis-ci.org/allanlei/python-zipstream)
+[![Coverage Status](https://coveralls.io/repos/allanlei/python-zipstream/badge.png)](https://coveralls.io/r/allanlei/python-zipstream)
+
+zipstream.py is a zip archive generator based on python 3.3's zipfile.py. It was created to
+generate a zip file generator for streaming (ie web apps). This is beneficial for when you 
+want to provide a downloadable archive of a large collection of regular files, which would be infeasible to
 generate the archive prior to downloading.
 
 The archive is generated as an iterator of strings, which, when joined, form
@@ -37,10 +39,63 @@ archives.
 
 ## Requirements
 
-  * Python >=2.6
+  * Python 2.6, 2.7, 3.2, 3.3, pypy
+
+## Examples
+
+### flask
+
+```python
+from flask import Response
+
+@app.route('/package.zip', methods=['GET'], endpoint='zipball')
+def zipball():
+    def generator():
+    	z = ZipStream(mode='w', compression=ZIP_DEFLATED)
+
+    	z.write('/path/to/file')
 
-## License
+    	for chunk in z:
+    		yield chunk
 
-This library was created by SpiderOak, Inc. and is released under the GPLv3.
-Copyright 2008-2013 SpiderOak Inc.
+    response = Response(generator(), mimetype='application/zip')
+    response.headers['Content-Disposition'] = 'attachment; filename={}'.format('files.zip')
+    return response
+
+# or
+
+@app.route('/package.zip', methods=['GET'], endpoint='zipball')
+def zipball():
+	z = ZipStream(mode='w', compression=ZIP_DEFLATED)
+	z.write('/path/to/file')
+
+    response = Response(z, mimetype='application/zip')
+    response.headers['Content-Disposition'] = 'attachment; filename={}'.format('files.zip')
+    return response
+```
+
+### django 1.5+
+
+```python
+from django.http import StreamingHttpResponse
 
+def zipball(request):
+	z = ZipStream(mode='w', compression=ZIP_DEFLATED)
+	z.write('/path/to/file')
+
+    response = StreamingHttpResponse(z, mimetype='application/zip')
+    response['Content-Disposition'] = 'attachment; filename={}'.format('files.zip')
+    return response
+```
+
+### webpy
+
+```python
+def GET(self):
+    path = '/path/to/dir/of/files'
+    zip_filename = 'files.zip'
+    web.header('Content-type' , 'application/zip')
+    web.header('Content-Disposition', 'attachment; filename="%s"' % (
+        zip_filename,))
+    return ZipStream(path)
+```

example.py

@@ -0,0 +1,25 @@
+import os
+import zipstream
+import zipfile
+
+
+f = open('test.zip', 'wb')
+
+with zipstream.ZipFile(mode='w', compression=zipstream.ZIP_DEFLATED) as z:
+    z.write('LICENSE')
+    z.write('LICENSE', arcname='stuff/LICENSE')
+
+    for root, directories, files in os.walk('zipstream'):
+        for filename in files:
+            path = os.path.join(root, filename)
+            z.write(path, path)
+
+    with open('test.zip', 'wb') as f:
+        for chunk in z:
+            f.write(chunk)
+
+f.close()
+
+
+with zipfile.ZipFile('test.zip') as z:
+    z.testzip()

setup.py

@@ -1,13 +1,18 @@
-#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from setuptools import setup, find_packages
+import zipstream
 
-from distutils.core import setup
 
 setup(
     name='zipstream',
-    version='1.0.1',
-    description='SpiderOak ZipStream Module',
-    author='SpiderOak Team',
-    author_email='[email protected]',
-    url='http://www.spideroak.com',
-    py_modules=['zipstream'],
+    version=zipstream.__version__,
+    description='Zipfile generator',
+    author='Allan Lei',
+    author_email='[email protected]',
+    url='https://github.com/allanlei/python-zipstream',
+    packages=find_packages(),
+    keywords='zip streaming',
+
+    test_suite='nose.collector',
+    tests_require = ['nose'],
 )

tests/__init__.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+import unittest
+import zipstream
+
+
+class PointerIOTestCase(unittest.TestCase):
+    def test_init_no_args(self):
+        zipstream.PointerIO()
+
+    def test_init_mode(self):
+        try:
+            zipstream.PointerIO('wb')
+        except Exception as err:
+            self.fail(err)
+
+        for mode in ['w', 'r', 'rb', 'a', 'ab']:
+            self.assertRaises(Exception, zipstream.PointerIO, mode=mode)
+
+        for mode in ['w', 'wb''r', 'rb', 'a', 'ab']:
+            self.assertRaises(Exception, zipstream.PointerIO, mode=mode + '+')
+
+    def test_has_fileobj_attrs(self):
+        fileobj = zipstream.PointerIO()
+
+        self.assertTrue(hasattr(fileobj, 'write'))
+        self.assertTrue(hasattr(fileobj, 'close'))
+        self.assertTrue(hasattr(fileobj, 'tell'))
+
+    def test_write_bytes(self):
+        fileobj = zipstream.PointerIO()
+        data = b'Im a little tea pot'
+        try:
+            fileobj.write(data)
+        except Exception as err:
+            self.fail(err)
+        self.assertEqual(fileobj.tell(), 19)
+
+    def test_write_unicode(self):
+        fileobj = zipstream.PointerIO()
+        data = 'Im a little tea pot'
+        try:
+            fileobj.write(data)
+        except Exception as err:
+            self.fail(err)
+        self.assertEqual(fileobj.tell(), 19)
+
+
+        fileobj = zipstream.PointerIO()
+        data = '幋 儳鑤 寱懤擨 拻敁柧'
+        try:
+            fileobj.write(data)
+        except Exception as err:
+            self.fail(err)
+        self.assertEqual(fileobj.tell(), 30)
+
+    def test_write_non_string_type(self):
+        fileobj = zipstream.PointerIO()
+        data = None
+        self.assertRaises(TypeError, fileobj.write, data)
+
+        fileobj = zipstream.PointerIO()
+        data = []
+        self.assertRaises(TypeError, fileobj.write, data)
+
+        fileobj = zipstream.PointerIO()
+        data = tuple()
+        self.assertRaises(TypeError, fileobj.write, data)
+
+        fileobj = zipstream.PointerIO()
+        data = 1
+        self.assertRaises(TypeError, fileobj.write, data)
+
+        fileobj = zipstream.PointerIO()
+        data = 1.00
+        self.assertRaises(TypeError, fileobj.write, data)
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pointerio.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals, print_function
+
+import os
+import tempfile
+import unittest
+import zipstream
+import zipfile
+
+
+class ZipInfoTestCase(unittest.TestCase):
+    pass
+
+
+class ZipStreamTestCase(unittest.TestCase):
+    def setUp(self):
+        self.fileobjs = [
+            tempfile.NamedTemporaryFile(delete=False, suffix='.txt'),
+            tempfile.NamedTemporaryFile(delete=False, suffix='.py'),
+        ]
+
+    def tearDown(self):
+        for fileobj in self.fileobjs:
+            fileobj.close()
+            os.remove(fileobj.name)
+
+    def test_init_no_args(self):
+        zipstream.ZipFile()
+
+    def test_init_mode(self):
+        try:
+            zipstream.ZipFile(mode='w')
+        except Exception as err:
+            self.fail(err)
+
+        for mode in ['wb', 'r', 'rb', 'a', 'ab']:
+            self.assertRaises(Exception, zipstream.ZipFile, mode=mode)
+
+        for mode in ['wb', 'r', 'rb', 'a', 'ab']:
+            self.assertRaises(Exception, zipstream.ZipFile, mode=mode + '+')
+
+    def test_write_file(self):
+        z = zipstream.ZipFile(mode='w')
+        for fileobj in self.fileobjs:
+            z.write(fileobj.name)
+
+        f = tempfile.NamedTemporaryFile(suffix='zip', delete=False)
+        for chunk in z:
+            f.write(chunk)
+        f.close()
+
+        z2 = zipfile.ZipFile(f.name, 'r')
+        z2.testzip()
+
+        os.remove(f.name)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_zipstream.py

@@ -0,0 +1,6 @@
+[tox]
+envlist = py26, py27, py32, py33, pypy
+
+[testenv]
+deps=nose
+commands = nosetests {posargs}