Fix chinese file names and add tests. Works under python 2 & 3

qiniustorage/backends.py

@@ -6,7 +6,7 @@
 import os
 import posixpath
 
-from six import BytesIO, string_types
+import six
 from six.moves import cStringIO as StringIO
 from six.moves.urllib_parse import urljoin, urlparse
 
@@ -29,7 +29,7 @@ def get_qiniu_config(name, default=None):
     """
     config = os.environ.get(name, getattr(settings, name, default))
     if config is not None:
-        if isinstance(config, string_types):
+        if isinstance(config, six.string_types):
             return config.strip()
         else:
             return config
@@ -46,7 +46,7 @@ def get_qiniu_config(name, default=None):
 QINIU_SECURE_URL = get_qiniu_config('QINIU_SECURE_URL', 'False')
 
 
-if isinstance(QINIU_SECURE_URL, string_types):
+if isinstance(QINIU_SECURE_URL, six.string_types):
     if QINIU_SECURE_URL.lower() in ('true', '1'):
         QINIU_SECURE_URL = True
     else:
@@ -139,14 +139,17 @@ def _read(self, name):
 
     def delete(self, name):
         name = self._normalize_name(self._clean_name(name))
+        if six.PY2:
+            name = name.encode('utf-8')
         ret, info = self.bucket_manager.delete(self.bucket_name, name)
 
         if ret is None or info.status_code == 612:
             raise QiniuError(info)
 
     def _file_stat(self, name, silent=False):
         name = self._normalize_name(self._clean_name(name))
-
+        if six.PY2:
+            name = name.encode('utf-8')
         ret, info = self.bucket_manager.stat(self.bucket_name, name)
         if ret is None and not silent:
             raise QiniuError(info)
@@ -206,7 +209,7 @@ def __init__(self, name, storage, mode):
         self._storage = storage
         self._name = name[len(self._storage.location):].lstrip('/')
         self._mode = mode
-        self.file = BytesIO()
+        self.file = six.BytesIO()
         self._is_dirty = False
         self._is_read = False
 
@@ -226,7 +229,7 @@ def size(self):
     def read(self, num_bytes=None):
         if not self._is_read:
             content = self._storage._read(self._name)
-            self.file = BytesIO(content)
+            self.file = six.BytesIO(content)
             self._is_read = True
 
         if num_bytes is None:

tests/test_storage.py

@@ -7,6 +7,7 @@
 import unittest
 import uuid
 
+import six
 import django
 import pytest
 
@@ -58,52 +59,57 @@ def test_write_to_read_only_file(self):
             fil.write("goto fail")
 
     def test_write_and_delete_file(self):
-        ASSET_FILE_NAME = 'jquery-1.11.1.min.js'
-        REMOTE_PATH = join(UNIQUE_PATH, ASSET_FILE_NAME)
-        assert self.storage.exists(REMOTE_PATH) == False
-        fil = QiniuFile(REMOTE_PATH, self.storage, mode='wb')
+        ASSET_FILE_NAMES = [u'jquery-2.1.1.min.js', u'jquery-2点壹点一.min.js']
+        for assert_file_name in ASSET_FILE_NAMES:
+            REMOTE_PATH = join(UNIQUE_PATH, assert_file_name)
+            assert self.storage.exists(REMOTE_PATH) == False
+            fil = QiniuFile(REMOTE_PATH, self.storage, mode='wb')
 
-        with open(join(dirname(__file__),'assets', ASSET_FILE_NAME), 'rb') as assset_file:
-            content = assset_file.read()
+            with open(join(dirname(__file__),'assets', assert_file_name), 'rb') as assset_file:
+                content = assset_file.read()
 
-            assset_file.seek(0, os.SEEK_END)
-            assset_file_size = assset_file.tell()
+                assset_file.seek(0, os.SEEK_END)
+                assset_file_size = assset_file.tell()
 
-            fil.write(content)
-            self.storage._save(REMOTE_PATH, fil)
+                fil.write(content)
+                self.storage._save(REMOTE_PATH, fil)
 
-            assert self.storage.exists(REMOTE_PATH)
+                assert self.storage.exists(REMOTE_PATH)
 
-        assert self.storage.size(REMOTE_PATH) == assset_file_size
+            assert self.storage.size(REMOTE_PATH) == assset_file_size
 
-        time_delta = datetime.now() - self.storage.modified_time(REMOTE_PATH)
-        assert time_delta.seconds < 60
+            now = datetime.utcnow()
+            modified_time = self.storage.modified_time(REMOTE_PATH)
+            # Datetime on Qiniu servers may be faster or slower than the local
+            # machine. Thus the absolute delta within 60s should be considered
+            # acceptable.
+            time_delta = max(now, modified_time) - min(now, modified_time)
+            assert time_delta.total_seconds() < 60
 
-        self.storage.delete(REMOTE_PATH)
-        assert self.storage.exists(REMOTE_PATH) == False
+            self.storage.delete(REMOTE_PATH)
+            assert self.storage.exists(REMOTE_PATH) == False
 
     def test_read_file(self):
-        ASSET_FILE_NAME = 'jquery-2.1.1.min.js'
-        REMOTE_PATH = join(UNIQUE_PATH, ASSET_FILE_NAME)
+        ASSET_FILE_NAMES =  [u'jquery-1.11.1.min.js', u'jquery-一点十一点壹.js']
+        for assert_file_name in ASSET_FILE_NAMES:
+            REMOTE_PATH = join(UNIQUE_PATH, assert_file_name)
 
-        with open(join(dirname(__file__),'assets', ASSET_FILE_NAME), 'rb') as assset_file:
-            self.storage.save(REMOTE_PATH, assset_file)
+            with open(join(dirname(__file__),'assets', assert_file_name), 'rb') as assset_file:
+                self.storage.save(REMOTE_PATH, assset_file)
 
-        fil = self.storage.open(REMOTE_PATH, 'r')
+            fil = self.storage.open(REMOTE_PATH, 'r')
 
-        assert fil._is_read == False
-
-        content = fil.read()
-        assert content.startswith(u"/*!")
+            assert fil._is_read == False
 
-        assert fil._is_read == True
-
-        # Test open mode
-        fil = self.storage.open(REMOTE_PATH, 'rb')
-        bin_content = fil.read()
-        assert bin_content.startswith(b"/*!")
+            content = fil.read()
+            assert content.startswith(u"/*!")
 
+            assert fil._is_read == True
 
+            # Test open mode
+            fil = self.storage.open(REMOTE_PATH, 'rb')
+            bin_content = fil.read()
+            assert bin_content.startswith(b"/*!")
 
     def test_dirty_file(self):
         ASSET_FILE_NAME = 'bootstrap.min.css'
@@ -125,10 +131,11 @@ def test_dirty_file(self):
         fil.close()
         assert self.storage.exists(REMOTE_PATH) == True
 
+
     @retry(AssertionError, tries=10)
     def test_listdir(self):
         dirnames = ['', 'foo', 'bar']
-        filenames = ['file1', 'file2', 'file3']
+        filenames = ['file1', 'file2', 'file3', u'file四']
         for dirname in dirnames:
             for filename in filenames:
                 fil = self.storage.open(join(UNIQUE_PATH, dirname, filename), 'w')
@@ -160,7 +167,8 @@ def teardown_class(cls):
 
             for item in ret['items']:
                 name = item['key']
-                print("Deleting %s ..." % name)
+                if six.PY2:
+                    name = name.encode('utf-8')
                 ret, info = bucket.delete(storage.bucket_name, name)
                 if ret is None:
                     print(info)