import json
import mock
import os
from os.path import sep
import io
import time
import random
from threading import Thread
from contextlib import contextmanager
import pytest
import string
import six
import posixpath
from functools import reduce, cmp_to_key

from filelike.wrappers import FileWrapper
from requests.packages.urllib3.exceptions import HTTPError

from bft.conftest import get_chroot
from ucc.abstract import AccountInfo, Permission
from ucc.baidu import BaiduClient
from ucc.local import LocalClient
from ucc.obj2path import ObjToPathDriveClient
from ucc.id2path import IdToPathDriveClient
from ucc.webdav import WebDAVDriveClient
from ucc.yandexdisk import YandexDiskClient
from ucc.hidrive import HiDriveClient
from ucc.dropbox import DropboxClient
from ucc.sharefile import ShareFileClient
from ucc.directcloud import DirectCloudClient
from ucc.rtrr import RtrrClient
from ucc.exception import (NotFoundError, ConflictError,
                           InvalidError, TooManyRequestsError,
                           ChangeNeedResetException, PermissionsError,
                           NetworkError)
from ucc.util import nfc_normalize
if os.name == "posix":
    from ucc.ratelimit import *
from ucc.common import super_len


def cmp(a, b):
    return (a > b) - (a < b)


@pytest.fixture(scope='session', autouse=True)
def test_folder(request, path_client):
    path = u'/test_path_drive'
    c = path_client
    c.mkdir(path)

    def fin():
        c.rmdirs(path)
    request.addfinalizer(fin)
    return path

def is_provider(client, expected_or_list):
    if not isinstance(expected_or_list, list):
        expected_or_list = [expected_or_list]
    for expected in expected_or_list:
        if expected == client.provider_type:
            return True
        if expected == client.provider_name:
            return True
        if client.__class__.__name__ == 'IdToPathDriveClient' and \
                        expected == client._client.__class__.__name__:
            return True
    return False

@contextmanager
def exception_raises(exception_list, client):
    name = client.__class__.__name__
    exception = None
    if client is not None:
        exception = exception_list.get(client.provider_type, '')

    if exception == '' and client is not None:
        exception = exception_list.get(client.provider_name, '')

    if exception == '' and name in ['IdToPathDriveClient', 'ObjToPathDriveClient'] and client is not None:
        exception = exception_list.get(client._client.__class__.__name__, '')

    if exception == '':
        exception = exception_list.get(name, '')

    if exception == '':
        exception = exception_list['']

    if exception is None:
        yield
    else:
        try:
            yield
        except Exception as e:
            assert (exception == e.__class__), 'excepted exception is different'
        else:
            assert exception == None, 'expected exceptoin is not raised'


def test_account_info(path_client):
    if isinstance(path_client, ObjToPathDriveClient):
        pytest.skip('Object storage client does not support account_info')
    c = path_client
    acc = c.get_account_info()
    assert isinstance(acc, AccountInfo)
    print(acc)


def test_get_metadata(path_client):
    c = path_client
    m = c.get_metadata(u'/')

    assert u'/' == m.path
    assert isinstance(m.path, six.text_type)
    assert isinstance(m.name, six.text_type)

    if isinstance(path_client, LocalClient):
        assert hasattr(m, 'stat')
        assert hasattr(m, 'is_sparse')
        assert hasattr(m, 'is_symbolic')


def test_get_metadata_not_exist(path_client):
    c = path_client
    with pytest.raises(NotFoundError):
        c.get_metadata(u'/test_get_metadata_not_exist')


def test_mkdir(path_client, test_folder, path=None):
    if not path:
        path = posixpath.join(test_folder, "mkdir%f" % time.time())

    c = path_client
    m1 = c.mkdir(path)
    print(m1)
    m2 = c.get_metadata(path)
    if m1.permission is None:  # Box return meta do not have permission
        del m2.permission
    if m1.client_mtime is None:  # Baidu return meta do not have client_mtime
        del m2.client_mtime
    assert m1 == m2
    assert os.path.basename(path) == m1.name
    assert path == m1.path
    assert isinstance(m1.path, six.text_type)
    assert isinstance(m1.name, six.text_type)
    assert not m1.is_deleted
    # TODO: Examine data struct in other test case?
    # assert m1.server_mtime  # TODO: dropbox don't have mtime for folder
    # assert not m1.client_mtime
    assert True is m1.is_dir
    assert not m1.hash

    return m1

def test_obj_dir(path_client, test_folder):
    if not isinstance(path_client, ObjToPathDriveClient):
        pytest.skip('Only for object storage')
    c = path_client
    if path_client.provider_type == 'filebase':
        # remove % in path since filebase doesn't accept
        basepath = posixpath.join(test_folder, 'obj~!@#$^&*()-+`"\';:?><,.dir')
    else:
        basepath = posixpath.join(test_folder, 'obj~!@#$%^&*()-+`"\';:?><,.dir')
    c.mkdir(basepath)
    result = c.listdir(basepath)
    if path_client.provider_type == 'azure':
        result = [r for r in result if r.name != '.azEmpty']
    elif path_client.provider_type == 'backblaze':
        result = [r for r in result if r.name != '.bzEmpty']
    assert 0 == len(result)

    dir1path = posixpath.join(basepath, 'dir1')
    c.mkdir(dir1path)
    result = c.listdir(basepath)
    if path_client.provider_type == 'azure':
        result = [r for r in result if r.name != '.azEmpty']
    elif path_client.provider_type == 'backblaze':
        result = [r for r in result if r.name != '.bzEmpty']
    assert 1 == len(result)
    d = result[0]
    assert 'dir1' == d.name
    assert dir1path == d.path
    assert d.is_dir

    file1path = posixpath.join(basepath, 'dir1/file1')
    c.upload(file1path, io.BytesIO(b'foobar'))
    result = c.listdir(basepath)
    if path_client.provider_type == 'azure':
        result = [r for r in result if r.name != '.azEmpty']
    elif path_client.provider_type == 'backblaze':
        result = [r for r in result if r.name != '.bzEmpty']
    assert 1 == len(result)
    d = result[0]
    assert 'dir1' == d.name
    assert dir1path == d.path
    assert d.is_dir


    result = list(c.scan(path=basepath))
    assert 2 == len(result)
    d = result[0]
    assert 'dir1' == d.name
    assert dir1path == d.path
    assert d.is_dir
    d = result[1]
    assert 'file1' == d.name
    assert file1path == d.path
    assert not d.is_dir

    c.delete(dir1path)
    result = c.listdir(basepath)
    assert 0 == len(result)

def test_obj_dir2(path_client, test_folder):
    if not isinstance(path_client, ObjToPathDriveClient):
        pytest.skip('Only for object storage')
    c = path_client
    basepath = posixpath.join(test_folder, 'test_obj_dir2')
    c.mkdir(posixpath.join(basepath, 'a/b/c/d'))
    result = list(c.scan(path=basepath))
    print([d.path for d in result])
    assert 4 == len(result)
    c.mkdir(posixpath.join(basepath, 'a'))
    result = list(c.scan(path=basepath))
    print([d.path for d in result])
    assert 4 == len(result)


