lib/subunit/python/subunit/tests/test_chunked.py

   1 #
   2 #  subunit: extensions to python unittest to get test results from subprocesses.
   3 #  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
   4 #  Copyright (C) 2011  Martin Pool <mbp@sourcefrog.net>
   5 #
   6 #  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
   7 #  license at the users choice. A copy of both licenses are available in the
   8 #  project source as Apache-2.0 and BSD. You may not use this file except in
   9 #  compliance with one of these two licences.
  10 #
  11 #  Unless required by applicable law or agreed to in writing, software
  12 #  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
  13 #  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
  14 #  license you chose for the specific language governing permissions and
  15 #  limitations under that license.
  16 #
  17
  18 import unittest
  19
  20 from testtools.compat import _b, BytesIO
  21
  22 import subunit.chunked
  23
  24
  25 def test_suite():
  26     loader = subunit.tests.TestUtil.TestLoader()
  27     result = loader.loadTestsFromName(__name__)
  28     return result
  29
  30
  31 class TestDecode(unittest.TestCase):
  32
  33     def setUp(self):
  34         unittest.TestCase.setUp(self)
  35         self.output = BytesIO()
  36         self.decoder = subunit.chunked.Decoder(self.output)
  37
  38     def test_close_read_length_short_errors(self):
  39         self.assertRaises(ValueError, self.decoder.close)
  40
  41     def test_close_body_short_errors(self):
  42         self.assertEqual(None, self.decoder.write(_b('2\r\na')))
  43         self.assertRaises(ValueError, self.decoder.close)
  44
  45     def test_close_body_buffered_data_errors(self):
  46         self.assertEqual(None, self.decoder.write(_b('2\r')))
  47         self.assertRaises(ValueError, self.decoder.close)
  48
  49     def test_close_after_finished_stream_safe(self):
  50         self.assertEqual(None, self.decoder.write(_b('2\r\nab')))
  51         self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
  52         self.decoder.close()
  53
  54     def test_decode_nothing(self):
  55         self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
  56         self.assertEqual(_b(''), self.output.getvalue())
  57
  58     def test_decode_serialised_form(self):
  59         self.assertEqual(None, self.decoder.write(_b("F\r\n")))
  60         self.assertEqual(None, self.decoder.write(_b("serialised\n")))
  61         self.assertEqual(_b(''), self.decoder.write(_b("form0\r\n")))
  62
  63     def test_decode_short(self):
  64         self.assertEqual(_b(''), self.decoder.write(_b('3\r\nabc0\r\n')))
  65         self.assertEqual(_b('abc'), self.output.getvalue())
  66
  67     def test_decode_combines_short(self):
  68         self.assertEqual(_b(''), self.decoder.write(_b('6\r\nabcdef0\r\n')))
  69         self.assertEqual(_b('abcdef'), self.output.getvalue())
  70
  71     def test_decode_excess_bytes_from_write(self):
  72         self.assertEqual(_b('1234'), self.decoder.write(_b('3\r\nabc0\r\n1234')))
  73         self.assertEqual(_b('abc'), self.output.getvalue())
  74
  75     def test_decode_write_after_finished_errors(self):
  76         self.assertEqual(_b('1234'), self.decoder.write(_b('3\r\nabc0\r\n1234')))
  77         self.assertRaises(ValueError, self.decoder.write, _b(''))
  78
  79     def test_decode_hex(self):
  80         self.assertEqual(_b(''), self.decoder.write(_b('A\r\n12345678900\r\n')))
  81         self.assertEqual(_b('1234567890'), self.output.getvalue())
  82
  83     def test_decode_long_ranges(self):
  84         self.assertEqual(None, self.decoder.write(_b('10000\r\n')))
  85         self.assertEqual(None, self.decoder.write(_b('1' * 65536)))
  86         self.assertEqual(None, self.decoder.write(_b('10000\r\n')))
  87         self.assertEqual(None, self.decoder.write(_b('2' * 65536)))
  88         self.assertEqual(_b(''), self.decoder.write(_b('0\r\n')))
  89         self.assertEqual(_b('1' * 65536 + '2' * 65536), self.output.getvalue())
  90
  91     def test_decode_newline_nonstrict(self):
  92         """Tolerate chunk markers with no CR character."""
  93         # From <http://pad.lv/505078>
  94         self.decoder = subunit.chunked.Decoder(self.output, strict=False)
  95         self.assertEqual(None, self.decoder.write(_b('a\n')))
  96         self.assertEqual(None, self.decoder.write(_b('abcdeabcde')))
  97         self.assertEqual(_b(''), self.decoder.write(_b('0\n')))
  98         self.assertEqual(_b('abcdeabcde'), self.output.getvalue())
  99
 100     def test_decode_strict_newline_only(self):
 101         """Reject chunk markers with no CR character in strict mode."""
 102         # From <http://pad.lv/505078>
 103         self.assertRaises(ValueError,
 104             self.decoder.write, _b('a\n'))
 105
 106     def test_decode_strict_multiple_crs(self):
 107         self.assertRaises(ValueError,
 108             self.decoder.write, _b('a\r\r\n'))
 109
 110     def test_decode_short_header(self):
 111         self.assertRaises(ValueError,
 112             self.decoder.write, _b('\n'))
 113
 114
 115 class TestEncode(unittest.TestCase):
 116
 117     def setUp(self):
 118         unittest.TestCase.setUp(self)
 119         self.output = BytesIO()
 120         self.encoder = subunit.chunked.Encoder(self.output)
 121
 122     def test_encode_nothing(self):
 123         self.encoder.close()
 124         self.assertEqual(_b('0\r\n'), self.output.getvalue())
 125
 126     def test_encode_empty(self):
 127         self.encoder.write(_b(''))
 128         self.encoder.close()
 129         self.assertEqual(_b('0\r\n'), self.output.getvalue())
 130
 131     def test_encode_short(self):
 132         self.encoder.write(_b('abc'))
 133         self.encoder.close()
 134         self.assertEqual(_b('3\r\nabc0\r\n'), self.output.getvalue())
 135
 136     def test_encode_combines_short(self):
 137         self.encoder.write(_b('abc'))
 138         self.encoder.write(_b('def'))
 139         self.encoder.close()
 140         self.assertEqual(_b('6\r\nabcdef0\r\n'), self.output.getvalue())
 141
 142     def test_encode_over_9_is_in_hex(self):
 143         self.encoder.write(_b('1234567890'))
 144         self.encoder.close()
 145         self.assertEqual(_b('A\r\n12345678900\r\n'), self.output.getvalue())
 146
 147     def test_encode_long_ranges_not_combined(self):
 148         self.encoder.write(_b('1' * 65536))
 149         self.encoder.write(_b('2' * 65536))
 150         self.encoder.close()
 151         self.assertEqual(_b('10000\r\n' + '1' * 65536 + '10000\r\n' +
 152             '2' * 65536 + '0\r\n'), self.output.getvalue())