django/tests/file_storage/tests.py

import os
import shutil
import sys
import tempfile
import threading
import time
import unittest
from datetime import datetime, timedelta
from io import StringIO
from pathlib import Path
from urllib.request import urlopen

from django.core.cache import cache
from django.core.exceptions import SuspiciousFileOperation
from django.core.files.base import ContentFile, File
from django.core.files.storage import (
    FileSystemStorage, Storage as BaseStorage, default_storage,
    get_storage_class,
)
from django.core.files.uploadedfile import (
    InMemoryUploadedFile, SimpleUploadedFile, TemporaryUploadedFile,
)
from django.db.models import FileField
from django.db.models.fields.files import FileDescriptor
from django.test import (
    LiveServerTestCase, SimpleTestCase, TestCase, override_settings,
)
from django.test.utils import requires_tz_support
from django.urls import NoReverseMatch, reverse_lazy
from django.utils import timezone
from django.utils._os import symlinks_supported

from .models import (
    Storage, callable_storage, temp_storage, temp_storage_location,
)

FILE_SUFFIX_REGEX = '[A-Za-z0-9]{7}'


class GetStorageClassTests(SimpleTestCase):

    def test_get_filesystem_storage(self):
        """
        get_storage_class returns the class for a storage backend name/path.
        """
        self.assertEqual(
            get_storage_class('django.core.files.storage.FileSystemStorage'),
            FileSystemStorage)

    def test_get_invalid_storage_module(self):
        """
        get_storage_class raises an error if the requested import don't exist.
        """
        with self.assertRaisesMessage(ImportError, "No module named 'storage'"):
            get_storage_class('storage.NonexistentStorage')

    def test_get_nonexistent_storage_class(self):
        """
        get_storage_class raises an error if the requested class don't exist.
        """
        with self.assertRaises(ImportError):
            get_storage_class('django.core.files.storage.NonexistentStorage')

    def test_get_nonexistent_storage_module(self):
        """
        get_storage_class raises an error if the requested module don't exist.
        """
        with self.assertRaisesMessage(ImportError, "No module named 'django.core.files.nonexistent_storage'"):
            get_storage_class('django.core.files.nonexistent_storage.NonexistentStorage')


class FileSystemStorageTests(unittest.TestCase):

    def test_deconstruction(self):
        path, args, kwargs = temp_storage.deconstruct()
        self.assertEqual(path, "django.core.files.storage.FileSystemStorage")
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'location': temp_storage_location})

        kwargs_orig = {
            'location': temp_storage_location,
            'base_url': 'http://myfiles.example.com/'
        }
        storage = FileSystemStorage(**kwargs_orig)
        path, args, kwargs = storage.deconstruct()
        self.assertEqual(kwargs, kwargs_orig)

    def test_lazy_base_url_init(self):
        """
        FileSystemStorage.__init__() shouldn't evaluate base_url.
        """
        storage = FileSystemStorage(base_url=reverse_lazy('app:url'))
        with self.assertRaises(NoReverseMatch):
            storage.url(storage.base_url)


