class ClientError(OssError):
    def __init__(self, message):
        OssError.__init__(self, OSS_CLIENT_ERROR_STATUS, {}, 'ClientError: ' + message, {})

    def __str__(self):
        return self._str_with_body() class RequestError(OssError):
    def __init__(self, e):
        OssError.__init__(self, OSS_REQUEST_ERROR_STATUS, {}, 'RequestError: ' + str(e), {})
        self.exception = e

    def __str__(self):
        return self._str_with_body() class NoSuchBucket(NotFound):
    status = 404
    code = 'NoSuchBucket' class OpenApiServerError(OssError):
    def __init__(self, status, request_id, message, error_code):
        OssError.__init__(self, status, {OSS_REQUEST_ID : request_id}, '', {'Code': error_code, 'Message': message}) class NotFound(ServerError):
    status = 404
    code = '' class NoSuchKey(NotFound):
    status = 404
    code = 'NoSuchKey' class Conflict(ServerError):
    status = 409
    code = '' class PositionNotEqualToLength(Conflict):
    status = 409
    code = 'PositionNotEqualToLength'

    def __init__(self, status, headers, body, details):
        super(PositionNotEqualToLength, self).__init__(status, headers, body, details)
        self.next_position = int(headers[OSS_NEXT_APPEND_POSITION]) class ObjectNotAppendable(Conflict):
    status = 409
    code = 'ObjectNotAppendable' def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input):
def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input): class Tagging(object):

    def __init__(self, tagging_rules=None):
        
        self.tag_set = tagging_rules or TaggingRule() 

    def __str__(self):

        tag_str = ""
        
        tagging_rule = self.tag_set.tagging_rule

        for key in tagging_rule:
            tag_str += key
            tag_str += "#" + tagging_rule[key] + " "

        return tag_str class TaggingRule(object):

    def __init__(self):
        self.tagging_rule = dict()

    def add(self, key, value):

        if key is None or key == '':
            raise ClientError("Tagging key should not be empty")

        if len(key) > _MAX_OBJECT_TAGGING_KEY_LENGTH:
            raise ClientError("Tagging key is too long")

        if len(value) > _MAX_OBJECT_TAGGING_VALUE_LENGTH:
            raise ClientError("Tagging value is too long")

        self.tagging_rule[key] = value

    def delete(self, key):
        del self.tagging_rule[key]

    def len(self):
        return len(self.tagging_rule)

    def to_query_string(self):
        query_string = ''

        for key in self.tagging_rule:
            query_string += urlquote(key)
            query_string += '='
            query_string += urlquote(self.tagging_rule[key])
            query_string += '&'

        if len(query_string) == 0:
            return ''
        else:
            query_string = query_string[:-1]

        return query_string OSS_OBJECT_TAGGING = "x-oss-tagging" OSS_OBJECT_TAGGING_COPY_DIRECTIVE = "x-oss-tagging-directive" def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input):
def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input):
import requests
import filecmp
import calendar
import json
import base64
from oss2.exceptions import (ClientError, RequestError, NoSuchBucket, OpenApiServerError,
        NotFound, NoSuchKey, Conflict, PositionNotEqualToLength, ObjectNotAppendable)