def test_upload(path_client, test_folder, fp=None, path=None):
    # TODO: test hash
    # TODO: test client mtime
    if not path:
        path = posixpath.join(test_folder, "upload%f" % time.time())

    c = path_client
    if not fp:
        data = b'Hello world.'
        fp = io.BytesIO(data)
    else:
        data = fp.read()
        fp.seek(0)

    def cb(u, t):
        print('progress=%s/%s' % (u, t))
    m1 = c.upload(path, fp, progress_callback=cb)
    m2 = c.get_metadata(path)
    m1.server_mtime = m2.server_mtime  # ACD: server mtime is diff
    _, basename = os.path.split(path)
    if m1.permission is None:  # Box return meta do not have permission
        del m2.permission
    assert m1 == m2
    assert basename == m1.name
    assert path == m1.path
    assert isinstance(m1.path, six.text_type)
    assert isinstance(m1.name, six.text_type)
    assert not m1.is_deleted
    assert m1.server_mtime
    assert False is m1.is_dir
    assert len(data) == m1.size
    return m1


def test_multipart_upload_threshold(path_client, test_folder):
    if path_client.provider_type == 'tencent':
        pytest.skip("This provider may not pass due to network being too slow")
    mb = 1024*1024
    size = 10*mb
    if is_provider(path_client, 'box'):
        size = 21*mb

    fp = io.BytesIO(b"."*size)
    test_upload(path_client, test_folder, fp=fp)


def test_upload_without_size(path_client, test_folder):
    if isinstance(path_client, LocalClient):
        pytest.skip('Local client wont happen')

    class NoSizeFileObject(FileWrapper):
        def __init__(self, fp):
            self._raw = fp
            super(NoSizeFileObject, self).__init__(fp)

    size = int(random.uniform(0, 500*1024))
    fp = NoSizeFileObject(io.BytesIO(b'.'*size))
    assert super_len(fp) is None, "fp has size"
    test_upload(path_client, test_folder, fp=fp)


def test_upload_zero_size(path_client, test_folder):
    fp = io.BytesIO()
    test_upload(path_client, test_folder, fp=fp)


def test_upload_delete_upload(path_client, test_folder):
    path = posixpath.join(test_folder, "test_upload_delete_upload")
    test_upload(path_client, test_folder, path=path)
    test_delete(path_client, test_folder, path=path)
    test_upload(path_client, test_folder, path=path)


def test_upload_delete_mkdir_upload(path_client, test_folder):
    path = posixpath.join(test_folder, "test_upload_delete_mkdir_upload")
    test_upload(path_client, test_folder, path=path)
    test_delete(path_client, test_folder, path=path)
    test_mkdir(path_client, test_folder, path=path)
    test_upload(path_client, test_folder, path=os.path.join('/', path, "child"))


def test_download(path_client, test_folder, path=None):
    if not path:
        path = posixpath.join(test_folder, "download%f.txt" % time.time())

    test_upload(path_client, test_folder, path=path)
    # TODO: test premature terminate
    c = path_client
    fp = c.download(path)
    assert b'Hello world.' == fp.read()
    assert b'' == fp.read()
    fp.close()
    with c.download(path) as fp:
        assert b'Hello world.' == fp.read()


def test_download_json_file(path_client, test_folder):
    path = posixpath.join(test_folder, "download%f.json" % time.time())
    test_upload(path_client, test_folder, path=path)
    # TODO: test premature terminate
    c = path_client
    fp = c.download(path)
    assert b'Hello world.' == fp.read()


def test_download_range(path_client, test_folder, path=None):
    if isinstance(path_client, LocalClient):
        pytest.skip('Local client not support range download')
    if isinstance(path_client, RtrrClient):
        pytest.skip('Rtrr client not support range download')
    if path_client.provider_name == 'storagemadeeasy':
        pytest.skip('SME not support range download')
    if path_client.provider_name == 'baidu_sdk':
        pytest.skip('baidu sdk not support range download')

    if not path:
        path = posixpath.join(test_folder, "download%f.txt" % time.time())

    data = b'1234567890'
    fp = io.BytesIO(data)
    test_upload(path_client, test_folder, fp=fp, path=path)

    c = path_client
    fp = c.download(path, start=5)
    assert b'67890' == fp.read()
    assert b'' == fp.read()
    fp.close()

    fp = c.download(path, start=5, size=2)
    assert b'67' == fp.read()
    assert b'' == fp.read()
    fp.close()

    fp = c.download(path, size=5)
    assert b'12345' == fp.read()
    assert b'' == fp.read()
    fp.close()


def test_resume_download(path_client, test_folder):
    if isinstance(path_client, LocalClient):
        pytest.skip('Local Client do not support range download')
    if isinstance(path_client, WebDAVDriveClient):
        pytest.skip('WebDav Client do not support range download')
    if isinstance(path_client, RtrrClient):
        pytest.skip('Rtrr client not support range download')
    if path_client.provider_name == 'baidu_sdk':
        pytest.skip('baidu sdk not support range download')

    def fp_read(fp):
        read_string = b''
        while True:
            data = fp.read()
            if data == b'':
                break
            read_string += data
        return read_string

    def mocked_fp(fp):
        old_read = fp._read
        error_count = [0]

        def new_read(size=None):
            error_count[0] += 1
            if error_count[0] == 2:
                raise HTTPError('UCC BFT test_resume_download')
            return old_read(1)
        fp._read = new_read

    path = posixpath.join(test_folder, "download%f" % time.time())

    data = b'1234567890'
    fp = io.BytesIO(data)
    test_upload(path_client, test_folder, fp=fp, path=path)
    c = path_client
    fp = c.download(path)
    mocked_fp(fp)  # let fp.read() will raise network error
    assert b'1234567890' == fp_read(fp)
    assert b'' == fp.read()
    fp.close()

    # test range download with resume download
    fp = c.download(path, start=3, size=5)
    mocked_fp(fp)  # let fp.read() will raise network error
    assert b'45678' == fp_read(fp)
    assert b'' == fp.read()
    fp.close()

    with pytest.raises(NetworkError):
        fp = c.download(path)
        mocked_fp(fp)  # let fp.read() will raise network error
        fp.read(1)
        data = b'9876543210'
        new_fp = io.BytesIO(data)
        test_upload(path_client, test_folder, fp=new_fp, path=path)
        fp_read(fp)


def test_download_not_exist(path_client):
    c = path_client
    with pytest.raises(NotFoundError):
        with c.download('/test_path_drive/foobar') as fp:
            assert b'Hello world.' == fp.read()


def test_download_dir(path_client, test_folder):
    exception_list = {
        '': InvalidError,
        'swift': InvalidError,
        'ObjToPathDriveClient': NotFoundError,
    }

    c = path_client
    with exception_raises(exception_list, c):
        with c.download(test_folder) as fp:
            assert b'Hello world.' == fp.read()


def test_listdir(path_client, test_folder):
    path = posixpath.join(test_folder, "test_listdir")
    test_mkdir(path_client, test_folder, path=path)

    file_amount = 3 # how many files to be tested in listdir
    expect = list()
    for i in range(0, file_amount):
        file_path = posixpath.join(path, "File%d" % i)
        meta = test_upload(path_client, test_folder, path=file_path)
        expect.append(meta)

    c = path_client
    result = c.listdir(path)
    if path_client.provider_type == 'azure':
        result = [r for r in result if r.name != '.azEmpty']
    elif path_client.provider_type == 'backblaze':
        result = [r for r in result if r.name != '.bzEmpty']
    elif path_client.provider_type in ['box', 'sharefile']:
        for r in result:
            r.permission = None  # box/sharefile listdir will append permission
    elif path_client.provider_type == 'baidu':
        for r in result:
            r.client_mtime = None
    assert file_amount == len(result)
    result.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
    expect.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
    assert expect == result

    result_is_unicode = [x for x in result if isinstance(x.name, str)]
    assert file_amount == len(result_is_unicode)

    result_is_unicode = [x for x in result if isinstance(x.path, str)]
    assert file_amount == len(result_is_unicode)


def test_listdir_with_pagination(request, auth_file_config, ucc_client, test_folder, path_client):
    if path_client.provider_type == 'filebase':
        pytest.skip('Filebase does not implement list objects completely')
    file_amount = 1001
    if path_client.provider_type in ['baidu']:
        file_amount = 10001

    provider_name = ucc_client.provider_name
    source_chroot = path_client._chroot
    if path_client.provider_type == 'webdav' and \
            len(source_chroot.split('/')) > 2:
        # chroot will fail this test for server_url in this pattern: http://${ip}/${path}
        pass
    else:
        path_client._chroot = get_chroot(auth_file_config, provider_name, path='/')
    path = '/bft_iterdir_with_pagination'
    def fin():
        path_client._chroot = source_chroot
    request.addfinalizer(fin)

    if not path_client.exists(path):
        path_client.mkdirs(path)
        for i in range(0, file_amount):
            file_path = posixpath.join(path, "Folder%d" % i)
            test_mkdir(path_client, test_folder, path=file_path)

    result = path_client.listdir(path)
    assert file_amount == len(result)


def test_listdir_server_url_encoding_name(path_client, test_folder):
    if path_client.provider_type not in ['amazons3', 'aliyun', 'tencent']:
        pytest.skip()
    path = posixpath.join(test_folder, "test_listdir_server_url_encoding_name")
    test_mkdir(path_client, test_folder, path=path)

    file_amount = 1
    expect = list()

    test_chars = ''.join([string.punctuation[:].replace("/", ''), ' '])
    test_chars = test_chars.replace(sep, '')
    file_name = ''.join(['test_', test_chars, 'file'])
    file_path = posixpath.join(path, file_name)
    meta = test_upload(path_client, test_folder, path=file_path)
    expect.append(meta)

    c = path_client
    result = c.listdir(path)
    assert file_amount == len(result)
    result.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
    expect.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
    assert expect == result

    result_is_unicode = [x for x in result if isinstance(x.name, six.text_type)]
    assert file_amount == len(result_is_unicode)

    result_is_unicode = [x for x in result if isinstance(x.path, six.text_type)]
    assert file_amount == len(result_is_unicode)


def test_listdir_not_exist(path_client):
    c = path_client
    with pytest.raises(NotFoundError):
        c.listdir('/test_path_drive/foobar')


def test_listdir_is_file(path_client, test_folder):
    exception_list = {
        '': InvalidError,
        'ObjToPathDriveClient': None,
    }
    c = path_client
    m = test_upload(c, test_folder)
    with exception_raises(exception_list, c):
        c.listdir(m.path)


def test_listdir_file_contains_space(path_client, test_folder):
    path = posixpath.join(test_folder, 'test_listdir_file_contains_space')
    test_mkdir(path_client, test_folder, path=path)

    file_path = posixpath.join(path, 'file 1')
    test_upload(path_client, test_folder, path=file_path)

    c = path_client
    result = c.listdir(path)
    assert 'file 1' in [meta.name for meta in result]


def test_mkdirs(path_client, test_folder, path=None):
    exception_list = {
        '': None,
    }
    if not path:
        path = posixpath.join(test_folder, "dir1/dir2/dir3%s" % time.time())

    c = path_client
    with exception_raises(exception_list, c):
        m1 = c.mkdirs(path)
        m2 = c.get_metadata(path)
        if m1.permission is None:  # Box return meta do not have permission
            del m2.permission
        assert m1 == m2
        assert os.path.basename(path) == m1.name
        assert path == m1.path
        assert not m1.is_deleted
        assert True is m1.is_dir
        assert not m1.hash
        return m1


def test_scan(path_client):
    if isinstance(path_client, IdToPathDriveClient) and \
            not is_provider(path_client, ['q2']):
        pytest.skip("IdToPathDriveClient scan() may take too long to finish.")
    c = path_client
    r = c.scan()
    count = 0
    for e in r:
        count += 1
        print(e)
        assert isinstance(e.name, six.text_type)
        assert isinstance(e.path, six.text_type)
    print(r.cursor)
    walk_count = len(list(c.walk('/')))
    assert walk_count <= count <= walk_count + 1


def test_walk(path_client, test_folder):
    c = path_client
    c.mkdirs('test_walk/dir1/dir2')
    assert 2 == len(list(c.walk('test_walk')))


# TODO: use this case when implementation is ready
# def test_scanner_top_path_not_exist(path_client):
#     with pytest.raises(NotFoundError):
#         path_client.init_scanner(posixpath.join("Nothing%f" % time.time()))


# invalid cursor should act like when cursor is None
def test_scanner_invalid_cursor(path_client):
    # prepare 5 files whose size is 1KB
    for i in range(5):
        fp = io.BytesIO(bytes(str(i), 'utf-8') * 1024)
        path_client.upload(u'upload' + str(i), fp)

    if path_client.provider_type in ['amazonclouddrive']:
        time.sleep(30)
    scanner = path_client.init_scanner('/')
    expected_item_count = 0
    for _ in scanner:
        expected_item_count += 1

    invalid_cursor = {'invalid_cursor': "not_found"}
    scanner2 = path_client.init_scanner('/', cursor=invalid_cursor)
    item_count = 0
    for _ in scanner2:
        item_count += 1

    assert item_count == expected_item_count