class FileStorageTests(SimpleTestCase):
    storage_class = FileSystemStorage

    def setUp(self):
        self.temp_dir = tempfile.mkdtemp()
        self.storage = self.storage_class(location=self.temp_dir, base_url='/test_media_url/')
        # Set up a second temporary directory which is ensured to have a mixed
        # case name.
        self.temp_dir2 = tempfile.mkdtemp(suffix='aBc')

    def tearDown(self):
        shutil.rmtree(self.temp_dir)
        shutil.rmtree(self.temp_dir2)

    def test_empty_location(self):
        """
        Makes sure an exception is raised if the location is empty
        """
        storage = self.storage_class(location='')
        self.assertEqual(storage.base_location, '')
        self.assertEqual(storage.location, os.getcwd())

    def test_file_access_options(self):
        """
        Standard file access options are available, and work as expected.
        """
        self.assertFalse(self.storage.exists('storage_test'))
        f = self.storage.open('storage_test', 'w')
        f.write('storage contents')
        f.close()
        self.assertTrue(self.storage.exists('storage_test'))

        f = self.storage.open('storage_test', 'r')
        self.assertEqual(f.read(), 'storage contents')
        f.close()

        self.storage.delete('storage_test')
        self.assertFalse(self.storage.exists('storage_test'))

    def _test_file_time_getter(self, getter):
        # Check for correct behavior under both USE_TZ=True and USE_TZ=False.
        # The tests are similar since they both set up a situation where the
        # system time zone, Django's TIME_ZONE, and UTC are distinct.
        self._test_file_time_getter_tz_handling_on(getter)
        self._test_file_time_getter_tz_handling_off(getter)

    @override_settings(USE_TZ=True, TIME_ZONE='Africa/Algiers')
    def _test_file_time_getter_tz_handling_on(self, getter):
        # Django's TZ (and hence the system TZ) is set to Africa/Algiers which
        # is UTC+1 and has no DST change. We can set the Django TZ to something
        # else so that UTC, Django's TIME_ZONE, and the system timezone are all
        # different.
        now_in_algiers = timezone.make_aware(datetime.now())

        with timezone.override(timezone.get_fixed_timezone(-300)):
            # At this point the system TZ is +1 and the Django TZ
            # is -5. The following will be aware in UTC.
            now = timezone.now()
            self.assertFalse(self.storage.exists('test.file.tz.on'))

            f = ContentFile('custom contents')
            f_name = self.storage.save('test.file.tz.on', f)
            self.addCleanup(self.storage.delete, f_name)
            dt = getter(f_name)
            # dt should be aware, in UTC
            self.assertTrue(timezone.is_aware(dt))
            self.assertEqual(now.tzname(), dt.tzname())

            # The three timezones are indeed distinct.
            naive_now = datetime.now()
            algiers_offset = now_in_algiers.tzinfo.utcoffset(naive_now)
            django_offset = timezone.get_current_timezone().utcoffset(naive_now)
            utc_offset = timezone.utc.utcoffset(naive_now)
            self.assertGreater(algiers_offset, utc_offset)
            self.assertLess(django_offset, utc_offset)

            # dt and now should be the same effective time.
            self.assertLess(abs(dt - now), timedelta(seconds=2))

    @override_settings(USE_TZ=False, TIME_ZONE='Africa/Algiers')
    def _test_file_time_getter_tz_handling_off(self, getter):
        # Django's TZ (and hence the system TZ) is set to Africa/Algiers which
        # is UTC+1 and has no DST change. We can set the Django TZ to something
        # else so that UTC, Django's TIME_ZONE, and the system timezone are all
        # different.
        now_in_algiers = timezone.make_aware(datetime.now())

        with timezone.override(timezone.get_fixed_timezone(-300)):
            # At this point the system TZ is +1 and the Django TZ
            # is -5.
            self.assertFalse(self.storage.exists('test.file.tz.off'))

            f = ContentFile('custom contents')
            f_name = self.storage.save('test.file.tz.off', f)
            self.addCleanup(self.storage.delete, f_name)
            dt = getter(f_name)
            # dt should be naive, in system (+1) TZ
            self.assertTrue(timezone.is_naive(dt))

            # The three timezones are indeed distinct.
            naive_now = datetime.now()
            algiers_offset = now_in_algiers.tzinfo.utcoffset(naive_now)
            django_offset = timezone.get_current_timezone().utcoffset(naive_now)
            utc_offset = timezone.utc.utcoffset(naive_now)
            self.assertGreater(algiers_offset, utc_offset)
            self.assertLess(django_offset, utc_offset)

            # dt and naive_now should be the same effective time.
            self.assertLess(abs(dt - naive_now), timedelta(seconds=2))
            # If we convert dt to an aware object using the Algiers
            # timezone then it should be the same effective time to
            # now_in_algiers.
            _dt = timezone.make_aware(dt, now_in_algiers.tzinfo)
            self.assertLess(abs(_dt - now_in_algiers), timedelta(seconds=2))

    def test_file_get_accessed_time(self):
        """
        File storage returns a Datetime object for the last accessed time of
        a file.
        """
        self.assertFalse(self.storage.exists('test.file'))

        f = ContentFile('custom contents')
        f_name = self.storage.save('test.file', f)
        self.addCleanup(self.storage.delete, f_name)
        atime = self.storage.get_accessed_time(f_name)

        self.assertEqual(atime, datetime.fromtimestamp(os.path.getatime(self.storage.path(f_name))))
        self.assertLess(timezone.now() - self.storage.get_accessed_time(f_name), timedelta(seconds=2))

    @requires_tz_support
    def test_file_get_accessed_time_timezone(self):
        self._test_file_time_getter(self.storage.get_accessed_time)

    def test_file_get_created_time(self):
        """
        File storage returns a datetime for the creation time of a file.
        """
        self.assertFalse(self.storage.exists('test.file'))

        f = ContentFile('custom contents')
        f_name = self.storage.save('test.file', f)
        self.addCleanup(self.storage.delete, f_name)
        ctime = self.storage.get_created_time(f_name)

        self.assertEqual(ctime, datetime.fromtimestamp(os.path.getctime(self.storage.path(f_name))))
        self.assertLess(timezone.now() - self.storage.get_created_time(f_name), timedelta(seconds=2))

    @requires_tz_support
    def test_file_get_created_time_timezone(self):
        self._test_file_time_getter(self.storage.get_created_time)

    def test_file_get_modified_time(self):
        """
        File storage returns a datetime for the last modified time of a file.
        """
        self.assertFalse(self.storage.exists('test.file'))

        f = ContentFile('custom contents')
        f_name = self.storage.save('test.file', f)
        self.addCleanup(self.storage.delete, f_name)
        mtime = self.storage.get_modified_time(f_name)

        self.assertEqual(mtime, datetime.fromtimestamp(os.path.getmtime(self.storage.path(f_name))))
        self.assertLess(timezone.now() - self.storage.get_modified_time(f_name), timedelta(seconds=2))

    @requires_tz_support
    def test_file_get_modified_time_timezone(self):
        self._test_file_time_getter(self.storage.get_modified_time)

    def test_file_save_without_name(self):
        """
        File storage extracts the filename from the content object if no
        name is given explicitly.
        """
        self.assertFalse(self.storage.exists('test.file'))

        f = ContentFile('custom contents')
        f.name = 'test.file'

        storage_f_name = self.storage.save(None, f)

        self.assertEqual(storage_f_name, f.name)

        self.assertTrue(os.path.exists(os.path.join(self.temp_dir, f.name)))

        self.storage.delete(storage_f_name)

    def test_file_save_with_path(self):
        """
        Saving a pathname should create intermediate directories as necessary.
        """
        self.assertFalse(self.storage.exists('path/to'))
        self.storage.save('path/to/test.file', ContentFile('file saved with path'))

        self.assertTrue(self.storage.exists('path/to'))
        with self.storage.open('path/to/test.file') as f:
            self.assertEqual(f.read(), b'file saved with path')

        self.assertTrue(os.path.exists(
            os.path.join(self.temp_dir, 'path', 'to', 'test.file')))

        self.storage.delete('path/to/test.file')

    @unittest.skipUnless(symlinks_supported(), 'Must be able to symlink to run this test.')
    def test_file_save_broken_symlink(self):
        """A new path is created on save when a broken symlink is supplied."""
        nonexistent_file_path = os.path.join(self.temp_dir, 'nonexistent.txt')
        broken_symlink_path = os.path.join(self.temp_dir, 'symlink.txt')
        os.symlink(nonexistent_file_path, broken_symlink_path)
        f = ContentFile('some content')
        f_name = self.storage.save(broken_symlink_path, f)
        self.assertIs(os.path.exists(os.path.join(self.temp_dir, f_name)), True)

    def test_save_doesnt_close(self):
        with TemporaryUploadedFile('test', 'text/plain', 1, 'utf8') as file:
            file.write(b'1')
            file.seek(0)
            self.assertFalse(file.closed)
            self.storage.save('path/to/test.file', file)
            self.assertFalse(file.closed)
            self.assertFalse(file.file.closed)

        file = InMemoryUploadedFile(StringIO('1'), '', 'test', 'text/plain', 1, 'utf8')
        with file:
            self.assertFalse(file.closed)
            self.storage.save('path/to/test.file', file)
            self.assertFalse(file.closed)
            self.assertFalse(file.file.closed)

    def test_file_path(self):
        """
        File storage returns the full path of a file
        """
        self.assertFalse(self.storage.exists('test.file'))

        f = ContentFile('custom contents')
        f_name = self.storage.save('test.file', f)

        self.assertEqual(self.storage.path(f_name), os.path.join(self.temp_dir, f_name))

        self.storage.delete(f_name)

    def test_file_url(self):
        """
        File storage returns a url to access a given file from the Web.
        """
        self.assertEqual(self.storage.url('test.file'), self.storage.base_url + 'test.file')

        # should encode special chars except ~!*()'
        # like encodeURIComponent() JavaScript function do
        self.assertEqual(
            self.storage.url(r"~!*()'@#$%^&*abc`+ =.file"),
            "/test_media_url/~!*()'%40%23%24%25%5E%26*abc%60%2B%20%3D.file"
        )
        self.assertEqual(self.storage.url("ab\0c"), "/test_media_url/ab%00c")

        # should translate os path separator(s) to the url path separator
        self.assertEqual(self.storage.url("""a/b\\c.file"""), "/test_media_url/a/b/c.file")

        # #25905: remove leading slashes from file names to prevent unsafe url output
        self.assertEqual(self.storage.url("/evil.com"), "/test_media_url/evil.com")
        self.assertEqual(self.storage.url(r"\evil.com"), "/test_media_url/evil.com")
        self.assertEqual(self.storage.url("///evil.com"), "/test_media_url/evil.com")
        self.assertEqual(self.storage.url(r"\\\evil.com"), "/test_media_url/evil.com")

        self.assertEqual(self.storage.url(None), "/test_media_url/")

    def test_base_url(self):
        """
        File storage returns a url even when its base_url is unset or modified.
        """
        self.storage.base_url = None
        with self.assertRaises(ValueError):
            self.storage.url('test.file')

        # #22717: missing ending slash in base_url should be auto-corrected
        storage = self.storage_class(location=self.temp_dir, base_url='/no_ending_slash')
        self.assertEqual(
            storage.url('test.file'),
            '%s%s' % (storage.base_url, 'test.file')
        )

    def test_listdir(self):
        """
        File storage returns a tuple containing directories and files.
        """
        self.assertFalse(self.storage.exists('storage_test_1'))
        self.assertFalse(self.storage.exists('storage_test_2'))
        self.assertFalse(self.storage.exists('storage_dir_1'))

        self.storage.save('storage_test_1', ContentFile('custom content'))
        self.storage.save('storage_test_2', ContentFile('custom content'))
        os.mkdir(os.path.join(self.temp_dir, 'storage_dir_1'))

        self.addCleanup(self.storage.delete, 'storage_test_1')
        self.addCleanup(self.storage.delete, 'storage_test_2')

        for directory in ('', Path('')):
            with self.subTest(directory=directory):
                dirs, files = self.storage.listdir(directory)
                self.assertEqual(set(dirs), {'storage_dir_1'})
                self.assertEqual(set(files), {'storage_test_1', 'storage_test_2'})

    def test_file_storage_prevents_directory_traversal(self):
        """
        File storage prevents directory traversal (files can only be accessed if
        they're below the storage location).
        """
        with self.assertRaises(SuspiciousFileOperation):
            self.storage.exists('..')
        with self.assertRaises(SuspiciousFileOperation):
            self.storage.exists('/etc/passwd')

    def test_file_storage_preserves_filename_case(self):
        """The storage backend should preserve case of filenames."""
        # Create a storage backend associated with the mixed case name
        # directory.
        other_temp_storage = self.storage_class(location=self.temp_dir2)
        # Ask that storage backend to store a file with a mixed case filename.
        mixed_case = 'CaSe_SeNsItIvE'
        file = other_temp_storage.open(mixed_case, 'w')
        file.write('storage contents')
        file.close()
        self.assertEqual(os.path.join(self.temp_dir2, mixed_case), other_temp_storage.path(mixed_case))
        other_temp_storage.delete(mixed_case)

    def test_makedirs_race_handling(self):
        """
        File storage should be robust against directory creation race conditions.
        """
        real_makedirs = os.makedirs

        # Monkey-patch os.makedirs, to simulate a normal call, a raced call,
        # and an error.
        def fake_makedirs(path, mode=0o777, exist_ok=False):
            if path == os.path.join(self.temp_dir, 'normal'):
                real_makedirs(path, mode, exist_ok)
            elif path == os.path.join(self.temp_dir, 'raced'):
                real_makedirs(path, mode, exist_ok)
                if not exist_ok:
                    raise FileExistsError()
            elif path == os.path.join(self.temp_dir, 'error'):
                raise PermissionError()
            else:
                self.fail('unexpected argument %r' % path)

        try:
            os.makedirs = fake_makedirs

            self.storage.save('normal/test.file', ContentFile('saved normally'))
            with self.storage.open('normal/test.file') as f:
                self.assertEqual(f.read(), b'saved normally')

            self.storage.save('raced/test.file', ContentFile('saved with race'))
            with self.storage.open('raced/test.file') as f:
                self.assertEqual(f.read(), b'saved with race')

            # Exceptions aside from FileExistsError are raised.
            with self.assertRaises(PermissionError):
                self.storage.save('error/test.file', ContentFile('not saved'))
        finally:
            os.makedirs = real_makedirs

    def test_remove_race_handling(self):
        """
        File storage should be robust against file removal race conditions.
        """
        real_remove = os.remove

        # Monkey-patch os.remove, to simulate a normal call, a raced call,
        # and an error.
        def fake_remove(path):
            if path == os.path.join(self.temp_dir, 'normal.file'):
                real_remove(path)
            elif path == os.path.join(self.temp_dir, 'raced.file'):
                real_remove(path)
                raise FileNotFoundError()
            elif path == os.path.join(self.temp_dir, 'error.file'):
                raise PermissionError()
            else:
                self.fail('unexpected argument %r' % path)

        try:
            os.remove = fake_remove

            self.storage.save('normal.file', ContentFile('delete normally'))
            self.storage.delete('normal.file')
            self.assertFalse(self.storage.exists('normal.file'))

            self.storage.save('raced.file', ContentFile('delete with race'))
            self.storage.delete('raced.file')
            self.assertFalse(self.storage.exists('normal.file'))

            # Exceptions aside from FileNotFoundError are raised.
            self.storage.save('error.file', ContentFile('delete with error'))
            with self.assertRaises(PermissionError):
                self.storage.delete('error.file')
        finally:
            os.remove = real_remove

    def test_file_chunks_error(self):
        """
        Test behavior when file.chunks() is raising an error
        """
        f1 = ContentFile('chunks fails')

        def failing_chunks():
            raise OSError
        f1.chunks = failing_chunks
        with self.assertRaises(OSError):
            self.storage.save('error.file', f1)

    def test_delete_no_name(self):
        """
        Calling delete with an empty name should not try to remove the base
        storage directory, but fail loudly (#20660).
        """
        msg = 'The name must be given to delete().'
        with self.assertRaisesMessage(ValueError, msg):
            self.storage.delete(None)
        with self.assertRaisesMessage(ValueError, msg):
            self.storage.delete('')

    def test_delete_deletes_directories(self):
        tmp_dir = tempfile.mkdtemp(dir=self.storage.location)
        self.storage.delete(tmp_dir)
        self.assertFalse(os.path.exists(tmp_dir))

    @override_settings(
        MEDIA_ROOT='media_root',
        MEDIA_URL='media_url/',
        FILE_UPLOAD_PERMISSIONS=0o777,
        FILE_UPLOAD_DIRECTORY_PERMISSIONS=0o777,
    )
    def test_setting_changed(self):
        """
        Properties using settings values as defaults should be updated on
        referenced settings change while specified values should be unchanged.
        """
        storage = self.storage_class(
            location='explicit_location',
            base_url='explicit_base_url/',
            file_permissions_mode=0o666,
            directory_permissions_mode=0o666,
        )
        defaults_storage = self.storage_class()
        settings = {
            'MEDIA_ROOT': 'overridden_media_root',
            'MEDIA_URL': '/overridden_media_url/',
            'FILE_UPLOAD_PERMISSIONS': 0o333,
            'FILE_UPLOAD_DIRECTORY_PERMISSIONS': 0o333,
        }
        with self.settings(**settings):
            self.assertEqual(storage.base_location, 'explicit_location')
            self.assertIn('explicit_location', storage.location)
            self.assertEqual(storage.base_url, 'explicit_base_url/')
            self.assertEqual(storage.file_permissions_mode, 0o666)
            self.assertEqual(storage.directory_permissions_mode, 0o666)
            self.assertEqual(defaults_storage.base_location, settings['MEDIA_ROOT'])
            self.assertIn(settings['MEDIA_ROOT'], defaults_storage.location)
            self.assertEqual(defaults_storage.base_url, settings['MEDIA_URL'])
            self.assertEqual(defaults_storage.file_permissions_mode, settings['FILE_UPLOAD_PERMISSIONS'])
            self.assertEqual(
                defaults_storage.directory_permissions_mode, settings['FILE_UPLOAD_DIRECTORY_PERMISSIONS']
            )

    def test_file_methods_pathlib_path(self):
        p = Path('test.file')
        self.assertFalse(self.storage.exists(p))
        f = ContentFile('custom contents')
        f_name = self.storage.save(p, f)
        # Storage basic methods.
        self.assertEqual(self.storage.path(p), os.path.join(self.temp_dir, p))
        self.assertEqual(self.storage.size(p), 15)
        self.assertEqual(self.storage.url(p), self.storage.base_url + f_name)
        with self.storage.open(p) as f:
            self.assertEqual(f.read(), b'custom contents')
        self.addCleanup(self.storage.delete, p)


