verify callback test

Author: longbai

qiniu/compat.py

@@ -42,14 +42,14 @@
     basestring = basestring  # noqa
     numeric_types = (int, long, float)  # noqa
 
-    def b(s):
-        return s
+    def b(data):
+        return data
 
-    def s(b):
-        return b
+    def s(data):
+        return data
 
-    def u(s):
-        return unicode(s, 'unicode_escape')  # noqa
+    def u(data):
+        return unicode(data, 'unicode_escape')  # noqa
 
 elif is_py3:
     from urllib.parse import urlparse  # noqa
@@ -63,15 +63,15 @@ def u(s):
     basestring = (str, bytes)
     numeric_types = (int, float)
 
-    def b(s):
-        if isinstance(s, str):
-            return s.encode('utf-8')
-        return s
+    def b(data):
+        if isinstance(data, str):
+            return data.encode('utf-8')
+        return data
 
-    def s(b):
-        if isinstance(b, bytes):
-            b = b.decode('utf-8')
-        return b
+    def s(data):
+        if isinstance(data, bytes):
+            data = data.decode('utf-8')
+        return data
 
-    def u(s):
-        return s
+    def u(data):
+        return data

qiniu/services/storage/uploader.py

@@ -29,13 +29,13 @@ def put_file(up_token, key, file_path, params=None, mime_type='application/octet
     return ret, info
 
 
-def _form_put(up_token, key, data, params, mime_type, crc32, is_file=False, progress_handler=None):
+def _form_put(up_token, key, data, params, mime_type, crc, is_file=False, progress_handler=None):
     fields = {}
     if params:
         for k, v in params.items():
             fields[k] = str(v)
-    if crc32:
-        fields['crc32'] = crc32
+    if crc:
+        fields['crc32'] = crc
     if key is not None:
         fields['key'] = key
     fields['token'] = up_token

qiniu/utils.py

@@ -56,8 +56,8 @@ def _etag(input_stream):
         data = array[0]
         prefix = b('\x16')
     else:
-        s = b('').join(array)
-        data = _sha1(s)
+        sha1_str = b('').join(array)
+        data = _sha1(sha1_str)
         prefix = b('\x96')
     return urlsafe_base64_encode(prefix + data)
 

test_qiniu.py

@@ -88,6 +88,18 @@ def test_deprecatedPolicy(self):
         with pytest.raises(ValueError):
             dummy_auth.upload_token('1', None, policy={'asyncOps': 1})
 
+    def test_token_of_request(self):
+        token = dummy_auth.token_of_request('http://www.qiniu.com?go=1', 'test', '')
+        assert token == 'abcdefghklmnopq:cFyRVoWrE3IugPIMP5YJFTO-O-Y='
+        token = dummy_auth.token_of_request('http://www.qiniu.com?go=1', 'test', 'application/x-www-form-urlencoded')
+        assert token == 'abcdefghklmnopq:svWRNcacOE-YMsc70nuIYdaa1e4='
+
+    def test_verify_callback(self):
+        body = 'name=sunflower.jpg&hash=Fn6qeQi4VDLQ347NiRm-RlQx_4O2&location=Shanghai&price=1500.00&uid=123'
+        url = 'test.qiniu.com/callback'
+        ok = dummy_auth.verify_callback('QBox abcdefghklmnopq:ZWyeM5ljWMRFwuPTPOwQ4RwSto4=', url, body)
+        assert ok
+
 
 class BucketTestCase(unittest.TestCase):
     q = Auth(access_key, secret_key)
@@ -226,7 +238,7 @@ def test_putInvalidCrc(self):
         data = 'hello bubby!'
         crc32 = 'wrong crc32'
         token = self.q.upload_token(bucket_name)
-        ret, info = _form_put(token, key, data, None, None, crc32=crc32)
+        ret, info = _form_put(token, key, data, None, None, crc=crc32)
         print(info)
         assert ret is None
         assert info.status_code == 400