Preserve sparse character positions in spooled text writes

boltons/ioutils.py

@@ -404,25 +404,12 @@ def write(self, s):
     def _traverse_codepoints(self, current_position, n):
         """Traverse from current position to the right n codepoints"""
         dest = current_position + n
-        while True:
-            if current_position == dest:
-                # By chance we've landed on the right position, break
-                break
-
-            # If the read would take us past the intended position then
-            # seek only enough to cover the offset
-            if current_position + READ_CHUNK_SIZE > dest:
-                self.read(dest - current_position)
-                break
-            else:
-                ret = self.read(READ_CHUNK_SIZE)
-
-            # Increment our current position
-            current_position += READ_CHUNK_SIZE
-
-            # If we kept reading but there was nothing here, break
-            # as we are at the end of the file
+        while current_position < dest:
+            ret = self.read(min(READ_CHUNK_SIZE, dest - current_position))
+            current_position += len(ret)
             if not ret:
+                # Each missing character will be a single-byte NUL on write.
+                self.buffer.seek(self.buffer.tell() + dest - current_position)
                 break
 
         return dest

tests/test_spooled_sparse_seek.py

@@ -0,0 +1,40 @@
+import io
+import os
+
+import pytest
+
+from boltons.ioutils import SpooledStringIO
+
+
+@pytest.mark.parametrize('rolled', [False, True])
+@pytest.mark.parametrize('whence,offset', [(os.SEEK_SET, 9),
+                                           (os.SEEK_CUR, 7),
+                                           (os.SEEK_END, 5)])
+def test_text_sparse_seek_matches_stringio(rolled, whence, offset):
+    value = 'a\N{LATIN SMALL LETTER E WITH ACUTE}bc'
+    with SpooledStringIO() as stream, io.StringIO() as reference:
+        for target in (stream, reference):
+            target.write(value)
+            target.seek(2)
+        if rolled:
+            stream.rollover()
+        destination = offset
+        if whence == os.SEEK_CUR:
+            destination += reference.tell()
+        elif whence == os.SEEK_END:
+            destination += len(value)
+        assert stream.seek(offset, whence) == reference.seek(destination)
+        assert stream.read() == reference.read() == ''
+        assert stream.write('\N{GREEK SMALL LETTER PI}') == reference.write('\N{GREEK SMALL LETTER PI}')
+        assert stream.tell() == reference.tell()
+        assert stream.getvalue() == reference.getvalue()
+
+
+def test_sparse_write_can_trigger_rollover():
+    with SpooledStringIO(max_size=6) as stream:
+        stream.write('a\N{LATIN SMALL LETTER E WITH ACUTE}bc')
+        assert not stream._rolled
+        stream.seek(9)
+        stream.write('\N{GREEK SMALL LETTER PI}')
+        assert stream._rolled
+        assert stream.getvalue() == 'a\N{LATIN SMALL LETTER E WITH ACUTE}bc' + '\0' * 5 + '\N{GREEK SMALL LETTER PI}'