class CustomStorage(FileSystemStorage):
    def get_available_name(self, name, max_length=None):
        """
        Append numbers to duplicate files rather than underscores, like Trac.
        """
        basename, *ext = os.path.splitext(name)
        number = 2
        while self.exists(name):
            name = ''.join([basename, '.', str(number)] + ext)
            number += 1

        return name


class CustomStorageTests(FileStorageTests):
    storage_class = CustomStorage

    def test_custom_get_available_name(self):
        first = self.storage.save('custom_storage', ContentFile('custom contents'))
        self.assertEqual(first, 'custom_storage')
        second = self.storage.save('custom_storage', ContentFile('more contents'))
        self.assertEqual(second, 'custom_storage.2')
        self.storage.delete(first)
        self.storage.delete(second)


class OverwritingStorage(FileSystemStorage):
    """
    Overwrite existing files instead of appending a suffix to generate an
    unused name.
    """
    # Mask out O_EXCL so os.open() doesn't raise OSError if the file exists.
    OS_OPEN_FLAGS = FileSystemStorage.OS_OPEN_FLAGS & ~os.O_EXCL

    def get_available_name(self, name, max_length=None):
        """Override the effort to find an used name."""
        return name


class OverwritingStorageTests(FileStorageTests):
    storage_class = OverwritingStorage

    def test_save_overwrite_behavior(self):
        """Saving to same file name twice overwrites the first file."""
        name = 'test.file'
        self.assertFalse(self.storage.exists(name))
        content_1 = b'content one'
        content_2 = b'second content'
        f_1 = ContentFile(content_1)
        f_2 = ContentFile(content_2)
        stored_name_1 = self.storage.save(name, f_1)
        try:
            self.assertEqual(stored_name_1, name)
            self.assertTrue(self.storage.exists(name))
            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
            with self.storage.open(name) as fp:
                self.assertEqual(fp.read(), content_1)
            stored_name_2 = self.storage.save(name, f_2)
            self.assertEqual(stored_name_2, name)
            self.assertTrue(self.storage.exists(name))
            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
            with self.storage.open(name) as fp:
                self.assertEqual(fp.read(), content_2)
        finally:
            self.storage.delete(name)