def test_scanner_top_path_is_root(path_client, test_folder):
    item_names = []
    # prepare 5 files whose size is 1KB
    for i in range(5):
        fp = io.BytesIO(bytes(str(i), 'utf-8')*1024)
        name = u'upload'+str(i)
        path_client.upload(name, fp)
        item_names.append(name)

    if path_client.provider_type in ['amazonclouddrive']:
        time.sleep(30)
    scanner = path_client.init_scanner('/')
    item_count = 0
    for meta in scanner:
        print(meta)
        name = os.path.basename(meta.name)
        try:
            item_names.remove(name)
        except ValueError:
            pass
        item_count += 1
    assert len(item_names) == 0

    scanner1 = path_client.init_scanner('/')
    if is_provider(path_client, ['azure', 'googlecloudstorage']):
        scanner1._scanner._max_results = 1
    first_part_item_count = 0
    cursor = None
    for node in scanner1:
        first_part_item_count += 1
        cursor = scanner1.cursor
        print(node)
        if first_part_item_count >= item_count / 2:
            break

    scanner2 = path_client.init_scanner('/', cursor=cursor)
    second_part_item_count = 0
    for node in scanner2:
        second_part_item_count += 1
        print(node)

    print('first_part_item_count=%d, second_part_item_count=%d, '
          'item_count=%d' % (first_part_item_count, second_part_item_count,
                             item_count))
    assert 0 < first_part_item_count < item_count
    assert 0 < second_part_item_count <= item_count
    assert first_part_item_count + second_part_item_count >= item_count


def test_scanner_top_path_not_root(path_client, test_folder):
    item_names = []
    # prepare 5 files whose size is 1KB
    for i in range(5):
        fp = io.BytesIO(bytes(str(i), 'utf-8')*1024)
        path = posixpath.join(test_folder, u'upload'+str(i))
        path_client.upload(path, fp)
        item_names.append(os.path.basename(path))

    if path_client.provider_type in ['amazonclouddrive']:
        time.sleep(30)
    scanner = path_client.init_scanner(test_folder)
    item_count = 0
    for meta in scanner:
        print(meta)
        name = os.path.basename(meta.name)
        try:
            item_names.remove(name)
        except ValueError:
            pass
        item_count += 1
    assert len(item_names) == 0

    scanner1 = path_client.init_scanner(test_folder)
    if is_provider(path_client, ['azure', 'googlecloudstorage']):
        scanner1._scanner._max_results = 1
    first_part_item_count = 0
    cursor = None
    for node in scanner1:
        first_part_item_count += 1
        cursor = scanner1.cursor
        print(node)
        if first_part_item_count >= item_count / 2:
            break

    scanner2 = path_client.init_scanner(test_folder, cursor=cursor)
    second_part_item_count = 0
    for node in scanner2:
        second_part_item_count += 1
        print(node)

    print('first_part_item_count=%d, second_part_item_count=%d, '
          'item_count=%d' % (first_part_item_count, second_part_item_count,
                             item_count))
    assert 0 < first_part_item_count < item_count
    assert 0 < second_part_item_count <= item_count
    assert first_part_item_count + second_part_item_count >= item_count


def test_copy(path_client, test_folder, src_path=None, dst_path=None):
    if path_client.provider_type in ['cubby', 'qnapclouddrive', 'backblaze',
                                     'amazonclouddrive', 'directcloud',
                                     'filebase']:
        pytest.skip('copy is not implemented for this provider')

    if not src_path:
        src_path = posixpath.join(test_folder, "test_copy_file%f" % time.time())

    if not dst_path:
        dst_path = src_path + "_copy"

    if path_client.provider_type == 'sharefile':
        if os.path.dirname(src_path) == os.path.dirname(dst_path):
            pytest.skip('not support copy in the same file')

    test_upload(path_client, test_folder, path=src_path)

    c = path_client
    m1 = c.copy(src_path, dst_path)
    m2 = c.get_metadata(dst_path)
    m3 = c.get_metadata(src_path)
    print(m1)
    print(m2)
    m1.server_mtime = m2.server_mtime  # ACD: server mtime is diff
    if m1.permission is None:  # Box return meta do not have permission
        del m2.permission
    assert m1 == m2
    assert isinstance(m1.name, six.text_type)
    assert isinstance(m1.path, six.text_type)
    assert os.path.basename(dst_path) == m2.name
    assert dst_path == m2.path
    assert m3.size == m2.size


def test_move(path_client, test_folder, src_path=None, dst_path=None):
    if path_client.provider_type in ['cubby', 'backblaze', 'filebase']:
        pytest.xfail('move is not implemented for this provider')

    if not src_path:
        src_path = posixpath.join(test_folder, "test_move_file%f" % time.time())

    if not dst_path:
        dst_path = src_path + "_move"

    test_upload(path_client, test_folder, path=src_path)

    c = path_client
    m2 = c.get_metadata(src_path)
    m1 = c.move(src_path, dst_path)
    m3 = c.get_metadata(dst_path)
    with pytest.raises(NotFoundError):
        c.get_metadata(src_path)
    if m1.permission is None:  # Box return meta do not have permission
        del m3.permission
    assert m1 == m3
    assert isinstance(m1.name, six.text_type)
    assert isinstance(m1.path, six.text_type)
    assert os.path.basename(dst_path) == m1.name
    assert dst_path == m1.path
    assert m2.size == m1.size


def test_delete(path_client, test_folder, path=None):
    if not path:
        path = posixpath.join(test_folder, "delete%f" % time.time())

    m = test_upload(path_client, test_folder, path=path)

    c = path_client
    c.delete(m.path)
    with pytest.raises(NotFoundError):
        print(c.get_metadata(m.path))


def test_delete_not_found(path_client):
    exception_list = {
        '': NotFoundError,
        'ObjToPathDriveClient': None,
    }
    c = path_client
    with exception_raises(exception_list, c):
        c.delete('/test_path_drive/testDir/test_delete_not_found')


def test_delete_folder(path_client, test_folder):
    exception_list = {
        '': InvalidError,
        'ObjToPathDriveClient': None,
    }
    c = path_client
    m = test_mkdir(c, test_folder)
    with exception_raises(exception_list, c):
        c.delete(m.path)


def test_delete_multi(path_client, test_folder):
    c = path_client
    try:
        c.delete_multi([])
    except NotImplementedError:
        pytest.skip("Do not support delete_multi")

    f1 = test_upload(c, test_folder)
    f2 = test_upload(c, test_folder)
    # d = test_mkdir(c, test_folder)  # TODO: Need define SPEC about folder
    not_exist_path = posixpath.join(test_folder, 'not_exist')
    delete_multi_paths = [not_exist_path, f1.path, f2.path]
    result = c.delete_multi(delete_multi_paths)

    delete_paths = set()
    error_paths = set()
    for path in delete_multi_paths:
        try:
            c.get_metadata(path)
            error_paths.add(path)
        except NotFoundError:
            delete_paths.add(path)
    assert delete_paths == set(result['deleted'])
    assert error_paths == set([e['path'] for e in result['error']])


def test_get_latest_change_info(path_client):  # TODO(harry): should have get latest change info
    if isinstance(path_client, (YandexDiskClient,
                                HiDriveClient,
                                LocalClient,
                                ObjToPathDriveClient,
                                WebDAVDriveClient,
                                ShareFileClient,
                                DirectCloudClient,
                                BaiduClient)):
        pytest.skip('get_latest_change_info is not implemented for this provider')
    if is_provider(path_client, ['amazonclouddrive', 'rtrr', 'q2']):
        pytest.skip('get_latest_change_info is not implemented for this provider')

    c = path_client
    change_info = c.get_latest_change_info()
    print(change_info)
    return change_info


