Add support for buffering in text mode

- support line buffering
- closes #549

Author: mrbean-bremen

CHANGES.md

@@ -4,8 +4,8 @@ The released versions correspond to PyPi releases.
 ## Version 4.2.0 (as yet unreleased)
 
 #### New Features
-  * add support for the `buffering` parameter in `open` (text line mode not
-    supported) (see [#549](../../issues/549))
+  * add support for the `buffering` parameter in `open` 
+    (see [#549](../../issues/549))
 
 #### Fixes
   * do not truncate file on failed flush

docs/usage.rst

@@ -540,6 +540,7 @@ and you may fail to create new files if the fake file system is full.
             with open('/foo/bar.txt', 'w') as f:
                 with self.assertRaises(OSError):
                     f.write('a' * 200)
+                    f.flush()
 
 To get the file system size, you may use ``get_disk_usage()``, which is
 modeled after ``shutil.disk_usage()``.

pyfakefs/fake_filesystem.py

@@ -139,7 +139,7 @@
     'r+': (True, True, True, False, False, False),
     'w+': (False, True, True, True, False, False),
     'a+': (False, True, True, False, True, False),
-    'x':  (False, False, True, False, False, True),
+    'x': (False, False, True, False, False, True),
     'x+': (False, True, True, False, False, True)
 }
 
@@ -1406,8 +1406,8 @@ def components_to_path():
                     path_components[len(normalized_components):])
             sep = self._path_separator(path)
             normalized_path = sep.join(normalized_components)
-            if (self._starts_with_sep(path) and not
-                    self._starts_with_sep(normalized_path)):
+            if (self._starts_with_sep(path)
+                    and not self._starts_with_sep(normalized_path)):
                 normalized_path = sep + normalized_path
             return normalized_path
 
@@ -4501,7 +4501,7 @@ class FakeFileWrapper:
     def __init__(self, file_object, file_path, update, read,
                  append, delete_on_close, filesystem,
                  newline, binary, closefd, encoding,
-                 errors, buffer_size, raw_io, is_stream=False):
+                 errors, buffering, raw_io, is_stream=False):
         self.file_object = file_object
         self.file_path = file_path
         self._append = append
@@ -4511,10 +4511,16 @@ def __init__(self, file_object, file_path, update, read,
         self._file_epoch = file_object.epoch
         self.raw_io = raw_io
         self._binary = binary
-        self._buffer_size = buffer_size
-        self._use_line_buffer = not binary and buffer_size == 1
         self.is_stream = is_stream
         self._changed = False
+        self._buffer_size = buffering
+        if self._buffer_size == 0 and not binary:
+            raise ValueError("can't have unbuffered text I/O")
+        # buffer_size is ignored in text mode
+        elif self._buffer_size == -1 or not binary:
+            self._buffer_size = io.DEFAULT_BUFFER_SIZE
+        self._use_line_buffer = not binary and buffering == 1
+
         contents = file_object.byte_contents
         self._encoding = encoding or locale.getpreferredencoding(False)
         errors = errors or 'strict'
@@ -4796,8 +4802,10 @@ def write_wrapper(*args, **kwargs):
             new_pos = self._io.tell()
 
             # if the buffer size is exceeded, we flush
-            if new_pos - self._flush_pos > self._buffer_size:
-                flush_all = new_pos - old_pos > self._buffer_size
+            use_line_buf = self._use_line_buffer and '\n' in args[0]
+            if new_pos - self._flush_pos > self._buffer_size or use_line_buf:
+                flush_all = (new_pos - old_pos > self._buffer_size or
+                             use_line_buf)
                 # if the current write does not exceed the buffer size,
                 # we revert to the previous position and flush that,
                 # otherwise we flush all
@@ -5079,13 +5087,6 @@ def call(self, file_, mode='r', buffering=-1, encoding=None,
         if not filedes:
             closefd = True
 
-        buffer_size = buffering
-        if buffer_size == -1:
-            buffer_size = io.DEFAULT_BUFFER_SIZE
-        elif buffer_size == 0:
-            if not binary:
-                raise ValueError("can't have unbuffered text I/O")
-
         if (open_modes.must_not_exist and
                 (file_object or self.filesystem.islink(file_path) and
                  not self.filesystem.is_windows_fs)):
@@ -5122,7 +5123,7 @@ def call(self, file_, mode='r', buffering=-1, encoding=None,
                                    closefd=closefd,
                                    encoding=encoding,
                                    errors=errors,
-                                   buffer_size=buffer_size,
+                                   buffering=buffering,
                                    raw_io=self.raw_io)
         if filedes is not None:
             fakefile.filedes = filedes

pyfakefs/tests/fake_open_test.py

@@ -925,7 +925,6 @@ def use_real_fs(self):
 
 
 class BufferingModeTest(FakeFileOpenTestBase):
-    # todo: check text mode, check append mode
     def test_no_buffering(self):
         file_path = self.make_path("buffertest.bin")
         with self.open(file_path, 'wb', buffering=0) as f:
@@ -989,6 +988,103 @@ def test_writing_with_specific_buffer(self):
                 # new buffer exceeded (600) -> all written
                 self.assertEqual(1700, len(x))
 
+    def test_writing_text_with_line_buffer(self):
+        file_path = self.make_path("buffertest.bin")
+        with self.open(file_path, 'w', buffering=1) as f:
+            f.write('test' * 100)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # no new line - not written
+                self.assertEqual(0, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # new line - buffer written
+                self.assertEqual(405, len(x))
+            f.write('test' * 10)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(405, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # new line - buffer written
+                self.assertEqual(450, len(x))
+
+    def test_writing_large_text_with_line_buffer(self):
+        file_path = self.make_path("buffertest.bin")
+        with self.open(file_path, 'w', buffering=1) as f:
+            f.write('test' * 4000)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer larger than default - written
+                self.assertEqual(16000, len(x))
+            f.write('test')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(16000, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # new line - buffer written
+                self.assertEqual(16009, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # another new line - buffer written
+                self.assertEqual(16014, len(x))
+
+    def test_writing_text_with_default_buffer(self):
+        file_path = self.make_path("buffertest.txt")
+        with self.open(file_path, 'w') as f:
+            f.write('test' * 5)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(0, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer exceeded, but new buffer (400) not - previous written
+                self.assertEqual(0, len(x))
+            f.write('test' * 10)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(0, len(x))
+            f.write('\ntest')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                self.assertEqual(0, len(x))
+
+    def test_writing_text_with_specific_buffer(self):
+        file_path = self.make_path("buffertest.txt")
+        with self.open(file_path, 'w', buffering=2) as f:
+            f.write('a' * 8000)
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(0, len(x))
+            f.write('test')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer exceeded, but new buffer (400) not - previous written
+                self.assertEqual(0, len(x))
+            f.write('test')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                # buffer not filled - not written
+                self.assertEqual(0, len(x))
+            f.write('test')
+            with self.open(file_path, "r") as r:
+                x = r.read()
+                self.assertEqual(0, len(x))
+        # with self.open(file_path, "r") as r:
+        #     x = r.read()
+        #     self.assertEqual(35, len(x))
+
     def test_append_with_specific_buffer(self):
         file_path = self.make_path("buffertest.bin")
         with self.open(file_path, 'wb', buffering=512) as f: