mmcv/tests/test_fileclient.py

import os
import os.path as osp
import sys
import tempfile
from contextlib import contextmanager
from copy import deepcopy
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

import mmcv
from mmcv import BaseStorageBackend, FileClient
from mmcv.utils import has_method

sys.modules['ceph'] = MagicMock()
sys.modules['petrel_client'] = MagicMock()
sys.modules['petrel_client.client'] = MagicMock()
sys.modules['mc'] = MagicMock()


@contextmanager
def build_temporary_directory():
    """Build a temporary directory containing many files to test
    ``FileClient.list_dir_or_file``.

    . \n
    | -- dir1 \n
    | -- | -- text3.txt \n
    | -- dir2 \n
    | -- | -- dir3 \n
    | -- | -- | -- text4.txt \n
    | -- | -- img.jpg \n
    | -- text1.txt \n
    | -- text2.txt \n
    """
    with tempfile.TemporaryDirectory() as tmp_dir:
        text1 = Path(tmp_dir) / 'text1.txt'
        text1.open('w').write('text1')
        text2 = Path(tmp_dir) / 'text2.txt'
        text2.open('w').write('text2')
        dir1 = Path(tmp_dir) / 'dir1'
        dir1.mkdir()
        text3 = dir1 / 'text3.txt'
        text3.open('w').write('text3')
        dir2 = Path(tmp_dir) / 'dir2'
        dir2.mkdir()
        jpg1 = dir2 / 'img.jpg'
        jpg1.open('wb').write(b'img')
        dir3 = dir2 / 'dir3'
        dir3.mkdir()
        text4 = dir3 / 'text4.txt'
        text4.open('w').write('text4')
        yield tmp_dir


@contextmanager
def delete_and_reset_method(obj, method):
    method_obj = deepcopy(getattr(type(obj), method))
    try:
        delattr(type(obj), method)
        yield
    finally:
        setattr(type(obj), method, method_obj)


class MockS3Client:

    def __init__(self, enable_mc=True):
        self.enable_mc = enable_mc

    def Get(self, filepath):
        with open(filepath, 'rb') as f:
            content = f.read()
        return content


class MockPetrelClient:

    def __init__(self, enable_mc=True, enable_multi_cluster=False):
        self.enable_mc = enable_mc
        self.enable_multi_cluster = enable_multi_cluster

    def Get(self, filepath):
        with open(filepath, 'rb') as f:
            content = f.read()
        return content

    def put(self):
        pass

    def delete(self):
        pass

    def contains(self):
        pass

    def isdir(self):
        pass

    def list(self, dir_path):
        for entry in os.scandir(dir_path):
            if not entry.name.startswith('.') and entry.is_file():
                yield entry.name
            elif osp.isdir(entry.path):
                yield entry.name + '/'


class MockMemcachedClient:

    def __init__(self, server_list_cfg, client_cfg):
        pass

    def Get(self, filepath, buffer):
        with open(filepath, 'rb') as f:
            buffer.content = f.read()