def test_change(path_client):
    if isinstance(path_client, IdToPathDriveClient):
        pytest.skip("IdToPathDriveClient scan() may take too long to finish.")
    if isinstance(path_client, (WebDAVDriveClient, ObjToPathDriveClient,
                  HiDriveClient, ShareFileClient, YandexDiskClient, DirectCloudClient,
                  BaiduClient)):
        return
    if is_provider(path_client, ['rtrr']):
        pytest.xfail('change is not implemented for this provider')

    c = path_client
    r = c.scan()
    for e in r:
        pass
    cursor = r.cursor
    t = Thread(target=c.mkdir, args=('/test_path_drive/testChange',))
    t.start()
    for e in c.change(cursor):
        print(e)
    t.join()
    print('test_change', e)
    assert e
    assert isinstance(e.name, six.text_type)
    assert isinstance(e.path, six.text_type)


def test_change_from_begin(path_client):
    if isinstance(path_client, IdToPathDriveClient):
        pytest.skip("IdToPathDriveClient scan() may take too long to finish.")

    if isinstance(path_client, (ObjToPathDriveClient, WebDAVDriveClient,
                                YandexDiskClient, HiDriveClient,
                                ShareFileClient, LocalClient,
                                DirectCloudClient, BaiduClient)):
        return
    if is_provider(path_client, ['rtrr']):
        pytest.xfail('change is not implemented for this provider')

    c = path_client
    info = {}
    with pytest.raises(ChangeNeedResetException):
        list(c.change(info))


def test_upload_big(path_client):
    if path_client.provider_type == 'tencent':
        pytest.skip("This provider may not pass due to network being too slow")
    c = path_client
    data = b'Hello world.' * 1024 * 1024
    fp = io.BytesIO(data)

    def cb(u, t):
        print('progress=%d/%d' % (u, t))
    m1 = c.upload(u'/test_path_drive/File_big', fp, progress_callback=cb)
    m2 = c.get_metadata(u'/test_path_drive/File_big')
    m1.server_mtime = m2.server_mtime  # ACD: server mtime is diff
    print(m1)
    print(m2)
    assert isinstance(m1.name, six.text_type)
    assert isinstance(m1.path, six.text_type)
    if m1.permission is None:  # Box return meta do not have permission
        del m2.permission
    assert m1 == m2
    assert u'File_big' == m1.name
    assert u'/test_path_drive/File_big' == m1.path
    assert not m1.is_deleted
    assert m1.server_mtime
    assert False is m1.is_dir
    assert len(data) == m1.size


def test_mkdir_no_parent(path_client):
    exception_list = {
        '': NotFoundError,
        'ObjToPathDriveClient': None,
        'drivehq': None,  # drivehq's mkdir performs mkdirs
        'box': NotFoundError,
    }

    c = path_client
    with exception_raises(exception_list, c):
        c.mkdir(u'/test_path_drive/test_mkdir_no_parent/TestDir')


# FIXME: Resolve conflict for googledrive
def test_mkdir_exists(path_client, test_folder):
    exception_list = {
        '': ConflictError
    }
    path = posixpath.join(test_folder, "test_mkdir_exists")
    if isinstance(path_client, ObjToPathDriveClient):
        pytest.skip("ObjToPathDriveClient mkdir won't raise ConflictError")
    c = path_client
    c.mkdir(path)
    with exception_raises(exception_list, c):
        c.mkdir(path)


def test_mkdirs_exists(path_client):
    if isinstance(path_client, IdToPathDriveClient):
        pytest.skip("IdToPathDriveClient does not implement mkdirs")
    exception_list = {
        '': ConflictError,
        'cubby': InvalidError,
        'ObjToPathDriveClient': None,
    }
    c = path_client
    c.mkdirs(u'/test_path_drive/test_mkdirs_exists/dir1/dir2')
    with exception_raises(exception_list, c):
        c.mkdirs(u'/test_path_drive/test_mkdirs_exists/dir1/dir2')


def test_mkdirs_parent_is_file(path_client, test_folder):
    exception_list = {
        '': InvalidError,
        'ObjToPathDriveClient': None,
        'storagemadeeasy': None,  # same name dir & file can be create/get_meta/list, but web not seen
        'LyveClient': InvalidError,
        'azure_hierarchical': NotFoundError
    }
    test_upload(path_client, test_folder, path=u'/test_path_drive/test_mkdirs_parent_is_file')
    c = path_client
    with exception_raises(exception_list, c):
        c.mkdirs(u'/test_path_drive/test_mkdirs_parent_is_file/dir1/dir2')


def test_upload_no_parent(path_client):
    exception_list = {
        '': NotFoundError,
        'ObjToPathDriveClient': None,
        'DropboxClient': None
    }
    # if isinstance(path_client, ObjToPathDriveClient):
    #     pytest.skip("ObjToPathDriveClient upload won't raise NotFoundError when parent not exists")
    c = path_client
    data = b'Hello world.'
    fp = io.BytesIO(data)
    with exception_raises(exception_list, c):
        c.upload(u'/test_path_drive/test_upload_no_parent/File_1', fp)


def test_upload_dir(path_client):
    exception_list = {
        '': InvalidError,
        'DropboxClient': None,
        'ObjToPathDriveClient': None,
        'dropbox': ConflictError,
        'directcloud': None,
        'ShareFileClient': None,
        'storagemadeeasy': None,  # SME allow file and dir same name
        'baidu': None,
        'baidu_sdk': None,
        'azure_hierarchical': InvalidError
    }
    c = path_client
    c.mkdir(u'/test_path_drive/test_upload_dir')
    data = b'Hello world.'
    fp = io.BytesIO(data)
    # with pytest.raises(InvalidError):
    with exception_raises(exception_list, c):
        c.upload(u'/test_path_drive/test_upload_dir', fp)


def test_mkdir_parent_is_file(path_client, test_folder):
    exception_list = {
        '': InvalidError,
        'ObjToPathDriveClient': None,
        'storagemadeeasy': None,
        'LyveClient': InvalidError,
        'azure_hierarchical': NotFoundError
    }

    test_upload(path_client, test_folder,
                path=u'/test_path_drive/test_mkdir_parent_is_file')
    c = path_client
    with exception_raises(exception_list, c):
        c.mkdir(u'/test_path_drive/test_mkdir_parent_is_file/foobar')


