@Test
public void testIsValid()
{

  String randomValue = RngUtils.randomString(100);
  ByteBuffer serializationBuffer = ByteBuffer.allocate(1000).order(_eventV1Factory.getByteOrder());
  try
  {
    DbusEventInfo eventInfo = new DbusEventInfo(DbusOpcode.UPSERT, 0L, (short)0, partitionId, timeStamp, srcId,
                                                schemaId, randomValue.getBytes(Charset.defaultCharset()), false, false);
    eventInfo.setEventSerializationVersion(DbusEventFactory.DBUS_EVENT_V1);  // make this explicit
    DbusEventFactory.serializeEvent(new DbusEventKey(key), serializationBuffer, eventInfo);
  }
  catch (KeyTypeNotImplementedException e2)
  {
    fail(e2.getLocalizedMessage());
  }
  DbusEventInternalReadable e = null;
  try
  {
    e = DbusEventFactoryForTesting.createReadOnlyDbusEventFromBuffer(serializationBuffer, 0, 200L,
                                                                     DbusEventFactory.DBUS_EVENT_V2);
  }
  catch (Exception e1)
  {
    fail("Not supposed to throw exception");
  }
  assertTrue(e.isValid());
  try
  {
    //start modifying events and check if isValid is still true;
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.LENGTH,e);
    assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.ERR);
    assertFalse(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.LENGTH,e);
    assertTrue(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.HEADERCRC,e);
    assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.ERR);
    assertFalse(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.HEADERCRC,e);
    assertTrue(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.PAYLOADCRC,e);
    //header crc uses payload crc as well!
    assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.ERR);
    assertFalse(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.PAYLOADCRC,e);
    assertTrue(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.PAYLOAD,e);
    assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.OK);
    assertTrue(e.scanEvent()== DbusEventInternalReadable.EventScanStatus.ERR);
    assertFalse(e.isValid());
    DbusEventCorrupter.toggleEventCorruption(EventCorruptionType.PAYLOAD,e);
    assertTrue(e.isValid());
  }
  catch (InvalidEventException e1)
  {
    fail("Not supposed to throw exception");
  }
  //now check partial events;
  int origLimit = serializationBuffer.limit();
  serializationBuffer.limit(30);
  assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.PARTIAL);
  assertTrue(e.isPartial());
  assertFalse(e.isValid());
  serializationBuffer.limit(origLimit);
  assertTrue(e.isValid());
  //another partial event - this time the value;
  serializationBuffer.limit(90);
  assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.OK);
  assertTrue(e.scanEvent()== DbusEventInternalReadable.EventScanStatus.PARTIAL);
  assertTrue(e.isPartial());
  assertFalse(e.isValid());
  serializationBuffer.limit(origLimit);
  assertTrue(e.isValid());
  int headerSize = 56;
  int preambleSize = 5;
  serializationBuffer.limit(headerSize);
  assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.PARTIAL);
  assertTrue(e.isPartial());
  //version byte + header crc;
  serializationBuffer.limit(headerSize+preambleSize);
  assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.OK);
  assertTrue(e.isPartial());
  serializationBuffer.limit(origLimit);
  assertTrue(e.isValid());
  //edge case
  serializationBuffer.limit(headerSize+preambleSize-1);
  assertTrue(e.scanHeader()== DbusEventInternalReadable.HeaderScanStatus.PARTIAL);
  assertTrue(e.isPartial());
  serializationBuffer.limit(origLimit);
  assertTrue(e.isValid());
}