BucketTestCase.test_batch_copy() - Code Metrics - Inspection of "fix put_data 400" - qiniu/python-sdk - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#210)

unknown

created 2016-07-08 15:21 UTC

BucketTestCase.test_batch_copy() A

↳ Parent: BucketTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
dl	0
loc	10
rs	9.4285
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# flake8: noqa
import os
import string
import random
import tempfile
import requests

import unittest
import pytest

from qiniu import Auth, set_default, etag, PersistentFop, build_op, op_save, Zone
from qiniu import put_data, put_file, put_stream
from qiniu import BucketManager, build_batch_copy, build_batch_rename, build_batch_move, build_batch_stat, build_batch_delete
from qiniu import urlsafe_base64_encode, urlsafe_base64_decode

from qiniu.compat import is_py2, is_py3, b

from qiniu.services.storage.uploader import _form_put

import qiniu.config

if is_py2:
    import sys
    import StringIO
    import urllib
    reload(sys)
    sys.setdefaultencoding('utf-8')
    StringIO = StringIO.StringIO
    urlopen = urllib.urlopen
elif is_py3:
    import io
    import urllib
    StringIO = io.StringIO
    urlopen = urllib.request.urlopen

access_key = os.getenv('QINIU_ACCESS_KEY')
secret_key = os.getenv('QINIU_SECRET_KEY')
bucket_name = os.getenv('QINIU_TEST_BUCKET')

dummy_access_key = 'abcdefghklmnopq'
dummy_secret_key = '1234567890'
dummy_auth = Auth(dummy_access_key, dummy_secret_key)


def rand_string(length):
    lib = string.ascii_uppercase
    return ''.join([random.choice(lib) for i in range(0, length)])


def create_temp_file(size):
    t = tempfile.mktemp()
    f = open(t, 'wb')
    f.seek(size-1)
    f.write(b('0'))
    f.close()
    return t


def remove_temp_file(file):
    try:
        os.remove(file)
    except OSError:
        pass


def is_travis():
    return os.environ['QINIU_TEST_ENV'] == 'travis'


class UtilsTest(unittest.TestCase):

    def test_urlsafe(self):
        a = 'hello\x96'
        u = urlsafe_base64_encode(a)
        assert b(a) == urlsafe_base64_decode(u)