def test_move_dest_exist(path_client, test_folder):
    if path_client.provider_type in ['cubby', 'backblaze', 'filebase']:
        pytest.xfail('move is not implemented for this provider')

    exception_list = {
        '': ConflictError,
        'ObjToPathDriveClient': None,
        # 'GoogleDriveClient': None,
        'storagemadeeasy': InvalidError
    }
    # if isinstance(path_client, ObjToPathDriveClient):
    #     pytest.skip("ObjToPathDriveClient move won't raise ConflictError when dest exists")
    c = path_client
    test_upload(path_client, test_folder, path=u'/test_path_drive/test_move_dest_exist')
    test_upload(path_client, test_folder, path=u'/test_path_drive/test_move_dest_exist2')
    # with pytest.raises(ConflictError):
    with exception_raises(exception_list, c):
        c.move(u'/test_path_drive/test_move_dest_exist',
               u'/test_path_drive/test_move_dest_exist2')

    exception_list.update({
        'storagemadeeasy': None,  # SME allow file and dir same name
        'webdav_windows': InvalidError,
        'azure_hierarchical': InvalidError
    })
    test_mkdir(path_client, test_folder, u'/test_path_drive/test_move_dest_exist_dir')
    # with pytest.raises(ConflictError):
    with exception_raises(exception_list, c):
        c.move(u'/test_path_drive/test_move_dest_exist2',
               u'/test_path_drive/test_move_dest_exist_dir')


def test_move_dir(path_client):
    if isinstance(path_client, ObjToPathDriveClient):
        pytest.skip("ObjToPathDriveClient does not support move dir")
    if path_client.provider_type in ['cubby']:
        pytest.xfail('move is not implemented for this provider')

    c = path_client
    m2 = c.mkdir(u'/test_path_drive/test_move_src_is_dir')
    m1 = c.move(u'/test_path_drive/test_move_src_is_dir',
                u'/test_path_drive/test_move_src_is_dir2')
    m3 = c.get_metadata(u'/test_path_drive/test_move_src_is_dir2')
    with pytest.raises(NotFoundError):
        c.get_metadata(u'/test_path_drive/test_move_src_is_dir')
    assert isinstance(m1.path, six.text_type)
    assert isinstance(m1.name, six.text_type)
    if m1.permission is None:  # Box return meta do not have permission
        del m3.permission
    assert m1 == m3
    assert u'test_move_src_is_dir2' == m1.name
    assert u'/test_path_drive/test_move_src_is_dir2' == m1.path
    assert m2.size == m1.size


def test_move_src_not_found(path_client):
    if path_client.provider_type in ['cubby', 'backblaze', 'filebase']:
        pytest.xfail('move is not implemented for this provider')

    exception_list = {
        '': NotFoundError,
        'drivehq': ConflictError,
        'storagemadeeasy': InvalidError,
    }

    c = path_client
    with exception_raises(exception_list, c):
        c.move(u'/test_path_drive/test_move_src_not_found',
               u'/test_path_drive/test_move_src_not_found1')


def test_copy_not_found(path_client):
    if path_client.provider_type in ['cubby', 'qnapclouddrive', 'backblaze',
                                     'amazonclouddrive', 'directcloud',
                                     'filebase']:
        pytest.skip('copy is not implemented for this provider')

    c = path_client
    with pytest.raises(NotFoundError):
        c.copy(u'/test_path_drive/testDir/test_copy_not_found',
               u'/test_path_drive/testDir/File_copy')


def test_copy_dest_exists(path_client, test_folder):
    if path_client.provider_type in ['cubby', 'qnapclouddrive', 'backblaze',
                                     'amazonclouddrive', 'directcloud',
                                     'filebase']:
        pytest.skip('copy is not implemented for this provider')

    exception_list = {
        '': ConflictError,
        # 'GoogleDriveClient': None,
        'ObjToPathDriveClient': None,
        'onedrivegraph-od': ConflictError,  # for OneDrive
        'onedrivegraph-odb': ConflictError  # for OneDriveforBusiness
    }
    c = path_client
    c.mkdir(u'/test_path_drive/test_copy_dest_exists')
    test_upload(path_client, test_folder,
                path=u'/test_path_drive/test_copy_dest_exists/File_1')
    test_upload(path_client, test_folder,
                path=u'/test_path_drive/test_copy_dest_exists/File_2')
    with exception_raises(exception_list, c):
        c.copy(u'/test_path_drive/test_copy_dest_exists/File_1',
               u'/test_path_drive/test_copy_dest_exists/File_2')


def test_copy_src_is_dir(path_client, test_folder):
    if isinstance(path_client, ObjToPathDriveClient):
        pytest.skip("ObjToPathDriveClient does not support copy dir")
    if path_client.provider_type in ['cubby', 'qnapclouddrive', 'backblaze', 'amazonclouddrive', 'directcloud']:
        pytest.skip('copy is not implemented for this provider')

    exception_list = {
        '': InvalidError,
        'DropboxClient': None,
        'box': None,
        'hidrive': None,
        'ShareFileClient': NotImplementedError,
        'googledrive': PermissionsError,
        'onedrivegraph': None,
        'onedrivegraph_germany': None,
    }
    c = path_client
    test_mkdir(path_client, test_folder,
               u'/test_path_drive/test_copy_src_is_dir')
    # with pytest.raises(InvalidError):
    with exception_raises(exception_list, c):
        c.copy(u'/test_path_drive/test_copy_src_is_dir',
               u'/test_path_drive/test_copy_src_is_dir2')


def test_rmdir(path_client):
    c = path_client
    c.mkdir('/test_path_drive/test_rmdir')
    c.get_metadata('/test_path_drive/test_rmdir')
    c.rmdir('/test_path_drive/test_rmdir')
    with pytest.raises(NotFoundError):
        c.get_metadata('/test_path_drive/test_rmdir')


def test_rmdir_file(path_client, test_folder):
    m = test_upload(path_client, test_folder)
    c = path_client
    with pytest.raises(InvalidError):
        c.rmdir(m.path)


def test_rmdir_not_found(path_client):
    exception_list = {
        '': NotFoundError,
        'box': NotFoundError,
    }
    c = path_client
    with exception_raises(exception_list, c):
        c.rmdir('/test_path_drive/test_rmdir_not_found')


def test_rmdir_not_empty(path_client, test_folder):
    exception_list = {
        '': ConflictError,
        'DropboxClient': ConflictError,
    }
    c = path_client
    c.mkdir(u'/test_path_drive/test_rmdir_not_empty')
    test_upload(path_client, test_folder,
                path=u'/test_path_drive/test_rmdir_not_empty/file1')
    time.sleep(2)  # amazon drive should wait to check folder empty or not
    with exception_raises(exception_list, c):
        c.rmdir(u'/test_path_drive/test_rmdir_not_empty')


def test_rmdirs(path_client, test_folder):
    c = path_client
    m = test_mkdirs(path_client, test_folder)
    if m is not None:
        print(c.rmdirs(m.path))
        with pytest.raises(NotFoundError):
            print('after deleted', c.get_metadata(m.path))


def test_rmdirs_not_found(path_client):
    exception_list = {
        '': NotFoundError,
        'ObjToPathDriveClient': None,
    }
    c = path_client
    with exception_raises(exception_list, c):
        c.rmdirs('/test_path_drive/test_rmdirs_not_found')


