Account for negative seek on empty file

Author: ddelange

smart_open/s3.py

@@ -586,7 +586,12 @@ def _open_body(self, start=None, stop=None):
                 raise
 
             actual_object_size = int(error_response.get('ActualObjectSize', 0))
-            if start >= actual_object_size:  # empty file or start is past end of file
+            if (
+                # empty file (==) or start is past end of file (>)
+                (start is not None and start >= actual_object_size)
+                # empty file (==) or negative seek like seek(-4, WHENCE_END) is past end of file (>)
+                or (start is None and stop is not None and stop >= actual_object_size)
+            ):
                 self._position = self._content_length = actual_object_size
                 self._body = io.BytesIO()
             else:  # stop is past end of file: request the correct remainder instead

tests/test_s3.py

@@ -319,6 +319,15 @@ def test_seek_past_end(self):
             seek = fin.seek(60)
             self.assertEqual(seek, len(self.body))
 
+    def test_seek_past_end_from_end(self):
+        """Test seeking from end with offset larger than file size."""
+        body_len = len(self.body)
+        with self.assertApiCalls(GetObject=1), patch_invalid_range_response(str(body_len)):
+            fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True)
+            seek = fin.seek(-(body_len + 10), whence=smart_open.constants.WHENCE_END)
+            self.assertEqual(seek, 0)  # Should clamp to start of file
+
+
     def test_detect_eof(self):
         with self.assertApiCalls(GetObject=1):
             fin = smart_open.s3.Reader(BUCKET_NAME, KEY_NAME)
@@ -470,6 +479,14 @@ def test_read_empty_file_no_actual_size(self):
                 # a subsequent read does not call _open_body
                 self.assertEqual(fin.read(), b'')
 
+    def test_seek_empty_file_from_end(self):
+        """Test seeking from end on an empty file."""
+        _resource('s3').Object(BUCKET_NAME, KEY_NAME).put(Body=b'')
+        with self.assertApiCalls(GetObject=1), patch_invalid_range_response('0'):
+            with smart_open.s3.Reader(BUCKET_NAME, KEY_NAME, defer_seek=True) as fin:
+                seek = fin.seek(-10, whence=smart_open.constants.WHENCE_END)
+                self.assertEqual(seek, 0)  # Should be at position 0 for empty file
+
 
 @mock_s3
 class MultipartWriterTest(unittest.TestCase):