add test

.travis.yml

@@ -5,5 +5,8 @@ python:
 before_script:
   - export QINIU_ACCESS_KEY="tGf47MBl1LyT9uaNv-NZV4XZe7sKxOIa9RE2Lp8B"
   - export QINIU_SECRET_KEY="zhbiA6gcQMEi22uZ8CBGvmbnD2sR8SO-5S8qlLCG"
-  - export QINIU_TEST_PIC_1="http://cheneya.qiniudn.com/hello_jpg"
-  - export QINIU_NOEXIST_PIC="http://cheneya.qiniudn.com/no_exist"
+  - export QINIU_PIC_KEY="hello_jpg"
+  - export QINIU_NOEXIST_PIC_KEY="no_exist"
+  - export QINIU_TEST_PIC_1="http://cheneya.qiniudn.com/$QINIU_PIC_KEY"
+  - export QINIU_NOEXIST_PIC="http://cheneya.qiniudn.com/$QINIU_NOEXIST_PIC_KEY"
+  - export QINIU_BUCKET_NAME="a"

qiniu/auth.py

@@ -36,13 +36,20 @@ def call_with(self, path, body):
 		return ret, None
 
 	def call_with_multipart(self, path, fields=None, files=None):
+		"""
+		 *  fields => [(key, value)]
+		 *  files => [(key, filename, mimeType, value)]
+		"""
 		content_type, body = self.encode_multipart_formdata(fields, files)
 		self.set_header("Content-Type", content_type)
 		self.set_header("Content-Length", len(body))
 		return self.call_with(path, body)
 
 	def call_with_form(self, path, ops):
-		self.set_header("Content-Type", "application/x-www-form-urlencoded")
+		"""
+		 * ops => {"key": value/list()}
+		"""
+
 		body = []
 		for i in ops:
 			if isinstance(ops[i], (list, tuple)):
@@ -51,7 +58,11 @@ def call_with_form(self, path, ops):
 				data = ops[i]
 
 			body.append('%s=%s' % (i, data))
-		return self.call_with(path, '&'.join(body))
+		body = '&'.join(body)
+
+		self.set_header("Content-Type", "application/x-www-form-urlencoded")
+		self.set_header("Content-Length", len(body))
+		return self.call_with(path, body)
 
 	def set_header(self, field, value):
 		self._header[field] = value