def xtest_too_many_request(path_client, path='/test_path_drive/testDir'):
    if isinstance(path_client, LocalClient):
        pytest.skip("LocalClient won't raise TooManyRequestsError")
    # mock HttpClient._session.send return dummy Response obj
    c = path_client
    resp = mock.MagicMock()
    resp.status_code = 429
    resp.content = '429'
    old_session_send = c._session.send
    c._session.send = mock.MagicMock(return_value=resp)
    c._options['max_retry_num'] = 1
    with pytest.raises(TooManyRequestsError):
        c.mkdir(path)
    c._session.send = old_session_send


def test_change_update_and_delete_folder(path_client, test_folder):
    '''test id2path client
    '''
    if isinstance(path_client, (ObjToPathDriveClient, WebDAVDriveClient,
                                YandexDiskClient, HiDriveClient,
                                ShareFileClient, LocalClient,
                                DirectCloudClient, BaiduClient)):
        pytest.skip('get_latest_change_info is not implemented for this provider')
    if is_provider(path_client, ['amazonclouddrive', 'rtrr', 'q2']):
        pytest.skip('get_latest_change_info is not implemented for this provider')

    c = path_client
    if isinstance(c, DropboxClient):
        r = c.scan()
        for e in r:
            pass
        latest_change_info = r.cursor
    else:
        latest_change_info = c.get_latest_change_info()
    print('latest_change_info: ', latest_change_info)

    m = test_mkdir(path_client, test_folder)
    src_path = posixpath.join(m.path, u'test_change_delete_folder_file')
    test_upload(path_client, test_folder, path=src_path)

    r = c.change(latest_change_info)
    for e in r:
        pass

    if isinstance(c, DropboxClient):
        test_folder_rename = u'%s_rename' % m.path
        c.move(m.path, test_folder_rename)
    else:
        _, test_folder_rename = os.path.split(m.path)
        test_folder_rename = u'%s_rename' % test_folder_rename
        node = c._resolve_path(c._abs_path(m.path))
        c._client.update_node(node.id, name=test_folder_rename)

    # wait for a while to get change after updating node
    if is_provider(c, ['box', 'googledrive']):
        time.sleep(5)

    cursor = r.cursor
    r = c.change(cursor)
    new_folder_name = posixpath.join(os.path.dirname(m.path), os.path.basename(test_folder_rename))
    new_file_name = posixpath.join(new_folder_name, os.path.basename(src_path))
    suc = {m.path: False, src_path: False, new_folder_name: False, new_file_name: False}
    for e in r:
        suc.update({e.path: True})
    assert reduce(lambda x, y: x and y, list(suc.values()))

    if isinstance(c, DropboxClient):
        c.rmdirs(new_folder_name)
    else:
        node = c._resolve_path(c._abs_path(new_folder_name))
        c._client.delete(node.id)
    if is_provider(c, 'box'):
        time.sleep(5)
    cursor = r.cursor
    r = c.change(cursor)
    suc = {new_folder_name: False, new_file_name: False}
    for e in r:
        suc.update({e.path: True})
    print(json.dumps(suc, indent=2))
    assert reduce(lambda x, y: x and y, list(suc.values()))


def test_get_permission(path_client):
    p = path_client.get_permission(u'/')
    assert isinstance(p, Permission)
    assert p.can_read in (True, False)
    assert p.can_write in (True, False)
    assert p.can_delete in (True, False)


def test_list_permission(path_client, test_folder):
    only_support_get_permission_provider = ['onedrivegraph']
    result = path_client.listdir(u'/')
    for meta in result:
        if path_client.provider_type in only_support_get_permission_provider:
            assert meta.permission is None
        else:
            assert meta.permission is not None


@pytest.mark.delete_with_version
class TestDeleteWithVersion:
    SUPPORTED_PROVIDERS = ['backblaze']

    def test_delete_ok(self, path_client, test_folder):
        if not is_provider(path_client, self.SUPPORTED_PROVIDERS):
            pytest.skip(f'Supported providers: {self.SUPPORTED_PROVIDERS}')

        meta = test_upload(path_client, test_folder)
        assert None is path_client.delete_with_version(meta.path, meta.version_id)
        with pytest.raises(NotFoundError):
            path_client.get_metadata(meta.path, meta.version_id)

    def test_path_not_exist(self, path_client, test_folder):
        if not is_provider(path_client, self.SUPPORTED_PROVIDERS):
            pytest.skip(f'Supported providers: {self.SUPPORTED_PROVIDERS}')
        meta = test_upload(path_client, test_folder)
        assert None is path_client.delete_with_version('test_path_drive/foo',
                                                       meta.version_id)

    def test_version_not_exist(self, path_client, test_folder):
        if not is_provider(path_client, self.SUPPORTED_PROVIDERS):
            pytest.skip(f'Supported providers: {self.SUPPORTED_PROVIDERS}')
        meta = test_upload(path_client, test_folder)
        assert None is path_client.delete_with_version(meta.path, 'foo')


@pytest.mark.str
class TestUnicode(object):
    def test_unicode_mkdir(self, path_client, test_folder, unicode_string):
        path = posixpath.join('/', unicode_string)
        test_mkdir(path_client, test_folder, path=path)
        return path

    def test_unicode_mkdirs(self, path_client, test_folder, unicode_string):
        path = u"/%s/%s" % (unicode_string, unicode_string)
        test_mkdirs(path_client, test_folder, path=path)

    def test_unicode_upload(self, path_client, test_folder, unicode_string):
        path = posixpath.join('/', unicode_string)
        test_upload(path_client, test_folder, path=path)

    def test_unicode_download(self, path_client, test_folder, unicode_string):
        path = posixpath.join('/', unicode_string)
        test_download(path_client, test_folder, path=path)

    def test_unicode_listdir(self, path_client, test_folder, unicode_string):
        path = self.test_unicode_mkdir(path_client, test_folder, unicode_string)
        expect = list()
        for n in range(0, 3):
            file_name = unicode_string + str(n)
            file_path = posixpath.join(path, file_name)
            meta = test_upload(path_client, test_folder, path=file_path)
            expect.append(meta)

        result = path_client.listdir(path)
        if path_client.provider_type == 'azure':
            result = [r for r in result if r.name != '.azEmpty']
        elif path_client.provider_type == 'backblaze':
            result = [r for r in result if r.name != '.bzEmpty']
        elif path_client.provider_type in ['box', 'sharefile']:
            for r in result:
                r.permission = None
        elif path_client.provider_type == 'baidu':
            for r in result:
                r.client_mtime = None
        assert 3 == len(result)
        result.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
        expect.sort(key=cmp_to_key(lambda x, y: cmp(x.name, y.name)))
        assert result == expect

    def test_unicode_copy(self, path_client, test_folder, unicode_string):
        if path_client.provider_type in ['filebase']:
            pytest.skip('copy is not implemented for this provider')
        path = posixpath.join('/', unicode_string)
        test_upload(path_client, test_folder, path=path)
        test_copy(path_client, test_folder,
                  src_path=path, dst_path=path+"copy")

    def test_unicode_move(self, path_client, test_folder, unicode_string):
        if path_client.provider_type in ['filebase']:
            pytest.xfail('move is not implemented for this provider')
        path = posixpath.join('/', unicode_string)
        test_upload(path_client, test_folder, path=path)
        test_move(path_client, test_folder,
                  src_path=path, dst_path=path+"move")

    def test_unicode_delete(self, path_client, test_folder, unicode_string):
        path = posixpath.join('/', unicode_string)
        test_upload(path_client, test_folder, path=path)
        path_client.delete(path)

    def test_unicode_delete_multi(self, path_client, test_folder, unicode_string):
        path = posixpath.join('/', unicode_string)
        test_upload(path_client, test_folder, path=path)
        path_client.delete(path)
        try:
            path_client.delete_multi([])
        except NotImplementedError:
            pytest.skip("Do not support delete_multi")

        result = path_client.delete_multi([path])
        assert [path] == result['deleted']