class DiscardingFalseContentStorage(FileSystemStorage):
    def _save(self, name, content):
        if content:
            return super()._save(name, content)
        return ''


class DiscardingFalseContentStorageTests(FileStorageTests):
    storage_class = DiscardingFalseContentStorage

    def test_custom_storage_discarding_empty_content(self):
        """
        When Storage.save() wraps a file-like object in File, it should include
        the name argument so that bool(file) evaluates to True (#26495).
        """
        output = StringIO('content')
        self.storage.save('tests/stringio', output)
        self.assertTrue(self.storage.exists('tests/stringio'))

        with self.storage.open('tests/stringio') as f:
            self.assertEqual(f.read(), b'content')


class FileFieldStorageTests(TestCase):
    def tearDown(self):
        shutil.rmtree(temp_storage_location)

    def _storage_max_filename_length(self, storage):
        """
        Query filesystem for maximum filename length (e.g. AUFS has 242).
        """
        dir_to_test = storage.location
        while not os.path.exists(dir_to_test):
            dir_to_test = os.path.dirname(dir_to_test)
        try:
            return os.pathconf(dir_to_test, 'PC_NAME_MAX')
        except Exception:
            return 255  # Should be safe on most backends

    def test_files(self):
        self.assertIsInstance(Storage.normal, FileDescriptor)

        # An object without a file has limited functionality.
        obj1 = Storage()
        self.assertEqual(obj1.normal.name, "")
        with self.assertRaises(ValueError):
            obj1.normal.size

        # Saving a file enables full functionality.
        obj1.normal.save("django_test.txt", ContentFile("content"))
        self.assertEqual(obj1.normal.name, "tests/django_test.txt")
        self.assertEqual(obj1.normal.size, 7)
        self.assertEqual(obj1.normal.read(), b"content")
        obj1.normal.close()

        # File objects can be assigned to FileField attributes, but shouldn't
        # get committed until the model it's attached to is saved.
        obj1.normal = SimpleUploadedFile("assignment.txt", b"content")
        dirs, files = temp_storage.listdir("tests")
        self.assertEqual(dirs, [])
        self.assertNotIn("assignment.txt", files)

        obj1.save()
        dirs, files = temp_storage.listdir("tests")
        self.assertEqual(sorted(files), ["assignment.txt", "django_test.txt"])

        # Save another file with the same name.
        obj2 = Storage()
        obj2.normal.save("django_test.txt", ContentFile("more content"))
        obj2_name = obj2.normal.name
        self.assertRegex(obj2_name, "tests/django_test_%s.txt" % FILE_SUFFIX_REGEX)
        self.assertEqual(obj2.normal.size, 12)
        obj2.normal.close()

        # Deleting an object does not delete the file it uses.
        obj2.delete()
        obj2.normal.save("django_test.txt", ContentFile("more content"))
        self.assertNotEqual(obj2_name, obj2.normal.name)
        self.assertRegex(obj2.normal.name, "tests/django_test_%s.txt" % FILE_SUFFIX_REGEX)
        obj2.normal.close()

    def test_filefield_read(self):
        # Files can be read in a little at a time, if necessary.
        obj = Storage.objects.create(
            normal=SimpleUploadedFile("assignment.txt", b"content"))
        obj.normal.open()
        self.assertEqual(obj.normal.read(3), b"con")
        self.assertEqual(obj.normal.read(), b"tent")
        self.assertEqual(list(obj.normal.chunks(chunk_size=2)), [b"co", b"nt", b"en", b"t"])
        obj.normal.close()

    def test_filefield_write(self):
        # Files can be written to.
        obj = Storage.objects.create(normal=SimpleUploadedFile('rewritten.txt', b'content'))
        with obj.normal as normal:
            normal.open('wb')
            normal.write(b'updated')
        obj.refresh_from_db()
        self.assertEqual(obj.normal.read(), b'updated')
        obj.normal.close()

    def test_filefield_reopen(self):
        obj = Storage.objects.create(normal=SimpleUploadedFile('reopen.txt', b'content'))
        with obj.normal as normal:
            normal.open()
        obj.normal.open()
        obj.normal.file.seek(0)
        obj.normal.close()

    def test_duplicate_filename(self):
        # Multiple files with the same name get _(7 random chars) appended to them.
        objs = [Storage() for i in range(2)]
        for o in objs:
            o.normal.save("multiple_files.txt", ContentFile("Same Content"))
        try:
            names = [o.normal.name for o in objs]
            self.assertEqual(names[0], "tests/multiple_files.txt")
            self.assertRegex(names[1], "tests/multiple_files_%s.txt" % FILE_SUFFIX_REGEX)
        finally:
            for o in objs:
                o.delete()

    def test_file_truncation(self):
        # Given the max_length is limited, when multiple files get uploaded
        # under the same name, then the filename get truncated in order to fit
        # in _(7 random chars). When most of the max_length is taken by
        # dirname + extension and there are not enough  characters in the
        # filename to truncate, an exception should be raised.
        objs = [Storage() for i in range(2)]
        filename = 'filename.ext'

        for o in objs:
            o.limited_length.save(filename, ContentFile('Same Content'))
        try:
            # Testing truncation.
            names = [o.limited_length.name for o in objs]
            self.assertEqual(names[0], 'tests/%s' % filename)
            self.assertRegex(names[1], 'tests/fi_%s.ext' % FILE_SUFFIX_REGEX)

            # Testing exception is raised when filename is too short to truncate.
            filename = 'short.longext'
            objs[0].limited_length.save(filename, ContentFile('Same Content'))
            with self.assertRaisesMessage(SuspiciousFileOperation, 'Storage can not find an available filename'):
                objs[1].limited_length.save(*(filename, ContentFile('Same Content')))
        finally:
            for o in objs:
                o.delete()

    @unittest.skipIf(
        sys.platform == 'win32',
        "Windows supports at most 260 characters in a path.",
    )
    def test_extended_length_storage(self):
        # Testing FileField with max_length > 255. Most systems have filename
        # length limitation of 255. Path takes extra chars.
        filename = (self._storage_max_filename_length(temp_storage) - 4) * 'a'  # 4 chars for extension.
        obj = Storage()
        obj.extended_length.save('%s.txt' % filename, ContentFile('Same Content'))
        self.assertEqual(obj.extended_length.name, 'tests/%s.txt' % filename)
        self.assertEqual(obj.extended_length.read(), b'Same Content')
        obj.extended_length.close()

    def test_filefield_default(self):
        # Default values allow an object to access a single file.
        temp_storage.save('tests/default.txt', ContentFile('default content'))
        obj = Storage.objects.create()
        self.assertEqual(obj.default.name, "tests/default.txt")
        self.assertEqual(obj.default.read(), b"default content")
        obj.default.close()

        # But it shouldn't be deleted, even if there are no more objects using
        # it.
        obj.delete()
        obj = Storage()
        self.assertEqual(obj.default.read(), b"default content")
        obj.default.close()

    def test_empty_upload_to(self):
        # upload_to can be empty, meaning it does not use subdirectory.
        obj = Storage()
        obj.empty.save('django_test.txt', ContentFile('more content'))
        self.assertEqual(obj.empty.name, "django_test.txt")
        self.assertEqual(obj.empty.read(), b"more content")
        obj.empty.close()

    def test_pathlib_upload_to(self):
        obj = Storage()
        obj.pathlib_callable.save('some_file1.txt', ContentFile('some content'))
        self.assertEqual(obj.pathlib_callable.name, 'bar/some_file1.txt')
        obj.pathlib_direct.save('some_file2.txt', ContentFile('some content'))
        self.assertEqual(obj.pathlib_direct.name, 'bar/some_file2.txt')
        obj.random.close()

    def test_random_upload_to(self):
        # Verify the fix for #5655, making sure the directory is only
        # determined once.
        obj = Storage()
        obj.random.save("random_file", ContentFile("random content"))
        self.assertTrue(obj.random.name.endswith("/random_file"))
        obj.random.close()

    def test_custom_valid_name_callable_upload_to(self):
        """
        Storage.get_valid_name() should be called when upload_to is a callable.
        """
        obj = Storage()
        obj.custom_valid_name.save("random_file", ContentFile("random content"))
        # CustomValidNameStorage.get_valid_name() appends '_valid' to the name
        self.assertTrue(obj.custom_valid_name.name.endswith("/random_file_valid"))
        obj.custom_valid_name.close()

    def test_filefield_pickling(self):
        # Push an object into the cache to make sure it pickles properly
        obj = Storage()
        obj.normal.save("django_test.txt", ContentFile("more content"))
        obj.normal.close()
        cache.set("obj", obj)
        self.assertEqual(cache.get("obj").normal.name, "tests/django_test.txt")

    def test_file_object(self):
        # Create sample file
        temp_storage.save('tests/example.txt', ContentFile('some content'))

        # Load it as Python file object
        with open(temp_storage.path('tests/example.txt')) as file_obj:
            # Save it using storage and read its content
            temp_storage.save('tests/file_obj', file_obj)
        self.assertTrue(temp_storage.exists('tests/file_obj'))
        with temp_storage.open('tests/file_obj') as f:
            self.assertEqual(f.read(), b'some content')

    def test_stringio(self):
        # Test passing StringIO instance as content argument to save
        output = StringIO()
        output.write('content')
        output.seek(0)

        # Save it and read written file
        temp_storage.save('tests/stringio', output)
        self.assertTrue(temp_storage.exists('tests/stringio'))
        with temp_storage.open('tests/stringio') as f:
            self.assertEqual(f.read(), b'content')


