# Licensed to the Apache Software Foundation (ASF) under one * # or more contributor license agreements. See the NOTICE file * # distributed with this work for additional information * # regarding copyright ownership. The ASF licenses this file * # to you under the Apache License, Version 2.0 (the * # "License"); you may not use this file except in compliance * # with the License. You may obtain a copy of the License at * # * # http://www.apache.org/licenses/LICENSE-2.0 * # * # Unless required by applicable law or agreed to in writing, * # software distributed under the License is distributed on an * # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * # KIND, either express or implied. See the License for the * # specific language governing permissions and limitations * # under the License. import unittest from etch.python.Types import * from etch.python.Exceptions import * from etch.util.FlexBuffer import * class Test_EtchUtilFlexBuffer(unittest.TestCase): def setUp(self): self.buf = FlexBuffer() def _checkBuf(self, length, index, avail): self.assertEqual(length, self.buf.length(), "Length") self.assertEqual(index, self.buf.index(), "Index") self.assertEqual(avail, self.buf.avail(), "Avail") def test_initial1(self): self._checkBuf(0, 0, 0) def test_initial2(self): self.buf = FlexBuffer(ByteArray(5)) self._checkBuf(5, 0, 5) def test_initial3(self): self.buf = FlexBuffer(ByteArray(5), 2) self._checkBuf(2, 0, 2) def test_initial4(self): self.buf = FlexBuffer(ByteArray(5), 1, 3) self._checkBuf(4, 1, 3) def test_setLength0(self): self.buf.setLength(0) self._checkBuf(0, 0, 0) def test_setLength1(self): self.buf.setLength(1) self._checkBuf(1, 0, 1) def test_setLength2(self): self.buf.setLength(2) self._checkBuf(2, 0, 2) def test_setLength3(self): self.buf.setLength(2) self.buf.setLength(0) self._checkBuf(0, 0, 0) def test_setLength4(self): self.assertRaises(IllegalArgumentException, self.buf.setLength, -1) def test_setIndex0(self): self.buf.setIndex(0) self._checkBuf(0, 0, 0) def test_setIndex1(self): self.assertRaises(IllegalArgumentException, self.buf.setIndex, 1) def test_setIndex2(self): self.assertRaises(IllegalArgumentException, self.buf.setIndex, -1) def test_setIndex3(self): self.buf.setLength(5) self.buf.setIndex(1) self._checkBuf(5, 1, 4) def test_setIndex4(self): self.buf.setLength(5) self.buf.setIndex(4) self._checkBuf(5, 4, 1) def test_setIndex5(self): self.buf.setLength(5) self.buf.setIndex(5) self._checkBuf(5, 5, 0) def test_setIndex5(self): self.buf.setLength(5) self.buf.setIndex(5) self.buf.setLength(0) self._checkBuf(0, 0, 0) def test_setIndex7(self): self.buf.setLength(5) self.buf.setIndex(5) self.buf.setLength(2) self._checkBuf(2, 2, 0) def test_reset(self): self.buf.setLength(2) self.buf.setIndex(1) self.buf.reset() self._checkBuf(0, 0, 0) def test_put1a(self): self.buf.put(1) self._checkBuf(1, 1, 0) def test_put1b(self): self.buf.put(1) self.buf.put(2) self._checkBuf(2, 2, 0) def _testPutBytes(self, bufLength): self.buf.put(1) if bufLength >= 0: self.buf.put(ByteArray(bufLength)) else: self.buf.put(None) self._checkBuf(bufLength+1, bufLength+1, 0) def test_put2a(self): self.assertRaises(NullPointerException, self._testPutBytes, -1) def test_put2b(self): self._testPutBytes(0) def test_put2c(self): self._testPutBytes(1) def test_put2d(self): self._testPutBytes(2) #// ------------------------------------------------------ #// cases for put( byte[] buf, int off, int len ): #// buf.length, off, len // failure reason #// --- group 3: #// null, 0, 0 // fail (buf is null) #// null, 0, 1 // fail (buf is null) #// null, 1, 0 // fail (buf is null) #// --- group 4: #// 0, 0, 0 #// 0, 0, 1 // fail (off+len > buf.length) #// 0, 1, 0 // fail (off > buf.length) #// --- group 5: #// 1, 0, 0 #// 1, 0, 1 #// 1, 0, 2 // fail (off+len > buf.length) #// 1, 1, 0 #// 1, 1, 1 // fail (off+len > buf.length) #// 1, 1, 2 // fail (off+len > buf.length) #// 1, 2, 0 // fail (off > buf.length) #// 1, 2, 1 // fail (off > buf.length) #// 1, 2, 2 // fail (off > buf.length) #// --- group 6: #// 2, 0, 0 #// 2, 0, 1 #// 2, 0, 2 #// 2, 1, 0 #// 2, 1, 1 #// 2, 1, 2 // fail (off+len > buf.length) #// 2, 2, 0 #// 2, 2, 1 // fail (off+len > buf.length) #// 2, 2, 2 // fail (off+len > buf.length) def _testPutBytesOffLen(self, bufLength, offset, length): self.buf.put(1) if bufLength >= 0: self.buf.put(ByteArray(bufLength), offset, length) else: self.buf.put(None, offset, length) self._checkBuf(length+1, length+1, 0) def test_put3a(self): self.assertRaises(NullPointerException, self._testPutBytesOffLen, -1, 0, 0) def test_put3b(self): self.assertRaises(NullPointerException, self._testPutBytesOffLen, -1, 0, 0) def test_put3c(self): self.assertRaises(NullPointerException, self._testPutBytesOffLen, -1, 1, 0) def test_put4a(self): self._testPutBytesOffLen(0, 0, 0) def test_put4b(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 0, 0, 1) def test_put4c(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 0, 1, 0) def test_put5a(self): self._testPutBytesOffLen(1, 0, 0) def test_put5b(self): self._testPutBytesOffLen(1, 0, 1) def test_put5c(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 0, 2) def test_put5d(self): self._testPutBytesOffLen(1, 1, 0) def test_put5e(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 1, 1) def test_put5f(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 1, 2) def test_put5g(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 2, 0) def test_put5h(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 2, 1) def test_put5i(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 1, 2, 2) def test_put6a(self): self._testPutBytesOffLen(2, 0, 0) def test_put6b(self): self._testPutBytesOffLen(2, 0, 1) def test_put6c(self): self._testPutBytesOffLen(2, 0, 2) def test_put6d(self): self._testPutBytesOffLen(2, 1, 0) def test_put6e(self): self._testPutBytesOffLen(2, 1, 1) def test_put6f(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 2, 1, 2) def test_put6g(self): self._testPutBytesOffLen(2, 2, 0) def test_put6h(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 2, 2, 1) def test_put6i(self): self.assertRaises(IllegalArgumentException, self._testPutBytesOffLen, 2, 2, 2) #// ------------------------------------------------------ #// cases for put(int), get(): #// put 0, get 0. // already tested above #// put 0, get 1. // fails with EOFException #// put 1, get 0. #// put 1, get 1. #// put 1, get 2. // fails with EOFException #// put 2, get 0. #// put 2, get 1. #// put 2, get 2. #// put 2, get 3. // fails with EOFException #// put 10000, get 10000. #// just like above, putting byte array instead #// just like above, getting byte array instead. #// just like above, putting and getting byte array. def _testPutGet(self, nPuts, nGets): for i in range(0, nPuts): self.buf.put(i) self.buf.setIndex(0) for i in range(0, nGets): self.assertEqual(Byte(i), Byte(self.buf.get())) self._checkBuf(nPuts, nGets, nPuts-nGets) def _testPutBytesGet(self, nPuts, nGets): buffer = ByteArray(nPuts) for i in range(0, nPuts): buffer[i] = Byte(i) self.buf.put(buffer) self.buf.setIndex(0) for i in range(0, nGets): self.assertEquals(Byte(i), Byte(self.buf.get())) self._checkBuf(nPuts, nGets, nPuts - nGets) def _testPutGetBytes(self, nPuts, nGets): for i in range(0, nPuts): #print "%s" % repr(i) self.buf.put(i) self.buf.setIndex(0) buffer = ByteArray(nGets) n = self.buf.get(buffer) self.assertEquals(min(nPuts, nGets), n) for i in range(0, n): self.assertEquals(Byte(i), buffer[i]) self._checkBuf(nPuts, n, nPuts-n) def _testPutBytesGetBytes(self, nPuts, nGets): buffer = ByteArray(nPuts) for i in range(0, nPuts): buffer[i] = Byte(i) self.buf.put(buffer) self.buf.setIndex(0) buffer = ByteArray(nGets) n = self.buf.get(buffer) self.assertEquals(min(nPuts, nGets), n) for i in range(0, n): self.assertEquals(Byte(i), buffer[i]) self._checkBuf(nPuts, n, nPuts-n) def test_get1a(self): self.assertRaises(EOFException, self._testPutGet, 0, 1) def test_get1b(self): self._testPutGet(1, 0) def test_get1c(self): self._testPutGet(1, 1) def test_get1d(self): self.assertRaises(EOFException, self._testPutGet, 1, 2) def test_get1e(self): self._testPutGet(2, 0) def test_get1f(self): self._testPutGet(2, 1) def test_get1g(self): self._testPutGet(2, 2) def test_get1h(self): self.assertRaises(EOFException, self._testPutGet, 2, 3) def test_get1i(self): self._testPutGet(10000, 9995) def test_get1j(self): self._testPutGet(10000, 10000) def test_get1k(self): self._testPutGet(100000, 100000) def test_get2a(self): self.assertRaises(EOFException, self._testPutBytesGet, 0, 1) def test_get2b(self): self._testPutBytesGet(1, 0) def test_get2d(self): self.assertRaises(EOFException, self._testPutBytesGet, 1, 2) def test_get2e(self): self._testPutBytesGet(2, 0) def test_get2f(self): self._testPutBytesGet(2, 1) def test_get2g(self): self._testPutBytesGet(2, 2) def test_get2h(self): self.assertRaises(EOFException, self._testPutBytesGet, 2, 3) def test_get2i(self): self._testPutBytesGet(10000, 9995) def test_get2j(self): self._testPutBytesGet(10000, 10000) def test_get2k(self): self._testPutBytesGet(100000, 100000) def test_get3a(self): self.assertRaises(EOFException, self._testPutBytesGet, 0, 1) def test_get3b(self): self._testPutGetBytes(1, 0) def test_get3c(self): self._testPutGetBytes(1, 1) def test_get3d(self): self._testPutGetBytes(1, 2) def test_get3e(self): self._testPutGetBytes(2, 0) def test_get3f(self): self._testPutGetBytes(2, 1) def test_get3g(self): self._testPutGetBytes(2, 2) def test_get3h(self): self._testPutGetBytes(2, 3) def test_get3i(self): self._testPutGetBytes(10000, 9995) def test_get3j(self): self._testPutGetBytes(10000, 10000) def test_get3k(self): self._testPutGetBytes(100000, 100000) def test_get4a(self): self.assertRaises(EOFException, self._testPutBytesGetBytes, 0, 1) def test_get4b(self): self._testPutBytesGetBytes(1, 0) def test_get4c(self): self._testPutBytesGetBytes(1, 1) def test_get4d(self): self._testPutBytesGetBytes(1, 2) def test_get4e(self): self._testPutBytesGetBytes(2, 0) def test_get4f(self): self._testPutBytesGetBytes(2, 1) def test_get4g(self): self._testPutBytesGetBytes(2, 2) def test_get4h(self): self._testPutBytesGetBytes(2, 3) def test_get4i(self): self._testPutBytesGetBytes(10000, 9995) def test_get4j(self): self._testPutBytesGetBytes(10000, 10000) def test_get4k(self): self._testPutBytesGetBytes(100000, 100000) def test_get5a(self): self.buf = FlexBuffer(ByteArray([1,2,3])) self.assertEquals(1, self.buf.get()) self.assertEquals(2, self.buf.get()) self.assertEquals(3, self.buf.get()) self._checkBuf(3, 3, 0) def test_get5b(self): self.buf = FlexBuffer(ByteArray([1,2,3]), 2) self.assertEquals(1, self.buf.get()) self.assertEquals(2, self.buf.get()) self._checkBuf(2, 2, 0) def test_get5c(self): self.buf = FlexBuffer(ByteArray([1,2,3,4,5]),1,2) self.assertEquals(2, self.buf.get()) self.assertEquals(3, self.buf.get()) self._checkBuf(3, 3, 0) def test_get6(self): self.buf = FlexBuffer(ByteArray([5,6,7,8])) buffer = ByteArray([11,12,13,14,15]) self.buf.get(buffer, 1, 3) self.assertEquals(11, buffer[0]) self.assertEquals(5, buffer[1]) self.assertEquals(6, buffer[2]) self.assertEquals(7, buffer[3]) self.assertEquals(15, buffer[4]) def test_compact(self): self.buf = FlexBuffer(ByteArray([1,2,3,4,5]), 1, 3) self.buf.compact() self._checkBuf(3, 0, 3) self.assertEquals(2, self.buf.get()) self.assertEquals(3, self.buf.get()) self.assertEquals(4, self.buf.get()) def test_getByte1(self): self.buf = FlexBuffer(ByteArray([127, 1, 0, -1, -128])) self.assertEquals(127, self.buf.getByte()) self.assertEquals(1, self.buf.getByte()) self.assertEquals(0, self.buf.getByte()) self.assertEquals(-1, self.buf.getByte()) self.assertEquals(-128, self.buf.getByte()) self._checkBuf(5, 5, 0) def test_getByte2(self): self.assertRaises(EOFException, self.buf.getByte) def test_getShort1(self): self.buf = FlexBuffer(ByteArray([1,2,-1,-2])) self.assertEquals(Short(0x0102), self.buf.getShort()) self.assertEquals(Short(0xfffe), self.buf.getShort()) self._checkBuf(4, 4, 0) def test_getShort2(self): self.assertRaises(EOFException, self.buf.getShort) def test_getInt1(self): self.buf = FlexBuffer(ByteArray([1,2,3,4,-1,-2,-3,-4])) self.assertEquals(Integer(0x01020304), self.buf.getInt()) self.assertEquals(Integer(0xfffefdfc), self.buf.getInt()) self._checkBuf(8, 8, 0) def test_getInt2(self): self.assertRaises(EOFException, self.buf.getInt) def test_getLong1(self): self.buf = FlexBuffer(ByteArray([1,2,3,4,5,6,7,8,-1,-2,-3,-4,-5,-6,-7,-8])) self.assertEquals(Long(0x0102030405060708), self.buf.getLong()) self.assertEquals(Long(0xfffefdfcfbfaf9f8), self.buf.getLong()) self._checkBuf(16, 16, 0) def test_getLong2(self): self.assertRaises(EOFException, self.buf.getLong) def test_getFloat1(self): self.buf = FlexBuffer(ByteArray([ 0, 0, 0, 0, 63, -128, 0, 0, 83, -113, 48, -37, 100, -98, -112, -120, 117, -120, 20, 25, -65, -128, 0, 0, -45, -113, 48, -37, -28, -98, -112, -120, -11, -120, 20, 25, 50, 83, 79, -7, 34, 44, -87, 90, 17, -38, -85, 115, -78, 83, 79, -7, -94, 44, -87, 90, -111, -38, -85, 115])) self.assertEquals( 0.0, self.buf.getFloat() ) self.assertEquals( 1.0, self.buf.getFloat() ) self.assertEquals( 123e10, self.buf.getFloat() ) self.assertEquals( 234e20, self.buf.getFloat() ) self.assertEquals( 345e30, self.buf.getFloat() ) self.assertEquals( -1.0, self.buf.getFloat() ) self.assertEquals( -123e10, self.buf.getFloat() ) self.assertEquals( -234e20, self.buf.getFloat() ) self.assertEquals( -345e30, self.buf.getFloat() ) self.assertEquals( 123e-10, self.buf.getFloat() ) self.assertEquals( 234e-20, self.buf.getFloat() ) self.assertEquals( 345e-30, self.buf.getFloat() ) self.assertEquals( -123e-10, self.buf.getFloat() ) self.assertEquals( -234e-20, self.buf.getFloat() ) self.assertEquals( -345e-30, self.buf.getFloat() ) self._checkBuf(15*4, 15*4, 0) def test_getFloat2(self): self.assertRaises(EOFException, self.buf.getFloat) def test_getDouble1(self): # TODO - double conversions appear to be off compared to Java self.buf = FlexBuffer(ByteArray([ 0, 0, 0, 0, 0, 0, 0, 0 , 63, -16, 0, 0, 0, 0, 0, 0 , 66, 113, -26, 27, 104, -64, 0, 0 , 85, 48, -73, 88, 68, 89, -6, -80 , 105, -4, 43, 112, 122, -31, 67, -116 , -65, -16, 0, 0, 0, 0, 0, 0 , -62, 113, -26, 27, 104, -64, 0, 0 , -43, 48, -73, 88, 68, 89, -6, -80 , -23, -4, 43, 112, 122, -31, 67, -116 , 62, 74, 105, -1, 27, 85, 80, 81 , 43, -87, -105, 64, -36, -23, -48, -66 , 22, -16, -127, 73, 111, 91, -75, -29 , -66, 74, 105, -1, 27, 85, 80, 81 , -85, -87, -105, 64, -36, -23, -48, -66 , -106, -16, -127, 73, 111, 91, -75, -29 ])) self.assertEquals( 0.0, self.buf.getDouble()) self.assertEquals( 1.0, self.buf.getDouble()) self.assertEquals( 123e10, self.buf.getDouble()) self.assertEquals( 234e100, self.buf.getDouble()) self.assertEquals( 345e200, self.buf.getDouble()) self.assertEquals( -1.0, self.buf.getDouble()) self.assertEquals( -123e10, self.buf.getDouble()) self.assertEquals( -234e100, self.buf.getDouble()) self.assertEquals( -345e200, self.buf.getDouble()) self.assertEquals( 123e-10, self.buf.getDouble()) self.assertEquals( 234e-100, self.buf.getDouble()) self.assertEquals( 345e-200, self.buf.getDouble()) self.assertEquals( -123e-10, self.buf.getDouble()) self.assertEquals( -234e-100, self.buf.getDouble()) self.assertEquals( -345e-200, self.buf.getDouble()) self._checkBuf(15*8, 15*8, 0) def test_getDouble2(self): self.assertRaises(EOFException, self.buf.getDouble) def test_getFully1(self): self.buf = FlexBuffer(ByteArray([1,2,3,4,5,6])) b = ByteArray(3) self.buf.getFully(b) self.assertEquals(1, b[0]) self.assertEquals(2, b[1]) self.assertEquals(3, b[2]) self.buf.getFully(b) self.assertEquals(4, b[0]) self.assertEquals(5, b[1]) self.assertEquals(6, b[2]) self._checkBuf(6, 6, 0) def test_getFully2(self): self.buf = FlexBuffer(ByteArray([1,2])) b = ByteArray(3) #self.buf = self.buf.getFully(b) self.assertRaises(EOFException, self.buf.getFully, b) # // ------------------------------------------------------ # // cases for FlexBuffer put( FlexBuffer ): # // put, put FlexBuffer with valid buffer # // put, put Flexbuffer to a Flexbuffer approaching max length, fails with IOException. def test_putFlexBuffer1(self): buf0 = FlexBuffer(ByteArray([1,2])) self.buf = FlexBuffer(ByteArray([])) self.buf.put(buf0) self._checkBuf(buf0.length(), buf0.length(), 0) def test_putFlexBuffer2(self): buf0 = FlexBuffer(ByteArray([1,2])) max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) # TODO - Confirm this is the correct test self.assertRaises(IOException, self.buf.put, buf0) #self._checkBuf(max + buf0.length(), max + buf0.length(), 0) #// ------------------------------------------------------ #// cases for FlexBuffer put( FlexBuffer, int ): #// put, put FlexBuffer with length larger than available bytes, fails with IllegalArgumentException #// put, put FlexBuffer with valid length value #// put, put Flexbuffer to a Flexbuffer approaching max length, fails with IOException. def test_putFlexBuffer3(self): buf0 = FlexBuffer(ByteArray([1,2])) self.buf = FlexBuffer(ByteArray([])) self.buf.setIndex(0) # TODO - confirm this is the correct test self.assertRaises(IllegalArgumentException, self.buf.put, buf0, 3) #self._checkBuf(3,3,0) def test_putFlexBuffer4(self): buf0 = FlexBuffer(ByteArray([1,2])) self.buf = FlexBuffer(ByteArray([])) self.buf.put(buf0, 1) self._checkBuf(1,1,0) def test_putFlexBuffer5(self): buf0 = FlexBuffer(ByteArray([1,2])) max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) self.assertRaises(IOException, self.buf.put, buf0, 1) #self._checkBuf(max+1, max+1, 0) #// ------------------------------------------------------ #// cases for void putByte( byte ): #// putByte, put byte value and expect getByte to return matched value. #// putByte to a Flexbuffer approaching max length, fails with IOException. def test_putByte1(self): self.buf = FlexBuffer(ByteArray()) self.buf.putByte(0x01) self._checkBuf(1, 1, 0) self.buf.setIndex(0) self.assertEquals(0x01, self.buf.getByte()) self._checkBuf(1, 1, 0) def test_putByte2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) self.assertRaises(IOException, self.buf.putByte, 0x01) #self._checkBuf(max+1, max+1, 0) #// ------------------------------------------------------ #// cases for void putDouble( double ): #// putDouble, put Double and expect getDouble to return matched value. #// putDouble to a Flexbuffer approaching max length, fails with IOException. def test_putDouble1(self): self.buf = FlexBuffer(ByteArray([])) self.buf.putDouble(Double.MAX_VALUE) self._checkBuf(8,8,0) self.buf.setIndex(0) self.assertEquals(Double.MAX_VALUE, self.buf.getDouble()) self._checkBuf(8,8,0) def test_putDouble2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) self.assertRaises(IOException, self.buf.putDouble, Double.MIN_VALUE) #self._checkBuf(max+8, max+8, 0) #// ------------------------------------------------------ #// cases for void putFloat( float ): #// putFloat, put Float and expect getFloat to return matched value. #// putFloat to a Flexbuffer approaching max length, fails with IOException. def test_putFloat1(self): self.buf = FlexBuffer(ByteArray([])) self.buf.putFloat(Float.MAX_VALUE) self._checkBuf(4, 4, 0) self.buf.setIndex(0) self.assertEquals(Float.MAX_VALUE, self.buf.getFloat()) self._checkBuf(4, 4, 0) def test_putFloat2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) # TODO - confirm this is the correct test self.assertRaises(IOException, self.buf.putFloat, Float.MIN_VALUE) #self._checkBuf(max+4, max+4, 0) #// ------------------------------------------------------ #// cases for void putInt( int ): #// putInt, put Int and expect getInt to return matched value. #// putInt to a Flexbuffer approaching max length, fails with IOException. def test_putInt1(self): self.buf = FlexBuffer(ByteArray([])) self.buf.putInt(Integer.MAX_VALUE) self._checkBuf(4, 4, 0) self.buf.setIndex(0) self.assertEquals(Integer.MAX_VALUE, self.buf.getInt()) self._checkBuf(4 ,4, 0) def test_putInt2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) # TODO - validate test self.assertRaises(IOException, self.buf.putInt, Integer.MIN_VALUE) #self._checkBuf(max+4, max+4, 0) #// ------------------------------------------------------ #// cases for void putLong( long ): #// putLong, put Long and expect getLong to return matched value. #// putShort to a Flexbuffer approaching max length, fails with IOException. def test_putLong1(self): self.buf = FlexBuffer(ByteArray([])) self.buf.putLong(Long.MAX_VALUE) self._checkBuf(8, 8, 0) self.buf.setIndex(0) self.assertEquals(Long.MAX_VALUE, self.buf.getLong()) self._checkBuf(8, 8, 0) def test_putLong2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) # TODO - confirm this is the correct test self.assertRaises(IOException, self.buf.putLong, Long.MIN_VALUE) #self._checkBuf(max+8, max+8, 0) #// ------------------------------------------------------ #// cases for void putShort( short ): #// putShort, put short and expect getShort to return matched value. #// putShort to a Flexbuffer approaching max length, fails with IOException. def test_putShort1(self): self.buf = FlexBuffer(ByteArray([])) self.buf.putShort(Short.MAX_VALUE) self._checkBuf(2, 2, 0) self.buf.setIndex(0) self.assertEquals(Short.MAX_VALUE, self.buf.getShort()) self._checkBuf(2, 2, 0) def test_putShort2(self): max = 4*1024*1024 self.buf = FlexBuffer(ByteArray(max)) self.buf.setIndex(max) # TODO - confirm this correct test self.assertRaises(IOException, self.buf.putShort, 1) #self._checkBuf(max+2, max+2, 0) #// ------------------------------------------------------ #// cases for FlexBuffer skip( int, boolean ): #// skip length < 0, put = false. // fails with illegal argument #// skip length > buffer length, put = false. // fails with EOFException #// skip length = 0, put = false. #// skip length = buffer length, put = false. #// skip length < buffer length, put = false. #// skip length > buffer length, put = true. #// skip length < buffer length, put = true. #// skip length = max buffer length, put = true. // fails with IOException def test_skip1(self): self.buf = FlexBuffer(ByteArray([])) self.assertRaises(IllegalArgumentException, self.buf.skip, -1, False) def test_skip2(self): self.buf = FlexBuffer(ByteArray([])) self.assertRaises(EOFException, self.buf.skip, 2, False) def test_skip3(self): self.buf = FlexBuffer(ByteArray([1, 2])) self.buf.skip(2, False) self._checkBuf(2, 2, 0) def test_skip4(self): self.buf = FlexBuffer(ByteArray([1,2])) self.buf.skip(2, False) self._checkBuf(2, 2, 0) def test_skip5(self): self.buf = FlexBuffer(ByteArray([1,2])) self.buf.skip(1, False) self._checkBuf(2, 1, 1) def test_skip6(self): self.buf = FlexBuffer(ByteArray([1,2])) self.buf.skip(1, True) self._checkBuf(2, 1, 1) def test_skip7(self): self.buf = FlexBuffer(ByteArray([1,2])) self.buf.skip(5, True) self._checkBuf(5, 5, 0) def test_skip8(self): self.buf = FlexBuffer(ByteArray([1,2])) max = 4*1024*1024 # TODO - confirm this is the correct test self.assertRaises(IOException, self.buf.skip, max+1, True) #self._checkBuf(max+1, max+1, 0) if __name__ == '__main__': unittest.main()