class AuthTestCase(unittest.TestCase):

    def test_token(self):
        token = dummy_auth.token('test')
        assert token == 'abcdefghklmnopq:mSNBTR7uS2crJsyFr2Amwv1LaYg='

    def test_token_with_data(self):
        token = dummy_auth.token_with_data('test')
        assert token == 'abcdefghklmnopq:-jP8eEV9v48MkYiBGs81aDxl60E=:dGVzdA=='

    def test_noKey(self):
        with pytest.raises(ValueError):
            Auth(None, None).token('nokey')
        with pytest.raises(ValueError):
            Auth('', '').token('nokey')

    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_deprecatedPolicy(self):
        with pytest.raises(ValueError):
            dummy_auth.upload_token('1', None, policy={'asyncOps': 1})

    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)
    bucket = BucketManager(q)

    def test_list(self):
        ret, eof, info = self.bucket.list(bucket_name, limit=4)
        print(info)
        assert eof is False
        assert len(ret.get('items')) == 4
        ret, eof, info = self.bucket.list(bucket_name, limit=100)
        print(info)
        assert eof is True

    def test_buckets(self):
        ret, info = self.bucket.buckets()
        print(info)
        assert bucket_name in ret

    def test_prefetch(self):
        ret, info = self.bucket.prefetch(bucket_name, 'python-sdk.html')
        print(info)
        assert ret['key'] == 'python-sdk.html'

    def test_fetch(self):
        ret, info = self.bucket.fetch('http://developer.qiniu.com/docs/v6/sdk/python-sdk.html', bucket_name, 'fetch.html')
        print(info)
        assert ret['key'] == 'fetch.html'
        assert 'hash' in ret

    def test_fetch_without_key(self):
        ret, info = self.bucket.fetch('http://developer.qiniu.com/docs/v6/sdk/python-sdk.html', bucket_name)
        print(info)
        assert ret['key'] == ret['hash']
        assert 'hash' in ret

    def test_stat(self):
        ret, info = self.bucket.stat(bucket_name, 'python-sdk.html')
        print(info)
        assert 'hash' in ret

    def test_delete(self):
        ret, info = self.bucket.delete(bucket_name, 'del')
        print(info)
        assert ret is None
        assert info.status_code == 612

    def test_rename(self):
        key = 'renameto'+rand_string(8)
        self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
        key2 = key + 'move'
        ret, info = self.bucket.rename(bucket_name, key, key2)
        print(info)
        assert ret == {}
        ret, info = self.bucket.delete(bucket_name, key2)
        print(info)
        assert ret == {}

    def test_copy(self):
        key = 'copyto'+rand_string(8)
        ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
        print(info)
        assert ret == {}
        ret, info = self.bucket.delete(bucket_name, key)
        print(info)
        assert ret == {}

    def test_change_mime(self):
        ret, info = self.bucket.change_mime(bucket_name, 'python-sdk.html', 'text/html')
        print(info)
        assert ret == {}

    def test_copy(self):
        key = 'copyto'+rand_string(8)
        ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
        print(info)
        assert ret == {}
        ret, info = self.bucket.delete(bucket_name, key)
        print(info)
        assert ret == {}

    def test_copy_force(self):
        ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, 'copyfrom', force='true')
        print(info)
        assert info.status_code == 200

    def test_change_mime(self):
        ret, info = self.bucket.change_mime(bucket_name, 'python-sdk.html', 'text/html')
        print(info)
        assert ret == {}

    def test_batch_copy(self):
        key = 'copyto'+rand_string(8)
        ops = build_batch_copy(bucket_name, {'copyfrom': key}, bucket_name)
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200
        ops = build_batch_delete(bucket_name, [key])
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200

    def test_batch_copy_force(self):
        ops = build_batch_copy(bucket_name, {'copyfrom': 'copyfrom'}, bucket_name, force='true')
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200

    def test_batch_move(self):

        key = 'moveto'+rand_string(8)
        self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
        key2 = key + 'move'
        ops = build_batch_move(bucket_name, {key: key2}, bucket_name)
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200
        ret, info = self.bucket.delete(bucket_name, key2)
        print(info)
        assert ret == {}

    def test_batch_move_force(self):

        ret,info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, 'copyfrom', force='true')
        print(info)
        assert info.status_code == 200
        ops = build_batch_move(bucket_name, {'copyfrom':'copyfrom'}, bucket_name,force='true')
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200

    def test_batch_rename(self):

        key = 'rename'+rand_string(8)
        self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
        key2 = key + 'rename'
        ops = build_batch_move(bucket_name, {key: key2}, bucket_name)
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200
        ret, info = self.bucket.delete(bucket_name, key2)
        print(info)
        assert ret == {}

    def test_batch_rename_force(self):

        ret,info = self.bucket.rename(bucket_name, 'copyfrom', 'copyfrom', force='true')
        print(info)
        assert info.status_code == 200
        ops = build_batch_rename(bucket_name, {'copyfrom':'copyfrom'}, force='true')
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200

    def test_batch_stat(self):
        ops = build_batch_stat(bucket_name, ['python-sdk.html'])
        ret, info = self.bucket.batch(ops)
        print(info)
        assert ret[0]['code'] == 200


