chiark - git - mdw - catacomb-python/blob - t/t-buffer.py

   1 ### -*- mode: python, coding: utf-8 -*-
   2 ###
   3 ### Test read and write buffers
   4 ###
   5 ### (c) 2019 Straylight/Edgeware
   6 ###
   7
   8 ###----- Licensing notice ---------------------------------------------------
   9 ###
  10 ### This file is part of the Python interface to Catacomb.
  11 ###
  12 ### Catacomb/Python is free software: you can redistribute it and/or
  13 ### modify it under the terms of the GNU General Public License as
  14 ### published by the Free Software Foundation; either version 2 of the
  15 ### License, or (at your option) any later version.
  16 ###
  17 ### Catacomb/Python is distributed in the hope that it will be useful, but
  18 ### WITHOUT ANY WARRANTY; without even the implied warranty of
  19 ### MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  20 ### General Public License for more details.
  21 ###
  22 ### You should have received a copy of the GNU General Public License
  23 ### along with Catacomb/Python.  If not, write to the Free Software
  24 ### Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
  25 ### USA.
  26
  27 ###--------------------------------------------------------------------------
  28 ### Imported modules.
  29
  30 import catacomb as C
  31 import unittest as U
  32 import testutils as T
  33
  34 ###--------------------------------------------------------------------------
  35 class TestReadBuffer (U.TestCase):
  36
  37   def check_getn(me, w, bigendp, getfn):
  38     """Check that `getuN' works."""
  39     buf = C.ReadBuffer(T.span(w + 2))
  40     me.assertEqual(buf.getu8(), 0x00)
  41     me.assertEqual(getfn(buf), T.bytes_as_int(w, bigendp))
  42     me.assertEqual(buf.getu8(), w + 1)
  43     me.assertTrue(buf.endp)
  44     me.assertRaises(C.BufferError, getfn, C.ReadBuffer(T.span(w - 1)))
  45     me.assertEqual(getfn(C.ReadBuffer(w*C.bytes("00"))), 0)
  46     me.assertEqual(getfn(C.ReadBuffer(w*C.bytes("ff"))), (1 << 8*w) - 1)
  47
  48   def check_getbufn(me, w, bigendp, blkfn, buffn):
  49     """Check that `getblkN' and `getbufN' work."""
  50
  51     ## Run tests for several different data sizes.
  52     for n in [0, 1, 7, 8, 19, 255, 12345, 65535, 123456]:
  53
  54       ## Make a sequence to parse.  If it's too large, then skip.
  55       if n >= 1 << 8*w: continue
  56       seq = T.prep_lenseq(w, n, bigendp, True)
  57
  58       ## Check `getblkN'.
  59       buf = C.ReadBuffer(seq)
  60       me.assertEqual(buf.getu8(), 0)
  61       me.assertEqual(blkfn(buf), T.span(n))
  62       me.assertEqual(buf.getu8(), 0xff)
  63       me.assertTrue(buf.endp)
  64
  65       ## Check `getbufN'.  Delete the outside buffer early, to make sure that
  66       ## the subbuffer keeps it alive.
  67       buf = C.ReadBuffer(seq)
  68       me.assertEqual(buf.getu8(), 0)
  69       b = buffn(buf)
  70       me.assertEqual(buf.getu8(), 0xff)
  71       me.assertTrue(buf.endp)
  72       del buf
  73       me.assertEqual(b.offset, 0)
  74       me.assertEqual(b.size, n)
  75       if n > 0:
  76         me.assertEqual(b.getu8(), 0x00)
  77         b.offset = n - 1
  78         me.assertEqual(b.getu8(), (n - 1)&0xff)
  79       me.assertTrue(b.endp)
  80
  81       ## Test invalid lengths.  This is going to work by setting the top bit
  82       ## of the length, so if it's already set, then that won't be any good.
  83       if n >= 1 << 8*w - 1: continue
  84       seq = T.prep_lenseq(w, n, bigendp, False)
  85
  86       ## Check `getblkN'.
  87       me.assertRaises(C.BufferError, blkfn, C.ReadBuffer(T.span(w - 1)))
  88       b = C.ReadBuffer(seq)
  89       me.assertEqual(b.getu8(), 0)
  90       me.assertRaises(C.BufferError, blkfn, b)
  91
  92       ## Check `getbufN'.
  93       me.assertRaises(C.BufferError, buffn, C.ReadBuffer(T.span(w - 1)))
  94       b = C.ReadBuffer(seq)
  95       me.assertEqual(b.getu8(), 0)
  96       me.assertRaises(C.BufferError, buffn, b)
  97
  98   def test_readbuffer(me):
  99
 100     ## Check `getuN'.
 101     me.check_getn(1, True, lambda buf: buf.getu8())
 102     me.check_getn(2, True, lambda buf: buf.getu16())
 103     me.check_getn(2, True, lambda buf: buf.getu16b())
 104     me.check_getn(2, False, lambda buf: buf.getu16l())
 105     me.check_getn(3, True, lambda buf: buf.getu24())
 106     me.check_getn(3, True, lambda buf: buf.getu24b())
 107     me.check_getn(3, False, lambda buf: buf.getu24l())
 108     me.check_getn(4, True, lambda buf: buf.getu32())
 109     me.check_getn(4, True, lambda buf: buf.getu32b())
 110     me.check_getn(4, False, lambda buf: buf.getu32l())
 111     if hasattr(C.ReadBuffer, "getu64"):
 112       me.check_getn(8, True, lambda buf: buf.getu64())
 113       me.check_getn(8, True, lambda buf: buf.getu64b())
 114       me.check_getn(8, False, lambda buf: buf.getu64l())
 115
 116     ## Check `getblkN' and `getbufN'.
 117     me.check_getbufn(1, True,
 118                      lambda buf: buf.getblk8(),
 119                      lambda buf: buf.getbuf8())
 120     me.check_getbufn(2, True,
 121                      lambda buf: buf.getblk16(),
 122                      lambda buf: buf.getbuf16())
 123     me.check_getbufn(2, True,
 124                      lambda buf: buf.getblk16b(),
 125                      lambda buf: buf.getbuf16b())
 126     me.check_getbufn(2, False,
 127                      lambda buf: buf.getblk16l(),
 128                      lambda buf: buf.getbuf16l())
 129     me.check_getbufn(3, True,
 130                      lambda buf: buf.getblk24(),
 131                      lambda buf: buf.getbuf24())
 132     me.check_getbufn(3, True,
 133                      lambda buf: buf.getblk24b(),
 134                      lambda buf: buf.getbuf24b())
 135     me.check_getbufn(3, False,
 136                      lambda buf: buf.getblk24l(),
 137                      lambda buf: buf.getbuf24l())
 138     me.check_getbufn(4, True,
 139                      lambda buf: buf.getblk32(),
 140                      lambda buf: buf.getbuf32())
 141     me.check_getbufn(4, True,
 142                      lambda buf: buf.getblk32b(),
 143                      lambda buf: buf.getbuf32b())
 144     me.check_getbufn(4, False,
 145                      lambda buf: buf.getblk32l(),
 146                      lambda buf: buf.getbuf32l())
 147     if hasattr(C.ReadBuffer, "getu64"):
 148       me.check_getbufn(8, True,
 149                        lambda buf: buf.getblk64(),
 150                        lambda buf: buf.getbuf64())
 151       me.check_getbufn(8, True,
 152                        lambda buf: buf.getblk64b(),
 153                        lambda buf: buf.getbuf64b())
 154       me.check_getbufn(8, False,
 155                        lambda buf: buf.getblk64l(),
 156                        lambda buf: buf.getbuf64l())
 157
 158     ## Check other `ReadBuffer' methods and properties.
 159     buf = C.ReadBuffer(T.span(256))
 160     me.assertEqual(buf.size, 256)
 161     me.assertEqual(buf.left, 256)
 162     me.assertEqual(buf.offset, 0)
 163     buf.offset = 52
 164     me.assertEqual(buf.left, 204)
 165     buf.skip(7)
 166     me.assertEqual(buf.offset, 59)
 167     me.assertEqual(buf.left, 197)
 168     me.assertRaises(C.BufferError, C.ReadBuffer(T.span(6)).skip, 7)
 169     me.assertEqual(buf.get(5), C.bytes("3b3c3d3e3f"))
 170     me.assertRaises(C.BufferError, C.ReadBuffer(T.span(4)).get, 5)
 171
 172 ###--------------------------------------------------------------------------
 173 class TestWriteBuffer (U.TestCase):
 174
 175   def check_putn(me, w, bigendp, putfn):
 176     """Check `putuN'."""
 177
 178     ## Check encoding an integer.
 179     buf = C.WriteBuffer()
 180     buf.putu8(0x00)
 181     putfn(buf, T.bytes_as_int(w, bigendp))
 182     buf.putu8(w + 1)
 183     me.assertEqual(buf.contents, T.span(w + 2))
 184     me.assertEqual(putfn(C.WriteBuffer(), (1 << 8*w) - 1).contents,
 185                    w*C.bytes("ff"))
 186     me.assertEqual(putfn(C.WriteBuffer(), C.MP(0)).contents, w*C.bytes("00"))
 187
 188     ## Check overflow detection.
 189     me.assertRaises((OverflowError, ValueError),
 190                     putfn, C.WriteBuffer(), 1 << 8*w)
 191
 192   def check_putbufn(me, w, bigendp, putfn):
 193     """Check `putblkN'."""
 194
 195     ## Go through a number of different sizes.
 196     for n in [0, 1, 7, 8, 19, 255, 12345, 65535, 123456]:
 197       if n >= 1 << 8*w: continue
 198       me.assertEqual(putfn(C.WriteBuffer().putu8(0x00),
 199                            T.span(n)).putu8(0xff).contents,
 200                      T.prep_lenseq(w, n, bigendp, True))
 201
 202     ## Check blocks which are too large for the length prefix.
 203     if w <= 3:
 204       me.assertRaises(ValueError, putfn,
 205                       C.WriteBuffer(), C.ByteString.zero(1 << 8*w))
 206
 207   def test_writebuffer(me):
 208
 209     ## Check `putuN'.
 210     me.check_putn(1, True, lambda buf, n: buf.putu8(n))
 211     me.check_putn(2, True, lambda buf, n: buf.putu16(n))
 212     me.check_putn(2, True, lambda buf, n: buf.putu16b(n))
 213     me.check_putn(2, False, lambda buf, n: buf.putu16l(n))
 214     me.check_putn(3, True, lambda buf, n: buf.putu24(n))
 215     me.check_putn(3, True, lambda buf, n: buf.putu24b(n))
 216     me.check_putn(3, False, lambda buf, n: buf.putu24l(n))
 217     me.check_putn(4, True, lambda buf, n: buf.putu32(n))
 218     me.check_putn(4, True, lambda buf, n: buf.putu32b(n))
 219     me.check_putn(4, False, lambda buf, n: buf.putu32l(n))
 220     if hasattr(C.WriteBuffer, "putu64"):
 221       me.check_putn(8, True, lambda buf, n: buf.putu64(n))
 222       me.check_putn(8, True, lambda buf, n: buf.putu64b(n))
 223       me.check_putn(8, False, lambda buf, n: buf.putu64l(n))
 224
 225     ## Check `putblkN".
 226     me.check_putbufn(1, True, lambda buf, x: buf.putblk8(x))
 227     me.check_putbufn(2, True, lambda buf, x: buf.putblk16(x))
 228     me.check_putbufn(2, True, lambda buf, x: buf.putblk16b(x))
 229     me.check_putbufn(2, False, lambda buf, x: buf.putblk16l(x))
 230     me.check_putbufn(3, True, lambda buf, x: buf.putblk24(x))
 231     me.check_putbufn(3, True, lambda buf, x: buf.putblk24b(x))
 232     me.check_putbufn(3, False, lambda buf, x: buf.putblk24l(x))
 233     me.check_putbufn(4, True, lambda buf, x: buf.putblk32(x))
 234     me.check_putbufn(4, True, lambda buf, x: buf.putblk32b(x))
 235     me.check_putbufn(4, False, lambda buf, x: buf.putblk32l(x))
 236     if hasattr(C.WriteBuffer, "putu64"):
 237       me.check_putbufn(8, True, lambda buf, x: buf.putblk64(x))
 238       me.check_putbufn(8, True, lambda buf, x: buf.putblk64b(x))
 239       me.check_putbufn(8, False, lambda buf, x: buf.putblk64l(x))
 240
 241     ## Check other methods and properties.
 242     buf = C.WriteBuffer()
 243     buf.zero(17)
 244     buf.put(T.span(23))
 245     me.assertEqual(buf.size, 40)
 246     me.assertEqual(buf.contents, C.ByteString.zero(17) + T.span(23))
 247
 248 ###----- That's all, folks --------------------------------------------------
 249
 250 if __name__ == "__main__": U.main()