class TestFileClient:

    @classmethod
    def setup_class(cls):
        cls.test_data_dir = Path(__file__).parent / 'data'
        cls.img_path = cls.test_data_dir / 'color.jpg'
        cls.img_shape = (300, 400, 3)
        cls.text_path = cls.test_data_dir / 'filelist.txt'

    def test_error(self):
        with pytest.raises(ValueError):
            FileClient('hadoop')

    def test_disk_backend(self):
        disk_backend = FileClient('disk')

        # test `name` attribute
        assert disk_backend.name == 'HardDiskBackend'
        # test `allow_symlink` attribute
        assert disk_backend.allow_symlink
        # test `get`
        # input path is Path object
        img_bytes = disk_backend.get(self.img_path)
        img = mmcv.imfrombytes(img_bytes)
        assert self.img_path.open('rb').read() == img_bytes
        assert img.shape == self.img_shape
        # input path is str
        img_bytes = disk_backend.get(str(self.img_path))
        img = mmcv.imfrombytes(img_bytes)
        assert self.img_path.open('rb').read() == img_bytes
        assert img.shape == self.img_shape

        # test `get_text`
        # input path is Path object
        value_buf = disk_backend.get_text(self.text_path)
        assert self.text_path.open('r').read() == value_buf
        # input path is str
        value_buf = disk_backend.get_text(str(self.text_path))
        assert self.text_path.open('r').read() == value_buf

        with tempfile.TemporaryDirectory() as tmp_dir:
            # test `put`
            filepath1 = Path(tmp_dir) / 'test.jpg'
            disk_backend.put(b'disk', filepath1)
            assert filepath1.open('rb').read() == b'disk'
            # test the `mkdir_or_exist` behavior in `put`
            _filepath1 = Path(tmp_dir) / 'not_existed_dir1' / 'test.jpg'
            disk_backend.put(b'disk', _filepath1)
            assert _filepath1.open('rb').read() == b'disk'

            # test `put_text`
            filepath2 = Path(tmp_dir) / 'test.txt'
            disk_backend.put_text('disk', filepath2)
            assert filepath2.open('r').read() == 'disk'
            # test the `mkdir_or_exist` behavior in `put_text`
            _filepath2 = Path(tmp_dir) / 'not_existed_dir2' / 'test.txt'
            disk_backend.put_text('disk', _filepath2)
            assert _filepath2.open('r').read() == 'disk'

            # test `isfile`
            assert disk_backend.isfile(filepath2)
            assert not disk_backend.isfile(Path(tmp_dir) / 'not/existed/path')

            # test `remove`
            disk_backend.remove(filepath2)

            # test `exists`
            assert not disk_backend.exists(filepath2)

            # test `get_local_path`
            # if the backend is disk, `get_local_path` just return the input
            with disk_backend.get_local_path(filepath1) as path:
                assert str(filepath1) == path
            assert osp.isfile(filepath1)

        # test `join_path`
        disk_dir = '/path/of/your/directory'
        assert disk_backend.join_path(disk_dir, 'file') == \
            osp.join(disk_dir, 'file')
        assert disk_backend.join_path(disk_dir, 'dir', 'file') == \
            osp.join(disk_dir, 'dir', 'file')

        # test `list_dir_or_file`
        with build_temporary_directory() as tmp_dir:
            # 1. list directories and files
            assert set(disk_backend.list_dir_or_file(tmp_dir)) == set(
                ['dir1', 'dir2', 'text1.txt', 'text2.txt'])
            # 2. list directories and files recursively
            assert set(disk_backend.list_dir_or_file(
                tmp_dir, recursive=True)) == set([
                    'dir1',
                    osp.join('dir1', 'text3.txt'), 'dir2',
                    osp.join('dir2', 'dir3'),
                    osp.join('dir2', 'dir3', 'text4.txt'),
                    osp.join('dir2', 'img.jpg'), 'text1.txt', 'text2.txt'
                ])
            # 3. only list directories
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_file=False)) == set(['dir1', 'dir2'])
            with pytest.raises(
                    TypeError,
                    match='`suffix` should be None when `list_dir` is True'):
                # Exception is raised among the `list_dir_or_file` of client,
                # so we need to invode the client to trigger the exception
                disk_backend.client.list_dir_or_file(
                    tmp_dir, list_file=False, suffix='.txt')
            # 4. only list directories recursively
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_file=False, recursive=True)) == set(
                        ['dir1', 'dir2',
                         osp.join('dir2', 'dir3')])
            # 5. only list files
            assert set(disk_backend.list_dir_or_file(
                tmp_dir, list_dir=False)) == set(['text1.txt', 'text2.txt'])
            # 6. only list files recursively
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_dir=False, recursive=True)) == set([
                        osp.join('dir1', 'text3.txt'),
                        osp.join('dir2', 'dir3', 'text4.txt'),
                        osp.join('dir2', 'img.jpg'), 'text1.txt', 'text2.txt'
                    ])
            # 7. only list files ending with suffix
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_dir=False,
                    suffix='.txt')) == set(['text1.txt', 'text2.txt'])
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_dir=False,
                    suffix=('.txt',
                            '.jpg'))) == set(['text1.txt', 'text2.txt'])
            with pytest.raises(
                    TypeError,
                    match='`suffix` must be a string or tuple of strings'):
                disk_backend.client.list_dir_or_file(
                    tmp_dir, list_dir=False, suffix=['.txt', '.jpg'])
            # 8. only list files ending with suffix recursively
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir, list_dir=False, suffix='.txt',
                    recursive=True)) == set([
                        osp.join('dir1', 'text3.txt'),
                        osp.join('dir2', 'dir3', 'text4.txt'), 'text1.txt',
                        'text2.txt'
                    ])
            # 7. only list files ending with suffix
            assert set(
                disk_backend.list_dir_or_file(
                    tmp_dir,
                    list_dir=False,
                    suffix=('.txt', '.jpg'),
                    recursive=True)) == set([
                        osp.join('dir1', 'text3.txt'),
                        osp.join('dir2', 'dir3', 'text4.txt'),
                        osp.join('dir2', 'img.jpg'), 'text1.txt', 'text2.txt'
                    ])

    @patch('ceph.S3Client', MockS3Client)
    def test_ceph_backend(self):
        ceph_backend = FileClient('ceph')

        # test `allow_symlink` attribute
        assert not ceph_backend.allow_symlink

        # input path is Path object
        with pytest.raises(NotImplementedError):
            ceph_backend.get_text(self.text_path)
        # input path is str
        with pytest.raises(NotImplementedError):
            ceph_backend.get_text(str(self.text_path))

        # input path is Path object
        img_bytes = ceph_backend.get(self.img_path)
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape
        # input path is str
        img_bytes = ceph_backend.get(str(self.img_path))
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape

        # `path_mapping` is either None or dict
        with pytest.raises(AssertionError):
            FileClient('ceph', path_mapping=1)
        # test `path_mapping`
        ceph_path = 's3://user/data'
        ceph_backend = FileClient(
            'ceph', path_mapping={str(self.test_data_dir): ceph_path})
        ceph_backend.client._client.Get = MagicMock(
            return_value=ceph_backend.client._client.Get(self.img_path))
        img_bytes = ceph_backend.get(self.img_path)
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape
        ceph_backend.client._client.Get.assert_called_with(
            str(self.img_path).replace(str(self.test_data_dir), ceph_path))

    @patch('petrel_client.client.Client', MockPetrelClient)
    @pytest.mark.parametrize('backend,prefix', [('petrel', None),
                                                (None, 's3')])
    def test_petrel_backend(self, backend, prefix):
        petrel_backend = FileClient(backend=backend, prefix=prefix)

        # test `allow_symlink` attribute
        assert not petrel_backend.allow_symlink

        # input path is Path object
        img_bytes = petrel_backend.get(self.img_path)
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape
        # input path is str
        img_bytes = petrel_backend.get(str(self.img_path))
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape

        # `path_mapping` is either None or dict
        with pytest.raises(AssertionError):
            FileClient('petrel', path_mapping=1)

        # test `_map_path`
        petrel_dir = 's3://user/data'
        petrel_backend = FileClient(
            'petrel', path_mapping={str(self.test_data_dir): petrel_dir})
        assert petrel_backend.client._map_path(str(self.img_path)) == \
            str(self.img_path).replace(str(self.test_data_dir), petrel_dir)

        petrel_path = f'{petrel_dir}/test.jpg'
        petrel_backend = FileClient('petrel')

        # test `_format_path`
        assert petrel_backend.client._format_path('s3://user\\data\\test.jpg')\
            == petrel_path

        # test `get`
        with patch.object(
                petrel_backend.client._client, 'Get',
                return_value=b'petrel') as mock_get:
            assert petrel_backend.get(petrel_path) == b'petrel'
            mock_get.assert_called_once_with(petrel_path)

        # test `get_text`
        with patch.object(
                petrel_backend.client._client, 'Get',
                return_value=b'petrel') as mock_get:
            assert petrel_backend.get_text(petrel_path) == 'petrel'
            mock_get.assert_called_once_with(petrel_path)

        # test `put`
        with patch.object(petrel_backend.client._client, 'put') as mock_put:
            petrel_backend.put(b'petrel', petrel_path)
            mock_put.assert_called_once_with(petrel_path, b'petrel')

        # test `put_text`
        with patch.object(petrel_backend.client._client, 'put') as mock_put:
            petrel_backend.put_text('petrel', petrel_path)
            mock_put.assert_called_once_with(petrel_path, b'petrel')

        # test `remove`
        assert has_method(petrel_backend.client._client, 'delete')
        # raise Exception if `delete` is not implemented
        with delete_and_reset_method(petrel_backend.client._client, 'delete'):
            assert not has_method(petrel_backend.client._client, 'delete')
            with pytest.raises(NotImplementedError):
                petrel_backend.remove(petrel_path)

        with patch.object(petrel_backend.client._client,
                          'delete') as mock_delete:
            petrel_backend.remove(petrel_path)
            mock_delete.assert_called_once_with(petrel_path)

        # test `exists`
        assert has_method(petrel_backend.client._client, 'contains')
        assert has_method(petrel_backend.client._client, 'isdir')
        # raise Exception if `delete` is not implemented
        with delete_and_reset_method(petrel_backend.client._client,
                                     'contains'), delete_and_reset_method(
                                         petrel_backend.client._client,
                                         'isdir'):
            assert not has_method(petrel_backend.client._client, 'contains')
            assert not has_method(petrel_backend.client._client, 'isdir')
            with pytest.raises(NotImplementedError):
                petrel_backend.exists(petrel_path)

        with patch.object(
                petrel_backend.client._client, 'contains',
                return_value=True) as mock_contains:
            assert petrel_backend.exists(petrel_path)
            mock_contains.assert_called_once_with(petrel_path)

        # test `isdir`
        assert has_method(petrel_backend.client._client, 'isdir')
        with delete_and_reset_method(petrel_backend.client._client, 'isdir'):
            assert not has_method(petrel_backend.client._client, 'isdir')
            with pytest.raises(NotImplementedError):
                petrel_backend.isdir(petrel_path)

        with patch.object(
                petrel_backend.client._client, 'isdir',
                return_value=True) as mock_isdir:
            assert petrel_backend.isdir(petrel_dir)
            mock_isdir.assert_called_once_with(petrel_dir)

        # test `isfile`
        assert has_method(petrel_backend.client._client, 'contains')
        with delete_and_reset_method(petrel_backend.client._client,
                                     'contains'):
            assert not has_method(petrel_backend.client._client, 'contains')
            with pytest.raises(NotImplementedError):
                petrel_backend.isfile(petrel_path)

        with patch.object(
                petrel_backend.client._client, 'contains',
                return_value=True) as mock_contains:
            assert petrel_backend.isfile(petrel_path)
            mock_contains.assert_called_once_with(petrel_path)

        # test `join_path`
        assert petrel_backend.join_path(petrel_dir, 'file') == \
            f'{petrel_dir}/file'
        assert petrel_backend.join_path(f'{petrel_dir}/', 'file') == \
            f'{petrel_dir}/file'
        assert petrel_backend.join_path(petrel_dir, 'dir', 'file') == \
            f'{petrel_dir}/dir/file'

        # test `get_local_path`
        with patch.object(petrel_backend.client._client, 'Get',
                          return_value=b'petrel') as mock_get, \
             patch.object(petrel_backend.client._client, 'contains',
                          return_value=True) as mock_contains:
            with petrel_backend.get_local_path(petrel_path) as path:
                assert Path(path).open('rb').read() == b'petrel'
            # exist the with block and path will be released
            assert not osp.isfile(path)
            mock_get.assert_called_once_with(petrel_path)
            mock_contains.assert_called_once_with(petrel_path)

        # test `list_dir_or_file`
        assert has_method(petrel_backend.client._client, 'list')
        with delete_and_reset_method(petrel_backend.client._client, 'list'):
            assert not has_method(petrel_backend.client._client, 'list')
            with pytest.raises(NotImplementedError):
                list(petrel_backend.list_dir_or_file(petrel_dir))

        with build_temporary_directory() as tmp_dir:
            # 1. list directories and files
            assert set(petrel_backend.list_dir_or_file(tmp_dir)) == set(
                ['dir1', 'dir2', 'text1.txt', 'text2.txt'])
            # 2. list directories and files recursively
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, recursive=True)) == set([
                        'dir1', '/'.join(('dir1', 'text3.txt')), 'dir2',
                        '/'.join(('dir2', 'dir3')), '/'.join(
                            ('dir2', 'dir3', 'text4.txt')), '/'.join(
                                ('dir2', 'img.jpg')), 'text1.txt', 'text2.txt'
                    ])
            # 3. only list directories
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_file=False)) == set(['dir1', 'dir2'])
            with pytest.raises(
                    TypeError,
                    match=('`list_dir` should be False when `suffix` is not '
                           'None')):
                # Exception is raised among the `list_dir_or_file` of client,
                # so we need to invode the client to trigger the exception
                petrel_backend.client.list_dir_or_file(
                    tmp_dir, list_file=False, suffix='.txt')
            # 4. only list directories recursively
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_file=False, recursive=True)) == set(
                        ['dir1', 'dir2', '/'.join(('dir2', 'dir3'))])
            # 5. only list files
            assert set(
                petrel_backend.list_dir_or_file(tmp_dir,
                                                list_dir=False)) == set(
                                                    ['text1.txt', 'text2.txt'])
            # 6. only list files recursively
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_dir=False, recursive=True)) == set([
                        '/'.join(('dir1', 'text3.txt')), '/'.join(
                            ('dir2', 'dir3', 'text4.txt')), '/'.join(
                                ('dir2', 'img.jpg')), 'text1.txt', 'text2.txt'
                    ])
            # 7. only list files ending with suffix
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_dir=False,
                    suffix='.txt')) == set(['text1.txt', 'text2.txt'])
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_dir=False,
                    suffix=('.txt',
                            '.jpg'))) == set(['text1.txt', 'text2.txt'])
            with pytest.raises(
                    TypeError,
                    match='`suffix` must be a string or tuple of strings'):
                petrel_backend.client.list_dir_or_file(
                    tmp_dir, list_dir=False, suffix=['.txt', '.jpg'])
            # 8. only list files ending with suffix recursively
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir, list_dir=False, suffix='.txt',
                    recursive=True)) == set([
                        '/'.join(('dir1', 'text3.txt')), '/'.join(
                            ('dir2', 'dir3', 'text4.txt')), 'text1.txt',
                        'text2.txt'
                    ])
            # 7. only list files ending with suffix
            assert set(
                petrel_backend.list_dir_or_file(
                    tmp_dir,
                    list_dir=False,
                    suffix=('.txt', '.jpg'),
                    recursive=True)) == set([
                        '/'.join(('dir1', 'text3.txt')), '/'.join(
                            ('dir2', 'dir3', 'text4.txt')), '/'.join(
                                ('dir2', 'img.jpg')), 'text1.txt', 'text2.txt'
                    ])

    @patch('mc.MemcachedClient.GetInstance', MockMemcachedClient)
    @patch('mc.pyvector', MagicMock)
    @patch('mc.ConvertBuffer', lambda x: x.content)
    def test_memcached_backend(self):
        mc_cfg = dict(server_list_cfg='', client_cfg='', sys_path=None)
        mc_backend = FileClient('memcached', **mc_cfg)

        # test `allow_symlink` attribute
        assert not mc_backend.allow_symlink

        # input path is Path object
        with pytest.raises(NotImplementedError):
            mc_backend.get_text(self.text_path)
        # input path is str
        with pytest.raises(NotImplementedError):
            mc_backend.get_text(str(self.text_path))

        # input path is Path object
        img_bytes = mc_backend.get(self.img_path)
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape
        # input path is str
        img_bytes = mc_backend.get(str(self.img_path))
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape

    def test_lmdb_backend(self):
        lmdb_path = self.test_data_dir / 'demo.lmdb'

        # db_path is Path object
        lmdb_backend = FileClient('lmdb', db_path=lmdb_path)

        # test `allow_symlink` attribute
        assert not lmdb_backend.allow_symlink

        with pytest.raises(NotImplementedError):
            lmdb_backend.get_text(self.text_path)

        img_bytes = lmdb_backend.get('baboon')
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == (120, 125, 3)

        # db_path is str
        lmdb_backend = FileClient('lmdb', db_path=str(lmdb_path))
        with pytest.raises(NotImplementedError):
            lmdb_backend.get_text(str(self.text_path))
        img_bytes = lmdb_backend.get('baboon')
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == (120, 125, 3)

    @pytest.mark.parametrize('backend,prefix', [('http', None),
                                                (None, 'http')])
    def test_http_backend(self, backend, prefix):
        http_backend = FileClient(backend=backend, prefix=prefix)
        img_url = 'https://raw.githubusercontent.com/open-mmlab/mmcv/' \
            'master/tests/data/color.jpg'
        text_url = 'https://raw.githubusercontent.com/open-mmlab/mmcv/' \
            'master/tests/data/filelist.txt'

        # test `allow_symlink` attribute
        assert not http_backend.allow_symlink

        # input is path or Path object
        with pytest.raises(Exception):
            http_backend.get(self.img_path)
        with pytest.raises(Exception):
            http_backend.get(str(self.img_path))
        with pytest.raises(Exception):
            http_backend.get_text(self.text_path)
        with pytest.raises(Exception):
            http_backend.get_text(str(self.text_path))

        # input url is http image
        img_bytes = http_backend.get(img_url)
        img = mmcv.imfrombytes(img_bytes)
        assert img.shape == self.img_shape

        # input url is http text
        value_buf = http_backend.get_text(text_url)
        assert self.text_path.open('r').read() == value_buf

        # test `_get_local_path`
        # exist the with block and path will be released
        with http_backend.get_local_path(img_url) as path:
            assert mmcv.imread(path).shape == self.img_shape
        assert not osp.isfile(path)

    def test_new_magic_method(self):

        class DummyBackend1(BaseStorageBackend):

            def get(self, filepath):
                return filepath

            def get_text(self, filepath, encoding='utf-8'):
                return filepath

        FileClient.register_backend('dummy_backend', DummyBackend1)
        client1 = FileClient(backend='dummy_backend')
        client2 = FileClient(backend='dummy_backend')
        assert client1 is client2

        # if a backend is overwrote, it will disable the singleton pattern for
        # the backend
        class DummyBackend2(BaseStorageBackend):

            def get(self, filepath):
                pass

            def get_text(self, filepath):
                pass

        FileClient.register_backend('dummy_backend', DummyBackend2, force=True)
        client3 = FileClient(backend='dummy_backend')
        client4 = FileClient(backend='dummy_backend')
        assert client3 is not client4

    def test_parse_uri_prefix(self):
        # input path is None
        with pytest.raises(AssertionError):
            FileClient.parse_uri_prefix(None)
        # input path is list
        with pytest.raises(AssertionError):
            FileClient.parse_uri_prefix([])

        # input path is Path object
        assert FileClient.parse_uri_prefix(self.img_path) is None
        # input path is str
        assert FileClient.parse_uri_prefix(str(self.img_path)) is None

        # input path starts with https
        img_url = 'https://raw.githubusercontent.com/open-mmlab/mmcv/' \
            'master/tests/data/color.jpg'
        assert FileClient.parse_uri_prefix(img_url) == 'https'

        # input path starts with s3
        img_url = 's3://your_bucket/img.png'
        assert FileClient.parse_uri_prefix(img_url) == 's3'

        # input path starts with clusterName:s3
        img_url = 'clusterName:s3://your_bucket/img.png'
        assert FileClient.parse_uri_prefix(img_url) == 's3'

    def test_infer_client(self):
        # HardDiskBackend
        file_client_args = {'backend': 'disk'}
        client = FileClient.infer_client(file_client_args)
        assert client.name == 'HardDiskBackend'
        client = FileClient.infer_client(uri=self.img_path)
        assert client.name == 'HardDiskBackend'

        # PetrelBackend
        file_client_args = {'backend': 'petrel'}
        client = FileClient.infer_client(file_client_args)
        assert client.name == 'PetrelBackend'
        uri = 's3://user_data'
        client = FileClient.infer_client(uri=uri)
        assert client.name == 'PetrelBackend'

    def test_register_backend(self):

        # name must be a string
        with pytest.raises(TypeError):

            class TestClass1:
                pass

            FileClient.register_backend(1, TestClass1)

        # module must be a class
        with pytest.raises(TypeError):
            FileClient.register_backend('int', 0)

        # module must be a subclass of BaseStorageBackend
        with pytest.raises(TypeError):

            class TestClass1:
                pass

            FileClient.register_backend('TestClass1', TestClass1)

        class ExampleBackend(BaseStorageBackend):

            def get(self, filepath):
                return filepath

            def get_text(self, filepath, encoding='utf-8'):
                return filepath

        FileClient.register_backend('example', ExampleBackend)
        example_backend = FileClient('example')
        assert example_backend.get(self.img_path) == self.img_path
        assert example_backend.get_text(self.text_path) == self.text_path
        assert 'example' in FileClient._backends

        class Example2Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes2'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text2'

        # force=False
        with pytest.raises(KeyError):
            FileClient.register_backend('example', Example2Backend)

        FileClient.register_backend('example', Example2Backend, force=True)
        example_backend = FileClient('example')
        assert example_backend.get(self.img_path) == b'bytes2'
        assert example_backend.get_text(self.text_path) == 'text2'

        @FileClient.register_backend(name='example3')
        class Example3Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes3'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text3'

        example_backend = FileClient('example3')
        assert example_backend.get(self.img_path) == b'bytes3'
        assert example_backend.get_text(self.text_path) == 'text3'
        assert 'example3' in FileClient._backends

        # force=False
        with pytest.raises(KeyError):

            @FileClient.register_backend(name='example3')
            class Example4Backend(BaseStorageBackend):

                def get(self, filepath):
                    return b'bytes4'

                def get_text(self, filepath, encoding='utf-8'):
                    return 'text4'

        @FileClient.register_backend(name='example3', force=True)
        class Example5Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes5'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text5'

        example_backend = FileClient('example3')
        assert example_backend.get(self.img_path) == b'bytes5'
        assert example_backend.get_text(self.text_path) == 'text5'

        # prefixes is a str
        class Example6Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes6'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text6'

        FileClient.register_backend(
            'example4',
            Example6Backend,
            force=True,
            prefixes='example4_prefix')
        example_backend = FileClient('example4')
        assert example_backend.get(self.img_path) == b'bytes6'
        assert example_backend.get_text(self.text_path) == 'text6'
        example_backend = FileClient(prefix='example4_prefix')
        assert example_backend.get(self.img_path) == b'bytes6'
        assert example_backend.get_text(self.text_path) == 'text6'
        example_backend = FileClient('example4', prefix='example4_prefix')
        assert example_backend.get(self.img_path) == b'bytes6'
        assert example_backend.get_text(self.text_path) == 'text6'

        # prefixes is a list of str
        class Example7Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes7'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text7'

        FileClient.register_backend(
            'example5',
            Example7Backend,
            force=True,
            prefixes=['example5_prefix1', 'example5_prefix2'])
        example_backend = FileClient('example5')
        assert example_backend.get(self.img_path) == b'bytes7'
        assert example_backend.get_text(self.text_path) == 'text7'
        example_backend = FileClient(prefix='example5_prefix1')
        assert example_backend.get(self.img_path) == b'bytes7'
        assert example_backend.get_text(self.text_path) == 'text7'
        example_backend = FileClient(prefix='example5_prefix2')
        assert example_backend.get(self.img_path) == b'bytes7'
        assert example_backend.get_text(self.text_path) == 'text7'

        # backend has a higher priority than prefixes
        class Example8Backend(BaseStorageBackend):

            def get(self, filepath):
                return b'bytes8'

            def get_text(self, filepath, encoding='utf-8'):
                return 'text8'

        FileClient.register_backend(
            'example6',
            Example8Backend,
            force=True,
            prefixes='example6_prefix')
        example_backend = FileClient('example6')
        assert example_backend.get(self.img_path) == b'bytes8'
        assert example_backend.get_text(self.text_path) == 'text8'
        example_backend = FileClient('example6', prefix='example4_prefix')
        assert example_backend.get(self.img_path) == b'bytes8'
        assert example_backend.get_text(self.text_path) == 'text8'