class FieldCallableFileStorageTests(SimpleTestCase):
    def setUp(self):
        self.temp_storage_location = tempfile.mkdtemp(suffix='filefield_callable_storage')

    def tearDown(self):
        shutil.rmtree(self.temp_storage_location)

    def test_callable_base_class_error_raises(self):
        class NotStorage:
            pass
        msg = 'FileField.storage must be a subclass/instance of django.core.files.storage.Storage'
        for invalid_type in (NotStorage, str, list, set, tuple):
            with self.subTest(invalid_type=invalid_type):
                with self.assertRaisesMessage(TypeError, msg):
                    FileField(storage=invalid_type)

    def test_file_field_storage_none_uses_default_storage(self):
        self.assertEqual(FileField().storage, default_storage)

    def test_callable_function_storage_file_field(self):
        storage = FileSystemStorage(location=self.temp_storage_location)

        def get_storage():
            return storage

        obj = FileField(storage=get_storage)
        self.assertEqual(obj.storage, storage)
        self.assertEqual(obj.storage.location, storage.location)

    def test_callable_class_storage_file_field(self):
        class GetStorage(FileSystemStorage):
            pass

        obj = FileField(storage=GetStorage)
        self.assertIsInstance(obj.storage, BaseStorage)

    def test_callable_storage_file_field_in_model(self):
        obj = Storage()
        self.assertEqual(obj.storage_callable.storage, temp_storage)
        self.assertEqual(obj.storage_callable.storage.location, temp_storage_location)
        self.assertIsInstance(obj.storage_callable_class.storage, BaseStorage)

    def test_deconstruction(self):
        """
        Deconstructing gives the original callable, not the evaluated value.
        """
        obj = Storage()
        *_, kwargs = obj._meta.get_field('storage_callable').deconstruct()
        storage = kwargs['storage']
        self.assertIs(storage, callable_storage)


