Python django.core.files.storage 模块，Storage() 实例源码

def exists(self, name):
        name = self._normalize_name(name)
        try:
            self.bucket.stat_object(name)
        except Error, e:
            if e[0] == 404:
                return False
            raise
        return True

    #def listdir(self, path):
    #    path = self._normalize_name(path)
    #    try:
    #        result = self.bucket.list(path=path)
    #        return [i.name for i in result]
    #    except Error, e:
    #        raise IOError('Storage Error: %s' % e.args)

def _open_read(self, name):
        name = self._normalize_name(name)
        class _:
            def __init__(self, chunks):
                self.buf = ''
            def read(self, num_bytes=None):
                if num_bytes is None:
                    num_bytes = sys.maxint
                try:
                    while len(self.buf) < num_bytes:
                        self.buf += chunks.next()
                except StopIteration:
                    pass
                except Error, e:
                    raise IOError('Storage Error: %s' % e.args)
                retval = self.buf[:num_bytes]
                self.buf = self.buf[num_bytes:]
                return retval
        chunks = self.bucket.get_object_contents(name, chunk_size=8192)
        return _(chunks)

def test_logo_path(self, file_exists, delete_called):
        """
        Test that the path of image file should beenterprise/branding/<model.id>/<model_id>_logo.<ext>.lower().

        Additionally, test that the correct backend actions are taken in regards to deleting existing data.
        """
        file_mock = self._make_file_mock()
        branding_config = EnterpriseCustomerBrandingConfiguration(
            id=1,
            enterprise_customer=factories.EnterpriseCustomerFactory(),
            logo=file_mock
        )

        storage_mock = mock.MagicMock(spec=Storage, name="StorageMock")
        storage_mock.exists.return_value = file_exists
        with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
            path = logo_path(branding_config, branding_config.logo.name)
            self.assertEqual(path, "enterprise/branding/1/1_logo.png")
            assert storage_mock.delete.call_count == (1 if delete_called else 0)
            if delete_called:
                storage_mock.delete.assert_called_once_with('enterprise/branding/1/1_logo.png')

def test_branding_configuration_saving_successfully(self):
        """
        Test enterprise customer branding configuration saving successfully.
        """
        storage_mock = mock.MagicMock(spec=Storage, name="StorageMock")
        branding_config_1 = EnterpriseCustomerBrandingConfiguration(
            enterprise_customer=factories.EnterpriseCustomerFactory(),
            logo="test1.png"
        )

        storage_mock.exists.return_value = True
        with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
            branding_config_1.save()
            self.assertEqual(EnterpriseCustomerBrandingConfiguration.objects.count(), 1)

        branding_config_2 = EnterpriseCustomerBrandingConfiguration(
            enterprise_customer=factories.EnterpriseCustomerFactory(),
            logo="test2.png"
        )

        storage_mock.exists.return_value = False
        with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
            branding_config_2.save()
            self.assertEqual(EnterpriseCustomerBrandingConfiguration.objects.count(), 2)

def _get_available_name(self, name, max_length=None):
        dir_name, file_name = os.path.split(name)
        file_root, file_ext = os.path.splitext(file_name)
        while self.exists(name) or (max_length and len(name) > max_length):
            # file_ext includes the dot.
            name = os.path.join(dir_name, "%s_%s%s" % (file_root, get_random_string(7), file_ext))
            if max_length is None:
                continue
            # Truncate file_root if max_length exceeded.
            truncation = len(name) - max_length
            if truncation > 0:
                file_root = file_root[:-truncation]
                # Entire file_root was truncated in attempt to find an available filename.
                if not file_root:
                    raise SuspiciousFileOperation(
                        'Storage can not find an available filename for "%s". '
                        'Please make sure that the corresponding file field '
                        'allows sufficient "max_length".' % name
                    )
                name = os.path.join(dir_name, "%s_%s%s" % (file_root, get_random_string(7), file_ext))
        return name

def path(self, name):
        """
        Return a local filesystem path where the file can be retrieved using
        Python's built-in open() function. Storage systems that can't be
        accessed using open() should *not* implement this method.
        """
        raise NotImplementedError("This backend doesn't support absolute paths.")

def listdir(self, path):
        """
        List the contents of the specified path. Return a 2-tuple of lists:
        the first item being directories, the second item being files.
        """
        raise NotImplementedError('subclasses of Storage must provide a listdir() method')

def test_storage_check(self):
        sm = MagicMock(spec=Storage, name='StorageMock')
        with patch('django.core.files.storage.default_storage._wrapped', sm):
            check = StorageCheck(name='TestStorageCheck')
            check.run_check()
            self.assertIn('Data mismatch', check.errors[0])
            self.assertEqual(len(check.errors), 1)

def path(self, name):
        """
        Returns a local filesystem path where the file can be retrieved using
        Python's built-in open() function. Storage systems that can't be
        accessed using open() should *not* implement this method.
        """
        raise NotImplementedError("This backend doesn't have paths.")

def get_available_name(self, name, max_length=None):
        if max_length and len(name) + 41 > max_length:
            raise SuspiciousFileOperation(
                'Storage can not find an available filename for "%s". '
                'Please make sure that the corresponding file field '
                'allows sufficient "max_length".' % name
            )
        return name

def _save(self, name, content):
        name = self._normalize_name(name)
        try:
            self.bucket.put_object(name, content)
        except Error, e:
            raise IOError('Storage Error: %s' % e.args)
        return name

def delete(self, name):
        name = self._normalize_name(name)
        try:
            self.bucket.delete_object(name)
        except Error, e:
            raise IOError('Storage Error: %s' % e.args)

def size(self, name):
        name = self._normalize_name(name)
        try:
            attrs = self.bucket.stat_object(name)
            return attrs.bytes
        except Error, e:
            raise IOError('Storage Error: %s' % e.args)

def __init__(self, storage=None, *args, **kwargs):
        if storage is not None:
            self.storage = storage
        if self.storage is None:
            raise ImproperlyConfigured("The staticfiles storage finder %r "
                                       "doesn't have a storage class "
                                       "assigned." % self.__class__)
        # Make sure we have an storage instance here.
        if not isinstance(self.storage, (Storage, LazyObject)):
            self.storage = self.storage()
        super(BaseStorageFinder, self).__init__(*args, **kwargs)