from oss2.compat import is_py2, is_py33
from oss2.models import Tagging, TaggingRule
from oss2.headers import OSS_OBJECT_TAGGING, OSS_OBJECT_TAGGING_COPY_DIRECTIVE
from oss2.compat import urlunquote, urlquote, to_bytes, to_string
from .common import *
        from oss2.models import (RestoreJobParameters, RestoreConfiguration, RESTORE_TIER_EXPEDITED,
                                 RESTORE_TIER_STANDARD, RESTORE_TIER_BULK)
        from oss2.models import RestoreJobParameters, RestoreConfiguration
        from oss2.models import RestoreJobParameters, RestoreConfiguration, RESTORE_TIER_BULK
        self.assertTrue(get_result.client_crc is None)
        self.assertTrue(get_result.server_crc)
        
        bucket.delete_object(key)
        
        # append
        append_result = bucket.append_object(key, 0, content)
        self.assertFalse(hasattr(append_result, 'get_crc'))
        self.assertTrue(append_result.crc is not None)
        
        append_result = bucket.append_object(key, len(content), content)
        self.assertFalse(hasattr(append_result, 'get_crc'))
        self.assertTrue(append_result.crc is not None)
        
        bucket.delete_object(key)
        
        # multipart
        upload_id = bucket.init_multipart_upload(key).upload_id

        parts = []
        result = bucket.upload_part(key, upload_id, 1, content)
        parts.append(oss2.models.PartInfo(1, result.etag))
        result = bucket.upload_part(key, upload_id, 2, content)
        parts.append(oss2.models.PartInfo(2, result.etag))

        bucket.complete_multipart_upload(key, upload_id, parts)

    def test_invalid_crc(self):
        key = self.random_key()
        content = random_bytes(512)

        try:
            self.bucket.append_object(key, 0, content, init_crc=1)
        except oss2.exceptions.InconsistentError as e:
            self.assertEqual(e.status, -3)
            self.assertTrue(e.body.startswith('InconsistentError'))
        else:
            self.assertTrue(False)

    def test_put_symlink(self):
        key  = self.random_key()
        symlink = self.random_key()
        content = 'hello'
        
        self.bucket.put_object(key, content)
        
        # put symlink normal
        self.bucket.put_symlink(key, symlink)
        
        head_result = self.bucket.head_object(symlink)
        self.assertEqual(head_result.content_length, len(content))
        self.assertEqual(head_result.etag, '5D41402ABC4B2A76B9719D911017C592')

        self.bucket.put_object(key, content)
        
        # put symlink with meta
        self.bucket.put_symlink(key, symlink, headers={'x-oss-meta-key1': 'value1',
                                                              'X-Oss-Meta-Key2': 'value2'})
        head_result = self.bucket.head_object(symlink)
        self.assertEqual(head_result.content_length, len(content))
        self.assertEqual(head_result.etag, '5D41402ABC4B2A76B9719D911017C592')
        self.assertEqual(head_result.headers['x-oss-meta-key1'], 'value1')
        self.assertEqual(head_result.headers['x-oss-meta-key2'], 'value2')

    def test_get_symlink(self):
        key = self.random_key()
        symlink = self.random_key()
        content = 'hello'
        
        # bucket no exist
        auth = oss2.Auth(OSS_ID, OSS_SECRET)
        bucket_name = OSS_BUCKET + "-test-get-symlink"
        bucket = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name)
        
        self.assertRaises(NoSuchBucket, bucket.get_symlink, symlink)
        
        # object no exist
        self.assertRaises(NoSuchKey, self.bucket.get_symlink, symlink)
        
        self.bucket.put_object(key, content)
        self.bucket.put_symlink(key, symlink)
        
        # get symlink normal
        result = self.bucket.get_symlink(symlink)
        self.assertEqual(result.target_key, key)

    def test_process_object(self):
        key = self.random_key(".jpg")
        result = self.bucket.put_object_from_file(key, "tests/example.jpg")
        self.assertEqual(result.status, 200)

        dest_key = self.random_key(".jpg")
        process = "image/resize,w_100|sys/saveas,o_{0},b_{1}".format(
            oss2.compat.to_string(base64.urlsafe_b64encode(oss2.compat.to_bytes(dest_key))),
            oss2.compat.to_string(base64.urlsafe_b64encode(oss2.compat.to_bytes(self.bucket.bucket_name))))
        result = self.bucket.process_object(key, process)
        self.assertEqual(result.status, 200)
        self.assertEqual(result.bucket, self.bucket.bucket_name)
        self.assertEqual(result.object, dest_key)
        result = self.bucket.object_exists(dest_key)
        self.assertEqual(result, True)


        # If bucket-name not specified, it is saved to the current bucket by default.
        dest_key = self.random_key(".jpg")
        process = "image/resize,w_100|sys/saveas,o_{0}".format(
            oss2.compat.to_string(base64.urlsafe_b64encode(oss2.compat.to_bytes(dest_key))))
        result = self.bucket.process_object(key, process)
        self.assertEqual(result.status, 200)
        self.assertEqual(result.bucket, "")
        self.assertEqual(result.object, dest_key)
        result = self.bucket.object_exists(dest_key)
        self.assertEqual(result, True)
    
    def test_object_tagging_client_error(self):

        rule = TaggingRule()
        self.assertRaises(oss2.exceptions.ClientError, rule.add, 129*'a', 'test')
        self.assertRaises(oss2.exceptions.ClientError, rule.add, 'test', 257*'a')
        self.assertRaises(oss2.exceptions.ClientError, rule.add, None, 'test')

        self.assertRaises(oss2.exceptions.ClientError, rule.add, '', 'test')