"""Test airfs.storage."""
from copy import deepcopy as _deepcopy
from os import urandom as _os_urandom
from time import time as _time
from uuid import uuid4 as _uuid

import pytest as _pytest
import requests as _requests


def _urandom(size):
    """Return random generated bytes.

    But avoid generating Null chars.

    Args:
        size (int):

    Returns:
        bytes: Generated bytes.
    """
    return _os_urandom(size).replace(b"\0", b"\x01")


class StorageTester:
    """Class that contain common set of tests for storage."""

    def __init__(
        self,
        system=None,
        raw_io=None,
        buffered_io=None,
        storage_mock=None,
        unsupported_operations=None,
        storage_info=None,
        system_parameters=None,
        root=None,
        path_prefix=None,
    ):
        """Init.

        Args:
            system (airfs._core.io_system.SystemBase instance): System to test.
            raw_io (airfs._core.io_raw.ObjectRawIOBase subclass): Raw IO class.
            buffered_io (airfs._core.io_buffered.ObjectBufferedIOBase subclass):
                Buffered IOclass.
            storage_mock (tests.storage_mock.ObjectStorageMock instance): Storage mock
                in use, if any.
            unsupported_operations: Unsupported operations.
            storage_info (dict): Storage information from airfs.mount.
            system_parameters: System parameters.
            root: Root.
            path_prefix (str): Prefix to append before any path.
        """
        if system is None:
            system = storage_info["system_cached"]
        if raw_io is None:
            raw_io = storage_info["raw"]
        if buffered_io is None:
            buffered_io = storage_info["buffered"]
        if system_parameters is None and storage_info:
            system_parameters = storage_info["system_parameters"]

        if not system_parameters:
            system_parameters = dict(storage_parameters=dict())
        else:
            system_parameters = _deepcopy(system_parameters)

        self._system_parameters = system_parameters
        self._system_parameters["storage_parameters"]["airfs.system_cached"] = system
        self._system = system
        self._raw_io = raw_io
        self._buffered_io = buffered_io
        self._storage_mock = storage_mock
        self._unsupported_operations = unsupported_operations or tuple()

        # Get storage root
        if not root:
            root = system.roots[0]

        # Defines randomized names for locator and objects
        self.locator = self._get_id()
        self.locator_url = "/".join((root, self.locator))
        self.base_dir_name = f"{self._get_id()}/"
        if path_prefix:
            dir_parts = (self.locator, path_prefix, self.base_dir_name)
        else:
            dir_parts = (self.locator, self.base_dir_name)
        self.base_dir_path = "/".join(dir_parts)
        self.base_dir_url = root + self.base_dir_path

        # Run test sequence
        self._objects = set()
        self._to_clean = self._objects.add

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, exc_traceback):
        from airfs._core.exceptions import (
            ObjectNotFoundError,
            ObjectUnsupportedOperation,
        )

        # Remove objects, and once empty the locator
        for obj in reversed(sorted(self._objects, key=str.lower)):
            self._objects.discard(obj)
            try:
                self._system.remove(obj, relative=True)
            except (ObjectNotFoundError, ObjectUnsupportedOperation):
                continue

    def test_common(self):
        """Common set of tests."""
        self._test_system_locator()
        self._test_system_objects()
        self._test_raw_io()
        self._test_buffered_io()
        # TODO: Add airfs public functions tests

        # Only if mocked
        if self._storage_mock is not None:
            self._test_mock_only()

    def _is_supported(self, feature):
        """Return True if a feature is supported.

        Args:
            feature (str): Feature to support.

        Returns:
            bool: Feature is supported.
        """
        return feature not in self._unsupported_operations

    @staticmethod
    def _get_id():
        """Return an unique ID.

        Returns:
            str: id
        """
        return f"airfs{str(_uuid()).replace('-', '')}"

    def _test_raw_io(self):
        """Tests raw IO."""
        from os import SEEK_END, SEEK_CUR
        from io import UnsupportedOperation

        size = 100
        file_name = "raw_file0.dat"
        file_path = self.base_dir_path + file_name
        self._to_clean(file_path)
        content = _urandom(size)

        # Open file in write mode
        assert not self._system.exists(file_path), "Raw write, file not exists"

        if self._is_supported("write"):
            with self._raw_io(file_path, "wb", **self._system_parameters) as file:
                assert (
                    self._system.getsize(file_path) == 0
                ), "Raw write, file must exist but empty"

                # Get file file type for later
                is_seekable = file.seekable()
                try:
                    max_flush_size = file.MAX_FLUSH_SIZE
                except AttributeError:
                    max_flush_size = 0

                # Test: Write blocs of data
                assert file.write(content[:10]) == 10, "Raw write, written size match"
                if is_seekable:
                    assert file.write(b"\0" * 10) == 10, "Raw write, written size match"
                else:
                    assert (
                        file.write(content[10:20]) == 10
                    ), "Raw write, written size match"
                assert file.write(content[20:]) == 80, "Raw write, written size match"

                # Test: tell
                if is_seekable:
                    assert file.tell() == size, "Raw write, tell match writen size"

                    # Test write seek back and write
                    assert file.seek(10) == 10, "Raw write, seek position match"
                    assert (
                        file.write(content[10:20]) == 10
                    ), "Raw write, written size match"
                    assert file.tell() == 20, "Raw write, tell match ending position"

                else:
                    # Test not seekable raises Unsupported exception
                    with _pytest.raises(UnsupportedOperation):
                        file.tell()

                    with _pytest.raises(UnsupportedOperation):
                        file.seek(0)

                # Test: read in write mode is not supported
                with _pytest.raises(UnsupportedOperation):
                    file.read()

                with _pytest.raises(UnsupportedOperation):
                    file.readinto(bytearray(100))

        else:
            is_seekable = False
            max_flush_size = 0

            # Test: Unsupported
            with _pytest.raises(UnsupportedOperation):
                self._raw_io(file_path, "wb", **self._system_parameters)

            # Create pre-existing file
            if self._storage_mock:
                self._storage_mock.put_object(
                    self.locator, self.base_dir_name + file_name, content
                )

        # Open file in read mode
        with self._raw_io(file_path, **self._system_parameters) as file:
            # Test: read_all
            assert file.readall() == content, "Raw read all, content match"
            assert file.tell() == size, "Raw read all, tell match"

            # Test: seek and read_all
            assert file.seek(10) == 10, "Raw seek 10 & read all, seek match"
            assert (
                file.readall() == content[10:]
            ), "Raw seek 10 & read all, content match"
            assert file.tell() == size, "Raw seek 10 & read all, tell match"

            # Test: seek from current position & read_all
            assert (
                file.seek(-50, SEEK_CUR) == 50
            ), "Raw seek from current & read all, seek match"
            assert (
                file.readall() == content[-50:]
            ), "Raw seek from current & read all, content match"
            assert file.tell() == size, "Raw seek from current & read all, tell match"

            # Test: seek with bad whence value
            with _pytest.raises(ValueError):
                file.seek(0, 10)

            # Test: Cannot write in read mode
            with _pytest.raises(UnsupportedOperation):
                file.write(b"0")

            # Test: Flush has no effect in read mode
            file.flush()

            # Test: _read_range
            assert file.seek(0) == 0, "Raw seek 0, seek match"
            buffer = bytearray(40)
            assert file.readinto(buffer) == 40, "Raw read into, returned size match"
            assert bytes(buffer) == content[:40], "Raw read into, content match"
            assert file.tell() == 40, "Raw read into, tell match"

            buffer = bytearray(40)
            assert (
                file.readinto(buffer) == 40
            ), "Raw read into from 40, returned size match"
            assert (
                bytes(buffer) == content[40:80]
            ), "Raw read into from 40, content match"
            assert file.tell() == 80, "Raw read into from 40, tell match"

            buffer = bytearray(40)
            assert (
                file.readinto(buffer) == 20
            ), "Raw read into partially over EOF, returned size match"
            assert (
                bytes(buffer) == content[80:] + b"\0" * 20
            ), "Raw read into partially over EOF, content match"
            assert file.tell() == size, "Raw read into partially over EOF, tell match"

            buffer = bytearray(40)
            assert (
                file.readinto(buffer) == 0
            ), "Raw read into over EOF, returned size match"
            assert bytes(buffer) == b"\0" * 40, "Raw read into over EOF, content match"
            assert file.tell() == size, "Raw read into over EOF, tell match"

            file.seek(-10, SEEK_END)
            buffer = bytearray(20)
            assert (
                file.readinto(buffer) == 10
            ), "Raw seek from end & read into, returned size match"
            assert (
                bytes(buffer) == content[90:] + b"\0" * 10
            ), "Raw seek from end & read into, content match"
            assert file.tell() == size, "Raw seek from end & read into, tell match"

        # Test: Append mode
        if self._is_supported("write"):
            # Test: Appending on existing file
            with self._raw_io(file_path, mode="ab", **self._system_parameters) as file:
                file.write(content)

            with self._raw_io(file_path, **self._system_parameters) as file:
                assert (
                    file.readall() == content + content
                ), "Raw append, previous content read match"

            # Test: Appending on not existing file
            file_name = "raw_file1.dat"
            file_path = self.base_dir_path + file_name
            self._to_clean(file_path)
            with self._raw_io(file_path, mode="ab", **self._system_parameters) as file:
                file.write(content)

            with self._raw_io(file_path, **self._system_parameters) as file:
                assert (
                    file.readall() == content
                ), "Raw append, file create content match"

        # Test: Seek out of file and write
        if is_seekable:
            file_name = "raw_file2.dat"
            file_path = self.base_dir_path + file_name
            self._to_clean(file_path)
            with self._raw_io(file_path, "wb", **self._system_parameters) as file:
                file.seek(256)
                file.write(b"\x01")

            with self._raw_io(file_path, "rb", **self._system_parameters) as file:
                assert (
                    file.readall() == b"\0" * 256 + b"\x01"
                ), "Raw seek, null padding read"

        # Test: write big file
        if self._is_supported("write") and max_flush_size:
            file_name = "raw_file3.dat"
            file_path = self.base_dir_path + file_name
            self._to_clean(file_path)

            size = max_flush_size * 4
            content = _urandom(size)

            with self._raw_io(file_path, "wb", **self._system_parameters) as file:
                file.write(content)

            with self._raw_io(file_path, "rb", **self._system_parameters) as file:
                assert file.readall() == content, "Raw Write big file, content match"

        # Test exclusive write mode
        if self._is_supported("write"):
            file_name = "raw_file4.dat"
            file_path = self.base_dir_path + file_name
            self._to_clean(file_path)

            # Create file
            with self._raw_io(file_path, "xb", **self._system_parameters):
                pass

            # File already exists
            with _pytest.raises(FileExistsError):
                self._raw_io(file_path, "xb", **self._system_parameters)

    def _test_buffered_io(self):
        """Tests buffered IO."""
        from airfs.io import ObjectBufferedIOBase
        from io import UnsupportedOperation

        # Set buffer size
        buffer_size = 16 * 1024

        # Test: write data, not multiple of buffer
        file_name = "buffered_file0.dat"
        file_path = self.base_dir_path + file_name
        self._to_clean(file_path)
        size = int(4.5 * buffer_size)
        content = _urandom(size)

        if self._is_supported("write"):
            with self._buffered_io(
                file_path, "wb", buffer_size=buffer_size, **self._system_parameters
            ) as file:
                file.write(content)
        else:
            # Test: Unsupported
            with _pytest.raises(UnsupportedOperation):
                self._buffered_io(
                    file_path, "wb", buffer_size=buffer_size, **self._system_parameters
                )

            # Create pre-existing file
            if self._storage_mock:
                self._storage_mock.put_object(
                    self.locator, self.base_dir_name + file_name, content
                )

        # Test: Read data, not multiple of buffer
        with self._buffered_io(
            file_path, "rb", buffer_size=buffer_size, **self._system_parameters
        ) as file:
            assert content == file.read(), "Buffered read, not multiple of buffer size"

        # Test: write data, multiple of buffer
        file_name = "buffered_file1.dat"
        file_path = self.base_dir_path + file_name
        self._to_clean(file_path)
        size = int(5 * buffer_size)
        content = _urandom(size)

        if self._is_supported("write"):
            with self._buffered_io(
                file_path, "wb", buffer_size=buffer_size, **self._system_parameters
            ) as file:
                file.write(content)

                # Test: Flush manually
                file.flush()

                # Test: read in write mode is not supported
                with _pytest.raises(UnsupportedOperation):
                    file.read()

                with _pytest.raises(UnsupportedOperation):
                    file.read1()

                with _pytest.raises(UnsupportedOperation):
                    file.readinto(bytearray(100))

                with _pytest.raises(UnsupportedOperation):
                    file.readinto1(bytearray(100))

                with _pytest.raises(UnsupportedOperation):
                    file.peek()

                # Test: Unsupported if not seekable
                if not file.seekable():
                    with _pytest.raises(UnsupportedOperation):
                        file.tell()

                    with _pytest.raises(UnsupportedOperation):
                        file.seek(0)
        else:
            # Create pre-existing file
            if self._storage_mock:
                self._storage_mock.put_object(
                    self.locator, self.base_dir_name + file_name, content
                )

        # Test: Read data, multiple of buffer
        with self._buffered_io(
            file_path, "rb", buffer_size=buffer_size, **self._system_parameters
        ) as file:
            # Test full data read
            assert content == file.read(), "Buffered read, multiple of buffer size"

            # Test: seek
            assert file.seek(10) == 10, "Buffered read, seek"
            assert file.tell() == 10, "Buffered read, tell match seek"

            # Test: peek:
            assert content[10:110] == file.peek(
                100
            ), "Buffered read, peek content match"
            assert file.tell() == 10, "Buffered read, peek tell match"

            # Test: Cannot write in read mode
            with _pytest.raises(UnsupportedOperation):
                file.write(b"0")

            # Test: Flush has no effect in read mode
            file.flush()

            # Check if airfs subclass
            is_rfs_subclass = isinstance(file, ObjectBufferedIOBase)

        # Test: Buffer limits and default values
        if is_rfs_subclass:
            with self._buffered_io(file_path, **self._system_parameters) as file:
                assert (
                    file._buffer_size == file.DEFAULT_BUFFER_SIZE
                ), "Buffered, Default buffer size"

                # Get limits values
                minimum_buffer_size = file.MINIMUM_BUFFER_SIZE
                maximum_buffer_size = file.MAXIMUM_BUFFER_SIZE

                # Get current max buffers
                calculated_max_buffers = file._max_buffers

            # Test: Minimum buffer size
            if minimum_buffer_size > 1:
                with self._buffered_io(
                    file_path,
                    buffer_size=minimum_buffer_size // 2,
                    **self._system_parameters,
                ) as file:
                    assert (
                        file._buffer_size == minimum_buffer_size
                    ), "Buffered, Minimum buffer size"

            # Test: Maximum buffer size
            if maximum_buffer_size:
                with self._buffered_io(
                    file_path,
                    buffer_size=maximum_buffer_size * 2,
                    **self._system_parameters,
                ) as file:
                    assert (
                        file._buffer_size == maximum_buffer_size
                    ), "Buffered, Maximum buffer size"

            # Test: Maximum buffer count
            assert calculated_max_buffers, "Buffered, calculated buffer count not 0"

            max_buffers = calculated_max_buffers * 2
            with self._buffered_io(
                file_path, mode="rb", max_buffers=max_buffers, **self._system_parameters
            ) as file:
                assert (
                    file._max_buffers == max_buffers
                ), "Buffered, Maximum buffer count"

    def _test_system_locator(self):
        """Test system internals related to locators."""
        from airfs._core.exceptions import ObjectUnsupportedOperation

        system = self._system

        # Test: Create locator
        if self._is_supported("mkdir"):
            system.make_dir(self.locator_url)
            self._to_clean(self.locator)
        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                system.make_dir(self.locator_url)

            # Create a preexisting locator
            if self._storage_mock:
                self._storage_mock.put_locator(self.locator)

        # Test: Check locator listed
        if self._is_supported("listdir") and self._is_supported("list_locator"):
            for name, header, _ in system._list_locators(None):
                if name == self.locator and isinstance(header, dict):
                    break
            else:
                _pytest.fail(f'Locator "{self.locator}" not found')

            # Test: Check locator header return a mapping
            assert hasattr(
                system.head(path=self.locator), "__getitem__"
            ), "List locators, header is mapping"
        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                system._list_locators(None)

        # Test: remove locator
        tmp_locator = self._get_id()
        self._to_clean(tmp_locator)
        if self._is_supported("mkdir"):
            system.make_dir(tmp_locator)
        elif self._storage_mock:
            self._storage_mock.put_locator(tmp_locator)

        if self._is_supported("remove"):
            if self._is_supported("listdir"):
                assert tmp_locator in [
                    name for name, _, _ in system._list_locators(None)
                ], "Remove locator, locator exists"

            system.remove(tmp_locator)

            if self._is_supported("listdir"):
                assert tmp_locator not in [
                    name for name, _, _ in system._list_locators(None)
                ], "Remove locator, locator not exists"
        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                system.remove(tmp_locator)

    def _test_system_objects(self):
        """Test system internals related to objects."""
        from airfs._core.exceptions import (
            ObjectNotFoundError,
            ObjectUnsupportedOperation,
            ObjectNotImplementedError,
            ObjectNotASymlinkError,
        )

        system = self._system

        if self._is_supported("mkdir"):
            # Create parent directory
            system.make_dir(self.base_dir_path)
            self._to_clean(self.base_dir_path)

            # Test: Make a directory (With trailing /)
            dir_name0 = "directory0/"
            dir_path0 = self.base_dir_path + dir_name0
            system.make_dir(dir_path0)
            self._to_clean(dir_path0)
            if self._is_supported("listdir"):
                assert (
                    dir_path0 in self._list_objects_names()
                ), 'Create directory, exists (with "/")'

            # Test: Check directory header
            assert hasattr(
                system.head(path=dir_path0), "__getitem__"
            ), "Head directory, header is mapping"

            # Test: Make a directory (Without trailing /)
            dir_name1 = "directory1"
            dir_path1 = self.base_dir_path + dir_name1
            system.make_dir(dir_path1)
            dir_path1 += "/"
            self._to_clean(dir_path1)

            if self._is_supported("listdir"):
                assert (
                    dir_path1 in self._list_objects_names()
                ), 'Create directory, exists (without "/")'

                # Test: Listing empty directory
                assert (
                    len(tuple(system.list_objects(dir_path0))) == 0
                ), "List objects, empty directory"

        # Write a sample file
        file_name = "sample_1K.dat"
        file_path = self.base_dir_path + file_name
        self._to_clean(file_path)
        file_url = self.base_dir_url + file_name
        size = 1024
        content = _urandom(size)

        if self._is_supported("write"):
            with self._raw_io(file_path, mode="w", **self._system_parameters) as file:
                # Write content
                file.write(content)

        elif self._storage_mock:
            # Create pre-existing file
            self._storage_mock.put_object(
                self.locator, self.base_dir_name + file_name, content
            )

        # Estimate creation time
        create_time = _time()

        # Test: Check file header
        assert hasattr(
            system.head(path=file_path), "__getitem__"
        ), "Head file, header is mapping"

        # Test: Check file size
        try:
            assert system.getsize(file_path) == size, "Head file, size match"
        except ObjectUnsupportedOperation:
            # May not be supported on all files, if supported
            if self._is_supported("getsize"):
                raise

        # Test: Check file modification time
        try:
            file_time = system.getmtime(file_path)
            if self._is_supported("write"):
                assert file_time == _pytest.approx(
                    create_time, 2
                ), "Head file, modification time match"
        except ObjectUnsupportedOperation:
            # May not be supported on all files, if supported
            if self._is_supported("getmtime"):
                raise

        # Test: Check file creation time
        try:
            file_time = system.getctime(file_path)
            if self._is_supported("write"):
                assert file_time == _pytest.approx(
                    create_time, 2
                ), "Head file, creation time match"
        except ObjectUnsupportedOperation:
            # May not be supported on all files, if supported
            if self._is_supported("getctime"):
                raise

        # Test: Check path and URL handling
        with self._raw_io(file_path, **self._system_parameters) as file:
            assert file.name == file_path, "Open file, path match"

        with self._raw_io(file_url, **self._system_parameters) as file:
            assert file.name == file_url, "Open file, URL match"

        # Write some files
        files = set()
        files.add(file_path)
        for i in range(11):
            if i < 10:
                # Files in directory
                file_name = f"file{i}.dat"
                path = self.base_dir_path + file_name
                rel_path = self.base_dir_name + file_name
            else:
                # File in locator
                rel_path = self._get_id() + ".dat"
                path = f"{self.locator}/{rel_path}"

            files.add(path)
            self._to_clean(path)
            if self._is_supported("write"):
                with self._raw_io(path, mode="w", **self._system_parameters) as file:
                    file.flush()
            elif self._storage_mock:
                # Create pre-existing file
                self._storage_mock.put_object(self.locator, rel_path, b"")

        # Test: List objects
        if self._is_supported("listdir"):
            objects = tuple(system.list_objects(self.locator))
            objects_list = set(f"{self.locator}/{name}" for name, _ in objects)
            for file in files:
                assert file in objects_list, "List objects, file name match"
            for _, header in objects:
                assert hasattr(
                    header, "__getitem__"
                ), "List objects, file header is mapping"

            # Test: List objects, with limited output
            max_results = 5
            entries = len(tuple(system.list_objects(max_results=max_results)))
            assert entries == max_results, "List objects, Number of entries match"

            # Test: List objects, no objects found
            with _pytest.raises(ObjectNotFoundError):
                list(system.list_objects(self.base_dir_path + "dir_not_exists/"))

            # Test: List objects on locator root, no objects found
            with _pytest.raises(ObjectNotFoundError):
                list(system.list_objects(self.locator + "/dir_not_exists/"))

            # Test: List objects, locator not found
            with _pytest.raises(ObjectNotFoundError):
                list(system.list_objects(self._get_id()))

        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                list(system.list_objects(self.base_dir_path))

        # Test: copy
        copy_path = file_path + ".copy"
        self._to_clean(copy_path)
        if self._is_supported("copy"):
            system.copy(file_path, copy_path)
            assert system.getsize(copy_path) == size, "Copy file, size match"
        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                system.copy(file_path, copy_path)

        # Test: Normal file is not symlink
        assert not system.islink(file_path), "Symlink, file is not symlink"

        # Test: Symlink
        if self._is_supported("symlink") and self._storage_mock:
            print(0)
            link_path = self.base_dir_path + "symlink"
            self._to_clean(link_path)

            try:
                system.symlink(file_path, link_path)
            except ObjectUnsupportedOperation:
                # Some systems only support symlinks for reading
                # Put a symlink directly on the mock in this case
                self._storage_mock.put_symlink(
                    self.locator, link_path.split("/", 1)[1], file_path
                )

            assert system.islink(link_path)
            assert system.read_link(link_path) == file_path

            with _pytest.raises(ObjectNotASymlinkError):
                system.read_link(self.locator)

        # Test: Shared file
        if self._is_supported("shareable_url"):
            shareable_url = system.shareable_url(file_path, 60)
            assert shareable_url.startswith("http"), "Shareable URL"

            if self._storage_mock is None:
                # Test file download
                response = _requests.get(shareable_url)
                response.raise_for_status()
                assert response.content == content

            # Test share locator (If supported)
            try:
                assert system.shareable_url(self.locator, 60).startswith(
                    "http"
                ), "Shareable locator URL"
            except ObjectNotImplementedError:
                pass

            # Test share directory (If supported)
            try:
                assert system.shareable_url(dir_path0, 60).startswith(
                    "http"
                ), "Shareable directory URL"
            except ObjectNotImplementedError:
                pass

        else:
            # Test: Unsupported
            with _pytest.raises(ObjectNotImplementedError):
                system.shareable_url(file_path, 60)

        # Test: Remove file
        if self._is_supported("remove"):
            if self._is_supported("listdir"):
                assert (
                    file_path in self._list_objects_names()
                ), "Remove file, file exists"
            system.remove(file_path)
            if self._is_supported("listdir"):
                assert (
                    file_path not in self._list_objects_names()
                ), "Remove file, file not exists"
        else:
            # Test: Unsupported
            with _pytest.raises(ObjectUnsupportedOperation):
                system.remove(file_path)

    def _test_mock_only(self):
        """Tests that can only be performed on mocks."""
        file_name = "mocked.dat"

        # Create a file
        file_path = self.base_dir_path + file_name
        self._to_clean(file_path)
        content = _urandom(20)

        if self._is_supported("write"):
            with self._raw_io(file_path, mode="w", **self._system_parameters) as file:
                file.write(content)
                file.flush()
        elif self._storage_mock:
            # Create pre-existing file
            self._storage_mock.put_object(
                self.locator, self.base_dir_name + file_name, content
            )

        # Test: Read not block other exceptions
        with self._storage_mock.raise_server_error():
            with _pytest.raises(Exception):
                self._raw_io(file_path, **self._system_parameters).read(10)

    def _list_objects_names(self):
        """List objects names.

        Returns:
            set of str: objects names.
        """
        return set(
            f"{self.locator}/{name}"
            for name, _ in self._system.list_objects(self.locator)
        )


def test_user_storage(storage_test_kwargs):
    """Test specified storage.

    Test cases are automatically generated base on user configuration, see
    "tests.conftest.pytest_generate_tests".

    Args:
        storage_test_kwargs (dict): Storage test keyword arguments.
    """
    # Get list of unsupported operations
    from importlib import import_module

    module = import_module(
        f"tests.test_storage_{storage_test_kwargs['storage_info']['storage']}"
    )
    try:
        unsupported_operations = module.UNSUPPORTED_OPERATIONS
    except AttributeError:
        unsupported_operations = None

    # Run tests
    with StorageTester(
        unsupported_operations=unsupported_operations, **storage_test_kwargs
    ) as tester:
        tester.test_common()


def test_mock_storage():
    """Run test sequence on the mocked storage."""
    from tests_storage_package.mock import MockSystem, MockBufferedIO, MockRawIO

    # Init mocked system
    system = MockSystem()

    # Tests
    with StorageTester(system, MockRawIO, MockBufferedIO, system.client) as tester:
        # Common tests
        tester.test_common()