class UploaderTestCase(unittest.TestCase):

    mime_type = "text/plain"
    params = {'x:a': 'a'}
    q = Auth(access_key, secret_key)

    def test_put(self):
        key = 'a\\b\\c"hello'
        data = 'hello bubby!'
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret['key'] == key

    def test_put_crc(self):
        key = ''
        data = 'hello bubby!'
        token = self.q.upload_token(bucket_name, key)
        ret, info = put_data(token, key, data, check_crc=True)
        print(info)
        assert ret['key'] == key

    def test_putfile(self):
        localfile = __file__
        key = 'test_file'

        token = self.q.upload_token(bucket_name, key)
        ret, info = put_file(token, key, localfile, mime_type=self.mime_type, check_crc=True)
        print(info)
        assert ret['key'] == key
        assert ret['hash'] == etag(localfile)

    def test_putInvalidCrc(self):
        key = 'test_invalid'
        data = 'hello bubby!'
        crc32 = 'wrong crc32'
        token = self.q.upload_token(bucket_name)
        ret, info = _form_put(token, key, data, None, None, crc=crc32)
        print(info)
        assert ret is None
        assert info.status_code == 400

    def test_putWithoutKey(self):
        key = None
        data = 'hello bubby!'
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret['hash'] == ret['key']

        data = 'hello bubby!'
        token = self.q.upload_token(bucket_name, 'nokey2')
        ret, info = put_data(token, None, data)
        print(info)
        assert ret is None
        assert info.status_code == 403  # key not match

    def test_withoutRead_withoutSeek_retry(self):

        key = 'retry'
        data = 'hello retry!'
        set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret['key'] == key
        assert ret['hash'] == 'FlYu0iBR1WpvYi4whKXiBuQpyLLk'
        qiniu.set_default(default_zone=qiniu.config.zone0)

    def test_hasRead_hasSeek_retry(self):

        key = 'withReadAndSeek_retry'
        data = StringIO('hello retry again!')
        set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret['key'] == key
        assert ret['hash'] == 'FuEbdt6JP2BqwQJi7PezYhmuVYOo'
        qiniu.set_default(default_zone=qiniu.config.zone0)

    def test_hasRead_withoutSeek_retry(self):
        key = 'withReadAndWithoutSeek_retry'
        data = ReadWithoutSeek('I only have read attribute!')
        set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret is None
        qiniu.set_default(default_zone=qiniu.config.zone0)

    def test_hasRead_WithoutSeek_retry2(self):
        key = 'withReadAndWithoutSeek_retry2'
        data = urlopen("http://www.qiniu.com")
        set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        token = self.q.upload_token(bucket_name)
        ret, info = put_data(token, key, data)
        print(info)
        assert ret is None
        qiniu.set_default(default_zone=qiniu.config.zone0)

    def test_putData_without_fname(self):
        if is_travis():
            return
        localfile = create_temp_file(30 * 1024 * 1024)
        key = 'test_putData_without_fname'
        with open(localfile, 'rb') as input_stream:
            token = self.q.upload_token(bucket_name)
            ret, info = put_data(token, key, input_stream)
            print(info)
            assert ret is not None

    def test_putData_without_fname1(self):
        if is_travis():
            return
        localfile = create_temp_file(30 * 1024 * 1024)
        key = 'test_putData_without_fname1'
        with open(localfile, 'rb') as input_stream:
            token = self.q.upload_token(bucket_name)
            ret, info = put_data(token, key, input_stream, self.params, self.mime_type, False, None, "")
            print(info)
            assert ret is not None

    def test_putData_without_fname2(self):
        if is_travis():
            return
        localfile = create_temp_file(30 * 1024 * 1024)
        key = 'test_putData_without_fname2'
        with open(localfile, 'rb') as input_stream:
            token = self.q.upload_token(bucket_name)
            ret, info = put_data(token, key, input_stream, self.params, self.mime_type, False, None, "  ")
            print(info)
            assert ret is not None


class ResumableUploaderTestCase(unittest.TestCase):

    mime_type = "text/plain"
    params = {'x:a': 'a'}
    q = Auth(access_key, secret_key)

    def test_put_stream(self):
        localfile = __file__
        key = 'test_file_r'
        size = os.stat(localfile).st_size
        with open(localfile, 'rb') as input_stream:
            token = self.q.upload_token(bucket_name, key)
            ret, info = put_stream(token, key, input_stream, os.path.basename(__file__), size, self.params, self.mime_type)
            print(info)
            assert ret['key'] == key

    def test_big_file(self):

        key = 'big'
        token = self.q.upload_token(bucket_name, key)
        localfile = create_temp_file(4 * 1024 * 1024 + 1)
        progress_handler = lambda progress, total: progress
        qiniu.set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        ret, info = put_file(token, key, localfile, self.params, self.mime_type, progress_handler=progress_handler)
        print(info)
        assert ret['key'] == key
        qiniu.set_default(default_zone=qiniu.config.zone0)
        remove_temp_file(localfile)

    def test_retry(self):

        localfile = __file__
        key = 'test_file_r_retry'
        qiniu.set_default(default_zone=Zone('a', 'upload.qiniu.com'))
        token = self.q.upload_token(bucket_name, key)
        ret, info = put_file(token, key, localfile, self.params, self.mime_type)
        print(info)
        assert ret['key'] == key
        assert ret['hash'] == etag(localfile)
        qiniu.set_default(default_zone=qiniu.config.zone0)


