Merge lp:~gz/brz/py3_chunk_writer into lp:brz

[Martin Packman]

Make chunk_writer Python 3 compatible

Simple enough changes, mostly annotating byte strings correctly.

[Jelmer Vernooij (jelmer)]

review: Approve

=== modified file 'breezy/chunk_writer.py'
--- breezy/chunk_writer.py	2011-12-19 13:23:58 +0000
+++ breezy/chunk_writer.py	2017-07-02 13:51:52 +0000
@@ -144,7 +144,7 @@
                                  % (self.bytes_out_len, self.chunk_size))
         nulls_needed = self.chunk_size - self.bytes_out_len
         if nulls_needed:
-            self.bytes_list.append("\x00" * nulls_needed)
+            self.bytes_list.append(b"\x00" * nulls_needed)
         return self.bytes_list, self.unused_bytes, nulls_needed
 
     def set_optimize(self, for_size=True):

=== modified file 'breezy/tests/test_chunk_writer.py'
--- breezy/tests/test_chunk_writer.py	2017-06-04 18:09:30 +0000
+++ breezy/tests/test_chunk_writer.py	2017-07-02 13:51:52 +0000
@@ -26,15 +26,15 @@
 class TestWriter(TestCaseWithTransport):
 
     def check_chunk(self, bytes_list, size):
-        bytes = ''.join(bytes_list)
-        self.assertEqual(size, len(bytes))
-        return zlib.decompress(bytes)
+        data = b''.join(bytes_list)
+        self.assertEqual(size, len(data))
+        return zlib.decompress(data)
 
     def test_chunk_writer_empty(self):
         writer = chunk_writer.ChunkWriter(4096)
         bytes_list, unused, padding = writer.finish()
         node_bytes = self.check_chunk(bytes_list, 4096)
-        self.assertEqual("", node_bytes)
+        self.assertEqual(b"", node_bytes)
         self.assertEqual(None, unused)
         # Only a zlib header.
         self.assertEqual(4088, padding)
@@ -59,22 +59,27 @@
 
     def test_some_data(self):
         writer = chunk_writer.ChunkWriter(4096)
-        writer.write("foo bar baz quux\n")
+        writer.write(b"foo bar baz quux\n")
         bytes_list, unused, padding = writer.finish()
         node_bytes = self.check_chunk(bytes_list, 4096)
-        self.assertEqual("foo bar baz quux\n", node_bytes)
+        self.assertEqual(b"foo bar baz quux\n", node_bytes)
         self.assertEqual(None, unused)
         # More than just the header..
         self.assertEqual(4073, padding)
 
+    @staticmethod
+    def _make_lines():
+        lines = []
+        for group in range(48):
+            offset = group * 50
+            numbers = list(range(offset, offset + 50))
+            # Create a line with this group
+            lines.append(b''.join(b'%d' % n for n in numbers) + b'\n')
+        return lines
+
     def test_too_much_data_does_not_exceed_size(self):
         # Generate enough data to exceed 4K
-        lines = []
-        for group in range(48):
-            offset = group * 50
-            numbers = list(range(offset, offset + 50))
-            # Create a line with this group
-            lines.append(''.join(map(str, numbers)) + '\n')
+        lines = self._make_lines()
         writer = chunk_writer.ChunkWriter(4096)
         for idx, line in enumerate(lines):
             if writer.write(line):
@@ -83,19 +88,14 @@
         bytes_list, unused, _ = writer.finish()
         node_bytes = self.check_chunk(bytes_list, 4096)
         # the first 46 lines should have been added
-        expected_bytes = ''.join(lines[:46])
+        expected_bytes = b''.join(lines[:46])
         self.assertEqualDiff(expected_bytes, node_bytes)
         # And the line that failed should have been saved for us
         self.assertEqual(lines[46], unused)
 
     def test_too_much_data_preserves_reserve_space(self):
         # Generate enough data to exceed 4K
-        lines = []
-        for group in range(48):
-            offset = group * 50
-            numbers = list(range(offset, offset + 50))
-            # Create a line with this group
-            lines.append(''.join(map(str, numbers)) + '\n')
+        lines = self._make_lines()
         writer = chunk_writer.ChunkWriter(4096, 256)
         for idx, line in enumerate(lines):
             if writer.write(line):
@@ -103,11 +103,11 @@
                 break
         else:
             self.fail('We were able to write all lines')
-        self.assertFalse(writer.write("A"*256, reserved=True))
+        self.assertFalse(writer.write(b"A"*256, reserved=True))
         bytes_list, unused, _ = writer.finish()
         node_bytes = self.check_chunk(bytes_list, 4096)
         # the first 44 lines should have been added
-        expected_bytes = ''.join(lines[:44]) + "A"*256
+        expected_bytes = b''.join(lines[:44]) + b"A"*256
         self.assertEqualDiff(expected_bytes, node_bytes)
         # And the line that failed should have been saved for us
         self.assertEqual(lines[44], unused)