Index: Lib/io.py
===================================================================
--- Lib/io.py	(revision 68372)
+++ Lib/io.py	(working copy)
@@ -1065,8 +1065,9 @@
                         # We've hit max_buffer_size. We have to accept a
                         # partial write and cut back our buffer.
                         overage = len(self._write_buf) - self.max_buffer_size
+                        written -= overage
                         self._write_buf = self._write_buf[:self.max_buffer_size]
-                        raise BlockingIOError(e.errno, e.strerror, overage)
+                        raise BlockingIOError(e.errno, e.strerror, written)
             return written
 
     def truncate(self, pos=None):
Index: Lib/test/test_io.py
===================================================================
--- Lib/test/test_io.py	(revision 68372)
+++ Lib/test/test_io.py	(working copy)
@@ -65,22 +65,38 @@
 
 class MockNonBlockWriterIO(io.RawIOBase):
 
-    def __init__(self, blocking_script):
-        self._blocking_script = list(blocking_script)
+    def __init__(self):
         self._write_stack = []
+        self._blocker_char = None
 
-    def write(self, b):
-        self._write_stack.append(b[:])
-        n = self._blocking_script.pop(0)
-        if (n < 0):
-            raise io.BlockingIOError(0, "test blocking", -n)
-        else:
-            return n
+    def pop_written(self):
+        s = b"".join(self._write_stack)
+        self._write_stack[:] = []
+        return s
 
+    def block_on(self, char):
+        """Block when a given char is encountered."""
+        self._blocker_char = char
+ 
     def writable(self):
         return True
 
+    def write(self, b):
+        b = bytes(b)
+        n = -1
+        if self._blocker_char:
+            try:
+                n = b.index(self._blocker_char)
+            except ValueError:
+                pass
+            else:
+                self._blocker_char = None
+                self._write_stack.append(b[:n])
+                raise io.BlockingIOError(0, "test blocking", n)
+        self._write_stack.append(b)
+        return len(b)
 
+
 class IOTest(unittest.TestCase):
 
     def tearDown(self):
@@ -481,25 +497,52 @@
         self.assertEquals(b"abcdefghijkl", writer._write_stack[0])
 
     def testWriteNonBlocking(self):
-        raw = MockNonBlockWriterIO((9, 2, 22, -6, 10, 12, 12))
-        bufio = io.BufferedWriter(raw, 8, 16)
+        raw = MockNonBlockWriterIO()
+        bufio = io.BufferedWriter(raw, 8, 8)
+
+        self.assertEquals(bufio.write(b"abcd"), 4)
+        self.assertEquals(bufio.write(b"efghi"), 5)
+        # 1 byte will be written, the rest will be buffered
+        raw.block_on(b"k")
+        self.assertEquals(bufio.write(b"jklmn"), 5)
+
+        # 8 bytes will be written, 8 will be buffered and the rest will be lost
+        raw.block_on(b"0")
+        try:
+            bufio.write(b"opqrwxyz0123456789")
+        except io.BlockingIOError as e:
+            written = e.characters_written
+        else:
+            self.fail("BlockingIOError should have been raised")
+        self.assertEquals(written, 16)
+        self.assertEquals(raw.pop_written(),
+            b"abcdefghijklmnopqrwxyz")
+
+        self.assertEquals(bufio.write(b"ABCDEFGHI"), 9)
+        s = raw.pop_written()
+        # Previously buffered bytes were flushed
+        self.assertTrue(s.startswith(b"01234567A"), s)
+
+        bufio_p = io.BufferedWriter(raw, 8, 16)
+        raw.block_on(b"1")
+        try:
+            bufio_p.write(b"12345678901234567890")
+        except io.BlockingIOError as e:
+            written = e.characters_written
+        else:
+            self.fail("BlockingIOError should have been raised")
+        self.assertEquals(written, 16)
+
+        raw.block_on(b"1")
+        try:
+            bufio_p.write(b"x")
+        except io.BlockingIOError as e:
+            written = e.characters_written
+        else:
+            self.fail("BlockingIOError should have been raised")
+        self.assertEquals(written, 0)
+        self.assertEquals(raw.pop_written(), b"")
 
-        bufio.write(b"asdf")
-        bufio.write(b"asdfa")
-        self.assertEquals(b"asdfasdfa", raw._write_stack[0])
-
-        bufio.write(b"asdfasdfasdf")
-        self.assertEquals(b"asdfasdfasdf", raw._write_stack[1])
-        bufio.write(b"asdfasdfasdf")
-        self.assertEquals(b"dfasdfasdf", raw._write_stack[2])
-        self.assertEquals(b"asdfasdfasdf", raw._write_stack[3])
-
-        bufio.write(b"asdfasdfasdf")
-
-        # XXX I don't like this test. It relies too heavily on how the
-        # algorithm actually works, which we might change. Refactor
-        # later.
-
     def testFileno(self):
         rawio = MockRawIO((b"abc", b"d", b"efg"))
         bufio = io.BufferedWriter(rawio)