← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_tuned_gzip.py

Turn completion assertions into separate methods.

Many common assertions used to be expressed as arguments to the complete
method.  This makes the checks more explicit, and the code easier to read.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_tuned_gzip.py
1
 
# Copyright (C) 2006, 2009, 2010, 2011 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
 
18
 
"""Tests for tuned_gzip."""
19
 
 
20
 
 
21
 
# do not use bzrlib test cases here - this should be suitable for sending
22
 
# upstream.
23
 
from cStringIO import StringIO
24
 
import zlib
25
 
 
26
 
 
27
 
from bzrlib import (
28
 
    symbol_versioning,
29
 
    tuned_gzip,
30
 
    tests,
31
 
    )
32
 
 
33
 
 
34
 
class FakeDecompress(object):
35
 
    """A fake decompressor for testing GzipFile."""
36
 
 
37
 
    def __init__(self):
38
 
        self.unused_data=''
39
 
 
40
 
    def decompress(self, buf):
41
 
        """Return an empty string as though we are at eof."""
42
 
        # note that the zlib module *overwrites* unused data
43
 
        # on writes after EOF.
44
 
        self.unused_data = buf
45
 
        return ''
46
 
 
47
 
 
48
 
class TestFakeDecompress(tests.TestCase):
49
 
    """We use a fake decompressor to test GzipFile.
50
 
 
51
 
    This class tests the behaviours we want from it.
52
 
    """
53
 
 
54
 
    def test_decompress(self):
55
 
        # decompressing returns no data.
56
 
        decompress = FakeDecompress()
57
 
        self.assertEqual('', decompress.decompress('0'))
58
 
 
59
 
    def test_unused_data(self):
60
 
        # after decompressing, we have 1 unused byte.
61
 
        # this is normally set by decompressors when they
62
 
        # detect the end of a compressed stream.
63
 
        decompress = FakeDecompress()
64
 
        decompress.decompress('0')
65
 
        self.assertEqual('0', decompress.unused_data)
66
 
        # decompressing again (when the short read is read)
67
 
        # will give us the latest input in the unused_data
68
 
        # this is arguably a bug in zlib but ...
69
 
        decompress.decompress('1234567')
70
 
        self.assertEqual('1234567', decompress.unused_data)
71
 
 
72
 
 
73
 
class TestGzip(tests.TestCase):
74
 
 
75
 
    def test__read_short_remainder(self):
76
 
        # a _read call at the end of a compressed hunk should
77
 
        # read more bytes if there is less than 8 bytes (the
78
 
        # gzip trailer) unread.
79
 
        stream = StringIO('\0\0\0\0\0\0\0\0')
80
 
        myfile = self.applyDeprecated(
81
 
            symbol_versioning.deprecated_in((2, 3, 0)),
82
 
            tuned_gzip.GzipFile, fileobj=stream)
83
 
        # disable the _new_member check, we are microtesting.
84
 
        myfile._new_member = False
85
 
        myfile.crc = zlib.crc32('')
86
 
        myfile.decompress = FakeDecompress()
87
 
        myfile.size = 0
88
 
        myfile._read(1)
89
 
        # all the data should have been read now
90
 
        self.assertEqual('', stream.read())
91
 
        # and it should be new member time in the stream.
92
 
        self.assertTrue(myfile._new_member)
93
 
 
94
 
    def test_negative_crc(self):
95
 
        """Content with a negative crc should not break when written"""
96
 
        sio = StringIO()
97
 
        gfile = self.applyDeprecated(
98
 
            symbol_versioning.deprecated_in((2, 3, 0)),
99
 
            tuned_gzip.GzipFile, mode="w", fileobj=sio)
100
 
        gfile.write("\xFF")
101
 
        gfile.close()
102
 
        self.assertEqual(gfile.crc & 0xFFFFFFFFL, 0xFF000000L)
103
 
        self.assertEqual(sio.getvalue()[-8:-4], "\x00\x00\x00\xFF")
104
 
 
105
 
 
106
 
class TestToGzip(tests.TestCase):
107
 
 
108
 
    def assertToGzip(self, chunks):
109
 
        bytes = ''.join(chunks)
110
 
        gzfromchunks = tuned_gzip.chunks_to_gzip(chunks)
111
 
        gzfrombytes = tuned_gzip.bytes_to_gzip(bytes)
112
 
        self.assertEqual(gzfrombytes, gzfromchunks)
113
 
        decoded = self.applyDeprecated(
114
 
            symbol_versioning.deprecated_in((2, 3, 0)),
115
 
            tuned_gzip.GzipFile, fileobj=StringIO(gzfromchunks)).read()
116
 
        self.assertEqual(bytes, decoded)
117
 
 
118
 
    def test_single_chunk(self):
119
 
        self.assertToGzip(['a modest chunk\nwith some various\nbits\n'])
120
 
 
121
 
    def test_simple_text(self):
122
 
        self.assertToGzip(['some\n', 'strings\n', 'to\n', 'process\n'])
123
 
 
124
 
    def test_large_chunks(self):
125
 
        self.assertToGzip(['a large string\n'*1024])
126
 
        self.assertToGzip(['a large string\n']*1024)
127
 
 
128
 
    def test_enormous_chunks(self):
129
 
        self.assertToGzip(['a large string\n'*1024*256])
130
 
        self.assertToGzip(['a large string\n']*1024*256)