@@ -65,6 +76,11 @@ def encode_multipart_formdata(self, fields, files):
 		 *  files => [(key, filename, mimeType, value)]
 		 *  return content_type, body
 		"""
+		if files is None:
+			files = []
+		if fields is None:
+			fields = []
+
 		BOUNDARY = '----------ThIs_Is_tHe_bouNdaRY_$'
 		CRLF = '\r\n'
 		L = []
@@ -75,7 +91,8 @@ def encode_multipart_formdata(self, fields, files):
 			L.append(value)
 		for (key, filename, mimeType, value) in files:
 			L.append('--' + BOUNDARY)
-			L.append('Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename))
+			disposition = "Content-Disposition: form-data;"
+			L.append('%s name="%s"; filename="%s"' % (disposition, key, filename))
 			L.append('Content-Type: %s' % mimeType)
 			L.append('')
 			L.append(value)
@@ -92,11 +109,3 @@ def sign(secret, data):
 def sign_json(access, secret, data):
 	data = urlsafe_b64encode(json.dumps(data, separators=(',',':')))
 	return '%s:%s:%s' % (access, sign(secret, data), data)
-
-if __name__ == "__main__":
-	ACCESS_KEY = "tGf47MBl1LyT9uaNv-NZV4XZe7sKxOIa9RE2Lp8B"
-	SECRET_KEY = "zhbiA6gcQMEi22uZ8CBGvmbnD2sR8SO-5S8qlLCG"
-	a = dict(a="b")
-	print sign_json(ACCESS_KEY, SECRET_KEY, a)
-	help(Client)
-

qiniu/auth_demo.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+import unittest
+
+import auth
+import config
+
+def round_tripper(client, method, path, body):
+	pass
+
+class ClsTestClient(auth.Client):
+	def round_tripper(self, method, path, body):
+		round_tripper(self, method, path, body)
+		return super(ClsTestClient, self).round_tripper(method, path, body)
+
+client = ClsTestClient(config.RS_HOST)
+
+class TestClient(unittest.TestCase):
+	def test_call(self):
+		global round_tripper
+
+		def tripper(client, method, path, body):
+			self.assertEqual(path, "/hello")
+			self.assertIsNone(body)
+
+		round_tripper = tripper
+		client.call("/hello")
+
+	def test_call_with(self):
+		global round_tripper
+		def tripper(client, method, path, body):
+			self.assertEqual(body, "body")
+
+		round_tripper = tripper
+		client.call_with("/hello", "body")
+
+	def test_call_with_multipart(self):
+		global round_tripper
+		def tripper(client, method, path, body):
+			self.assertEqual(len(body), client._header["Content-Length"])
+			target_type = "multipart/form-data"
+			self.assertTrue(client._header["Content-Type"].startswith(target_type))
+			start_index = client._header["Content-Type"].find("boundary")
+			boundary = client._header["Content-Type"][start_index + 9: ]
+			dispostion = 'Content-Disposition: form-data; name="auth"'
+			tpl = "--%s\r\n%s\r\n\r\n%s\r\n--%s--\r\n" % (boundary, dispostion,
+					"auth_string", boundary)
+			self.assertEqual(tpl, body)
+
+		round_tripper = tripper
+		client.call_with_multipart("/hello", fields=[("auth", "auth_string")])
+
+	def test_call_with_form(self):
+		global round_tripper
+		def tripper(client, method, path, body):
+			self.assertEqual(body, "action=a&op=a&op=b")
+			target_type = "application/x-www-form-urlencoded"
+			self.assertEqual(client._header["Content-Type"], target_type)
+			self.assertEqual(client._header["Content-Length"], len(body))
+
+		round_tripper = tripper
+		client.call_with_form("/hello", dict(op=["a", "b"], action="a"))
+
+if __name__ == "__main__":
+	unittest.main()

qiniu/auth_digest.py

@@ -4,10 +4,6 @@
 
 import auth
 
-def setup():
-	pass
-
-
 class Client(auth.Client):
 	def __init__(self):
 		super(Client, self).__init__(config.RS_HOST)
@@ -31,7 +27,3 @@ def round_tripper(self, method, path, body):
 		digest = self.make_signal(path, body)
 		self.set_header("Authorization", "QBox %s:%s" % (config.ACCESS_KEY, digest))
 		return super(Client, self).round_tripper(method, path, body)
-
-
-if __name__ == "__main__":
-	c = Client()

qiniu/auth_token.py

@@ -47,19 +47,12 @@ def token(self):
 class GetPolicy(object):
 	scope = None
 	expires = 3600
+	def __init__(self, scope):
+		self.scope = scope
 
 	def token(self):
 		token = dict(
 			S = self.scope,
 			E = self.expires + int(time.time())
 		)
 		return auth.sign_json(config.ACCESS_KEY, config.SECRET_KEY, token)
-
-if __name__ == "__main__":
-	config.ACCESS_KEY = "tGf47MBl1LyT9uaNv-NZV4XZe7sKxOIa9RE2Lp8B"
-	config.SECRET_KEY = "zhbiA6gcQMEi22uZ8CBGvmbnD2sR8SO-5S8qlLCG"
-
-	p = PutPolicy("a")
-	# eyJzY29wZSI6ImEiLCJkZWFkbGluZSI6MTExMjIyMn0=
-	print p.token()
-	# tGf47MBl1LyT9uaNv-NZV4XZe7sKxOIa9RE2Lp8B:B6-2V-iikhegLkra9IvKnm8j3DE=:eyJzY29wZSI6ImEiLCJkZWFkbGluZSI6MTM2NDI2NDk5MX0=

qiniu/auth_token_demo.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+import unittest
+import config
+import os
+import json
+from base64 import urlsafe_b64decode as decode
+from base64 import urlsafe_b64encode as encode
+from hashlib import sha1
+import hmac
+
+import auth
+import auth_token
+
+config.ACCESS_KEY = os.getenv("QINIU_ACCESS_KEY")
+config.SECRET_KEY = os.getenv("QINIU_SECRET_KEY")
+bucket_name = os.getenv("QINIU_BUCKET_NAME")
+
+class TestToken(unittest.TestCase):
+	def test_put_policy(self):
+		policy = auth_token.PutPolicy(bucket_name)
+		policy.customer = "hello!"
+		tokens = policy.token().split(':')
+		self.assertEqual(config.ACCESS_KEY, tokens[0])
+		data = json.loads(decode(tokens[2]))
+		self.assertEqual(data["scope"], bucket_name)
+		self.assertEqual(data["customer"], policy.customer)
+
+		new_hmac = encode(hmac.new(config.SECRET_KEY, tokens[2], sha1).digest())
+		self.assertEqual(new_hmac, tokens[1])
+
+	def test_get_policy(self):
+		policy = auth_token.GetPolicy(bucket_name)
+		tokens = policy.token().split(':')
+		data = json.loads(decode(tokens[2]))
+		self.assertEqual(data["S"], bucket_name)
+
+if __name__ == "__main__":
+	unittest.main()

qiniu/fop_demo.py

@@ -0,0 +1,52 @@
+import unittest
+import os
+import fop
+
+access_key = os.getenv("QINIU_ACCESS_KEY")
+secret_key = os.getenv("QINIU_SECRET_KEY")
+pic = os.getenv("QINIU_TEST_PIC_1")
+noexist_pic = os.getenv("QINIU_NOEXIST_PIC")
+
+class TestFop(unittest.TestCase):
+	def test_imageExif(self):
+		ie = fop.ImageExif()
+		ret, err = ie.call(pic)
+		self.assertIsNone(err)
+		self.assertIsNotNone(ret)
+
+		# error
+		ret, err = ie.call(noexist_pic)
+		self.assertIsNotNone(err)
+		self.assertIsNone(ret)
+
+	def test_imageView(self):
+		iv = fop.ImageView()
+		iv.height = 100
+		ret = iv.make_request(pic)
+		self.assertEqual(ret, "%s?imageView/1/h/100/" % pic)
+
+		iv.quality = 20
+		iv.format = "png"
+		ret = iv.make_request(pic)
+		self.assertEqual(ret, "%s?imageView/1/h/100/q/20/format/png/" % pic)
+
+	def test_imageInfo(self):
+		ii = fop.ImageInfo()
+		ret, err = ii.call(pic)
+		self.assertIsNone(err)
+		self.assertIsNotNone(ret)
+
+		# error
+		ret, err = ii.call(noexist_pic)
+		self.assertIsNotNone(err)
+
+	def test_imageMogr(self):
+		im = fop.ImageMogr()
+		im.auto_orient = True
+		im.quality = 200
+		im.rotate = 90
+		target = "%s?imageMogr/auto-orient/quality/200/rotate/90" % pic
+		self.assertEqual(im.make_request(pic), target)
+
+if __name__ == '__main__':
+	unittest.main()

qiniu/rs.py

@@ -37,16 +37,18 @@ def batch_delete(self, entries):
 			ops.append(uri_delete(entry.bucket, entry.key))
 		return self.batch(ops)
 
-	def batch_move(self, entryies):
+	def batch_move(self, entries):
 		ops = []
 		for entry in entries:
-			ops.append(uri_move(entry.bucket, entry.key))
+			ops.append(uri_move(entry.src.bucket, entry.src.key, 
+				entry.dest.bucket, entry.dest.key))
 		return self.batch(ops)
 
-	def batch_copy(self, entryies):
+	def batch_copy(self, entries):
 		ops = []
 		for entry in entries:
-			ops.append(uri_copy(entry.bucket, entry.key))
+			ops.append(uri_copy(entry.src.bucket, entry.src.key, 
+				entry.dest.bucket, entry.dest.key))
 		return self.batch(ops)
 
 class EntryPath(object):
@@ -78,17 +80,3 @@ def uri_copy(bucket_src, key_src, bucket_dest, key_dest):
 	src = urlsafe_b64encode("%s:%s" % (bucket_src, key_src))
 	dest = urlsafe_b64encode("%s:%s" % (bucket_dest, key_dest))
 	return "/copy/%s/%s" % (src, dest)
-
-if __name__ == "__main__":
-	config.ACCESS_KEY = "tGf47MBl1LyT9uaNv-NZV4XZe7sKxOIa9RE2Lp8B"
-	config.SECRET_KEY = "zhbiA6gcQMEi22uZ8CBGvmbnD2sR8SO-5S8qlLCG"
-	rs = Rs()
-	entries = [
-		EntryPath("a", "ffdfd_9"),
-		EntryPath("a", "hello_jpg"),
-	]
-	ret, err = rs.batch_stat(entries)
-	if not err is None:
-		print 'error:', err
-		exit()
-	print ret