class DownloadTestCase(unittest.TestCase):

    q = Auth(access_key, secret_key)

    def test_private_url(self):
        private_bucket = 'private-res'
        private_key = 'gogopher.jpg'
        base_url = 'http://%s/%s' % (private_bucket+'.qiniudn.com', private_key)
        private_url = self.q.private_download_url(base_url, expires=3600)
        print(private_url)
        r = requests.get(private_url)
        assert r.status_code == 200


class MediaTestCase(unittest.TestCase):
    def test_pfop(self):
        q = Auth(access_key, secret_key)
        pfop = PersistentFop(q, 'testres', 'sdktest')
        op = op_save('avthumb/m3u8/segtime/10/vcodec/libx264/s/320x240', 'pythonsdk', 'pfoptest')
        ops = []
        ops.append(op)
        ret, info = pfop.execute('sintel_trailer.mp4', ops, 1)
        print(info)
        assert ret['persistentId'] is not None


class EtagTestCase(unittest.TestCase):
    def test_zero_size(self):
        open("x", 'a').close()
        hash = etag("x")
        assert hash == 'Fto5o-5ea0sNMlW_75VgGJCv2AcJ'
        remove_temp_file("x")
    def test_small_size(self):
        localfile = create_temp_file(1024 * 1024)
        hash = etag(localfile)
        assert hash == 'FnlAdmDasGTQOIgrU1QIZaGDv_1D'
        remove_temp_file(localfile)
    def test_large_size(self):
        localfile = create_temp_file(4 * 1024 * 1024 + 1)
        hash = etag(localfile)
        assert hash == 'ljF323utglY3GI6AvLgawSJ4_dgk'
        remove_temp_file(localfile)


class ReadWithoutSeek(object):
    def __init__(self, str):
        self.str = str
        pass

    def read(self):
        print(self.str)

if __name__ == '__main__':
    unittest.main()