@pytest.mark.str
class TestUnicodeNFC:
    # https://en.wikipedia.org/wiki/Unicode_equivalence#Example
    # NFD code point: 0065 0301
    # NFD character:    e   '́
    # NFC code point:    00e9
    # NFC character:      é
    NFC_PATH = 'verziós mentése q'

    def test_nfc_upload(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.delete(path)
        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        path_client.get_metadata(path)

    def test_nfc_listdir(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.delete(path)

        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        has_nfc_file = False
        files = path_client.listdir(test_folder)
        for file in files:
            if file.path == path:
                has_nfc_file = True
            if is_provider(path_client, ['dropbox', 'directcloud']) and \
                    file.path == nfc_normalize(path):
                has_nfc_file = True
        assert has_nfc_file, [file.path for file in files]

    def test_nfc_mkdir(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.rmdir(path)
        request.addfinalizer(fin)

        path_client.mkdir(path)
        path_client.get_metadata(path)

    def test_nfc_mkdirs(self, path_client, test_folder):
        path = posixpath.join(test_folder, 'test_nfc_mkdirs', self.NFC_PATH)
        path_client.mkdirs(path)
        path_client.get_metadata(path)

    def test_nfc_rmdirs(self, path_client, test_folder):
        path = posixpath.join(test_folder, 'test_nfc_rmdirs', self.NFC_PATH)
        path_client.mkdirs(path)
        path_client.rmdirs(path)

    def test_nfc_download(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.delete(path)

        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        fp = path_client.download(path)
        assert fp.read() == b'foobar'

    def test_nfc_move(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)
        new_path = f'{path}-move'

        def fin():
            path_client.delete(new_path)

        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        path_client.move(path, new_path)
        path_client.get_metadata(new_path)

    def test_nfc_scan(self, path_client, test_folder, request):
        if isinstance(path_client, IdToPathDriveClient) and \
                not is_provider(path_client, ['q2']):
            pytest.skip(
                "IdToPathDriveClient scan() may take too long to finish.")

        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.delete(path)

        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        has_nfc_file = False
        files = path_client.scan()
        for file in files:
            if file.path == path:
                has_nfc_file = True
            if is_provider(path_client, ['dropbox', 'directcloud']) and \
                    file.path == nfc_normalize(path):
                has_nfc_file = True
        assert has_nfc_file, [file.path for file in files]

    def test_nfc_walk(self, path_client, test_folder, request):
        path = posixpath.join(test_folder, self.NFC_PATH)

        def fin():
            path_client.delete(path)

        request.addfinalizer(fin)

        path_client.upload(path, io.BytesIO(b'foobar'))
        has_nfc_file = False
        files = path_client.walk(test_folder)
        for file in files:
            if file.path == path:
                has_nfc_file = True
            if is_provider(path_client, ['dropbox', 'directcloud']) and \
                    file.path == nfc_normalize(path):
                has_nfc_file = True
        assert has_nfc_file, [file.path for file in files]

@pytest.mark.ratelimit
class TestRatelimit(object):
    def test_ratelimit_upload_file(self, path_client, bandwidth):
        c = path_client
        path = "/test_path_drive/ratelimit_upload_file"
        try:
            c.get_metadata(os.path.dirname(path))
        except NotFoundError:
            c.mkdir(os.path.dirname(path))

        monkey_patch()
        set_upload_rate(bandwidth)
        fp = io.BytesIO(b'.'*bandwidth)
        c.upload(path, fp)

        fp = io.BytesIO(b'.'* (bandwidth // 2))
        c.upload(path, fp)
        set_upload_rate(0)

    def test_ratelimit_download_file(self, path_client, bandwidth):
        c = path_client
        path = "/test_path_drive/ratelimit_download_file"
        try:
            c.get_metadata(os.path.dirname(path))
        except NotFoundError:
            c.mkdir(os.path.dirname(path))

        fp = io.BytesIO(b'.'*bandwidth)
        c.upload(path, fp)

        monkey_patch()
        set_download_rate(bandwidth)
        with c.download(path) as fp:
            assert len(fp.read()) == bandwidth


    @pytest.mark.multipartsize
    def test_multipartsize_threshold(self, path_client):
        if isinstance(path_client, (DropboxClient, YandexDiskClient,
                                    ShareFileClient, WebDAVDriveClient,
                                    HiDriveClient)):
            ucc_client = path_client;
        else:
            ucc_client = path_client._client;

        # WebDAVDriveClient does not have _options
        if not hasattr(ucc_client, '_options'):
            pytest.skip('%s has no _options attributes' %
                        ucc_client.__class__.__name__)

        options = ucc_client._options;
        if 'multipart_upload_threshold' not in options:
            pytest.skip('%s has no multipart_upload_threshold in options' %
                        ucc_client.__class__.__name__)

        multipart_upload_threshold = options['multipart_upload_threshold']

        c = path_client
        path = "/test_path_drive/test_multipartsize_threshold"
        try:
            c.get_metadata(os.path.dirname(path))
        except NotFoundError:
            c.mkdir(os.path.dirname(path))

        class MemoryFile:
            def __init__ (self, length):
                self.length = 0;
                self.read_bytes = 0;

            def __enter__(self, *args, **kwargs):
                return self

            def read(self, size=None):
                if size == -1:
                    size = None
                if size is None:
                    size = self.length
                if size > self.length - self.read_bytes:
                    size = self.length - self.read_bytes

                self.read_bytes += size
                return b'.' * size

            def tell(self):
                return self.read_bytes

        file_size = multipart_upload_threshold + 1
        _file = MemoryFile(file_size)
        c.upload(path, _file)

        file_size = multipart_upload_threshold
        _file = MemoryFile(file_size)
        c.upload(path, _file)

        file_size = multipart_upload_threshold - 1
        _file = MemoryFile(file_size)
        c.upload(path, _file)

# move invalid (dst invalid char)
# test shared folder