# Tests for a race condition on file saving (#4948).
# This is written in such a way that it'll always pass on platforms
# without threading.

class SlowFile(ContentFile):
    def chunks(self):
        time.sleep(1)
        return super().chunks()


class FileSaveRaceConditionTest(SimpleTestCase):
    def setUp(self):
        self.storage_dir = tempfile.mkdtemp()
        self.storage = FileSystemStorage(self.storage_dir)
        self.thread = threading.Thread(target=self.save_file, args=['conflict'])

    def tearDown(self):
        shutil.rmtree(self.storage_dir)

    def save_file(self, name):
        name = self.storage.save(name, SlowFile(b"Data"))

    def test_race_condition(self):
        self.thread.start()
        self.save_file('conflict')
        self.thread.join()
        files = sorted(os.listdir(self.storage_dir))
        self.assertEqual(files[0], 'conflict')
        self.assertRegex(files[1], 'conflict_%s' % FILE_SUFFIX_REGEX)


@unittest.skipIf(sys.platform == 'win32', "Windows only partially supports umasks and chmod.")
class FileStoragePermissions(unittest.TestCase):
    def setUp(self):
        self.umask = 0o027
        self.old_umask = os.umask(self.umask)
        self.storage_dir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self.storage_dir)
        os.umask(self.old_umask)

    @override_settings(FILE_UPLOAD_PERMISSIONS=0o654)
    def test_file_upload_permissions(self):
        self.storage = FileSystemStorage(self.storage_dir)
        name = self.storage.save("the_file", ContentFile("data"))
        actual_mode = os.stat(self.storage.path(name))[0] & 0o777
        self.assertEqual(actual_mode, 0o654)

    @override_settings(FILE_UPLOAD_PERMISSIONS=None)
    def test_file_upload_default_permissions(self):
        self.storage = FileSystemStorage(self.storage_dir)
        fname = self.storage.save("some_file", ContentFile("data"))
        mode = os.stat(self.storage.path(fname))[0] & 0o777
        self.assertEqual(mode, 0o666 & ~self.umask)

    @override_settings(FILE_UPLOAD_DIRECTORY_PERMISSIONS=0o765)
    def test_file_upload_directory_permissions(self):
        self.storage = FileSystemStorage(self.storage_dir)
        name = self.storage.save('the_directory/subdir/the_file', ContentFile('data'))
        file_path = Path(self.storage.path(name))
        self.assertEqual(file_path.parent.stat().st_mode & 0o777, 0o765)
        self.assertEqual(file_path.parent.parent.stat().st_mode & 0o777, 0o765)

    @override_settings(FILE_UPLOAD_DIRECTORY_PERMISSIONS=None)
    def test_file_upload_directory_default_permissions(self):
        self.storage = FileSystemStorage(self.storage_dir)
        name = self.storage.save('the_directory/subdir/the_file', ContentFile('data'))
        file_path = Path(self.storage.path(name))
        expected_mode = 0o777 & ~self.umask
        self.assertEqual(file_path.parent.stat().st_mode & 0o777, expected_mode)
        self.assertEqual(file_path.parent.parent.stat().st_mode & 0o777, expected_mode)