1		# -- coding: utf-8 --
2		# flake8: noqa
3		import os
4		import string
5		import random
6		import tempfile
7		import requests
8
9		import unittest
10		import pytest
11
12		from qiniu import Auth, set_default, etag, PersistentFop, build_op, op_save, Zone
13		from qiniu import put_data, put_file, put_stream
14		from qiniu import BucketManager, build_batch_copy, build_batch_rename, build_batch_move, build_batch_stat, build_batch_delete
15		from qiniu import urlsafe_base64_encode, urlsafe_base64_decode
16
17		from qiniu.compat import is_py2, is_py3, b
18
19		from qiniu.services.storage.uploader import _form_put
20
21		import qiniu.config
22
23		if is_py2:
24		import sys
25		import StringIO
26		import urllib
27		reload(sys)
28		sys.setdefaultencoding('utf-8')
29		StringIO = StringIO.StringIO
30		urlopen = urllib.urlopen
31		elif is_py3:
32		import io
33		import urllib
34		StringIO = io.StringIO
35		urlopen = urllib.request.urlopen
36
37		access_key = os.getenv('QINIU_ACCESS_KEY')
38		secret_key = os.getenv('QINIU_SECRET_KEY')
39		bucket_name = os.getenv('QINIU_TEST_BUCKET')
40
41		dummy_access_key = 'abcdefghklmnopq'
42		dummy_secret_key = '1234567890'
43		dummy_auth = Auth(dummy_access_key, dummy_secret_key)
44
45
46		def rand_string(length):
47		lib = string.ascii_uppercase
48		return ''.join([random.choice(lib) for i in range(0, length)])
49
50
51		def create_temp_file(size):
52		t = tempfile.mktemp()
53		f = open(t, 'wb')
54		f.seek(size-1)
55		f.write(b('0'))
56		f.close()
57		return t
58
59
60		def remove_temp_file(file):
61		try:
62		os.remove(file)
63		except OSError:
64		pass
65
66
67		def is_travis():
68		return os.environ['QINIU_TEST_ENV'] == 'travis'
69
70
71		class UtilsTest(unittest.TestCase):
72
73		def test_urlsafe(self):
74		a = 'hello\x96'
75		u = urlsafe_base64_encode(a)
76		assert b(a) == urlsafe_base64_decode(u)
77
78
79		class AuthTestCase(unittest.TestCase):
80
81		def test_token(self):
82		token = dummy_auth.token('test')
83		assert token == 'abcdefghklmnopq:mSNBTR7uS2crJsyFr2Amwv1LaYg='
84
85		def test_token_with_data(self):
86		token = dummy_auth.token_with_data('test')
87		assert token == 'abcdefghklmnopq:-jP8eEV9v48MkYiBGs81aDxl60E=:dGVzdA=='
88
89		def test_noKey(self):
90		with pytest.raises(ValueError):
91		Auth(None, None).token('nokey')
92		with pytest.raises(ValueError):
93		Auth('', '').token('nokey')
94
95		def test_token_of_request(self):
96		token = dummy_auth.token_of_request('http://www.qiniu.com?go=1', 'test', '')
97		assert token == 'abcdefghklmnopq:cFyRVoWrE3IugPIMP5YJFTO-O-Y='
98		token = dummy_auth.token_of_request('http://www.qiniu.com?go=1', 'test', 'application/x-www-form-urlencoded')
99		assert token == 'abcdefghklmnopq:svWRNcacOE-YMsc70nuIYdaa1e4='
100
101		def test_deprecatedPolicy(self):
102		with pytest.raises(ValueError):
103		dummy_auth.upload_token('1', None, policy={'asyncOps': 1})
104
105		def test_verify_callback(self):
106		body = 'name=sunflower.jpg&hash=Fn6qeQi4VDLQ347NiRm-RlQx_4O2&location=Shanghai&price=1500.00&uid=123'
107		url = 'test.qiniu.com/callback'
108		ok = dummy_auth.verify_callback('QBox abcdefghklmnopq:ZWyeM5ljWMRFwuPTPOwQ4RwSto4=', url, body)
109		assert ok
110
111
112		class BucketTestCase(unittest.TestCase):
113		q = Auth(access_key, secret_key)
114		bucket = BucketManager(q)
115
116		def test_list(self):
117		ret, eof, info = self.bucket.list(bucket_name, limit=4)
118		print(info)
119		assert eof is False
120		assert len(ret.get('items')) == 4
121		ret, eof, info = self.bucket.list(bucket_name, limit=100)
122		print(info)
123		assert eof is True
124
125		def test_buckets(self):
126		ret, info = self.bucket.buckets()
127		print(info)
128		assert bucket_name in ret
129
130		def test_prefetch(self):
131		ret, info = self.bucket.prefetch(bucket_name, 'python-sdk.html')
132		print(info)
133		assert ret['key'] == 'python-sdk.html'
134
135		def test_fetch(self):
136		ret, info = self.bucket.fetch('http://developer.qiniu.com/docs/v6/sdk/python-sdk.html', bucket_name, 'fetch.html')
137		print(info)
138		assert ret['key'] == 'fetch.html'
139		assert 'hash' in ret
140
141		def test_fetch_without_key(self):
142		ret, info = self.bucket.fetch('http://developer.qiniu.com/docs/v6/sdk/python-sdk.html', bucket_name)
143		print(info)
144		assert ret['key'] == ret['hash']
145		assert 'hash' in ret
146
147		def test_stat(self):
148		ret, info = self.bucket.stat(bucket_name, 'python-sdk.html')
149		print(info)
150		assert 'hash' in ret
151
152		def test_delete(self):
153		ret, info = self.bucket.delete(bucket_name, 'del')
154		print(info)
155		assert ret is None
156		assert info.status_code == 612
157
158		def test_rename(self):
159		key = 'renameto'+rand_string(8)
160		self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
161		key2 = key + 'move'
162		ret, info = self.bucket.rename(bucket_name, key, key2)
163		print(info)
164		assert ret == {}
165		ret, info = self.bucket.delete(bucket_name, key2)
166		print(info)
167		assert ret == {}
168
169		def test_copy(self):
170		key = 'copyto'+rand_string(8)
171		ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
172		print(info)
173		assert ret == {}
174		ret, info = self.bucket.delete(bucket_name, key)
175		print(info)
176		assert ret == {}
177
178		def test_change_mime(self):
179		ret, info = self.bucket.change_mime(bucket_name, 'python-sdk.html', 'text/html')
180		print(info)
181		assert ret == {}
182
183		def test_copy(self):
184		key = 'copyto'+rand_string(8)
185		ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
186		print(info)
187		assert ret == {}
188		ret, info = self.bucket.delete(bucket_name, key)
189		print(info)
190		assert ret == {}
191
192		def test_copy_force(self):
193		ret, info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, 'copyfrom', force='true')
194		print(info)
195		assert info.status_code == 200
196
197		def test_change_mime(self):
198		ret, info = self.bucket.change_mime(bucket_name, 'python-sdk.html', 'text/html')
199		print(info)
200		assert ret == {}
201
202		def test_batch_copy(self):
203		key = 'copyto'+rand_string(8)
204		ops = build_batch_copy(bucket_name, {'copyfrom': key}, bucket_name)
205		ret, info = self.bucket.batch(ops)
206		print(info)
207		assert ret[0]['code'] == 200
208		ops = build_batch_delete(bucket_name, [key])
209		ret, info = self.bucket.batch(ops)
210		print(info)
211		assert ret[0]['code'] == 200
212
213		def test_batch_copy_force(self):
214		ops = build_batch_copy(bucket_name, {'copyfrom': 'copyfrom'}, bucket_name, force='true')
215		ret, info = self.bucket.batch(ops)
216		print(info)
217		assert ret[0]['code'] == 200
218
219	View Code Duplication	def test_batch_move(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
220		key = 'moveto'+rand_string(8)
221		self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
222		key2 = key + 'move'
223		ops = build_batch_move(bucket_name, {key: key2}, bucket_name)
224		ret, info = self.bucket.batch(ops)
225		print(info)
226		assert ret[0]['code'] == 200
227		ret, info = self.bucket.delete(bucket_name, key2)
228		print(info)
229		assert ret == {}
230
231	View Code Duplication	def test_batch_move_force(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
232		ret,info = self.bucket.copy(bucket_name, 'copyfrom', bucket_name, 'copyfrom', force='true')
233		print(info)
234		assert info.status_code == 200
235		ops = build_batch_move(bucket_name, {'copyfrom':'copyfrom'}, bucket_name,force='true')
236		ret, info = self.bucket.batch(ops)
237		print(info)
238		assert ret[0]['code'] == 200
239
240	View Code Duplication	def test_batch_rename(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
241		key = 'rename'+rand_string(8)
242		self.bucket.copy(bucket_name, 'copyfrom', bucket_name, key)
243		key2 = key + 'rename'
244		ops = build_batch_move(bucket_name, {key: key2}, bucket_name)
245		ret, info = self.bucket.batch(ops)
246		print(info)
247		assert ret[0]['code'] == 200
248		ret, info = self.bucket.delete(bucket_name, key2)
249		print(info)
250		assert ret == {}
251
252	View Code Duplication	def test_batch_rename_force(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
253		ret,info = self.bucket.rename(bucket_name, 'copyfrom', 'copyfrom', force='true')
254		print(info)
255		assert info.status_code == 200
256		ops = build_batch_rename(bucket_name, {'copyfrom':'copyfrom'}, force='true')
257		ret, info = self.bucket.batch(ops)
258		print(info)
259		assert ret[0]['code'] == 200
260
261		def test_batch_stat(self):
262		ops = build_batch_stat(bucket_name, ['python-sdk.html'])
263		ret, info = self.bucket.batch(ops)
264		print(info)
265		assert ret[0]['code'] == 200
266
267
268		class UploaderTestCase(unittest.TestCase):
269
270		mime_type = "text/plain"
271		params = {'x:a': 'a'}
272		q = Auth(access_key, secret_key)
273
274		def test_put(self):
275		key = 'a\\b\\c"hello'
276		data = 'hello bubby!'
277		token = self.q.upload_token(bucket_name)
278		ret, info = put_data(token, key, data)
279		print(info)
280		assert ret['key'] == key
281
282		def test_put_crc(self):
283		key = ''
284		data = 'hello bubby!'
285		token = self.q.upload_token(bucket_name, key)
286		ret, info = put_data(token, key, data, check_crc=True)
287		print(info)
288		assert ret['key'] == key
289
290		def test_putfile(self):
291		localfile = __file__
292		key = 'test_file'
293
294		token = self.q.upload_token(bucket_name, key)
295		ret, info = put_file(token, key, localfile, mime_type=self.mime_type, check_crc=True)
296		print(info)
297		assert ret['key'] == key
298		assert ret['hash'] == etag(localfile)
299
300		def test_putInvalidCrc(self):
301		key = 'test_invalid'
302		data = 'hello bubby!'
303		crc32 = 'wrong crc32'
304		token = self.q.upload_token(bucket_name)
305		ret, info = _form_put(token, key, data, None, None, crc=crc32)
306		print(info)
307		assert ret is None
308		assert info.status_code == 400
309
310		def test_putWithoutKey(self):
311		key = None
312		data = 'hello bubby!'
313		token = self.q.upload_token(bucket_name)
314		ret, info = put_data(token, key, data)
315		print(info)
316		assert ret['hash'] == ret['key']
317
318		data = 'hello bubby!'
319		token = self.q.upload_token(bucket_name, 'nokey2')
320		ret, info = put_data(token, None, data)
321		print(info)
322		assert ret is None
323		assert info.status_code == 403 # key not match
324
325	View Code Duplication	def test_withoutRead_withoutSeek_retry(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
326		key = 'retry'
327		data = 'hello retry!'
328		set_default(default_zone=Zone('a', 'upload.qiniu.com'))
329		token = self.q.upload_token(bucket_name)
330		ret, info = put_data(token, key, data)
331		print(info)
332		assert ret['key'] == key
333		assert ret['hash'] == 'FlYu0iBR1WpvYi4whKXiBuQpyLLk'
334		qiniu.set_default(default_zone=qiniu.config.zone0)
335
336	View Code Duplication	def test_hasRead_hasSeek_retry(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
337		key = 'withReadAndSeek_retry'
338		data = StringIO('hello retry again!')
339		set_default(default_zone=Zone('a', 'upload.qiniu.com'))
340		token = self.q.upload_token(bucket_name)
341		ret, info = put_data(token, key, data)
342		print(info)
343		assert ret['key'] == key
344		assert ret['hash'] == 'FuEbdt6JP2BqwQJi7PezYhmuVYOo'
345		qiniu.set_default(default_zone=qiniu.config.zone0)
346
347		def test_hasRead_withoutSeek_retry(self):
348		key = 'withReadAndWithoutSeek_retry'
349		data = ReadWithoutSeek('I only have read attribute!')
350		set_default(default_zone=Zone('a', 'upload.qiniu.com'))
351		token = self.q.upload_token(bucket_name)
352		ret, info = put_data(token, key, data)
353		print(info)
354		assert ret is None
355		qiniu.set_default(default_zone=qiniu.config.zone0)
356
357		def test_hasRead_WithoutSeek_retry2(self):
358		key = 'withReadAndWithoutSeek_retry2'
359		data = urlopen("http://www.qiniu.com")
360		set_default(default_zone=Zone('a', 'upload.qiniu.com'))
361		token = self.q.upload_token(bucket_name)
362		ret, info = put_data(token, key, data)
363		print(info)
364		assert ret is None
365		qiniu.set_default(default_zone=qiniu.config.zone0)
366
367		def test_putData_without_fname(self):
368		if is_travis():
369		return
370		localfile = create_temp_file(30 * 1024 * 1024)
371		key = 'test_putData_without_fname'
372		with open(localfile, 'rb') as input_stream:
373		token = self.q.upload_token(bucket_name)
374		ret, info = put_data(token, key, input_stream)
375		print(info)
376		assert ret is not None
377
378		def test_putData_without_fname1(self):
379		if is_travis():
380		return
381		localfile = create_temp_file(30 * 1024 * 1024)
382		key = 'test_putData_without_fname1'
383		with open(localfile, 'rb') as input_stream:
384		token = self.q.upload_token(bucket_name)
385		ret, info = put_data(token, key, input_stream, self.params, self.mime_type, False, None, "")
386		print(info)
387		assert ret is not None
388
389		def test_putData_without_fname2(self):
390		if is_travis():
391		return
392		localfile = create_temp_file(30 * 1024 * 1024)
393		key = 'test_putData_without_fname2'
394		with open(localfile, 'rb') as input_stream:
395		token = self.q.upload_token(bucket_name)
396		ret, info = put_data(token, key, input_stream, self.params, self.mime_type, False, None, " ")
397		print(info)
398		assert ret is not None
399
400
401		class ResumableUploaderTestCase(unittest.TestCase):
402
403		mime_type = "text/plain"
404		params = {'x:a': 'a'}
405		q = Auth(access_key, secret_key)
406
407		def test_put_stream(self):
408		localfile = __file__
409		key = 'test_file_r'
410		size = os.stat(localfile).st_size
411		with open(localfile, 'rb') as input_stream:
412		token = self.q.upload_token(bucket_name, key)
413		ret, info = put_stream(token, key, input_stream, os.path.basename(__file__), size, self.params, self.mime_type)
414		print(info)
415		assert ret['key'] == key
416
417	View Code Duplication	def test_big_file(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
418		key = 'big'
419		token = self.q.upload_token(bucket_name, key)
420		localfile = create_temp_file(4 * 1024 * 1024 + 1)
421		progress_handler = lambda progress, total: progress
422		qiniu.set_default(default_zone=Zone('a', 'upload.qiniu.com'))
423		ret, info = put_file(token, key, localfile, self.params, self.mime_type, progress_handler=progress_handler)
424		print(info)
425		assert ret['key'] == key
426		qiniu.set_default(default_zone=qiniu.config.zone0)
427		remove_temp_file(localfile)
428
429	View Code Duplication	def test_retry(self):
		0 ignored issues – show Duplication introduced 2016-07-08 15:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
430		localfile = __file__
431		key = 'test_file_r_retry'
432		qiniu.set_default(default_zone=Zone('a', 'upload.qiniu.com'))
433		token = self.q.upload_token(bucket_name, key)
434		ret, info = put_file(token, key, localfile, self.params, self.mime_type)
435		print(info)
436		assert ret['key'] == key
437		assert ret['hash'] == etag(localfile)
438		qiniu.set_default(default_zone=qiniu.config.zone0)
439
440
441		class DownloadTestCase(unittest.TestCase):
442
443		q = Auth(access_key, secret_key)
444
445		def test_private_url(self):
446		private_bucket = 'private-res'
447		private_key = 'gogopher.jpg'
448		base_url = 'http://%s/%s' % (private_bucket+'.qiniudn.com', private_key)
449		private_url = self.q.private_download_url(base_url, expires=3600)
450		print(private_url)
451		r = requests.get(private_url)
452		assert r.status_code == 200
453
454
455		class MediaTestCase(unittest.TestCase):
456		def test_pfop(self):
457		q = Auth(access_key, secret_key)
458		pfop = PersistentFop(q, 'testres', 'sdktest')
459		op = op_save('avthumb/m3u8/segtime/10/vcodec/libx264/s/320x240', 'pythonsdk', 'pfoptest')
460		ops = []
461		ops.append(op)
462		ret, info = pfop.execute('sintel_trailer.mp4', ops, 1)
463		print(info)
464		assert ret['persistentId'] is not None
465
466
467		class EtagTestCase(unittest.TestCase):
468		def test_zero_size(self):
469		open("x", 'a').close()
470		hash = etag("x")
471		assert hash == 'Fto5o-5ea0sNMlW_75VgGJCv2AcJ'
472		remove_temp_file("x")
473		def test_small_size(self):
474		localfile = create_temp_file(1024 * 1024)
475		hash = etag(localfile)
476		assert hash == 'FnlAdmDasGTQOIgrU1QIZaGDv_1D'
477		remove_temp_file(localfile)
478		def test_large_size(self):
479		localfile = create_temp_file(4 * 1024 * 1024 + 1)
480		hash = etag(localfile)
481		assert hash == 'ljF323utglY3GI6AvLgawSJ4_dgk'
482		remove_temp_file(localfile)
483
484
485		class ReadWithoutSeek(object):
486		def __init__(self, str):
487		self.str = str
488		pass
489
490		def read(self):
491		print(self.str)
492
493		if __name__ == '__main__':
494		unittest.main()
495

qiniu / python-sdk

Pull Request — master (#210)

BucketTestCase.test_batch_copy() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like