/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_chunk_writer.py

  • Committer: Jelmer Vernooij
  • Date: 2017-07-23 22:06:41 UTC
  • mfrom: (6738 trunk)
  • mto: This revision was merged to the branch mainline in revision 6739.
  • Revision ID: jelmer@jelmer.uk-20170723220641-69eczax9bmv8d6kk
Merge trunk, address review comments.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_chunk_writer.py
26
26
class TestWriter(TestCaseWithTransport):
27
27
 
28
28
    def check_chunk(self, bytes_list, size):
29
 
        bytes = ''.join(bytes_list)
30
 
        self.assertEqual(size, len(bytes))
31
 
        return zlib.decompress(bytes)
 
29
        data = b''.join(bytes_list)
 
30
        self.assertEqual(size, len(data))
 
31
        return zlib.decompress(data)
32
32
 
33
33
    def test_chunk_writer_empty(self):
34
34
        writer = chunk_writer.ChunkWriter(4096)
35
35
        bytes_list, unused, padding = writer.finish()
36
36
        node_bytes = self.check_chunk(bytes_list, 4096)
37
 
        self.assertEqual("", node_bytes)
 
37
        self.assertEqual(b"", node_bytes)
38
38
        self.assertEqual(None, unused)
39
39
        # Only a zlib header.
40
40
        self.assertEqual(4088, padding)
59
59
 
60
60
    def test_some_data(self):
61
61
        writer = chunk_writer.ChunkWriter(4096)
62
 
        writer.write("foo bar baz quux\n")
 
62
        writer.write(b"foo bar baz quux\n")
63
63
        bytes_list, unused, padding = writer.finish()
64
64
        node_bytes = self.check_chunk(bytes_list, 4096)
65
 
        self.assertEqual("foo bar baz quux\n", node_bytes)
 
65
        self.assertEqual(b"foo bar baz quux\n", node_bytes)
66
66
        self.assertEqual(None, unused)
67
67
        # More than just the header..
68
68
        self.assertEqual(4073, padding)
69
69
 
 
70
    @staticmethod
 
71
    def _make_lines():
 
72
        lines = []
 
73
        for group in range(48):
 
74
            offset = group * 50
 
75
            numbers = list(range(offset, offset + 50))
 
76
            # Create a line with this group
 
77
            lines.append(b''.join(b'%d' % n for n in numbers) + b'\n')
 
78
        return lines
 
79
 
70
80
    def test_too_much_data_does_not_exceed_size(self):
71
81
        # Generate enough data to exceed 4K
72
 
        lines = []
73
 
        for group in range(48):
74
 
            offset = group * 50
75
 
            numbers = list(range(offset, offset + 50))
76
 
            # Create a line with this group
77
 
            lines.append(''.join(map(str, numbers)) + '\n')
 
82
        lines = self._make_lines()
78
83
        writer = chunk_writer.ChunkWriter(4096)
79
84
        for idx, line in enumerate(lines):
80
85
            if writer.write(line):
83
88
        bytes_list, unused, _ = writer.finish()
84
89
        node_bytes = self.check_chunk(bytes_list, 4096)
85
90
        # the first 46 lines should have been added
86
 
        expected_bytes = ''.join(lines[:46])
 
91
        expected_bytes = b''.join(lines[:46])
87
92
        self.assertEqualDiff(expected_bytes, node_bytes)
88
93
        # And the line that failed should have been saved for us
89
94
        self.assertEqual(lines[46], unused)
90
95
 
91
96
    def test_too_much_data_preserves_reserve_space(self):
92
97
        # Generate enough data to exceed 4K
93
 
        lines = []
94
 
        for group in range(48):
95
 
            offset = group * 50
96
 
            numbers = list(range(offset, offset + 50))
97
 
            # Create a line with this group
98
 
            lines.append(''.join(map(str, numbers)) + '\n')
 
98
        lines = self._make_lines()
99
99
        writer = chunk_writer.ChunkWriter(4096, 256)
100
100
        for idx, line in enumerate(lines):
101
101
            if writer.write(line):
103
103
                break
104
104
        else:
105
105
            self.fail('We were able to write all lines')
106
 
        self.assertFalse(writer.write("A"*256, reserved=True))
 
106
        self.assertFalse(writer.write(b"A"*256, reserved=True))
107
107
        bytes_list, unused, _ = writer.finish()
108
108
        node_bytes = self.check_chunk(bytes_list, 4096)
109
109
        # the first 44 lines should have been added
110
 
        expected_bytes = ''.join(lines[:44]) + "A"*256
 
110
        expected_bytes = b''.join(lines[:44]) + b"A"*256
111
111
        self.assertEqualDiff(expected_bytes, node_bytes)
112
112
        # And the line that failed should have been saved for us
113
113
        self.assertEqual(lines[44], unused)