Skip to content

Commit

Permalink
Only seek in S3Boto3Storage._save if content is seekable
Browse files Browse the repository at this point in the history
Also add tests to verify the behavior. This fixes issue #860.
  • Loading branch information
vainu-arto committed Sep 16, 2021
1 parent a7a2d2b commit 304b1b0
Show file tree
Hide file tree
Showing 2 changed files with 47 additions and 1 deletion.
3 changes: 2 additions & 1 deletion storages/backends/s3boto3.py
Original file line number Diff line number Diff line change
Expand Up @@ -463,7 +463,8 @@ def _save(self, name, content):
name = self._normalize_name(cleaned_name)
params = self._get_write_parameters(name, content)

content.seek(0, os.SEEK_SET)
if content.seekable():
content.seek(0, os.SEEK_SET)
if (self.gzip and
params['ContentType'] in self.gzip_content_types and
'ContentEncoding' not in params):
Expand Down
45 changes: 45 additions & 0 deletions tests/test_s3boto3.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,18 @@ def setUp(self):
self.storage._connections.connection = mock.MagicMock()


class NonSeekableContentFile(ContentFile):

def open(self, mode=None):
return self

def seekable(self):
return False

def seek(self, pos, whence=0):
raise AttributeError()


class S3Boto3StorageTests(S3Boto3TestCase):

def test_clean_name(self):
Expand Down Expand Up @@ -121,6 +133,23 @@ def test_storage_save(self):
}
)

def test_storage_save_non_seekable(self):
"""
Test saving a non-seekable file
"""
name = 'test_storage_save.txt'
content = NonSeekableContentFile('new content')
self.storage.save(name, content)
self.storage.bucket.Object.assert_called_once_with(name)

obj = self.storage.bucket.Object.return_value
obj.upload_fileobj.assert_called_with(
content,
ExtraArgs={
'ContentType': 'text/plain',
}
)

def test_storage_save_with_default_acl(self):
"""
Test saving a file with user defined ACL.
Expand Down Expand Up @@ -194,6 +223,22 @@ def test_storage_save_gzipped(self):
}
)

def test_storage_save_gzipped_non_seekable(self):
"""
Test saving a gzipped file
"""
name = 'test_storage_save.gz'
content = NonSeekableContentFile("I am gzip'd")
self.storage.save(name, content)
obj = self.storage.bucket.Object.return_value
obj.upload_fileobj.assert_called_with(
content,
ExtraArgs={
'ContentType': 'application/octet-stream',
'ContentEncoding': 'gzip',
}
)

def test_storage_save_gzip(self):
"""
Test saving a file with gzip enabled.
Expand Down

0 comments on commit 304b1b0

Please sign in to comment.