class FileStoragePathParsing(SimpleTestCase):
    def setUp(self):
        self.storage_dir = tempfile.mkdtemp()
        self.storage = FileSystemStorage(self.storage_dir)

    def tearDown(self):
        shutil.rmtree(self.storage_dir)

    def test_directory_with_dot(self):
        """Regression test for #9610.

        If the directory name contains a dot and the file name doesn't, make
        sure we still mangle the file name instead of the directory name.
        """

        self.storage.save('dotted.path/test', ContentFile("1"))
        self.storage.save('dotted.path/test', ContentFile("2"))

        files = sorted(os.listdir(os.path.join(self.storage_dir, 'dotted.path')))
        self.assertFalse(os.path.exists(os.path.join(self.storage_dir, 'dotted_.path')))
        self.assertEqual(files[0], 'test')
        self.assertRegex(files[1], 'test_%s' % FILE_SUFFIX_REGEX)

    def test_first_character_dot(self):
        """
        File names with a dot as their first character don't have an extension,
        and the underscore should get added to the end.
        """
        self.storage.save('dotted.path/.test', ContentFile("1"))
        self.storage.save('dotted.path/.test', ContentFile("2"))

        files = sorted(os.listdir(os.path.join(self.storage_dir, 'dotted.path')))
        self.assertFalse(os.path.exists(os.path.join(self.storage_dir, 'dotted_.path')))
        self.assertEqual(files[0], '.test')
        self.assertRegex(files[1], '.test_%s' % FILE_SUFFIX_REGEX)


class ContentFileStorageTestCase(unittest.TestCase):

    def setUp(self):
        self.storage_dir = tempfile.mkdtemp()
        self.storage = FileSystemStorage(self.storage_dir)

    def tearDown(self):
        shutil.rmtree(self.storage_dir)

    def test_content_saving(self):
        """
        ContentFile can be saved correctly with the filesystem storage,
        if it was initialized with either bytes or unicode content.
        """
        self.storage.save('bytes.txt', ContentFile(b"content"))
        self.storage.save('unicode.txt', ContentFile("español"))


@override_settings(ROOT_URLCONF='file_storage.urls')
class FileLikeObjectTestCase(LiveServerTestCase):
    """
    Test file-like objects (#15644).
    """

    available_apps = []

    def setUp(self):
        self.temp_dir = tempfile.mkdtemp()
        self.storage = FileSystemStorage(location=self.temp_dir)

    def tearDown(self):
        shutil.rmtree(self.temp_dir)

    def test_urllib_request_urlopen(self):
        """
        Test the File storage API with a file-like object coming from
        urllib.request.urlopen().
        """
        file_like_object = urlopen(self.live_server_url + '/')
        f = File(file_like_object)
        stored_filename = self.storage.save("remote_file.html", f)

        remote_file = urlopen(self.live_server_url + '/')
        with self.storage.open(stored_filename) as stored_file:
            self.assertEqual(stored_file.read(), remote_file.read())